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