2 * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
3 * 2018 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
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.
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
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.
25 #include <sys/types.h>
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>
36 #include "health-sessiond.h"
39 #include "lttng-sessiond.h"
42 * Send a data payload using a given consumer socket of size len.
44 * The consumer socket lock MUST be acquired before calling this since this
45 * function can change the fd value.
47 * Return 0 on success else a negative value on error.
49 int consumer_socket_send(struct consumer_socket
*socket
, void *msg
, size_t len
)
55 assert(socket
->fd_ptr
);
58 /* Consumer socket is invalid. Stopping. */
64 size
= lttcomm_send_unix_sock(fd
, msg
, len
);
66 /* The above call will print a PERROR on error. */
67 DBG("Error when sending data to consumer on sock %d", fd
);
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.
73 /* This call will PERROR on error. */
74 (void) lttcomm_close_unix_sock(fd
);
86 * Receive a data payload using a given consumer socket of size len.
88 * The consumer socket lock MUST be acquired before calling this since this
89 * function can change the fd value.
91 * Return 0 on success else a negative value on error.
93 int consumer_socket_recv(struct consumer_socket
*socket
, void *msg
, size_t len
)
99 assert(socket
->fd_ptr
);
102 /* Consumer socket is invalid. Stopping. */
103 fd
= *socket
->fd_ptr
;
108 size
= lttcomm_recv_unix_sock(fd
, msg
, len
);
110 /* The above call will print a PERROR on error. */
111 DBG("Error when receiving data from the consumer socket %d", fd
);
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.
117 /* This call will PERROR on error. */
118 (void) lttcomm_close_unix_sock(fd
);
119 *socket
->fd_ptr
= -1;
130 * Receive a reply command status message from the consumer. Consumer socket
131 * lock MUST be acquired before calling this function.
133 * Return 0 on success, -1 on recv error or a negative lttng error code which
134 * was possibly returned by the consumer.
136 int consumer_recv_status_reply(struct consumer_socket
*sock
)
139 struct lttcomm_consumer_status_msg reply
;
143 ret
= consumer_socket_recv(sock
, &reply
, sizeof(reply
));
148 if (reply
.ret_code
== LTTCOMM_CONSUMERD_SUCCESS
) {
152 ret
= -reply
.ret_code
;
153 DBG("Consumer ret code %d", ret
);
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
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.
168 int consumer_recv_status_channel(struct consumer_socket
*sock
,
169 uint64_t *key
, unsigned int *stream_count
)
172 struct lttcomm_consumer_status_channel reply
;
175 assert(stream_count
);
178 ret
= consumer_socket_recv(sock
, &reply
, sizeof(reply
));
183 /* An error is possible so don't touch the key and stream_count. */
184 if (reply
.ret_code
!= LTTCOMM_CONSUMERD_SUCCESS
) {
190 *stream_count
= reply
.stream_count
;
198 * Send destroy relayd command to consumer.
200 * On success return positive value. On error, negative value.
202 int consumer_send_destroy_relayd(struct consumer_socket
*sock
,
203 struct consumer_output
*consumer
)
206 struct lttcomm_consumer_msg msg
;
211 DBG2("Sending destroy relayd command to consumer sock %d", *sock
->fd_ptr
);
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
;
217 pthread_mutex_lock(sock
->lock
);
218 ret
= consumer_socket_send(sock
, &msg
, sizeof(msg
));
223 /* Don't check the return value. The caller will do it. */
224 ret
= consumer_recv_status_reply(sock
);
226 DBG2("Consumer send destroy relayd command done");
229 pthread_mutex_unlock(sock
->lock
);
234 * For each consumer socket in the consumer output object, send a destroy
237 void consumer_output_send_destroy_relayd(struct consumer_output
*consumer
)
239 struct lttng_ht_iter iter
;
240 struct consumer_socket
*socket
;
244 /* Destroy any relayd connection */
245 if (consumer
->type
== CONSUMER_DST_NET
) {
247 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
251 /* Send destroy relayd command */
252 ret
= consumer_send_destroy_relayd(socket
, consumer
);
254 DBG("Unable to send destroy relayd command to consumer");
255 /* Continue since we MUST delete everything at this point. */
263 * From a consumer_data structure, allocate and add a consumer socket to the
266 * Return 0 on success, else negative value on error
268 int consumer_create_socket(struct consumer_data
*data
,
269 struct consumer_output
*output
)
272 struct consumer_socket
*socket
;
276 if (output
== NULL
|| data
->cmd_sock
< 0) {
278 * Not an error. Possible there is simply not spawned consumer or it's
279 * disabled for the tracing session asking the socket.
285 socket
= consumer_find_socket(data
->cmd_sock
, output
);
287 if (socket
== NULL
) {
288 socket
= consumer_allocate_socket(&data
->cmd_sock
);
289 if (socket
== NULL
) {
294 socket
->registered
= 0;
295 socket
->lock
= &data
->lock
;
297 consumer_add_socket(socket
, output
);
301 socket
->type
= data
->type
;
303 DBG3("Consumer socket created (fd: %d) and added to output",
311 * Return the consumer socket from the given consumer output with the right
312 * bitness. On error, returns NULL.
314 * The caller MUST acquire a rcu read side lock and keep it until the socket
315 * object reference is not needed anymore.
317 struct consumer_socket
*consumer_find_socket_by_bitness(int bits
,
318 const struct consumer_output
*consumer
)
321 struct consumer_socket
*socket
= NULL
;
325 consumer_fd
= uatomic_read(&ust_consumerd64_fd
);
328 consumer_fd
= uatomic_read(&ust_consumerd32_fd
);
335 socket
= consumer_find_socket(consumer_fd
, consumer
);
337 ERR("Consumer socket fd %d not found in consumer obj %p",
338 consumer_fd
, consumer
);
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.
350 struct consumer_socket
*consumer_find_socket(int key
,
351 const struct consumer_output
*consumer
)
353 struct lttng_ht_iter iter
;
354 struct lttng_ht_node_ulong
*node
;
355 struct consumer_socket
*socket
= NULL
;
357 /* Negative keys are lookup failures */
358 if (key
< 0 || consumer
== NULL
) {
362 lttng_ht_lookup(consumer
->socks
, (void *)((unsigned long) key
),
364 node
= lttng_ht_iter_get_node_ulong(&iter
);
366 socket
= caa_container_of(node
, struct consumer_socket
, node
);
373 * Allocate a new consumer_socket and return the pointer.
375 struct consumer_socket
*consumer_allocate_socket(int *fd
)
377 struct consumer_socket
*socket
= NULL
;
381 socket
= zmalloc(sizeof(struct consumer_socket
));
382 if (socket
== NULL
) {
383 PERROR("zmalloc consumer socket");
388 lttng_ht_node_init_ulong(&socket
->node
, *fd
);
395 * Add consumer socket to consumer output object. Read side lock must be
396 * acquired before calling this function.
398 void consumer_add_socket(struct consumer_socket
*sock
,
399 struct consumer_output
*consumer
)
404 lttng_ht_add_unique_ulong(consumer
->socks
, &sock
->node
);
408 * Delete consumer socket to consumer output object. Read side lock must be
409 * acquired before calling this function.
411 void consumer_del_socket(struct consumer_socket
*sock
,
412 struct consumer_output
*consumer
)
415 struct lttng_ht_iter iter
;
420 iter
.iter
.node
= &sock
->node
.node
;
421 ret
= lttng_ht_del(consumer
->socks
, &iter
);
426 * RCU destroy call function.
428 static void destroy_socket_rcu(struct rcu_head
*head
)
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
);
439 * Destroy and free socket pointer in a call RCU. Read side lock must be
440 * acquired before calling this function.
442 void consumer_destroy_socket(struct consumer_socket
*sock
)
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,
451 if (sock
->registered
) {
452 DBG3("Consumer socket was registered. Closing fd %d", *sock
->fd_ptr
);
453 lttcomm_close_unix_sock(*sock
->fd_ptr
);
456 call_rcu(&sock
->node
.head
, destroy_socket_rcu
);
460 * Allocate and assign data to a consumer_output object.
462 * Return pointer to structure.
464 struct consumer_output
*consumer_create_output(enum consumer_dst_type type
)
466 struct consumer_output
*output
= NULL
;
468 output
= zmalloc(sizeof(struct consumer_output
));
469 if (output
== NULL
) {
470 PERROR("zmalloc consumer_output");
474 /* By default, consumer output is enabled */
477 output
->net_seq_index
= (uint64_t) -1ULL;
478 urcu_ref_init(&output
->ref
);
480 output
->socks
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
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.
491 void consumer_destroy_output_sockets(struct consumer_output
*obj
)
493 struct lttng_ht_iter iter
;
494 struct consumer_socket
*socket
;
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
);
509 * Delete the consumer_output object from the list and free the ptr.
511 * Should *NOT* be called with RCU read-side lock held.
513 static void consumer_release_output(struct urcu_ref
*ref
)
515 struct consumer_output
*obj
=
516 caa_container_of(ref
, struct consumer_output
, ref
);
518 consumer_destroy_output_sockets(obj
);
521 /* Finally destroy HT */
522 ht_cleanup_push(obj
->socks
);
529 * Get the consumer_output object.
531 void consumer_output_get(struct consumer_output
*obj
)
533 urcu_ref_get(&obj
->ref
);
537 * Put the consumer_output object.
539 * Should *NOT* be called with RCU read-side lock held.
541 void consumer_output_put(struct consumer_output
*obj
)
546 urcu_ref_put(&obj
->ref
, consumer_release_output
);
550 * Copy consumer output and returned the newly allocated copy.
552 * Should *NOT* be called with RCU read-side lock held.
554 struct consumer_output
*consumer_copy_output(struct consumer_output
*src
)
557 struct consumer_output
*output
;
561 output
= consumer_create_output(src
->type
);
562 if (output
== NULL
) {
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
);
581 consumer_output_put(output
);
586 * Copy consumer sockets from src to dst.
588 * Return 0 on success or else a negative value.
590 int consumer_copy_sockets(struct consumer_output
*dst
,
591 struct consumer_output
*src
)
594 struct lttng_ht_iter iter
;
595 struct consumer_socket
*socket
, *copy_sock
;
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
);
608 /* Create new socket object. */
609 copy_sock
= consumer_allocate_socket(socket
->fd_ptr
);
610 if (copy_sock
== NULL
) {
616 copy_sock
->registered
= socket
->registered
;
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
622 copy_sock
->lock
= socket
->lock
;
623 consumer_add_socket(copy_sock
, dst
);
632 * Set network URI to the consumer output.
634 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
637 int consumer_set_network_uri(const struct ltt_session
*session
,
638 struct consumer_output
*output
,
639 struct lttng_uri
*uri
)
642 struct lttng_uri
*dst_uri
= NULL
;
644 /* Code flow error safety net. */
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
;
656 if (output
->dst
.net
.data_isset
&& uri
->port
==
657 output
->dst
.net
.data
.port
) {
658 ret
= -LTTNG_ERR_INVALID
;
662 DBG3("Consumer control URI set with port %d", uri
->port
);
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
;
671 if (output
->dst
.net
.control_isset
&& uri
->port
==
672 output
->dst
.net
.control
.port
) {
673 ret
= -LTTNG_ERR_INVALID
;
677 DBG3("Consumer data URI set with port %d", uri
->port
);
680 ERR("Set network uri type unknown %d", uri
->stype
);
681 ret
= -LTTNG_ERR_INVALID
;
685 ret
= uri_compare(dst_uri
, uri
);
687 /* Same URI, don't touch it and return success. */
688 DBG3("URI network compare are the same");
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. */
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
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
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
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
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.
723 if (strstr(uri
->subdir
, "../")) {
724 ERR("Network URI subdirs are not allowed to walk up the path hierarchy");
725 ret
= -LTTNG_ERR_INVALID
;
728 ret
= snprintf(output
->dst
.net
.base_dir
,
729 sizeof(output
->dst
.net
.base_dir
),
730 "/%s/%s/", session
->hostname
, uri
->subdir
);
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
,
738 char session_creation_datetime
[16];
742 timeinfo
= localtime(&session
->creation_time
);
744 ret
= -LTTNG_ERR_FATAL
;
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
;
755 ret
= snprintf(output
->dst
.net
.base_dir
,
756 sizeof(output
->dst
.net
.base_dir
),
757 "/%s/%s-%s/", session
->hostname
,
759 session_creation_datetime
);
762 if (ret
>= sizeof(output
->dst
.net
.base_dir
)) {
763 ret
= -LTTNG_ERR_INVALID
;
764 ERR("Truncation occurred while setting network output base directory");
766 } else if (ret
== -1) {
767 ret
= -LTTNG_ERR_INVALID
;
768 PERROR("Error occurred while setting network output base directory");
772 DBG3("Consumer set network uri base_dir path %s",
773 output
->dst
.net
.base_dir
);
784 * Send file descriptor to consumer via sock.
786 * The consumer socket lock must be held by the caller.
788 int consumer_send_fds(struct consumer_socket
*sock
, const int *fds
,
796 assert(pthread_mutex_trylock(sock
->lock
) == EBUSY
);
798 ret
= lttcomm_send_fds_unix_sock(*sock
->fd_ptr
, fds
, nb_fd
);
800 /* The above call will print a PERROR on error. */
801 DBG("Error when sending consumer fds on sock %d", *sock
->fd_ptr
);
805 ret
= consumer_recv_status_reply(sock
);
811 * Consumer send communication message structure to consumer.
813 * The consumer socket lock must be held by the caller.
815 int consumer_send_msg(struct consumer_socket
*sock
,
816 struct lttcomm_consumer_msg
*msg
)
822 assert(pthread_mutex_trylock(sock
->lock
) == EBUSY
);
824 ret
= consumer_socket_send(sock
, msg
, sizeof(struct lttcomm_consumer_msg
));
829 ret
= consumer_recv_status_reply(sock
);
836 * Consumer send channel communication message structure to consumer.
838 * The consumer socket lock must be held by the caller.
840 int consumer_send_channel(struct consumer_socket
*sock
,
841 struct lttcomm_consumer_msg
*msg
)
848 ret
= consumer_send_msg(sock
, msg
);
858 * Populate the given consumer msg structure with the ask_channel command
861 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg
*msg
,
862 uint64_t subbuf_size
,
865 unsigned int switch_timer_interval
,
866 unsigned int read_timer_interval
,
867 unsigned int live_timer_interval
,
868 unsigned int monitor_timer_interval
,
872 const char *pathname
,
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 const struct lttng_credentials
*buffer_credentials
)
891 /* Zeroed structure */
892 memset(msg
, 0, sizeof(struct lttcomm_consumer_msg
));
893 msg
->u
.ask_channel
.buffer_credentials
.uid
= UINT32_MAX
;
894 msg
->u
.ask_channel
.buffer_credentials
.gid
= UINT32_MAX
;
898 enum lttng_trace_chunk_status chunk_status
;
900 chunk_status
= lttng_trace_chunk_get_id(trace_chunk
, &chunk_id
);
901 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
902 LTTNG_OPTIONAL_SET(&msg
->u
.ask_channel
.chunk_id
, chunk_id
);
904 msg
->u
.ask_channel
.buffer_credentials
.uid
= buffer_credentials
->uid
;
905 msg
->u
.ask_channel
.buffer_credentials
.gid
= buffer_credentials
->gid
;
907 msg
->cmd_type
= LTTNG_CONSUMER_ASK_CHANNEL_CREATION
;
908 msg
->u
.ask_channel
.subbuf_size
= subbuf_size
;
909 msg
->u
.ask_channel
.num_subbuf
= num_subbuf
;
910 msg
->u
.ask_channel
.overwrite
= overwrite
;
911 msg
->u
.ask_channel
.switch_timer_interval
= switch_timer_interval
;
912 msg
->u
.ask_channel
.read_timer_interval
= read_timer_interval
;
913 msg
->u
.ask_channel
.live_timer_interval
= live_timer_interval
;
914 msg
->u
.ask_channel
.monitor_timer_interval
= monitor_timer_interval
;
915 msg
->u
.ask_channel
.output
= output
;
916 msg
->u
.ask_channel
.type
= type
;
917 msg
->u
.ask_channel
.session_id
= session_id
;
918 msg
->u
.ask_channel
.session_id_per_pid
= session_id_per_pid
;
919 msg
->u
.ask_channel
.relayd_id
= relayd_id
;
920 msg
->u
.ask_channel
.key
= key
;
921 msg
->u
.ask_channel
.chan_id
= chan_id
;
922 msg
->u
.ask_channel
.tracefile_size
= tracefile_size
;
923 msg
->u
.ask_channel
.tracefile_count
= tracefile_count
;
924 msg
->u
.ask_channel
.monitor
= monitor
;
925 msg
->u
.ask_channel
.ust_app_uid
= ust_app_uid
;
926 msg
->u
.ask_channel
.blocking_timeout
= blocking_timeout
;
928 memcpy(msg
->u
.ask_channel
.uuid
, uuid
, sizeof(msg
->u
.ask_channel
.uuid
));
931 strncpy(msg
->u
.ask_channel
.pathname
, pathname
,
932 sizeof(msg
->u
.ask_channel
.pathname
));
933 msg
->u
.ask_channel
.pathname
[sizeof(msg
->u
.ask_channel
.pathname
)-1] = '\0';
936 strncpy(msg
->u
.ask_channel
.name
, name
, sizeof(msg
->u
.ask_channel
.name
));
937 msg
->u
.ask_channel
.name
[sizeof(msg
->u
.ask_channel
.name
) - 1] = '\0';
940 strncpy(msg
->u
.ask_channel
.root_shm_path
, root_shm_path
,
941 sizeof(msg
->u
.ask_channel
.root_shm_path
));
942 msg
->u
.ask_channel
.root_shm_path
[sizeof(msg
->u
.ask_channel
.root_shm_path
) - 1] = '\0';
945 strncpy(msg
->u
.ask_channel
.shm_path
, shm_path
,
946 sizeof(msg
->u
.ask_channel
.shm_path
));
947 msg
->u
.ask_channel
.shm_path
[sizeof(msg
->u
.ask_channel
.shm_path
) - 1] = '\0';
952 * Init channel communication message structure.
954 void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg
*msg
,
955 uint64_t channel_key
,
957 const char *pathname
,
962 unsigned int nb_init_streams
,
963 enum lttng_event_output output
,
965 uint64_t tracefile_size
,
966 uint64_t tracefile_count
,
967 unsigned int monitor
,
968 unsigned int live_timer_interval
,
969 unsigned int monitor_timer_interval
,
970 struct lttng_trace_chunk
*trace_chunk
)
974 /* Zeroed structure */
975 memset(msg
, 0, sizeof(struct lttcomm_consumer_msg
));
979 enum lttng_trace_chunk_status chunk_status
;
981 chunk_status
= lttng_trace_chunk_get_id(trace_chunk
, &chunk_id
);
982 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
983 LTTNG_OPTIONAL_SET(&msg
->u
.channel
.chunk_id
, chunk_id
);
987 msg
->cmd_type
= LTTNG_CONSUMER_ADD_CHANNEL
;
988 msg
->u
.channel
.channel_key
= channel_key
;
989 msg
->u
.channel
.session_id
= session_id
;
990 msg
->u
.channel
.relayd_id
= relayd_id
;
991 msg
->u
.channel
.nb_init_streams
= nb_init_streams
;
992 msg
->u
.channel
.output
= output
;
993 msg
->u
.channel
.type
= type
;
994 msg
->u
.channel
.tracefile_size
= tracefile_size
;
995 msg
->u
.channel
.tracefile_count
= tracefile_count
;
996 msg
->u
.channel
.monitor
= monitor
;
997 msg
->u
.channel
.live_timer_interval
= live_timer_interval
;
998 msg
->u
.channel
.monitor_timer_interval
= monitor_timer_interval
;
1000 strncpy(msg
->u
.channel
.pathname
, pathname
,
1001 sizeof(msg
->u
.channel
.pathname
));
1002 msg
->u
.channel
.pathname
[sizeof(msg
->u
.channel
.pathname
) - 1] = '\0';
1004 strncpy(msg
->u
.channel
.name
, name
, sizeof(msg
->u
.channel
.name
));
1005 msg
->u
.channel
.name
[sizeof(msg
->u
.channel
.name
) - 1] = '\0';
1009 * Init stream communication message structure.
1011 void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg
*msg
,
1012 uint64_t channel_key
,
1013 uint64_t stream_key
,
1018 memset(msg
, 0, sizeof(struct lttcomm_consumer_msg
));
1020 msg
->cmd_type
= LTTNG_CONSUMER_ADD_STREAM
;
1021 msg
->u
.stream
.channel_key
= channel_key
;
1022 msg
->u
.stream
.stream_key
= stream_key
;
1023 msg
->u
.stream
.cpu
= cpu
;
1026 void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg
*msg
,
1027 enum lttng_consumer_command cmd
,
1028 uint64_t channel_key
, uint64_t net_seq_idx
)
1032 memset(msg
, 0, sizeof(struct lttcomm_consumer_msg
));
1034 msg
->cmd_type
= cmd
;
1035 msg
->u
.sent_streams
.channel_key
= channel_key
;
1036 msg
->u
.sent_streams
.net_seq_idx
= net_seq_idx
;
1040 * Send stream communication structure to the consumer.
1042 int consumer_send_stream(struct consumer_socket
*sock
,
1043 struct consumer_output
*dst
, struct lttcomm_consumer_msg
*msg
,
1044 const int *fds
, size_t nb_fd
)
1053 ret
= consumer_send_msg(sock
, msg
);
1058 ret
= consumer_send_fds(sock
, fds
, nb_fd
);
1068 * Send relayd socket to consumer associated with a session name.
1070 * The consumer socket lock must be held by the caller.
1072 * On success return positive value. On error, negative value.
1074 int consumer_send_relayd_socket(struct consumer_socket
*consumer_sock
,
1075 struct lttcomm_relayd_sock
*rsock
, struct consumer_output
*consumer
,
1076 enum lttng_stream_type type
, uint64_t session_id
,
1077 const char *session_name
, const char *hostname
,
1078 int session_live_timer
, const uint64_t *current_chunk_id
,
1079 time_t session_creation_time
)
1082 struct lttcomm_consumer_msg msg
;
1084 /* Code flow error. Safety net. */
1087 assert(consumer_sock
);
1089 memset(&msg
, 0, sizeof(msg
));
1090 /* Bail out if consumer is disabled */
1091 if (!consumer
->enabled
) {
1096 if (type
== LTTNG_STREAM_CONTROL
) {
1097 ret
= relayd_create_session(rsock
,
1098 &msg
.u
.relayd_sock
.relayd_session_id
,
1099 session_name
, hostname
, session_live_timer
,
1100 consumer
->snapshot
, session_id
,
1101 sessiond_uuid
, current_chunk_id
,
1102 session_creation_time
);
1104 /* Close the control socket. */
1105 (void) relayd_close(rsock
);
1110 msg
.cmd_type
= LTTNG_CONSUMER_ADD_RELAYD_SOCKET
;
1112 * Assign network consumer output index using the temporary consumer since
1113 * this call should only be made from within a set_consumer_uri() function
1114 * call in the session daemon.
1116 msg
.u
.relayd_sock
.net_index
= consumer
->net_seq_index
;
1117 msg
.u
.relayd_sock
.type
= type
;
1118 msg
.u
.relayd_sock
.session_id
= session_id
;
1119 memcpy(&msg
.u
.relayd_sock
.sock
, rsock
, sizeof(msg
.u
.relayd_sock
.sock
));
1121 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock
->fd_ptr
);
1122 ret
= consumer_send_msg(consumer_sock
, &msg
);
1127 DBG3("Sending relayd socket file descriptor to consumer");
1128 ret
= consumer_send_fds(consumer_sock
, &rsock
->sock
.fd
, 1);
1133 DBG2("Consumer relayd socket sent");
1140 int consumer_send_pipe(struct consumer_socket
*consumer_sock
,
1141 enum lttng_consumer_command cmd
, int pipe
)
1144 struct lttcomm_consumer_msg msg
;
1145 const char *pipe_name
;
1146 const char *command_name
;
1149 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE
:
1150 pipe_name
= "channel monitor";
1151 command_name
= "SET_CHANNEL_MONITOR_PIPE";
1154 ERR("Unexpected command received in %s (cmd = %d)", __func__
,
1159 /* Code flow error. Safety net. */
1161 memset(&msg
, 0, sizeof(msg
));
1164 pthread_mutex_lock(consumer_sock
->lock
);
1165 DBG3("Sending %s command to consumer", command_name
);
1166 ret
= consumer_send_msg(consumer_sock
, &msg
);
1171 DBG3("Sending %s pipe %d to consumer on socket %d",
1173 pipe
, *consumer_sock
->fd_ptr
);
1174 ret
= consumer_send_fds(consumer_sock
, &pipe
, 1);
1179 DBG2("%s pipe successfully sent", pipe_name
);
1181 pthread_mutex_unlock(consumer_sock
->lock
);
1185 int consumer_send_channel_monitor_pipe(struct consumer_socket
*consumer_sock
,
1188 return consumer_send_pipe(consumer_sock
,
1189 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE
, pipe
);
1193 * Ask the consumer if the data is pending for the specific session id.
1194 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1196 int consumer_is_data_pending(uint64_t session_id
,
1197 struct consumer_output
*consumer
)
1200 int32_t ret_code
= 0; /* Default is that the data is NOT pending */
1201 struct consumer_socket
*socket
;
1202 struct lttng_ht_iter iter
;
1203 struct lttcomm_consumer_msg msg
;
1207 DBG3("Consumer data pending for id %" PRIu64
, session_id
);
1209 memset(&msg
, 0, sizeof(msg
));
1210 msg
.cmd_type
= LTTNG_CONSUMER_DATA_PENDING
;
1211 msg
.u
.data_pending
.session_id
= session_id
;
1213 /* Send command for each consumer */
1215 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1217 pthread_mutex_lock(socket
->lock
);
1218 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1220 pthread_mutex_unlock(socket
->lock
);
1225 * No need for a recv reply status because the answer to the command is
1226 * the reply status message.
1229 ret
= consumer_socket_recv(socket
, &ret_code
, sizeof(ret_code
));
1231 pthread_mutex_unlock(socket
->lock
);
1234 pthread_mutex_unlock(socket
->lock
);
1236 if (ret_code
== 1) {
1242 DBG("Consumer data is %s pending for session id %" PRIu64
,
1243 ret_code
== 1 ? "" : "NOT", session_id
);
1252 * Send a flush command to consumer using the given channel key.
1254 * Return 0 on success else a negative value.
1256 int consumer_flush_channel(struct consumer_socket
*socket
, uint64_t key
)
1259 struct lttcomm_consumer_msg msg
;
1263 DBG2("Consumer flush channel key %" PRIu64
, key
);
1265 memset(&msg
, 0, sizeof(msg
));
1266 msg
.cmd_type
= LTTNG_CONSUMER_FLUSH_CHANNEL
;
1267 msg
.u
.flush_channel
.key
= key
;
1269 pthread_mutex_lock(socket
->lock
);
1270 health_code_update();
1272 ret
= consumer_send_msg(socket
, &msg
);
1278 health_code_update();
1279 pthread_mutex_unlock(socket
->lock
);
1284 * Send a clear quiescent command to consumer using the given channel key.
1286 * Return 0 on success else a negative value.
1288 int consumer_clear_quiescent_channel(struct consumer_socket
*socket
, uint64_t key
)
1291 struct lttcomm_consumer_msg msg
;
1295 DBG2("Consumer clear quiescent channel key %" PRIu64
, key
);
1297 memset(&msg
, 0, sizeof(msg
));
1298 msg
.cmd_type
= LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL
;
1299 msg
.u
.clear_quiescent_channel
.key
= key
;
1301 pthread_mutex_lock(socket
->lock
);
1302 health_code_update();
1304 ret
= consumer_send_msg(socket
, &msg
);
1310 health_code_update();
1311 pthread_mutex_unlock(socket
->lock
);
1316 * Send a close metadata command to consumer using the given channel key.
1317 * Called with registry lock held.
1319 * Return 0 on success else a negative value.
1321 int consumer_close_metadata(struct consumer_socket
*socket
,
1322 uint64_t metadata_key
)
1325 struct lttcomm_consumer_msg msg
;
1329 DBG2("Consumer close metadata channel key %" PRIu64
, metadata_key
);
1331 memset(&msg
, 0, sizeof(msg
));
1332 msg
.cmd_type
= LTTNG_CONSUMER_CLOSE_METADATA
;
1333 msg
.u
.close_metadata
.key
= metadata_key
;
1335 pthread_mutex_lock(socket
->lock
);
1336 health_code_update();
1338 ret
= consumer_send_msg(socket
, &msg
);
1344 health_code_update();
1345 pthread_mutex_unlock(socket
->lock
);
1350 * Send a setup metdata command to consumer using the given channel key.
1352 * Return 0 on success else a negative value.
1354 int consumer_setup_metadata(struct consumer_socket
*socket
,
1355 uint64_t metadata_key
)
1358 struct lttcomm_consumer_msg msg
;
1362 DBG2("Consumer setup metadata channel key %" PRIu64
, metadata_key
);
1364 memset(&msg
, 0, sizeof(msg
));
1365 msg
.cmd_type
= LTTNG_CONSUMER_SETUP_METADATA
;
1366 msg
.u
.setup_metadata
.key
= metadata_key
;
1368 pthread_mutex_lock(socket
->lock
);
1369 health_code_update();
1371 ret
= consumer_send_msg(socket
, &msg
);
1377 health_code_update();
1378 pthread_mutex_unlock(socket
->lock
);
1383 * Send metadata string to consumer.
1384 * RCU read-side lock must be held to guarantee existence of socket.
1386 * Return 0 on success else a negative value.
1388 int consumer_push_metadata(struct consumer_socket
*socket
,
1389 uint64_t metadata_key
, char *metadata_str
, size_t len
,
1390 size_t target_offset
, uint64_t version
)
1393 struct lttcomm_consumer_msg msg
;
1397 DBG2("Consumer push metadata to consumer socket %d", *socket
->fd_ptr
);
1399 pthread_mutex_lock(socket
->lock
);
1401 memset(&msg
, 0, sizeof(msg
));
1402 msg
.cmd_type
= LTTNG_CONSUMER_PUSH_METADATA
;
1403 msg
.u
.push_metadata
.key
= metadata_key
;
1404 msg
.u
.push_metadata
.target_offset
= target_offset
;
1405 msg
.u
.push_metadata
.len
= len
;
1406 msg
.u
.push_metadata
.version
= version
;
1408 health_code_update();
1409 ret
= consumer_send_msg(socket
, &msg
);
1410 if (ret
< 0 || len
== 0) {
1414 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket
->fd_ptr
,
1417 ret
= consumer_socket_send(socket
, metadata_str
, len
);
1422 health_code_update();
1423 ret
= consumer_recv_status_reply(socket
);
1429 pthread_mutex_unlock(socket
->lock
);
1430 health_code_update();
1435 * Ask the consumer to snapshot a specific channel using the key.
1437 * Returns LTTNG_OK on success or else an LTTng error code.
1439 enum lttng_error_code
consumer_snapshot_channel(struct consumer_socket
*socket
,
1440 uint64_t key
, const struct consumer_output
*output
, int metadata
,
1441 uid_t uid
, gid_t gid
, const char *channel_path
, int wait
,
1442 uint64_t nb_packets_per_stream
)
1445 enum lttng_error_code status
= LTTNG_OK
;
1446 struct lttcomm_consumer_msg msg
;
1451 DBG("Consumer snapshot channel key %" PRIu64
, key
);
1453 memset(&msg
, 0, sizeof(msg
));
1454 msg
.cmd_type
= LTTNG_CONSUMER_SNAPSHOT_CHANNEL
;
1455 msg
.u
.snapshot_channel
.key
= key
;
1456 msg
.u
.snapshot_channel
.nb_packets_per_stream
= nb_packets_per_stream
;
1457 msg
.u
.snapshot_channel
.metadata
= metadata
;
1459 if (output
->type
== CONSUMER_DST_NET
) {
1460 msg
.u
.snapshot_channel
.relayd_id
=
1461 output
->net_seq_index
;
1462 msg
.u
.snapshot_channel
.use_relayd
= 1;
1464 msg
.u
.snapshot_channel
.relayd_id
= (uint64_t) -1ULL;
1466 ret
= lttng_strncpy(msg
.u
.snapshot_channel
.pathname
,
1468 sizeof(msg
.u
.snapshot_channel
.pathname
));
1470 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1471 sizeof(msg
.u
.snapshot_channel
.pathname
),
1472 strlen(channel_path
),
1474 status
= LTTNG_ERR_SNAPSHOT_FAIL
;
1478 health_code_update();
1479 pthread_mutex_lock(socket
->lock
);
1480 ret
= consumer_send_msg(socket
, &msg
);
1481 pthread_mutex_unlock(socket
->lock
);
1484 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND
:
1485 status
= LTTNG_ERR_CHAN_NOT_FOUND
;
1488 status
= LTTNG_ERR_SNAPSHOT_FAIL
;
1495 health_code_update();
1500 * Ask the consumer the number of discarded events for a channel.
1502 int consumer_get_discarded_events(uint64_t session_id
, uint64_t channel_key
,
1503 struct consumer_output
*consumer
, uint64_t *discarded
)
1506 struct consumer_socket
*socket
;
1507 struct lttng_ht_iter iter
;
1508 struct lttcomm_consumer_msg msg
;
1512 DBG3("Consumer discarded events id %" PRIu64
, session_id
);
1514 memset(&msg
, 0, sizeof(msg
));
1515 msg
.cmd_type
= LTTNG_CONSUMER_DISCARDED_EVENTS
;
1516 msg
.u
.discarded_events
.session_id
= session_id
;
1517 msg
.u
.discarded_events
.channel_key
= channel_key
;
1521 /* Send command for each consumer */
1523 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1525 uint64_t consumer_discarded
= 0;
1526 pthread_mutex_lock(socket
->lock
);
1527 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1529 pthread_mutex_unlock(socket
->lock
);
1534 * No need for a recv reply status because the answer to the
1535 * command is the reply status message.
1537 ret
= consumer_socket_recv(socket
, &consumer_discarded
,
1538 sizeof(consumer_discarded
));
1540 ERR("get discarded events");
1541 pthread_mutex_unlock(socket
->lock
);
1544 pthread_mutex_unlock(socket
->lock
);
1545 *discarded
+= consumer_discarded
;
1548 DBG("Consumer discarded %" PRIu64
" events in session id %" PRIu64
,
1549 *discarded
, session_id
);
1557 * Ask the consumer the number of lost packets for a channel.
1559 int consumer_get_lost_packets(uint64_t session_id
, uint64_t channel_key
,
1560 struct consumer_output
*consumer
, uint64_t *lost
)
1563 struct consumer_socket
*socket
;
1564 struct lttng_ht_iter iter
;
1565 struct lttcomm_consumer_msg msg
;
1569 DBG3("Consumer lost packets id %" PRIu64
, session_id
);
1571 memset(&msg
, 0, sizeof(msg
));
1572 msg
.cmd_type
= LTTNG_CONSUMER_LOST_PACKETS
;
1573 msg
.u
.lost_packets
.session_id
= session_id
;
1574 msg
.u
.lost_packets
.channel_key
= channel_key
;
1578 /* Send command for each consumer */
1580 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1582 uint64_t consumer_lost
= 0;
1583 pthread_mutex_lock(socket
->lock
);
1584 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1586 pthread_mutex_unlock(socket
->lock
);
1591 * No need for a recv reply status because the answer to the
1592 * command is the reply status message.
1594 ret
= consumer_socket_recv(socket
, &consumer_lost
,
1595 sizeof(consumer_lost
));
1597 ERR("get lost packets");
1598 pthread_mutex_unlock(socket
->lock
);
1601 pthread_mutex_unlock(socket
->lock
);
1602 *lost
+= consumer_lost
;
1605 DBG("Consumer lost %" PRIu64
" packets in session id %" PRIu64
,
1614 * Ask the consumer to rotate a channel.
1616 * The new_chunk_id is the session->rotate_count that has been incremented
1617 * when the rotation started. On the relay, this allows to keep track in which
1618 * chunk each stream is currently writing to (for the rotate_pending operation).
1620 int consumer_rotate_channel(struct consumer_socket
*socket
, uint64_t key
,
1621 uid_t uid
, gid_t gid
, struct consumer_output
*output
,
1622 bool is_metadata_channel
)
1625 struct lttcomm_consumer_msg msg
;
1629 DBG("Consumer rotate channel key %" PRIu64
, key
);
1631 pthread_mutex_lock(socket
->lock
);
1632 memset(&msg
, 0, sizeof(msg
));
1633 msg
.cmd_type
= LTTNG_CONSUMER_ROTATE_CHANNEL
;
1634 msg
.u
.rotate_channel
.key
= key
;
1635 msg
.u
.rotate_channel
.metadata
= !!is_metadata_channel
;
1637 if (output
->type
== CONSUMER_DST_NET
) {
1638 msg
.u
.rotate_channel
.relayd_id
= output
->net_seq_index
;
1640 msg
.u
.rotate_channel
.relayd_id
= (uint64_t) -1ULL;
1643 health_code_update();
1644 ret
= consumer_send_msg(socket
, &msg
);
1647 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND
:
1648 ret
= -LTTNG_ERR_CHAN_NOT_FOUND
;
1651 ret
= -LTTNG_ERR_ROTATION_FAIL_CONSUMER
;
1657 pthread_mutex_unlock(socket
->lock
);
1658 health_code_update();
1662 int consumer_init(struct consumer_socket
*socket
,
1663 const lttng_uuid sessiond_uuid
)
1666 struct lttcomm_consumer_msg msg
= {
1667 .cmd_type
= LTTNG_CONSUMER_INIT
,
1672 DBG("Sending consumer initialization command");
1673 lttng_uuid_copy(msg
.u
.init
.sessiond_uuid
, sessiond_uuid
);
1675 health_code_update();
1676 ret
= consumer_send_msg(socket
, &msg
);
1682 health_code_update();
1687 * Ask the consumer to create a new chunk for a given session.
1689 * Called with the consumer socket lock held.
1691 int consumer_create_trace_chunk(struct consumer_socket
*socket
,
1692 uint64_t relayd_id
, uint64_t session_id
,
1693 struct lttng_trace_chunk
*chunk
)
1696 enum lttng_trace_chunk_status chunk_status
;
1697 struct lttng_credentials chunk_credentials
;
1698 const struct lttng_directory_handle
*chunk_directory_handle
;
1700 const char *chunk_name
;
1701 bool chunk_name_overriden
;
1703 time_t creation_timestamp
;
1704 char creation_timestamp_buffer
[ISO8601_STR_LEN
];
1705 const char *creation_timestamp_str
= "(none)";
1706 const bool chunk_has_local_output
= relayd_id
== -1ULL;
1707 struct lttcomm_consumer_msg msg
= {
1708 .cmd_type
= LTTNG_CONSUMER_CREATE_TRACE_CHUNK
,
1709 .u
.create_trace_chunk
.session_id
= session_id
,
1715 if (relayd_id
!= -1ULL) {
1716 LTTNG_OPTIONAL_SET(&msg
.u
.create_trace_chunk
.relayd_id
,
1720 chunk_status
= lttng_trace_chunk_get_name(chunk
, &chunk_name
,
1721 &chunk_name_overriden
);
1722 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
&&
1723 chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_NONE
) {
1724 ERR("Failed to get name of trace chunk");
1725 ret
= -LTTNG_ERR_FATAL
;
1728 if (chunk_name_overriden
) {
1729 ret
= lttng_strncpy(msg
.u
.create_trace_chunk
.override_name
,
1731 sizeof(msg
.u
.create_trace_chunk
.override_name
));
1733 ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1735 ret
= -LTTNG_ERR_FATAL
;
1740 chunk_status
= lttng_trace_chunk_get_creation_timestamp(chunk
,
1741 &creation_timestamp
);
1742 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1743 ret
= -LTTNG_ERR_FATAL
;
1746 msg
.u
.create_trace_chunk
.creation_timestamp
=
1747 (uint64_t) creation_timestamp
;
1748 /* Only used for logging purposes. */
1749 ret
= time_to_iso8601_str(creation_timestamp
,
1750 creation_timestamp_buffer
,
1751 sizeof(creation_timestamp_buffer
));
1752 creation_timestamp_str
= !ret
? creation_timestamp_buffer
:
1753 "(formatting error)";
1755 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1756 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1758 * Anonymous trace chunks should never be transmitted
1759 * to remote peers (consumerd and relayd). They are used
1760 * internally for backward-compatibility purposes.
1762 ret
= -LTTNG_ERR_FATAL
;
1765 msg
.u
.create_trace_chunk
.chunk_id
= chunk_id
;
1767 if (chunk_has_local_output
) {
1768 chunk_status
= lttng_trace_chunk_get_chunk_directory_handle(
1769 chunk
, &chunk_directory_handle
);
1770 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1771 ret
= -LTTNG_ERR_FATAL
;
1776 * This will only compile on platforms that support
1777 * dirfd (POSIX.2008). This is fine as the session daemon
1778 * is only built for such platforms.
1780 * The ownership of the chunk directory handle's is maintained
1781 * by the trace chunk.
1783 chunk_dirfd
= lttng_directory_handle_get_dirfd(
1784 chunk_directory_handle
);
1785 assert(chunk_dirfd
>= 0);
1787 chunk_status
= lttng_trace_chunk_get_credentials(
1788 chunk
, &chunk_credentials
);
1789 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1791 * Not associating credentials to a sessiond chunk is a
1792 * fatal internal error.
1794 ret
= -LTTNG_ERR_FATAL
;
1797 msg
.u
.create_trace_chunk
.credentials
.value
.uid
=
1798 chunk_credentials
.uid
;
1799 msg
.u
.create_trace_chunk
.credentials
.value
.gid
=
1800 chunk_credentials
.gid
;
1801 msg
.u
.create_trace_chunk
.credentials
.is_set
= 1;
1804 DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1805 ", session_id = %" PRIu64
", chunk_id = %" PRIu64
1806 ", creation_timestamp = %s",
1807 relayd_id
, session_id
, chunk_id
,
1808 creation_timestamp_str
);
1809 health_code_update();
1810 ret
= consumer_send_msg(socket
, &msg
);
1811 health_code_update();
1813 ERR("Trace chunk creation error on consumer");
1814 ret
= -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER
;
1818 if (chunk_has_local_output
) {
1819 DBG("Sending trace chunk directory fd to consumer");
1820 health_code_update();
1821 ret
= consumer_send_fds(socket
, &chunk_dirfd
, 1);
1822 health_code_update();
1824 ERR("Trace chunk creation error on consumer");
1825 ret
= -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER
;
1834 * Ask the consumer to close a trace chunk for a given session.
1836 * Called with the consumer socket lock held.
1838 int consumer_close_trace_chunk(struct consumer_socket
*socket
,
1839 uint64_t relayd_id
, uint64_t session_id
,
1840 struct lttng_trace_chunk
*chunk
)
1843 enum lttng_trace_chunk_status chunk_status
;
1844 struct lttcomm_consumer_msg msg
= {
1845 .cmd_type
= LTTNG_CONSUMER_CLOSE_TRACE_CHUNK
,
1846 .u
.close_trace_chunk
.session_id
= session_id
,
1849 time_t close_timestamp
;
1850 enum lttng_trace_chunk_command_type close_command
;
1851 const char *close_command_name
= "none";
1855 if (relayd_id
!= -1ULL) {
1857 &msg
.u
.close_trace_chunk
.relayd_id
, relayd_id
);
1860 chunk_status
= lttng_trace_chunk_get_close_command(
1861 chunk
, &close_command
);
1862 switch (chunk_status
) {
1863 case LTTNG_TRACE_CHUNK_STATUS_OK
:
1864 LTTNG_OPTIONAL_SET(&msg
.u
.close_trace_chunk
.close_command
,
1865 (uint32_t) close_command
);
1867 case LTTNG_TRACE_CHUNK_STATUS_NONE
:
1870 ERR("Failed to get trace chunk close command");
1875 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1877 * Anonymous trace chunks should never be transmitted to remote peers
1878 * (consumerd and relayd). They are used internally for
1879 * backward-compatibility purposes.
1881 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
1882 msg
.u
.close_trace_chunk
.chunk_id
= chunk_id
;
1884 chunk_status
= lttng_trace_chunk_get_close_timestamp(chunk
,
1887 * A trace chunk should be closed locally before being closed remotely.
1888 * Otherwise, the close timestamp would never be transmitted to the
1891 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
1892 msg
.u
.close_trace_chunk
.close_timestamp
= (uint64_t) close_timestamp
;
1894 if (msg
.u
.close_trace_chunk
.close_command
.is_set
) {
1895 close_command_name
= lttng_trace_chunk_command_type_get_name(
1898 DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
1899 ", session_id = %" PRIu64
", chunk_id = %" PRIu64
1900 ", close command = \"%s\"",
1901 relayd_id
, session_id
, chunk_id
, close_command_name
);
1903 health_code_update();
1904 ret
= consumer_send_msg(socket
, &msg
);
1906 ret
= -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER
;
1911 health_code_update();
1916 * Ask the consumer if a trace chunk exists.
1918 * Called with the consumer socket lock held.
1919 * Returns 0 on success, or a negative value on error.
1921 int consumer_trace_chunk_exists(struct consumer_socket
*socket
,
1922 uint64_t relayd_id
, uint64_t session_id
,
1923 struct lttng_trace_chunk
*chunk
,
1924 enum consumer_trace_chunk_exists_status
*result
)
1927 enum lttng_trace_chunk_status chunk_status
;
1928 struct lttcomm_consumer_msg msg
= {
1929 .cmd_type
= LTTNG_CONSUMER_TRACE_CHUNK_EXISTS
,
1930 .u
.trace_chunk_exists
.session_id
= session_id
,
1933 const char *consumer_reply_str
;
1937 if (relayd_id
!= -1ULL) {
1938 LTTNG_OPTIONAL_SET(&msg
.u
.trace_chunk_exists
.relayd_id
,
1942 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1943 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1945 * Anonymous trace chunks should never be transmitted
1946 * to remote peers (consumerd and relayd). They are used
1947 * internally for backward-compatibility purposes.
1949 ret
= -LTTNG_ERR_FATAL
;
1952 msg
.u
.trace_chunk_exists
.chunk_id
= chunk_id
;
1954 DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
1955 ", session_id = %" PRIu64
1956 ", chunk_id = %" PRIu64
, relayd_id
, session_id
, chunk_id
);
1958 health_code_update();
1959 ret
= consumer_send_msg(socket
, &msg
);
1961 case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK
:
1962 consumer_reply_str
= "unknown trace chunk";
1963 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
;
1965 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL
:
1966 consumer_reply_str
= "trace chunk exists locally";
1967 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL
;
1969 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE
:
1970 consumer_reply_str
= "trace chunk exists on remote peer";
1971 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE
;
1974 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
1978 DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
1979 consumer_reply_str
);
1982 health_code_update();