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 const char *base_path
, int session_live_timer
,
1079 const uint64_t *current_chunk_id
, time_t session_creation_time
,
1080 bool session_name_contains_creation_time
)
1083 struct lttcomm_consumer_msg msg
;
1085 /* Code flow error. Safety net. */
1088 assert(consumer_sock
);
1090 memset(&msg
, 0, sizeof(msg
));
1091 /* Bail out if consumer is disabled */
1092 if (!consumer
->enabled
) {
1097 if (type
== LTTNG_STREAM_CONTROL
) {
1098 ret
= relayd_create_session(rsock
,
1099 &msg
.u
.relayd_sock
.relayd_session_id
,
1100 session_name
, hostname
, base_path
,
1102 consumer
->snapshot
, session_id
,
1103 sessiond_uuid
, current_chunk_id
,
1104 session_creation_time
,
1105 session_name_contains_creation_time
);
1107 /* Close the control socket. */
1108 (void) relayd_close(rsock
);
1113 msg
.cmd_type
= LTTNG_CONSUMER_ADD_RELAYD_SOCKET
;
1115 * Assign network consumer output index using the temporary consumer since
1116 * this call should only be made from within a set_consumer_uri() function
1117 * call in the session daemon.
1119 msg
.u
.relayd_sock
.net_index
= consumer
->net_seq_index
;
1120 msg
.u
.relayd_sock
.type
= type
;
1121 msg
.u
.relayd_sock
.session_id
= session_id
;
1122 memcpy(&msg
.u
.relayd_sock
.sock
, rsock
, sizeof(msg
.u
.relayd_sock
.sock
));
1124 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock
->fd_ptr
);
1125 ret
= consumer_send_msg(consumer_sock
, &msg
);
1130 DBG3("Sending relayd socket file descriptor to consumer");
1131 ret
= consumer_send_fds(consumer_sock
, &rsock
->sock
.fd
, 1);
1136 DBG2("Consumer relayd socket sent");
1143 int consumer_send_pipe(struct consumer_socket
*consumer_sock
,
1144 enum lttng_consumer_command cmd
, int pipe
)
1147 struct lttcomm_consumer_msg msg
;
1148 const char *pipe_name
;
1149 const char *command_name
;
1152 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE
:
1153 pipe_name
= "channel monitor";
1154 command_name
= "SET_CHANNEL_MONITOR_PIPE";
1157 ERR("Unexpected command received in %s (cmd = %d)", __func__
,
1162 /* Code flow error. Safety net. */
1164 memset(&msg
, 0, sizeof(msg
));
1167 pthread_mutex_lock(consumer_sock
->lock
);
1168 DBG3("Sending %s command to consumer", command_name
);
1169 ret
= consumer_send_msg(consumer_sock
, &msg
);
1174 DBG3("Sending %s pipe %d to consumer on socket %d",
1176 pipe
, *consumer_sock
->fd_ptr
);
1177 ret
= consumer_send_fds(consumer_sock
, &pipe
, 1);
1182 DBG2("%s pipe successfully sent", pipe_name
);
1184 pthread_mutex_unlock(consumer_sock
->lock
);
1188 int consumer_send_channel_monitor_pipe(struct consumer_socket
*consumer_sock
,
1191 return consumer_send_pipe(consumer_sock
,
1192 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE
, pipe
);
1196 * Ask the consumer if the data is pending for the specific session id.
1197 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1199 int consumer_is_data_pending(uint64_t session_id
,
1200 struct consumer_output
*consumer
)
1203 int32_t ret_code
= 0; /* Default is that the data is NOT pending */
1204 struct consumer_socket
*socket
;
1205 struct lttng_ht_iter iter
;
1206 struct lttcomm_consumer_msg msg
;
1210 DBG3("Consumer data pending for id %" PRIu64
, session_id
);
1212 memset(&msg
, 0, sizeof(msg
));
1213 msg
.cmd_type
= LTTNG_CONSUMER_DATA_PENDING
;
1214 msg
.u
.data_pending
.session_id
= session_id
;
1216 /* Send command for each consumer */
1218 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1220 pthread_mutex_lock(socket
->lock
);
1221 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1223 pthread_mutex_unlock(socket
->lock
);
1228 * No need for a recv reply status because the answer to the command is
1229 * the reply status message.
1232 ret
= consumer_socket_recv(socket
, &ret_code
, sizeof(ret_code
));
1234 pthread_mutex_unlock(socket
->lock
);
1237 pthread_mutex_unlock(socket
->lock
);
1239 if (ret_code
== 1) {
1245 DBG("Consumer data is %s pending for session id %" PRIu64
,
1246 ret_code
== 1 ? "" : "NOT", session_id
);
1255 * Send a flush command to consumer using the given channel key.
1257 * Return 0 on success else a negative value.
1259 int consumer_flush_channel(struct consumer_socket
*socket
, uint64_t key
)
1262 struct lttcomm_consumer_msg msg
;
1266 DBG2("Consumer flush channel key %" PRIu64
, key
);
1268 memset(&msg
, 0, sizeof(msg
));
1269 msg
.cmd_type
= LTTNG_CONSUMER_FLUSH_CHANNEL
;
1270 msg
.u
.flush_channel
.key
= key
;
1272 pthread_mutex_lock(socket
->lock
);
1273 health_code_update();
1275 ret
= consumer_send_msg(socket
, &msg
);
1281 health_code_update();
1282 pthread_mutex_unlock(socket
->lock
);
1287 * Send a clear quiescent command to consumer using the given channel key.
1289 * Return 0 on success else a negative value.
1291 int consumer_clear_quiescent_channel(struct consumer_socket
*socket
, uint64_t key
)
1294 struct lttcomm_consumer_msg msg
;
1298 DBG2("Consumer clear quiescent channel key %" PRIu64
, key
);
1300 memset(&msg
, 0, sizeof(msg
));
1301 msg
.cmd_type
= LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL
;
1302 msg
.u
.clear_quiescent_channel
.key
= key
;
1304 pthread_mutex_lock(socket
->lock
);
1305 health_code_update();
1307 ret
= consumer_send_msg(socket
, &msg
);
1313 health_code_update();
1314 pthread_mutex_unlock(socket
->lock
);
1319 * Send a close metadata command to consumer using the given channel key.
1320 * Called with registry lock held.
1322 * Return 0 on success else a negative value.
1324 int consumer_close_metadata(struct consumer_socket
*socket
,
1325 uint64_t metadata_key
)
1328 struct lttcomm_consumer_msg msg
;
1332 DBG2("Consumer close metadata channel key %" PRIu64
, metadata_key
);
1334 memset(&msg
, 0, sizeof(msg
));
1335 msg
.cmd_type
= LTTNG_CONSUMER_CLOSE_METADATA
;
1336 msg
.u
.close_metadata
.key
= metadata_key
;
1338 pthread_mutex_lock(socket
->lock
);
1339 health_code_update();
1341 ret
= consumer_send_msg(socket
, &msg
);
1347 health_code_update();
1348 pthread_mutex_unlock(socket
->lock
);
1353 * Send a setup metdata command to consumer using the given channel key.
1355 * Return 0 on success else a negative value.
1357 int consumer_setup_metadata(struct consumer_socket
*socket
,
1358 uint64_t metadata_key
)
1361 struct lttcomm_consumer_msg msg
;
1365 DBG2("Consumer setup metadata channel key %" PRIu64
, metadata_key
);
1367 memset(&msg
, 0, sizeof(msg
));
1368 msg
.cmd_type
= LTTNG_CONSUMER_SETUP_METADATA
;
1369 msg
.u
.setup_metadata
.key
= metadata_key
;
1371 pthread_mutex_lock(socket
->lock
);
1372 health_code_update();
1374 ret
= consumer_send_msg(socket
, &msg
);
1380 health_code_update();
1381 pthread_mutex_unlock(socket
->lock
);
1386 * Send metadata string to consumer.
1387 * RCU read-side lock must be held to guarantee existence of socket.
1389 * Return 0 on success else a negative value.
1391 int consumer_push_metadata(struct consumer_socket
*socket
,
1392 uint64_t metadata_key
, char *metadata_str
, size_t len
,
1393 size_t target_offset
, uint64_t version
)
1396 struct lttcomm_consumer_msg msg
;
1400 DBG2("Consumer push metadata to consumer socket %d", *socket
->fd_ptr
);
1402 pthread_mutex_lock(socket
->lock
);
1404 memset(&msg
, 0, sizeof(msg
));
1405 msg
.cmd_type
= LTTNG_CONSUMER_PUSH_METADATA
;
1406 msg
.u
.push_metadata
.key
= metadata_key
;
1407 msg
.u
.push_metadata
.target_offset
= target_offset
;
1408 msg
.u
.push_metadata
.len
= len
;
1409 msg
.u
.push_metadata
.version
= version
;
1411 health_code_update();
1412 ret
= consumer_send_msg(socket
, &msg
);
1413 if (ret
< 0 || len
== 0) {
1417 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket
->fd_ptr
,
1420 ret
= consumer_socket_send(socket
, metadata_str
, len
);
1425 health_code_update();
1426 ret
= consumer_recv_status_reply(socket
);
1432 pthread_mutex_unlock(socket
->lock
);
1433 health_code_update();
1438 * Ask the consumer to snapshot a specific channel using the key.
1440 * Returns LTTNG_OK on success or else an LTTng error code.
1442 enum lttng_error_code
consumer_snapshot_channel(struct consumer_socket
*socket
,
1443 uint64_t key
, const struct consumer_output
*output
, int metadata
,
1444 uid_t uid
, gid_t gid
, const char *channel_path
, int wait
,
1445 uint64_t nb_packets_per_stream
)
1448 enum lttng_error_code status
= LTTNG_OK
;
1449 struct lttcomm_consumer_msg msg
;
1454 DBG("Consumer snapshot channel key %" PRIu64
, key
);
1456 memset(&msg
, 0, sizeof(msg
));
1457 msg
.cmd_type
= LTTNG_CONSUMER_SNAPSHOT_CHANNEL
;
1458 msg
.u
.snapshot_channel
.key
= key
;
1459 msg
.u
.snapshot_channel
.nb_packets_per_stream
= nb_packets_per_stream
;
1460 msg
.u
.snapshot_channel
.metadata
= metadata
;
1462 if (output
->type
== CONSUMER_DST_NET
) {
1463 msg
.u
.snapshot_channel
.relayd_id
=
1464 output
->net_seq_index
;
1465 msg
.u
.snapshot_channel
.use_relayd
= 1;
1467 msg
.u
.snapshot_channel
.relayd_id
= (uint64_t) -1ULL;
1469 ret
= lttng_strncpy(msg
.u
.snapshot_channel
.pathname
,
1471 sizeof(msg
.u
.snapshot_channel
.pathname
));
1473 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1474 sizeof(msg
.u
.snapshot_channel
.pathname
),
1475 strlen(channel_path
),
1477 status
= LTTNG_ERR_SNAPSHOT_FAIL
;
1481 health_code_update();
1482 pthread_mutex_lock(socket
->lock
);
1483 ret
= consumer_send_msg(socket
, &msg
);
1484 pthread_mutex_unlock(socket
->lock
);
1487 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND
:
1488 status
= LTTNG_ERR_CHAN_NOT_FOUND
;
1491 status
= LTTNG_ERR_SNAPSHOT_FAIL
;
1498 health_code_update();
1503 * Ask the consumer the number of discarded events for a channel.
1505 int consumer_get_discarded_events(uint64_t session_id
, uint64_t channel_key
,
1506 struct consumer_output
*consumer
, uint64_t *discarded
)
1509 struct consumer_socket
*socket
;
1510 struct lttng_ht_iter iter
;
1511 struct lttcomm_consumer_msg msg
;
1515 DBG3("Consumer discarded events id %" PRIu64
, session_id
);
1517 memset(&msg
, 0, sizeof(msg
));
1518 msg
.cmd_type
= LTTNG_CONSUMER_DISCARDED_EVENTS
;
1519 msg
.u
.discarded_events
.session_id
= session_id
;
1520 msg
.u
.discarded_events
.channel_key
= channel_key
;
1524 /* Send command for each consumer */
1526 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1528 uint64_t consumer_discarded
= 0;
1529 pthread_mutex_lock(socket
->lock
);
1530 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1532 pthread_mutex_unlock(socket
->lock
);
1537 * No need for a recv reply status because the answer to the
1538 * command is the reply status message.
1540 ret
= consumer_socket_recv(socket
, &consumer_discarded
,
1541 sizeof(consumer_discarded
));
1543 ERR("get discarded events");
1544 pthread_mutex_unlock(socket
->lock
);
1547 pthread_mutex_unlock(socket
->lock
);
1548 *discarded
+= consumer_discarded
;
1551 DBG("Consumer discarded %" PRIu64
" events in session id %" PRIu64
,
1552 *discarded
, session_id
);
1560 * Ask the consumer the number of lost packets for a channel.
1562 int consumer_get_lost_packets(uint64_t session_id
, uint64_t channel_key
,
1563 struct consumer_output
*consumer
, uint64_t *lost
)
1566 struct consumer_socket
*socket
;
1567 struct lttng_ht_iter iter
;
1568 struct lttcomm_consumer_msg msg
;
1572 DBG3("Consumer lost packets id %" PRIu64
, session_id
);
1574 memset(&msg
, 0, sizeof(msg
));
1575 msg
.cmd_type
= LTTNG_CONSUMER_LOST_PACKETS
;
1576 msg
.u
.lost_packets
.session_id
= session_id
;
1577 msg
.u
.lost_packets
.channel_key
= channel_key
;
1581 /* Send command for each consumer */
1583 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1585 uint64_t consumer_lost
= 0;
1586 pthread_mutex_lock(socket
->lock
);
1587 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1589 pthread_mutex_unlock(socket
->lock
);
1594 * No need for a recv reply status because the answer to the
1595 * command is the reply status message.
1597 ret
= consumer_socket_recv(socket
, &consumer_lost
,
1598 sizeof(consumer_lost
));
1600 ERR("get lost packets");
1601 pthread_mutex_unlock(socket
->lock
);
1604 pthread_mutex_unlock(socket
->lock
);
1605 *lost
+= consumer_lost
;
1608 DBG("Consumer lost %" PRIu64
" packets in session id %" PRIu64
,
1617 * Ask the consumer to rotate a channel.
1619 * The new_chunk_id is the session->rotate_count that has been incremented
1620 * when the rotation started. On the relay, this allows to keep track in which
1621 * chunk each stream is currently writing to (for the rotate_pending operation).
1623 int consumer_rotate_channel(struct consumer_socket
*socket
, uint64_t key
,
1624 uid_t uid
, gid_t gid
, struct consumer_output
*output
,
1625 bool is_metadata_channel
)
1628 struct lttcomm_consumer_msg msg
;
1632 DBG("Consumer rotate channel key %" PRIu64
, key
);
1634 pthread_mutex_lock(socket
->lock
);
1635 memset(&msg
, 0, sizeof(msg
));
1636 msg
.cmd_type
= LTTNG_CONSUMER_ROTATE_CHANNEL
;
1637 msg
.u
.rotate_channel
.key
= key
;
1638 msg
.u
.rotate_channel
.metadata
= !!is_metadata_channel
;
1640 if (output
->type
== CONSUMER_DST_NET
) {
1641 msg
.u
.rotate_channel
.relayd_id
= output
->net_seq_index
;
1643 msg
.u
.rotate_channel
.relayd_id
= (uint64_t) -1ULL;
1646 health_code_update();
1647 ret
= consumer_send_msg(socket
, &msg
);
1650 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND
:
1651 ret
= -LTTNG_ERR_CHAN_NOT_FOUND
;
1654 ret
= -LTTNG_ERR_ROTATION_FAIL_CONSUMER
;
1660 pthread_mutex_unlock(socket
->lock
);
1661 health_code_update();
1665 int consumer_init(struct consumer_socket
*socket
,
1666 const lttng_uuid sessiond_uuid
)
1669 struct lttcomm_consumer_msg msg
= {
1670 .cmd_type
= LTTNG_CONSUMER_INIT
,
1675 DBG("Sending consumer initialization command");
1676 lttng_uuid_copy(msg
.u
.init
.sessiond_uuid
, sessiond_uuid
);
1678 health_code_update();
1679 ret
= consumer_send_msg(socket
, &msg
);
1685 health_code_update();
1690 * Ask the consumer to create a new chunk for a given session.
1692 * Called with the consumer socket lock held.
1694 int consumer_create_trace_chunk(struct consumer_socket
*socket
,
1695 uint64_t relayd_id
, uint64_t session_id
,
1696 struct lttng_trace_chunk
*chunk
)
1699 enum lttng_trace_chunk_status chunk_status
;
1700 struct lttng_credentials chunk_credentials
;
1701 const struct lttng_directory_handle
*chunk_directory_handle
;
1703 const char *chunk_name
;
1704 bool chunk_name_overridden
;
1706 time_t creation_timestamp
;
1707 char creation_timestamp_buffer
[ISO8601_STR_LEN
];
1708 const char *creation_timestamp_str
= "(none)";
1709 const bool chunk_has_local_output
= relayd_id
== -1ULL;
1710 struct lttcomm_consumer_msg msg
= {
1711 .cmd_type
= LTTNG_CONSUMER_CREATE_TRACE_CHUNK
,
1712 .u
.create_trace_chunk
.session_id
= session_id
,
1718 if (relayd_id
!= -1ULL) {
1719 LTTNG_OPTIONAL_SET(&msg
.u
.create_trace_chunk
.relayd_id
,
1723 chunk_status
= lttng_trace_chunk_get_name(chunk
, &chunk_name
,
1724 &chunk_name_overridden
);
1725 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
&&
1726 chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_NONE
) {
1727 ERR("Failed to get name of trace chunk");
1728 ret
= -LTTNG_ERR_FATAL
;
1731 if (chunk_name_overridden
) {
1732 ret
= lttng_strncpy(msg
.u
.create_trace_chunk
.override_name
,
1734 sizeof(msg
.u
.create_trace_chunk
.override_name
));
1736 ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1738 ret
= -LTTNG_ERR_FATAL
;
1743 chunk_status
= lttng_trace_chunk_get_creation_timestamp(chunk
,
1744 &creation_timestamp
);
1745 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1746 ret
= -LTTNG_ERR_FATAL
;
1749 msg
.u
.create_trace_chunk
.creation_timestamp
=
1750 (uint64_t) creation_timestamp
;
1751 /* Only used for logging purposes. */
1752 ret
= time_to_iso8601_str(creation_timestamp
,
1753 creation_timestamp_buffer
,
1754 sizeof(creation_timestamp_buffer
));
1755 creation_timestamp_str
= !ret
? creation_timestamp_buffer
:
1756 "(formatting error)";
1758 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1759 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1761 * Anonymous trace chunks should never be transmitted
1762 * to remote peers (consumerd and relayd). They are used
1763 * internally for backward-compatibility purposes.
1765 ret
= -LTTNG_ERR_FATAL
;
1768 msg
.u
.create_trace_chunk
.chunk_id
= chunk_id
;
1770 if (chunk_has_local_output
) {
1771 chunk_status
= lttng_trace_chunk_get_chunk_directory_handle(
1772 chunk
, &chunk_directory_handle
);
1773 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1774 ret
= -LTTNG_ERR_FATAL
;
1779 * This will only compile on platforms that support
1780 * dirfd (POSIX.2008). This is fine as the session daemon
1781 * is only built for such platforms.
1783 * The ownership of the chunk directory handle's is maintained
1784 * by the trace chunk.
1786 chunk_dirfd
= lttng_directory_handle_get_dirfd(
1787 chunk_directory_handle
);
1788 assert(chunk_dirfd
>= 0);
1790 chunk_status
= lttng_trace_chunk_get_credentials(
1791 chunk
, &chunk_credentials
);
1792 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1794 * Not associating credentials to a sessiond chunk is a
1795 * fatal internal error.
1797 ret
= -LTTNG_ERR_FATAL
;
1800 msg
.u
.create_trace_chunk
.credentials
.value
.uid
=
1801 chunk_credentials
.uid
;
1802 msg
.u
.create_trace_chunk
.credentials
.value
.gid
=
1803 chunk_credentials
.gid
;
1804 msg
.u
.create_trace_chunk
.credentials
.is_set
= 1;
1807 DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1808 ", session_id = %" PRIu64
", chunk_id = %" PRIu64
1809 ", creation_timestamp = %s",
1810 relayd_id
, session_id
, chunk_id
,
1811 creation_timestamp_str
);
1812 health_code_update();
1813 ret
= consumer_send_msg(socket
, &msg
);
1814 health_code_update();
1816 ERR("Trace chunk creation error on consumer");
1817 ret
= -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER
;
1821 if (chunk_has_local_output
) {
1822 DBG("Sending trace chunk directory fd to consumer");
1823 health_code_update();
1824 ret
= consumer_send_fds(socket
, &chunk_dirfd
, 1);
1825 health_code_update();
1827 ERR("Trace chunk creation error on consumer");
1828 ret
= -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER
;
1837 * Ask the consumer to close a trace chunk for a given session.
1839 * Called with the consumer socket lock held.
1841 int consumer_close_trace_chunk(struct consumer_socket
*socket
,
1842 uint64_t relayd_id
, uint64_t session_id
,
1843 struct lttng_trace_chunk
*chunk
)
1846 enum lttng_trace_chunk_status chunk_status
;
1847 struct lttcomm_consumer_msg msg
= {
1848 .cmd_type
= LTTNG_CONSUMER_CLOSE_TRACE_CHUNK
,
1849 .u
.close_trace_chunk
.session_id
= session_id
,
1852 time_t close_timestamp
;
1853 enum lttng_trace_chunk_command_type close_command
;
1854 const char *close_command_name
= "none";
1858 if (relayd_id
!= -1ULL) {
1860 &msg
.u
.close_trace_chunk
.relayd_id
, relayd_id
);
1863 chunk_status
= lttng_trace_chunk_get_close_command(
1864 chunk
, &close_command
);
1865 switch (chunk_status
) {
1866 case LTTNG_TRACE_CHUNK_STATUS_OK
:
1867 LTTNG_OPTIONAL_SET(&msg
.u
.close_trace_chunk
.close_command
,
1868 (uint32_t) close_command
);
1870 case LTTNG_TRACE_CHUNK_STATUS_NONE
:
1873 ERR("Failed to get trace chunk close command");
1878 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1880 * Anonymous trace chunks should never be transmitted to remote peers
1881 * (consumerd and relayd). They are used internally for
1882 * backward-compatibility purposes.
1884 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
1885 msg
.u
.close_trace_chunk
.chunk_id
= chunk_id
;
1887 chunk_status
= lttng_trace_chunk_get_close_timestamp(chunk
,
1890 * A trace chunk should be closed locally before being closed remotely.
1891 * Otherwise, the close timestamp would never be transmitted to the
1894 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
1895 msg
.u
.close_trace_chunk
.close_timestamp
= (uint64_t) close_timestamp
;
1897 if (msg
.u
.close_trace_chunk
.close_command
.is_set
) {
1898 close_command_name
= lttng_trace_chunk_command_type_get_name(
1901 DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
1902 ", session_id = %" PRIu64
", chunk_id = %" PRIu64
1903 ", close command = \"%s\"",
1904 relayd_id
, session_id
, chunk_id
, close_command_name
);
1906 health_code_update();
1907 ret
= consumer_send_msg(socket
, &msg
);
1909 ret
= -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER
;
1914 health_code_update();
1919 * Ask the consumer if a trace chunk exists.
1921 * Called with the consumer socket lock held.
1922 * Returns 0 on success, or a negative value on error.
1924 int consumer_trace_chunk_exists(struct consumer_socket
*socket
,
1925 uint64_t relayd_id
, uint64_t session_id
,
1926 struct lttng_trace_chunk
*chunk
,
1927 enum consumer_trace_chunk_exists_status
*result
)
1930 enum lttng_trace_chunk_status chunk_status
;
1931 struct lttcomm_consumer_msg msg
= {
1932 .cmd_type
= LTTNG_CONSUMER_TRACE_CHUNK_EXISTS
,
1933 .u
.trace_chunk_exists
.session_id
= session_id
,
1936 const char *consumer_reply_str
;
1940 if (relayd_id
!= -1ULL) {
1941 LTTNG_OPTIONAL_SET(&msg
.u
.trace_chunk_exists
.relayd_id
,
1945 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1946 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1948 * Anonymous trace chunks should never be transmitted
1949 * to remote peers (consumerd and relayd). They are used
1950 * internally for backward-compatibility purposes.
1952 ret
= -LTTNG_ERR_FATAL
;
1955 msg
.u
.trace_chunk_exists
.chunk_id
= chunk_id
;
1957 DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
1958 ", session_id = %" PRIu64
1959 ", chunk_id = %" PRIu64
, relayd_id
, session_id
, chunk_id
);
1961 health_code_update();
1962 ret
= consumer_send_msg(socket
, &msg
);
1964 case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK
:
1965 consumer_reply_str
= "unknown trace chunk";
1966 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
;
1968 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL
:
1969 consumer_reply_str
= "trace chunk exists locally";
1970 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL
;
1972 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE
:
1973 consumer_reply_str
= "trace chunk exists on remote peer";
1974 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE
;
1977 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
1981 DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
1982 consumer_reply_str
);
1985 health_code_update();