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