b6c9d3055705b04bea1e848750f76cf1a1d86912
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
1 /*
2 * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
3 * 2018 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 *
5 * This program is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License, version 2 only, as
7 * published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
13 *
14 * You should have received a copy of the GNU General Public License along with
15 * this program; if not, write to the Free Software Foundation, Inc., 51
16 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19 #define _LGPL_SOURCE
20 #include <assert.h>
21 #include <stdio.h>
22 #include <stdlib.h>
23 #include <string.h>
24 #include <sys/stat.h>
25 #include <sys/types.h>
26 #include <unistd.h>
27 #include <inttypes.h>
28
29 #include <common/common.h>
30 #include <common/defaults.h>
31 #include <common/uri.h>
32 #include <common/relayd/relayd.h>
33
34 #include "consumer.h"
35 #include "health-sessiond.h"
36 #include "ust-app.h"
37 #include "utils.h"
38
39 /*
40 * Send a data payload using a given consumer socket of size len.
41 *
42 * The consumer socket lock MUST be acquired before calling this since this
43 * function can change the fd value.
44 *
45 * Return 0 on success else a negative value on error.
46 */
47 int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len)
48 {
49 int fd;
50 ssize_t size;
51
52 assert(socket);
53 assert(socket->fd_ptr);
54 assert(msg);
55
56 /* Consumer socket is invalid. Stopping. */
57 fd = *socket->fd_ptr;
58 if (fd < 0) {
59 goto error;
60 }
61
62 size = lttcomm_send_unix_sock(fd, msg, len);
63 if (size < 0) {
64 /* The above call will print a PERROR on error. */
65 DBG("Error when sending data to consumer on sock %d", fd);
66 /*
67 * At this point, the socket is not usable anymore thus closing it and
68 * setting the file descriptor to -1 so it is not reused.
69 */
70
71 /* This call will PERROR on error. */
72 (void) lttcomm_close_unix_sock(fd);
73 *socket->fd_ptr = -1;
74 goto error;
75 }
76
77 return 0;
78
79 error:
80 return -1;
81 }
82
83 /*
84 * Receive a data payload using a given consumer socket of size len.
85 *
86 * The consumer socket lock MUST be acquired before calling this since this
87 * function can change the fd value.
88 *
89 * Return 0 on success else a negative value on error.
90 */
91 int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
92 {
93 int fd;
94 ssize_t size;
95
96 assert(socket);
97 assert(socket->fd_ptr);
98 assert(msg);
99
100 /* Consumer socket is invalid. Stopping. */
101 fd = *socket->fd_ptr;
102 if (fd < 0) {
103 goto error;
104 }
105
106 size = lttcomm_recv_unix_sock(fd, msg, len);
107 if (size <= 0) {
108 /* The above call will print a PERROR on error. */
109 DBG("Error when receiving data from the consumer socket %d", fd);
110 /*
111 * At this point, the socket is not usable anymore thus closing it and
112 * setting the file descriptor to -1 so it is not reused.
113 */
114
115 /* This call will PERROR on error. */
116 (void) lttcomm_close_unix_sock(fd);
117 *socket->fd_ptr = -1;
118 goto error;
119 }
120
121 return 0;
122
123 error:
124 return -1;
125 }
126
127 /*
128 * Receive a reply command status message from the consumer. Consumer socket
129 * lock MUST be acquired before calling this function.
130 *
131 * Return 0 on success, -1 on recv error or a negative lttng error code which
132 * was possibly returned by the consumer.
133 */
134 int consumer_recv_status_reply(struct consumer_socket *sock)
135 {
136 int ret;
137 struct lttcomm_consumer_status_msg reply;
138
139 assert(sock);
140
141 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
142 if (ret < 0) {
143 goto end;
144 }
145
146 if (reply.ret_code == LTTCOMM_CONSUMERD_SUCCESS) {
147 /* All good. */
148 ret = 0;
149 } else {
150 ret = -reply.ret_code;
151 DBG("Consumer ret code %d", ret);
152 }
153
154 end:
155 return ret;
156 }
157
158 /*
159 * Once the ASK_CHANNEL command is sent to the consumer, the channel
160 * information are sent back. This call receives that data and populates key
161 * and stream_count.
162 *
163 * On success return 0 and both key and stream_count are set. On error, a
164 * negative value is sent back and both parameters are untouched.
165 */
166 int consumer_recv_status_channel(struct consumer_socket *sock,
167 uint64_t *key, unsigned int *stream_count)
168 {
169 int ret;
170 struct lttcomm_consumer_status_channel reply;
171
172 assert(sock);
173 assert(stream_count);
174 assert(key);
175
176 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
177 if (ret < 0) {
178 goto end;
179 }
180
181 /* An error is possible so don't touch the key and stream_count. */
182 if (reply.ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
183 ret = -1;
184 goto end;
185 }
186
187 *key = reply.key;
188 *stream_count = reply.stream_count;
189 ret = 0;
190
191 end:
192 return ret;
193 }
194
195 /*
196 * Send destroy relayd command to consumer.
197 *
198 * On success return positive value. On error, negative value.
199 */
200 int consumer_send_destroy_relayd(struct consumer_socket *sock,
201 struct consumer_output *consumer)
202 {
203 int ret;
204 struct lttcomm_consumer_msg msg;
205
206 assert(consumer);
207 assert(sock);
208
209 DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
210
211 memset(&msg, 0, sizeof(msg));
212 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
213 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
214
215 pthread_mutex_lock(sock->lock);
216 ret = consumer_socket_send(sock, &msg, sizeof(msg));
217 if (ret < 0) {
218 goto error;
219 }
220
221 /* Don't check the return value. The caller will do it. */
222 ret = consumer_recv_status_reply(sock);
223
224 DBG2("Consumer send destroy relayd command done");
225
226 error:
227 pthread_mutex_unlock(sock->lock);
228 return ret;
229 }
230
231 /*
232 * For each consumer socket in the consumer output object, send a destroy
233 * relayd command.
234 */
235 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
236 {
237 struct lttng_ht_iter iter;
238 struct consumer_socket *socket;
239
240 assert(consumer);
241
242 /* Destroy any relayd connection */
243 if (consumer->type == CONSUMER_DST_NET) {
244 rcu_read_lock();
245 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
246 node.node) {
247 int ret;
248
249 /* Send destroy relayd command */
250 ret = consumer_send_destroy_relayd(socket, consumer);
251 if (ret < 0) {
252 DBG("Unable to send destroy relayd command to consumer");
253 /* Continue since we MUST delete everything at this point. */
254 }
255 }
256 rcu_read_unlock();
257 }
258 }
259
260 /*
261 * From a consumer_data structure, allocate and add a consumer socket to the
262 * consumer output.
263 *
264 * Return 0 on success, else negative value on error
265 */
266 int consumer_create_socket(struct consumer_data *data,
267 struct consumer_output *output)
268 {
269 int ret = 0;
270 struct consumer_socket *socket;
271
272 assert(data);
273
274 if (output == NULL || data->cmd_sock < 0) {
275 /*
276 * Not an error. Possible there is simply not spawned consumer or it's
277 * disabled for the tracing session asking the socket.
278 */
279 goto error;
280 }
281
282 rcu_read_lock();
283 socket = consumer_find_socket(data->cmd_sock, output);
284 rcu_read_unlock();
285 if (socket == NULL) {
286 socket = consumer_allocate_socket(&data->cmd_sock);
287 if (socket == NULL) {
288 ret = -1;
289 goto error;
290 }
291
292 socket->registered = 0;
293 socket->lock = &data->lock;
294 rcu_read_lock();
295 consumer_add_socket(socket, output);
296 rcu_read_unlock();
297 }
298
299 socket->type = data->type;
300
301 DBG3("Consumer socket created (fd: %d) and added to output",
302 data->cmd_sock);
303
304 error:
305 return ret;
306 }
307
308 /*
309 * Return the consumer socket from the given consumer output with the right
310 * bitness. On error, returns NULL.
311 *
312 * The caller MUST acquire a rcu read side lock and keep it until the socket
313 * object reference is not needed anymore.
314 */
315 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
316 struct consumer_output *consumer)
317 {
318 int consumer_fd;
319 struct consumer_socket *socket = NULL;
320
321 switch (bits) {
322 case 64:
323 consumer_fd = uatomic_read(&ust_consumerd64_fd);
324 break;
325 case 32:
326 consumer_fd = uatomic_read(&ust_consumerd32_fd);
327 break;
328 default:
329 assert(0);
330 goto end;
331 }
332
333 socket = consumer_find_socket(consumer_fd, consumer);
334 if (!socket) {
335 ERR("Consumer socket fd %d not found in consumer obj %p",
336 consumer_fd, consumer);
337 }
338
339 end:
340 return socket;
341 }
342
343 /*
344 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
345 * be acquired before calling this function and across use of the
346 * returned consumer_socket.
347 */
348 struct consumer_socket *consumer_find_socket(int key,
349 struct consumer_output *consumer)
350 {
351 struct lttng_ht_iter iter;
352 struct lttng_ht_node_ulong *node;
353 struct consumer_socket *socket = NULL;
354
355 /* Negative keys are lookup failures */
356 if (key < 0 || consumer == NULL) {
357 return NULL;
358 }
359
360 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
361 &iter);
362 node = lttng_ht_iter_get_node_ulong(&iter);
363 if (node != NULL) {
364 socket = caa_container_of(node, struct consumer_socket, node);
365 }
366
367 return socket;
368 }
369
370 /*
371 * Allocate a new consumer_socket and return the pointer.
372 */
373 struct consumer_socket *consumer_allocate_socket(int *fd)
374 {
375 struct consumer_socket *socket = NULL;
376
377 assert(fd);
378
379 socket = zmalloc(sizeof(struct consumer_socket));
380 if (socket == NULL) {
381 PERROR("zmalloc consumer socket");
382 goto error;
383 }
384
385 socket->fd_ptr = fd;
386 lttng_ht_node_init_ulong(&socket->node, *fd);
387
388 error:
389 return socket;
390 }
391
392 /*
393 * Add consumer socket to consumer output object. Read side lock must be
394 * acquired before calling this function.
395 */
396 void consumer_add_socket(struct consumer_socket *sock,
397 struct consumer_output *consumer)
398 {
399 assert(sock);
400 assert(consumer);
401
402 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
403 }
404
405 /*
406 * Delte consumer socket to consumer output object. Read side lock must be
407 * acquired before calling this function.
408 */
409 void consumer_del_socket(struct consumer_socket *sock,
410 struct consumer_output *consumer)
411 {
412 int ret;
413 struct lttng_ht_iter iter;
414
415 assert(sock);
416 assert(consumer);
417
418 iter.iter.node = &sock->node.node;
419 ret = lttng_ht_del(consumer->socks, &iter);
420 assert(!ret);
421 }
422
423 /*
424 * RCU destroy call function.
425 */
426 static void destroy_socket_rcu(struct rcu_head *head)
427 {
428 struct lttng_ht_node_ulong *node =
429 caa_container_of(head, struct lttng_ht_node_ulong, head);
430 struct consumer_socket *socket =
431 caa_container_of(node, struct consumer_socket, node);
432
433 free(socket);
434 }
435
436 /*
437 * Destroy and free socket pointer in a call RCU. Read side lock must be
438 * acquired before calling this function.
439 */
440 void consumer_destroy_socket(struct consumer_socket *sock)
441 {
442 assert(sock);
443
444 /*
445 * We DO NOT close the file descriptor here since it is global to the
446 * session daemon and is closed only if the consumer dies or a custom
447 * consumer was registered,
448 */
449 if (sock->registered) {
450 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
451 lttcomm_close_unix_sock(*sock->fd_ptr);
452 }
453
454 call_rcu(&sock->node.head, destroy_socket_rcu);
455 }
456
457 /*
458 * Allocate and assign data to a consumer_output object.
459 *
460 * Return pointer to structure.
461 */
462 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
463 {
464 struct consumer_output *output = NULL;
465
466 output = zmalloc(sizeof(struct consumer_output));
467 if (output == NULL) {
468 PERROR("zmalloc consumer_output");
469 goto error;
470 }
471
472 /* By default, consumer output is enabled */
473 output->enabled = 1;
474 output->type = type;
475 output->net_seq_index = (uint64_t) -1ULL;
476 urcu_ref_init(&output->ref);
477
478 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
479
480 error:
481 return output;
482 }
483
484 /*
485 * Iterate over the consumer output socket hash table and destroy them. The
486 * socket file descriptor are only closed if the consumer output was
487 * registered meaning it's an external consumer.
488 */
489 void consumer_destroy_output_sockets(struct consumer_output *obj)
490 {
491 struct lttng_ht_iter iter;
492 struct consumer_socket *socket;
493
494 if (!obj->socks) {
495 return;
496 }
497
498 rcu_read_lock();
499 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
500 consumer_del_socket(socket, obj);
501 consumer_destroy_socket(socket);
502 }
503 rcu_read_unlock();
504 }
505
506 /*
507 * Delete the consumer_output object from the list and free the ptr.
508 *
509 * Should *NOT* be called with RCU read-side lock held.
510 */
511 static void consumer_release_output(struct urcu_ref *ref)
512 {
513 struct consumer_output *obj =
514 caa_container_of(ref, struct consumer_output, ref);
515
516 consumer_destroy_output_sockets(obj);
517
518 if (obj->socks) {
519 /* Finally destroy HT */
520 ht_cleanup_push(obj->socks);
521 }
522
523 free(obj);
524 }
525
526 /*
527 * Get the consumer_output object.
528 */
529 void consumer_output_get(struct consumer_output *obj)
530 {
531 urcu_ref_get(&obj->ref);
532 }
533
534 /*
535 * Put the consumer_output object.
536 *
537 * Should *NOT* be called with RCU read-side lock held.
538 */
539 void consumer_output_put(struct consumer_output *obj)
540 {
541 if (!obj) {
542 return;
543 }
544 urcu_ref_put(&obj->ref, consumer_release_output);
545 }
546
547 /*
548 * Copy consumer output and returned the newly allocated copy.
549 *
550 * Should *NOT* be called with RCU read-side lock held.
551 */
552 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
553 {
554 int ret;
555 struct consumer_output *output;
556
557 assert(obj);
558
559 output = consumer_create_output(obj->type);
560 if (output == NULL) {
561 goto end;
562 }
563 output->enabled = obj->enabled;
564 output->net_seq_index = obj->net_seq_index;
565 memcpy(output->subdir, obj->subdir, sizeof(output->subdir));
566 output->snapshot = obj->snapshot;
567 output->relay_major_version = obj->relay_major_version;
568 output->relay_minor_version = obj->relay_minor_version;
569 memcpy(&output->dst, &obj->dst, sizeof(output->dst));
570 ret = consumer_copy_sockets(output, obj);
571 if (ret < 0) {
572 goto error_put;
573 }
574 end:
575 return output;
576
577 error_put:
578 consumer_output_put(output);
579 return NULL;
580 }
581
582 /*
583 * Copy consumer sockets from src to dst.
584 *
585 * Return 0 on success or else a negative value.
586 */
587 int consumer_copy_sockets(struct consumer_output *dst,
588 struct consumer_output *src)
589 {
590 int ret = 0;
591 struct lttng_ht_iter iter;
592 struct consumer_socket *socket, *copy_sock;
593
594 assert(dst);
595 assert(src);
596
597 rcu_read_lock();
598 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
599 /* Ignore socket that are already there. */
600 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
601 if (copy_sock) {
602 continue;
603 }
604
605 /* Create new socket object. */
606 copy_sock = consumer_allocate_socket(socket->fd_ptr);
607 if (copy_sock == NULL) {
608 rcu_read_unlock();
609 ret = -ENOMEM;
610 goto error;
611 }
612
613 copy_sock->registered = socket->registered;
614 /*
615 * This is valid because this lock is shared accross all consumer
616 * object being the global lock of the consumer data structure of the
617 * session daemon.
618 */
619 copy_sock->lock = socket->lock;
620 consumer_add_socket(copy_sock, dst);
621 }
622 rcu_read_unlock();
623
624 error:
625 return ret;
626 }
627
628 /*
629 * Set network URI to the consumer output object.
630 *
631 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
632 * error.
633 */
634 int consumer_set_network_uri(struct consumer_output *obj,
635 struct lttng_uri *uri)
636 {
637 int ret;
638 char tmp_path[PATH_MAX];
639 char hostname[HOST_NAME_MAX];
640 struct lttng_uri *dst_uri = NULL;
641
642 /* Code flow error safety net. */
643 assert(obj);
644 assert(uri);
645
646 switch (uri->stype) {
647 case LTTNG_STREAM_CONTROL:
648 dst_uri = &obj->dst.net.control;
649 obj->dst.net.control_isset = 1;
650 if (uri->port == 0) {
651 /* Assign default port. */
652 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
653 } else {
654 if (obj->dst.net.data_isset && uri->port ==
655 obj->dst.net.data.port) {
656 ret = -LTTNG_ERR_INVALID;
657 goto error;
658 }
659 }
660 DBG3("Consumer control URI set with port %d", uri->port);
661 break;
662 case LTTNG_STREAM_DATA:
663 dst_uri = &obj->dst.net.data;
664 obj->dst.net.data_isset = 1;
665 if (uri->port == 0) {
666 /* Assign default port. */
667 uri->port = DEFAULT_NETWORK_DATA_PORT;
668 } else {
669 if (obj->dst.net.control_isset && uri->port ==
670 obj->dst.net.control.port) {
671 ret = -LTTNG_ERR_INVALID;
672 goto error;
673 }
674 }
675 DBG3("Consumer data URI set with port %d", uri->port);
676 break;
677 default:
678 ERR("Set network uri type unknown %d", uri->stype);
679 ret = -LTTNG_ERR_INVALID;
680 goto error;
681 }
682
683 ret = uri_compare(dst_uri, uri);
684 if (!ret) {
685 /* Same URI, don't touch it and return success. */
686 DBG3("URI network compare are the same");
687 goto equal;
688 }
689
690 /* URIs were not equal, replacing it. */
691 memset(dst_uri, 0, sizeof(struct lttng_uri));
692 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
693 obj->type = CONSUMER_DST_NET;
694
695 /* Handle subdir and add hostname in front. */
696 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
697 /* Get hostname to append it in the pathname */
698 ret = gethostname(hostname, sizeof(hostname));
699 if (ret < 0) {
700 PERROR("gethostname. Fallback on default localhost");
701 strncpy(hostname, "localhost", sizeof(hostname));
702 }
703 hostname[sizeof(hostname) - 1] = '\0';
704
705 /* Setup consumer subdir if none present in the control URI */
706 if (strlen(dst_uri->subdir) == 0) {
707 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
708 hostname, obj->subdir);
709 } else {
710 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
711 hostname, dst_uri->subdir);
712 }
713 if (ret < 0) {
714 PERROR("snprintf set consumer uri subdir");
715 ret = -LTTNG_ERR_NOMEM;
716 goto error;
717 }
718
719 if (lttng_strncpy(obj->dst.net.base_dir, tmp_path,
720 sizeof(obj->dst.net.base_dir))) {
721 ret = -LTTNG_ERR_INVALID;
722 goto error;
723 }
724 DBG3("Consumer set network uri base_dir path %s", tmp_path);
725 }
726
727 return 0;
728 equal:
729 return 1;
730 error:
731 return ret;
732 }
733
734 /*
735 * Send file descriptor to consumer via sock.
736 *
737 * The consumer socket lock must be held by the caller.
738 */
739 int consumer_send_fds(struct consumer_socket *sock, const int *fds,
740 size_t nb_fd)
741 {
742 int ret;
743
744 assert(fds);
745 assert(sock);
746 assert(nb_fd > 0);
747 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
748
749 ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
750 if (ret < 0) {
751 /* The above call will print a PERROR on error. */
752 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
753 goto error;
754 }
755
756 ret = consumer_recv_status_reply(sock);
757 error:
758 return ret;
759 }
760
761 /*
762 * Consumer send communication message structure to consumer.
763 *
764 * The consumer socket lock must be held by the caller.
765 */
766 int consumer_send_msg(struct consumer_socket *sock,
767 struct lttcomm_consumer_msg *msg)
768 {
769 int ret;
770
771 assert(msg);
772 assert(sock);
773 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
774
775 ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
776 if (ret < 0) {
777 goto error;
778 }
779
780 ret = consumer_recv_status_reply(sock);
781
782 error:
783 return ret;
784 }
785
786 /*
787 * Consumer send channel communication message structure to consumer.
788 *
789 * The consumer socket lock must be held by the caller.
790 */
791 int consumer_send_channel(struct consumer_socket *sock,
792 struct lttcomm_consumer_msg *msg)
793 {
794 int ret;
795
796 assert(msg);
797 assert(sock);
798
799 ret = consumer_send_msg(sock, msg);
800 if (ret < 0) {
801 goto error;
802 }
803
804 error:
805 return ret;
806 }
807
808 /*
809 * Populate the given consumer msg structure with the ask_channel command
810 * information.
811 */
812 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
813 uint64_t subbuf_size,
814 uint64_t num_subbuf,
815 int overwrite,
816 unsigned int switch_timer_interval,
817 unsigned int read_timer_interval,
818 unsigned int live_timer_interval,
819 unsigned int monitor_timer_interval,
820 int output,
821 int type,
822 uint64_t session_id,
823 const char *pathname,
824 const char *name,
825 uid_t uid,
826 gid_t gid,
827 uint64_t relayd_id,
828 uint64_t key,
829 unsigned char *uuid,
830 uint32_t chan_id,
831 uint64_t tracefile_size,
832 uint64_t tracefile_count,
833 uint64_t session_id_per_pid,
834 unsigned int monitor,
835 uint32_t ust_app_uid,
836 int64_t blocking_timeout,
837 const char *root_shm_path,
838 const char *shm_path,
839 uint64_t trace_archive_id)
840 {
841 assert(msg);
842
843 /* Zeroed structure */
844 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
845
846 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
847 msg->u.ask_channel.subbuf_size = subbuf_size;
848 msg->u.ask_channel.num_subbuf = num_subbuf ;
849 msg->u.ask_channel.overwrite = overwrite;
850 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
851 msg->u.ask_channel.read_timer_interval = read_timer_interval;
852 msg->u.ask_channel.live_timer_interval = live_timer_interval;
853 msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
854 msg->u.ask_channel.output = output;
855 msg->u.ask_channel.type = type;
856 msg->u.ask_channel.session_id = session_id;
857 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
858 msg->u.ask_channel.uid = uid;
859 msg->u.ask_channel.gid = gid;
860 msg->u.ask_channel.relayd_id = relayd_id;
861 msg->u.ask_channel.key = key;
862 msg->u.ask_channel.chan_id = chan_id;
863 msg->u.ask_channel.tracefile_size = tracefile_size;
864 msg->u.ask_channel.tracefile_count = tracefile_count;
865 msg->u.ask_channel.monitor = monitor;
866 msg->u.ask_channel.ust_app_uid = ust_app_uid;
867 msg->u.ask_channel.blocking_timeout = blocking_timeout;
868 msg->u.ask_channel.trace_archive_id = trace_archive_id;
869
870 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
871
872 if (pathname) {
873 strncpy(msg->u.ask_channel.pathname, pathname,
874 sizeof(msg->u.ask_channel.pathname));
875 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
876 }
877
878 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
879 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
880
881 if (root_shm_path) {
882 strncpy(msg->u.ask_channel.root_shm_path, root_shm_path,
883 sizeof(msg->u.ask_channel.root_shm_path));
884 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] = '\0';
885 }
886 if (shm_path) {
887 strncpy(msg->u.ask_channel.shm_path, shm_path,
888 sizeof(msg->u.ask_channel.shm_path));
889 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
890 }
891 }
892
893 /*
894 * Init channel communication message structure.
895 */
896 void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg,
897 uint64_t channel_key,
898 uint64_t session_id,
899 const char *pathname,
900 uid_t uid,
901 gid_t gid,
902 uint64_t relayd_id,
903 const char *name,
904 unsigned int nb_init_streams,
905 enum lttng_event_output output,
906 int type,
907 uint64_t tracefile_size,
908 uint64_t tracefile_count,
909 unsigned int monitor,
910 unsigned int live_timer_interval,
911 unsigned int monitor_timer_interval)
912 {
913 assert(msg);
914
915 /* Zeroed structure */
916 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
917
918 /* Send channel */
919 msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
920 msg->u.channel.channel_key = channel_key;
921 msg->u.channel.session_id = session_id;
922 msg->u.channel.uid = uid;
923 msg->u.channel.gid = gid;
924 msg->u.channel.relayd_id = relayd_id;
925 msg->u.channel.nb_init_streams = nb_init_streams;
926 msg->u.channel.output = output;
927 msg->u.channel.type = type;
928 msg->u.channel.tracefile_size = tracefile_size;
929 msg->u.channel.tracefile_count = tracefile_count;
930 msg->u.channel.monitor = monitor;
931 msg->u.channel.live_timer_interval = live_timer_interval;
932 msg->u.channel.monitor_timer_interval = monitor_timer_interval;
933
934 strncpy(msg->u.channel.pathname, pathname,
935 sizeof(msg->u.channel.pathname));
936 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
937
938 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
939 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
940 }
941
942 /*
943 * Init stream communication message structure.
944 */
945 void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg,
946 uint64_t channel_key,
947 uint64_t stream_key,
948 int32_t cpu,
949 uint64_t trace_archive_id)
950 {
951 assert(msg);
952
953 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
954
955 msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM;
956 msg->u.stream.channel_key = channel_key;
957 msg->u.stream.stream_key = stream_key;
958 msg->u.stream.cpu = cpu;
959 msg->u.stream.trace_archive_id = trace_archive_id;
960 }
961
962 void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
963 enum lttng_consumer_command cmd,
964 uint64_t channel_key, uint64_t net_seq_idx)
965 {
966 assert(msg);
967
968 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
969
970 msg->cmd_type = cmd;
971 msg->u.sent_streams.channel_key = channel_key;
972 msg->u.sent_streams.net_seq_idx = net_seq_idx;
973 }
974
975 /*
976 * Send stream communication structure to the consumer.
977 */
978 int consumer_send_stream(struct consumer_socket *sock,
979 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
980 const int *fds, size_t nb_fd)
981 {
982 int ret;
983
984 assert(msg);
985 assert(dst);
986 assert(sock);
987 assert(fds);
988
989 ret = consumer_send_msg(sock, msg);
990 if (ret < 0) {
991 goto error;
992 }
993
994 ret = consumer_send_fds(sock, fds, nb_fd);
995 if (ret < 0) {
996 goto error;
997 }
998
999 error:
1000 return ret;
1001 }
1002
1003 /*
1004 * Send relayd socket to consumer associated with a session name.
1005 *
1006 * The consumer socket lock must be held by the caller.
1007 *
1008 * On success return positive value. On error, negative value.
1009 */
1010 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
1011 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
1012 enum lttng_stream_type type, uint64_t session_id,
1013 char *session_name, char *hostname, int session_live_timer)
1014 {
1015 int ret;
1016 struct lttcomm_consumer_msg msg;
1017
1018 /* Code flow error. Safety net. */
1019 assert(rsock);
1020 assert(consumer);
1021 assert(consumer_sock);
1022
1023 memset(&msg, 0, sizeof(msg));
1024 /* Bail out if consumer is disabled */
1025 if (!consumer->enabled) {
1026 ret = LTTNG_OK;
1027 goto error;
1028 }
1029
1030 if (type == LTTNG_STREAM_CONTROL) {
1031 ret = relayd_create_session(rsock,
1032 &msg.u.relayd_sock.relayd_session_id,
1033 session_name, hostname, session_live_timer,
1034 consumer->snapshot);
1035 if (ret < 0) {
1036 /* Close the control socket. */
1037 (void) relayd_close(rsock);
1038 goto error;
1039 }
1040 }
1041
1042 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1043 /*
1044 * Assign network consumer output index using the temporary consumer since
1045 * this call should only be made from within a set_consumer_uri() function
1046 * call in the session daemon.
1047 */
1048 msg.u.relayd_sock.net_index = consumer->net_seq_index;
1049 msg.u.relayd_sock.type = type;
1050 msg.u.relayd_sock.session_id = session_id;
1051 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
1052
1053 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
1054 ret = consumer_send_msg(consumer_sock, &msg);
1055 if (ret < 0) {
1056 goto error;
1057 }
1058
1059 DBG3("Sending relayd socket file descriptor to consumer");
1060 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
1061 if (ret < 0) {
1062 goto error;
1063 }
1064
1065 DBG2("Consumer relayd socket sent");
1066
1067 error:
1068 return ret;
1069 }
1070
1071 static
1072 int consumer_send_pipe(struct consumer_socket *consumer_sock,
1073 enum lttng_consumer_command cmd, int pipe)
1074 {
1075 int ret;
1076 struct lttcomm_consumer_msg msg;
1077 const char *pipe_name;
1078 const char *command_name;
1079
1080 switch (cmd) {
1081 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1082 pipe_name = "channel monitor";
1083 command_name = "SET_CHANNEL_MONITOR_PIPE";
1084 break;
1085 default:
1086 ERR("Unexpected command received in %s (cmd = %d)", __func__,
1087 (int) cmd);
1088 abort();
1089 }
1090
1091 /* Code flow error. Safety net. */
1092
1093 memset(&msg, 0, sizeof(msg));
1094 msg.cmd_type = cmd;
1095
1096 pthread_mutex_lock(consumer_sock->lock);
1097 DBG3("Sending %s command to consumer", command_name);
1098 ret = consumer_send_msg(consumer_sock, &msg);
1099 if (ret < 0) {
1100 goto error;
1101 }
1102
1103 DBG3("Sending %s pipe %d to consumer on socket %d",
1104 pipe_name,
1105 pipe, *consumer_sock->fd_ptr);
1106 ret = consumer_send_fds(consumer_sock, &pipe, 1);
1107 if (ret < 0) {
1108 goto error;
1109 }
1110
1111 DBG2("%s pipe successfully sent", pipe_name);
1112 error:
1113 pthread_mutex_unlock(consumer_sock->lock);
1114 return ret;
1115 }
1116
1117 int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock,
1118 int pipe)
1119 {
1120 return consumer_send_pipe(consumer_sock,
1121 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe);
1122 }
1123
1124 /*
1125 * Set consumer subdirectory using the session name and a generated datetime if
1126 * needed. This is appended to the current subdirectory.
1127 */
1128 int consumer_set_subdir(struct consumer_output *consumer,
1129 const char *session_name)
1130 {
1131 int ret = 0;
1132 unsigned int have_default_name = 0;
1133 char datetime[16], tmp_path[PATH_MAX];
1134 time_t rawtime;
1135 struct tm *timeinfo;
1136
1137 assert(consumer);
1138 assert(session_name);
1139
1140 memset(tmp_path, 0, sizeof(tmp_path));
1141
1142 /* Flag if we have a default session. */
1143 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
1144 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
1145 have_default_name = 1;
1146 } else {
1147 /* Get date and time for session path */
1148 time(&rawtime);
1149 timeinfo = localtime(&rawtime);
1150 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
1151 }
1152
1153 if (have_default_name) {
1154 ret = snprintf(tmp_path, sizeof(tmp_path),
1155 "%s/%s", consumer->subdir, session_name);
1156 } else {
1157 ret = snprintf(tmp_path, sizeof(tmp_path),
1158 "%s/%s-%s/", consumer->subdir, session_name, datetime);
1159 }
1160 if (ret < 0) {
1161 PERROR("snprintf session name date");
1162 goto error;
1163 }
1164
1165 if (lttng_strncpy(consumer->subdir, tmp_path,
1166 sizeof(consumer->subdir))) {
1167 ret = -EINVAL;
1168 goto error;
1169 }
1170 DBG2("Consumer subdir set to %s", consumer->subdir);
1171
1172 error:
1173 return ret;
1174 }
1175
1176 /*
1177 * Ask the consumer if the data is pending for the specific session id.
1178 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1179 */
1180 int consumer_is_data_pending(uint64_t session_id,
1181 struct consumer_output *consumer)
1182 {
1183 int ret;
1184 int32_t ret_code = 0; /* Default is that the data is NOT pending */
1185 struct consumer_socket *socket;
1186 struct lttng_ht_iter iter;
1187 struct lttcomm_consumer_msg msg;
1188
1189 assert(consumer);
1190
1191 DBG3("Consumer data pending for id %" PRIu64, session_id);
1192
1193 memset(&msg, 0, sizeof(msg));
1194 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1195 msg.u.data_pending.session_id = session_id;
1196
1197 /* Send command for each consumer */
1198 rcu_read_lock();
1199 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1200 node.node) {
1201 pthread_mutex_lock(socket->lock);
1202 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1203 if (ret < 0) {
1204 pthread_mutex_unlock(socket->lock);
1205 goto error_unlock;
1206 }
1207
1208 /*
1209 * No need for a recv reply status because the answer to the command is
1210 * the reply status message.
1211 */
1212
1213 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1214 if (ret < 0) {
1215 pthread_mutex_unlock(socket->lock);
1216 goto error_unlock;
1217 }
1218 pthread_mutex_unlock(socket->lock);
1219
1220 if (ret_code == 1) {
1221 break;
1222 }
1223 }
1224 rcu_read_unlock();
1225
1226 DBG("Consumer data is %s pending for session id %" PRIu64,
1227 ret_code == 1 ? "" : "NOT", session_id);
1228 return ret_code;
1229
1230 error_unlock:
1231 rcu_read_unlock();
1232 return -1;
1233 }
1234
1235 /*
1236 * Send a flush command to consumer using the given channel key.
1237 *
1238 * Return 0 on success else a negative value.
1239 */
1240 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1241 {
1242 int ret;
1243 struct lttcomm_consumer_msg msg;
1244
1245 assert(socket);
1246
1247 DBG2("Consumer flush channel key %" PRIu64, key);
1248
1249 memset(&msg, 0, sizeof(msg));
1250 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1251 msg.u.flush_channel.key = key;
1252
1253 pthread_mutex_lock(socket->lock);
1254 health_code_update();
1255
1256 ret = consumer_send_msg(socket, &msg);
1257 if (ret < 0) {
1258 goto end;
1259 }
1260
1261 end:
1262 health_code_update();
1263 pthread_mutex_unlock(socket->lock);
1264 return ret;
1265 }
1266
1267 /*
1268 * Send a clear quiescent command to consumer using the given channel key.
1269 *
1270 * Return 0 on success else a negative value.
1271 */
1272 int consumer_clear_quiescent_channel(struct consumer_socket *socket, uint64_t key)
1273 {
1274 int ret;
1275 struct lttcomm_consumer_msg msg;
1276
1277 assert(socket);
1278
1279 DBG2("Consumer clear quiescent channel key %" PRIu64, key);
1280
1281 memset(&msg, 0, sizeof(msg));
1282 msg.cmd_type = LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL;
1283 msg.u.clear_quiescent_channel.key = key;
1284
1285 pthread_mutex_lock(socket->lock);
1286 health_code_update();
1287
1288 ret = consumer_send_msg(socket, &msg);
1289 if (ret < 0) {
1290 goto end;
1291 }
1292
1293 end:
1294 health_code_update();
1295 pthread_mutex_unlock(socket->lock);
1296 return ret;
1297 }
1298
1299 /*
1300 * Send a close metadata command to consumer using the given channel key.
1301 * Called with registry lock held.
1302 *
1303 * Return 0 on success else a negative value.
1304 */
1305 int consumer_close_metadata(struct consumer_socket *socket,
1306 uint64_t metadata_key)
1307 {
1308 int ret;
1309 struct lttcomm_consumer_msg msg;
1310
1311 assert(socket);
1312
1313 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1314
1315 memset(&msg, 0, sizeof(msg));
1316 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1317 msg.u.close_metadata.key = metadata_key;
1318
1319 pthread_mutex_lock(socket->lock);
1320 health_code_update();
1321
1322 ret = consumer_send_msg(socket, &msg);
1323 if (ret < 0) {
1324 goto end;
1325 }
1326
1327 end:
1328 health_code_update();
1329 pthread_mutex_unlock(socket->lock);
1330 return ret;
1331 }
1332
1333 /*
1334 * Send a setup metdata command to consumer using the given channel key.
1335 *
1336 * Return 0 on success else a negative value.
1337 */
1338 int consumer_setup_metadata(struct consumer_socket *socket,
1339 uint64_t metadata_key)
1340 {
1341 int ret;
1342 struct lttcomm_consumer_msg msg;
1343
1344 assert(socket);
1345
1346 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1347
1348 memset(&msg, 0, sizeof(msg));
1349 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1350 msg.u.setup_metadata.key = metadata_key;
1351
1352 pthread_mutex_lock(socket->lock);
1353 health_code_update();
1354
1355 ret = consumer_send_msg(socket, &msg);
1356 if (ret < 0) {
1357 goto end;
1358 }
1359
1360 end:
1361 health_code_update();
1362 pthread_mutex_unlock(socket->lock);
1363 return ret;
1364 }
1365
1366 /*
1367 * Send metadata string to consumer.
1368 * RCU read-side lock must be held to guarantee existence of socket.
1369 *
1370 * Return 0 on success else a negative value.
1371 */
1372 int consumer_push_metadata(struct consumer_socket *socket,
1373 uint64_t metadata_key, char *metadata_str, size_t len,
1374 size_t target_offset, uint64_t version)
1375 {
1376 int ret;
1377 struct lttcomm_consumer_msg msg;
1378
1379 assert(socket);
1380
1381 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1382
1383 pthread_mutex_lock(socket->lock);
1384
1385 memset(&msg, 0, sizeof(msg));
1386 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1387 msg.u.push_metadata.key = metadata_key;
1388 msg.u.push_metadata.target_offset = target_offset;
1389 msg.u.push_metadata.len = len;
1390 msg.u.push_metadata.version = version;
1391
1392 health_code_update();
1393 ret = consumer_send_msg(socket, &msg);
1394 if (ret < 0 || len == 0) {
1395 goto end;
1396 }
1397
1398 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1399 len);
1400
1401 ret = consumer_socket_send(socket, metadata_str, len);
1402 if (ret < 0) {
1403 goto end;
1404 }
1405
1406 health_code_update();
1407 ret = consumer_recv_status_reply(socket);
1408 if (ret < 0) {
1409 goto end;
1410 }
1411
1412 end:
1413 pthread_mutex_unlock(socket->lock);
1414 health_code_update();
1415 return ret;
1416 }
1417
1418 /*
1419 * Ask the consumer to snapshot a specific channel using the key.
1420 *
1421 * Return 0 on success or else a negative error.
1422 */
1423 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1424 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1425 const char *session_path, int wait, uint64_t nb_packets_per_stream,
1426 uint64_t trace_archive_id)
1427 {
1428 int ret;
1429 struct lttcomm_consumer_msg msg;
1430
1431 assert(socket);
1432 assert(output);
1433 assert(output->consumer);
1434
1435 DBG("Consumer snapshot channel key %" PRIu64, key);
1436
1437 memset(&msg, 0, sizeof(msg));
1438 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1439 msg.u.snapshot_channel.key = key;
1440 msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
1441 msg.u.snapshot_channel.metadata = metadata;
1442 msg.u.snapshot_channel.trace_archive_id = trace_archive_id;
1443
1444 if (output->consumer->type == CONSUMER_DST_NET) {
1445 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1446 msg.u.snapshot_channel.use_relayd = 1;
1447 ret = snprintf(msg.u.snapshot_channel.pathname,
1448 sizeof(msg.u.snapshot_channel.pathname),
1449 "%s/%s/%s-%s-%" PRIu64 "%s",
1450 output->consumer->dst.net.base_dir,
1451 output->consumer->subdir,
1452 output->name, output->datetime,
1453 output->nb_snapshot,
1454 session_path);
1455 if (ret < 0) {
1456 ret = -LTTNG_ERR_NOMEM;
1457 goto error;
1458 } else if (ret >= sizeof(msg.u.snapshot_channel.pathname)) {
1459 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%i bytes required) with path \"%s/%s/%s-%s-%" PRIu64 "%s\"",
1460 sizeof(msg.u.snapshot_channel.pathname),
1461 ret, output->consumer->dst.net.base_dir,
1462 output->consumer->subdir,
1463 output->name, output->datetime,
1464 output->nb_snapshot,
1465 session_path);
1466 ret = -LTTNG_ERR_SNAPSHOT_FAIL;
1467 goto error;
1468 }
1469 } else {
1470 ret = snprintf(msg.u.snapshot_channel.pathname,
1471 sizeof(msg.u.snapshot_channel.pathname),
1472 "%s/%s-%s-%" PRIu64 "%s",
1473 output->consumer->dst.session_root_path,
1474 output->name, output->datetime,
1475 output->nb_snapshot,
1476 session_path);
1477 if (ret < 0) {
1478 ret = -LTTNG_ERR_NOMEM;
1479 goto error;
1480 } else if (ret >= sizeof(msg.u.snapshot_channel.pathname)) {
1481 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%i bytes required) with path \"%s/%s-%s-%" PRIu64 "%s\"",
1482 sizeof(msg.u.snapshot_channel.pathname),
1483 ret, output->consumer->dst.session_root_path,
1484 output->name, output->datetime, output->nb_snapshot,
1485 session_path);
1486 ret = -LTTNG_ERR_SNAPSHOT_FAIL;
1487 goto error;
1488 }
1489
1490 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1491
1492 /* Create directory. Ignore if exist. */
1493 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1494 S_IRWXU | S_IRWXG, uid, gid);
1495 if (ret < 0) {
1496 if (errno != EEXIST) {
1497 ERR("Trace directory creation error");
1498 goto error;
1499 }
1500 }
1501 }
1502
1503 health_code_update();
1504 pthread_mutex_lock(socket->lock);
1505 ret = consumer_send_msg(socket, &msg);
1506 pthread_mutex_unlock(socket->lock);
1507 if (ret < 0) {
1508 goto error;
1509 }
1510
1511 error:
1512 health_code_update();
1513 return ret;
1514 }
1515
1516 /*
1517 * Ask the consumer the number of discarded events for a channel.
1518 */
1519 int consumer_get_discarded_events(uint64_t session_id, uint64_t channel_key,
1520 struct consumer_output *consumer, uint64_t *discarded)
1521 {
1522 int ret;
1523 struct consumer_socket *socket;
1524 struct lttng_ht_iter iter;
1525 struct lttcomm_consumer_msg msg;
1526
1527 assert(consumer);
1528
1529 DBG3("Consumer discarded events id %" PRIu64, session_id);
1530
1531 memset(&msg, 0, sizeof(msg));
1532 msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1533 msg.u.discarded_events.session_id = session_id;
1534 msg.u.discarded_events.channel_key = channel_key;
1535
1536 *discarded = 0;
1537
1538 /* Send command for each consumer */
1539 rcu_read_lock();
1540 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1541 node.node) {
1542 uint64_t consumer_discarded = 0;
1543 pthread_mutex_lock(socket->lock);
1544 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1545 if (ret < 0) {
1546 pthread_mutex_unlock(socket->lock);
1547 goto end;
1548 }
1549
1550 /*
1551 * No need for a recv reply status because the answer to the
1552 * command is the reply status message.
1553 */
1554 ret = consumer_socket_recv(socket, &consumer_discarded,
1555 sizeof(consumer_discarded));
1556 if (ret < 0) {
1557 ERR("get discarded events");
1558 pthread_mutex_unlock(socket->lock);
1559 goto end;
1560 }
1561 pthread_mutex_unlock(socket->lock);
1562 *discarded += consumer_discarded;
1563 }
1564 ret = 0;
1565 DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64,
1566 *discarded, session_id);
1567
1568 end:
1569 rcu_read_unlock();
1570 return ret;
1571 }
1572
1573 /*
1574 * Ask the consumer the number of lost packets for a channel.
1575 */
1576 int consumer_get_lost_packets(uint64_t session_id, uint64_t channel_key,
1577 struct consumer_output *consumer, uint64_t *lost)
1578 {
1579 int ret;
1580 struct consumer_socket *socket;
1581 struct lttng_ht_iter iter;
1582 struct lttcomm_consumer_msg msg;
1583
1584 assert(consumer);
1585
1586 DBG3("Consumer lost packets id %" PRIu64, session_id);
1587
1588 memset(&msg, 0, sizeof(msg));
1589 msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1590 msg.u.lost_packets.session_id = session_id;
1591 msg.u.lost_packets.channel_key = channel_key;
1592
1593 *lost = 0;
1594
1595 /* Send command for each consumer */
1596 rcu_read_lock();
1597 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1598 node.node) {
1599 uint64_t consumer_lost = 0;
1600 pthread_mutex_lock(socket->lock);
1601 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1602 if (ret < 0) {
1603 pthread_mutex_unlock(socket->lock);
1604 goto end;
1605 }
1606
1607 /*
1608 * No need for a recv reply status because the answer to the
1609 * command is the reply status message.
1610 */
1611 ret = consumer_socket_recv(socket, &consumer_lost,
1612 sizeof(consumer_lost));
1613 if (ret < 0) {
1614 ERR("get lost packets");
1615 pthread_mutex_unlock(socket->lock);
1616 goto end;
1617 }
1618 pthread_mutex_unlock(socket->lock);
1619 *lost += consumer_lost;
1620 }
1621 ret = 0;
1622 DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64,
1623 *lost, session_id);
1624
1625 end:
1626 rcu_read_unlock();
1627 return ret;
1628 }
1629
1630 /*
1631 * Ask the consumer to rotate a channel.
1632 * domain_path contains "/kernel" for kernel or the complete path for UST
1633 * (ex: /ust/uid/1000/64-bit);
1634 *
1635 * The new_chunk_id is the session->rotate_count that has been incremented
1636 * when the rotation started. On the relay, this allows to keep track in which
1637 * chunk each stream is currently writing to (for the rotate_pending operation).
1638 */
1639 int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key,
1640 uid_t uid, gid_t gid, struct consumer_output *output,
1641 char *domain_path, bool is_metadata_channel,
1642 uint64_t new_chunk_id)
1643 {
1644 int ret;
1645 struct lttcomm_consumer_msg msg;
1646
1647 assert(socket);
1648
1649 DBG("Consumer rotate channel key %" PRIu64, key);
1650
1651 pthread_mutex_lock(socket->lock);
1652 memset(&msg, 0, sizeof(msg));
1653 msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL;
1654 msg.u.rotate_channel.key = key;
1655 msg.u.rotate_channel.metadata = !!is_metadata_channel;
1656 msg.u.rotate_channel.new_chunk_id = new_chunk_id;
1657
1658 if (output->type == CONSUMER_DST_NET) {
1659 msg.u.rotate_channel.relayd_id = output->net_seq_index;
1660 ret = snprintf(msg.u.rotate_channel.pathname,
1661 sizeof(msg.u.rotate_channel.pathname), "%s%s%s",
1662 output->dst.net.base_dir,
1663 output->chunk_path, domain_path);
1664 if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) {
1665 ERR("Failed to format channel path name when asking consumer to rotate channel");
1666 ret = -LTTNG_ERR_INVALID;
1667 goto error;
1668 }
1669 } else {
1670 msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL;
1671 ret = snprintf(msg.u.rotate_channel.pathname,
1672 sizeof(msg.u.rotate_channel.pathname), "%s%s%s",
1673 output->dst.session_root_path,
1674 output->chunk_path, domain_path);
1675 if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) {
1676 ERR("Failed to format channel path name when asking consumer to rotate channel");
1677 ret = -LTTNG_ERR_INVALID;
1678 goto error;
1679 }
1680 }
1681
1682 health_code_update();
1683 ret = consumer_send_msg(socket, &msg);
1684 if (ret < 0) {
1685 switch (-ret) {
1686 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1687 ret = -LTTNG_ERR_CHAN_NOT_FOUND;
1688 break;
1689 default:
1690 ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER;
1691 break;
1692 }
1693 goto error;
1694 }
1695 error:
1696 pthread_mutex_unlock(socket->lock);
1697 health_code_update();
1698 return ret;
1699 }
1700
1701 int consumer_rotate_rename(struct consumer_socket *socket, uint64_t session_id,
1702 const struct consumer_output *output, const char *old_path,
1703 const char *new_path, uid_t uid, gid_t gid)
1704 {
1705 int ret;
1706 struct lttcomm_consumer_msg msg;
1707 size_t old_path_length, new_path_length;
1708
1709 assert(socket);
1710 assert(old_path);
1711 assert(new_path);
1712
1713 DBG("Consumer rotate rename session %" PRIu64 ", old path = \"%s\", new_path = \"%s\"",
1714 session_id, old_path, new_path);
1715
1716 old_path_length = strlen(old_path);
1717 if (old_path_length >= sizeof(msg.u.rotate_rename.old_path)) {
1718 ERR("consumer_rotate_rename: old path length (%zu bytes) exceeds the maximal length allowed by the consumer protocol (%zu bytes)",
1719 old_path_length + 1, sizeof(msg.u.rotate_rename.old_path));
1720 ret = -LTTNG_ERR_INVALID;
1721 goto error;
1722 }
1723
1724 new_path_length = strlen(new_path);
1725 if (new_path_length >= sizeof(msg.u.rotate_rename.new_path)) {
1726 ERR("consumer_rotate_rename: new path length (%zu bytes) exceeds the maximal length allowed by the consumer protocol (%zu bytes)",
1727 new_path_length + 1, sizeof(msg.u.rotate_rename.new_path));
1728 ret = -LTTNG_ERR_INVALID;
1729 goto error;
1730 }
1731
1732 memset(&msg, 0, sizeof(msg));
1733 msg.cmd_type = LTTNG_CONSUMER_ROTATE_RENAME;
1734 msg.u.rotate_rename.session_id = session_id;
1735 msg.u.rotate_rename.uid = uid;
1736 msg.u.rotate_rename.gid = gid;
1737 strcpy(msg.u.rotate_rename.old_path, old_path);
1738 strcpy(msg.u.rotate_rename.new_path, new_path);
1739
1740 if (output->type == CONSUMER_DST_NET) {
1741 msg.u.rotate_rename.relayd_id = output->net_seq_index;
1742 } else {
1743 msg.u.rotate_rename.relayd_id = -1ULL;
1744 }
1745
1746 health_code_update();
1747 ret = consumer_send_msg(socket, &msg);
1748 if (ret < 0) {
1749 ret = -LTTNG_ERR_ROTATE_RENAME_FAIL_CONSUMER;
1750 goto error;
1751 }
1752
1753 error:
1754 health_code_update();
1755 return ret;
1756 }
1757
1758 /*
1759 * Ask the consumer if a rotation is locally pending. Must be called with the
1760 * socket lock held.
1761 *
1762 * Return 1 if the rotation is still pending, 0 if finished, a negative value
1763 * on error.
1764 */
1765 int consumer_check_rotation_pending_local(struct consumer_socket *socket,
1766 uint64_t session_id, uint64_t chunk_id)
1767 {
1768 int ret;
1769 struct lttcomm_consumer_msg msg;
1770 uint32_t pending = 0;
1771
1772 assert(socket);
1773
1774 DBG("Asking consumer to locally check for pending rotation for session %" PRIu64 ", chunk id %" PRIu64,
1775 session_id, chunk_id);
1776
1777 memset(&msg, 0, sizeof(msg));
1778 msg.cmd_type = LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL;
1779 msg.u.check_rotation_pending_local.session_id = session_id;
1780 msg.u.check_rotation_pending_local.chunk_id = chunk_id;
1781
1782 health_code_update();
1783 ret = consumer_send_msg(socket, &msg);
1784 if (ret < 0) {
1785 ret = -LTTNG_ERR_ROTATION_PENDING_LOCAL_FAIL_CONSUMER;
1786 goto error;
1787 }
1788
1789 ret = consumer_socket_recv(socket, &pending, sizeof(pending));
1790 if (ret < 0) {
1791 goto error;
1792 }
1793
1794 ret = pending;
1795
1796 error:
1797 health_code_update();
1798 return ret;
1799 }
1800
1801 /*
1802 * Ask the consumer if a rotation is pending on the relayd. Must be called with
1803 * the socket lock held.
1804 *
1805 * Return 1 if the rotation is still pending, 0 if finished, a negative value
1806 * on error.
1807 */
1808 int consumer_check_rotation_pending_relay(struct consumer_socket *socket,
1809 const struct consumer_output *output, uint64_t session_id,
1810 uint64_t chunk_id)
1811 {
1812 int ret;
1813 struct lttcomm_consumer_msg msg;
1814 uint32_t pending = 0;
1815
1816 assert(socket);
1817
1818 DBG("Asking consumer to check for pending rotation on relay for session %" PRIu64 ", chunk id %" PRIu64,
1819 session_id, chunk_id);
1820 assert(output->type == CONSUMER_DST_NET);
1821
1822 memset(&msg, 0, sizeof(msg));
1823 msg.cmd_type = LTTNG_CONSUMER_CHECK_ROTATION_PENDING_RELAY;
1824 msg.u.check_rotation_pending_relay.session_id = session_id;
1825 msg.u.check_rotation_pending_relay.relayd_id = output->net_seq_index;
1826 msg.u.check_rotation_pending_relay.chunk_id = chunk_id;
1827
1828 health_code_update();
1829 ret = consumer_send_msg(socket, &msg);
1830 if (ret < 0) {
1831 ret = -LTTNG_ERR_ROTATION_PENDING_RELAY_FAIL_CONSUMER;
1832 goto error;
1833 }
1834
1835 ret = consumer_socket_recv(socket, &pending, sizeof(pending));
1836 if (ret < 0) {
1837 goto error;
1838 }
1839
1840 ret = pending;
1841
1842 error:
1843 health_code_update();
1844 return ret;
1845 }
1846
1847 /*
1848 * Ask the consumer to create a directory.
1849 *
1850 * Called with the consumer socket lock held.
1851 */
1852 int consumer_mkdir(struct consumer_socket *socket, uint64_t session_id,
1853 const struct consumer_output *output, const char *path,
1854 uid_t uid, gid_t gid)
1855 {
1856 int ret;
1857 struct lttcomm_consumer_msg msg;
1858
1859 assert(socket);
1860
1861 DBG("Consumer mkdir %s in session %" PRIu64, path, session_id);
1862
1863 memset(&msg, 0, sizeof(msg));
1864 msg.cmd_type = LTTNG_CONSUMER_MKDIR;
1865 msg.u.mkdir.session_id = session_id;
1866 msg.u.mkdir.uid = uid;
1867 msg.u.mkdir.gid = gid;
1868 ret = snprintf(msg.u.mkdir.path, sizeof(msg.u.mkdir.path), "%s", path);
1869 if (ret < 0 || ret >= sizeof(msg.u.mkdir.path)) {
1870 ERR("Format path");
1871 ret = -LTTNG_ERR_INVALID;
1872 goto error;
1873 }
1874
1875 if (output->type == CONSUMER_DST_NET) {
1876 msg.u.mkdir.relayd_id = output->net_seq_index;
1877 } else {
1878 msg.u.mkdir.relayd_id = -1ULL;
1879 }
1880
1881 health_code_update();
1882 ret = consumer_send_msg(socket, &msg);
1883 if (ret < 0) {
1884 ret = -LTTNG_ERR_MKDIR_FAIL_CONSUMER;
1885 goto error;
1886 }
1887
1888 error:
1889 health_code_update();
1890 return ret;
1891 }
This page took 0.117433 seconds and 4 git commands to generate.