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