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