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
)
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
, base_path
,
1101 consumer
->snapshot
, session_id
,
1102 sessiond_uuid
, current_chunk_id
,
1103 session_creation_time
);
1105 /* Close the control socket. */
1106 (void) relayd_close(rsock
);
1111 msg
.cmd_type
= LTTNG_CONSUMER_ADD_RELAYD_SOCKET
;
1113 * Assign network consumer output index using the temporary consumer since
1114 * this call should only be made from within a set_consumer_uri() function
1115 * call in the session daemon.
1117 msg
.u
.relayd_sock
.net_index
= consumer
->net_seq_index
;
1118 msg
.u
.relayd_sock
.type
= type
;
1119 msg
.u
.relayd_sock
.session_id
= session_id
;
1120 memcpy(&msg
.u
.relayd_sock
.sock
, rsock
, sizeof(msg
.u
.relayd_sock
.sock
));
1122 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock
->fd_ptr
);
1123 ret
= consumer_send_msg(consumer_sock
, &msg
);
1128 DBG3("Sending relayd socket file descriptor to consumer");
1129 ret
= consumer_send_fds(consumer_sock
, &rsock
->sock
.fd
, 1);
1134 DBG2("Consumer relayd socket sent");
1141 int consumer_send_pipe(struct consumer_socket
*consumer_sock
,
1142 enum lttng_consumer_command cmd
, int pipe
)
1145 struct lttcomm_consumer_msg msg
;
1146 const char *pipe_name
;
1147 const char *command_name
;
1150 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE
:
1151 pipe_name
= "channel monitor";
1152 command_name
= "SET_CHANNEL_MONITOR_PIPE";
1155 ERR("Unexpected command received in %s (cmd = %d)", __func__
,
1160 /* Code flow error. Safety net. */
1162 memset(&msg
, 0, sizeof(msg
));
1165 pthread_mutex_lock(consumer_sock
->lock
);
1166 DBG3("Sending %s command to consumer", command_name
);
1167 ret
= consumer_send_msg(consumer_sock
, &msg
);
1172 DBG3("Sending %s pipe %d to consumer on socket %d",
1174 pipe
, *consumer_sock
->fd_ptr
);
1175 ret
= consumer_send_fds(consumer_sock
, &pipe
, 1);
1180 DBG2("%s pipe successfully sent", pipe_name
);
1182 pthread_mutex_unlock(consumer_sock
->lock
);
1186 int consumer_send_channel_monitor_pipe(struct consumer_socket
*consumer_sock
,
1189 return consumer_send_pipe(consumer_sock
,
1190 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE
, pipe
);
1194 * Ask the consumer if the data is pending for the specific session id.
1195 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1197 int consumer_is_data_pending(uint64_t session_id
,
1198 struct consumer_output
*consumer
)
1201 int32_t ret_code
= 0; /* Default is that the data is NOT pending */
1202 struct consumer_socket
*socket
;
1203 struct lttng_ht_iter iter
;
1204 struct lttcomm_consumer_msg msg
;
1208 DBG3("Consumer data pending for id %" PRIu64
, session_id
);
1210 memset(&msg
, 0, sizeof(msg
));
1211 msg
.cmd_type
= LTTNG_CONSUMER_DATA_PENDING
;
1212 msg
.u
.data_pending
.session_id
= session_id
;
1214 /* Send command for each consumer */
1216 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1218 pthread_mutex_lock(socket
->lock
);
1219 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1221 pthread_mutex_unlock(socket
->lock
);
1226 * No need for a recv reply status because the answer to the command is
1227 * the reply status message.
1230 ret
= consumer_socket_recv(socket
, &ret_code
, sizeof(ret_code
));
1232 pthread_mutex_unlock(socket
->lock
);
1235 pthread_mutex_unlock(socket
->lock
);
1237 if (ret_code
== 1) {
1243 DBG("Consumer data is %s pending for session id %" PRIu64
,
1244 ret_code
== 1 ? "" : "NOT", session_id
);
1253 * Send a flush command to consumer using the given channel key.
1255 * Return 0 on success else a negative value.
1257 int consumer_flush_channel(struct consumer_socket
*socket
, uint64_t key
)
1260 struct lttcomm_consumer_msg msg
;
1264 DBG2("Consumer flush channel key %" PRIu64
, key
);
1266 memset(&msg
, 0, sizeof(msg
));
1267 msg
.cmd_type
= LTTNG_CONSUMER_FLUSH_CHANNEL
;
1268 msg
.u
.flush_channel
.key
= key
;
1270 pthread_mutex_lock(socket
->lock
);
1271 health_code_update();
1273 ret
= consumer_send_msg(socket
, &msg
);
1279 health_code_update();
1280 pthread_mutex_unlock(socket
->lock
);
1285 * Send a clear quiescent command to consumer using the given channel key.
1287 * Return 0 on success else a negative value.
1289 int consumer_clear_quiescent_channel(struct consumer_socket
*socket
, uint64_t key
)
1292 struct lttcomm_consumer_msg msg
;
1296 DBG2("Consumer clear quiescent channel key %" PRIu64
, key
);
1298 memset(&msg
, 0, sizeof(msg
));
1299 msg
.cmd_type
= LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL
;
1300 msg
.u
.clear_quiescent_channel
.key
= key
;
1302 pthread_mutex_lock(socket
->lock
);
1303 health_code_update();
1305 ret
= consumer_send_msg(socket
, &msg
);
1311 health_code_update();
1312 pthread_mutex_unlock(socket
->lock
);
1317 * Send a close metadata command to consumer using the given channel key.
1318 * Called with registry lock held.
1320 * Return 0 on success else a negative value.
1322 int consumer_close_metadata(struct consumer_socket
*socket
,
1323 uint64_t metadata_key
)
1326 struct lttcomm_consumer_msg msg
;
1330 DBG2("Consumer close metadata channel key %" PRIu64
, metadata_key
);
1332 memset(&msg
, 0, sizeof(msg
));
1333 msg
.cmd_type
= LTTNG_CONSUMER_CLOSE_METADATA
;
1334 msg
.u
.close_metadata
.key
= metadata_key
;
1336 pthread_mutex_lock(socket
->lock
);
1337 health_code_update();
1339 ret
= consumer_send_msg(socket
, &msg
);
1345 health_code_update();
1346 pthread_mutex_unlock(socket
->lock
);
1351 * Send a setup metdata command to consumer using the given channel key.
1353 * Return 0 on success else a negative value.
1355 int consumer_setup_metadata(struct consumer_socket
*socket
,
1356 uint64_t metadata_key
)
1359 struct lttcomm_consumer_msg msg
;
1363 DBG2("Consumer setup metadata channel key %" PRIu64
, metadata_key
);
1365 memset(&msg
, 0, sizeof(msg
));
1366 msg
.cmd_type
= LTTNG_CONSUMER_SETUP_METADATA
;
1367 msg
.u
.setup_metadata
.key
= metadata_key
;
1369 pthread_mutex_lock(socket
->lock
);
1370 health_code_update();
1372 ret
= consumer_send_msg(socket
, &msg
);
1378 health_code_update();
1379 pthread_mutex_unlock(socket
->lock
);
1384 * Send metadata string to consumer.
1385 * RCU read-side lock must be held to guarantee existence of socket.
1387 * Return 0 on success else a negative value.
1389 int consumer_push_metadata(struct consumer_socket
*socket
,
1390 uint64_t metadata_key
, char *metadata_str
, size_t len
,
1391 size_t target_offset
, uint64_t version
)
1394 struct lttcomm_consumer_msg msg
;
1398 DBG2("Consumer push metadata to consumer socket %d", *socket
->fd_ptr
);
1400 pthread_mutex_lock(socket
->lock
);
1402 memset(&msg
, 0, sizeof(msg
));
1403 msg
.cmd_type
= LTTNG_CONSUMER_PUSH_METADATA
;
1404 msg
.u
.push_metadata
.key
= metadata_key
;
1405 msg
.u
.push_metadata
.target_offset
= target_offset
;
1406 msg
.u
.push_metadata
.len
= len
;
1407 msg
.u
.push_metadata
.version
= version
;
1409 health_code_update();
1410 ret
= consumer_send_msg(socket
, &msg
);
1411 if (ret
< 0 || len
== 0) {
1415 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket
->fd_ptr
,
1418 ret
= consumer_socket_send(socket
, metadata_str
, len
);
1423 health_code_update();
1424 ret
= consumer_recv_status_reply(socket
);
1430 pthread_mutex_unlock(socket
->lock
);
1431 health_code_update();
1436 * Ask the consumer to snapshot a specific channel using the key.
1438 * Returns LTTNG_OK on success or else an LTTng error code.
1440 enum lttng_error_code
consumer_snapshot_channel(struct consumer_socket
*socket
,
1441 uint64_t key
, const struct consumer_output
*output
, int metadata
,
1442 uid_t uid
, gid_t gid
, const char *channel_path
, int wait
,
1443 uint64_t nb_packets_per_stream
)
1446 enum lttng_error_code status
= LTTNG_OK
;
1447 struct lttcomm_consumer_msg msg
;
1452 DBG("Consumer snapshot channel key %" PRIu64
, key
);
1454 memset(&msg
, 0, sizeof(msg
));
1455 msg
.cmd_type
= LTTNG_CONSUMER_SNAPSHOT_CHANNEL
;
1456 msg
.u
.snapshot_channel
.key
= key
;
1457 msg
.u
.snapshot_channel
.nb_packets_per_stream
= nb_packets_per_stream
;
1458 msg
.u
.snapshot_channel
.metadata
= metadata
;
1460 if (output
->type
== CONSUMER_DST_NET
) {
1461 msg
.u
.snapshot_channel
.relayd_id
=
1462 output
->net_seq_index
;
1463 msg
.u
.snapshot_channel
.use_relayd
= 1;
1465 msg
.u
.snapshot_channel
.relayd_id
= (uint64_t) -1ULL;
1467 ret
= lttng_strncpy(msg
.u
.snapshot_channel
.pathname
,
1469 sizeof(msg
.u
.snapshot_channel
.pathname
));
1471 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1472 sizeof(msg
.u
.snapshot_channel
.pathname
),
1473 strlen(channel_path
),
1475 status
= LTTNG_ERR_SNAPSHOT_FAIL
;
1479 health_code_update();
1480 pthread_mutex_lock(socket
->lock
);
1481 ret
= consumer_send_msg(socket
, &msg
);
1482 pthread_mutex_unlock(socket
->lock
);
1485 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND
:
1486 status
= LTTNG_ERR_CHAN_NOT_FOUND
;
1489 status
= LTTNG_ERR_SNAPSHOT_FAIL
;
1496 health_code_update();
1501 * Ask the consumer the number of discarded events for a channel.
1503 int consumer_get_discarded_events(uint64_t session_id
, uint64_t channel_key
,
1504 struct consumer_output
*consumer
, uint64_t *discarded
)
1507 struct consumer_socket
*socket
;
1508 struct lttng_ht_iter iter
;
1509 struct lttcomm_consumer_msg msg
;
1513 DBG3("Consumer discarded events id %" PRIu64
, session_id
);
1515 memset(&msg
, 0, sizeof(msg
));
1516 msg
.cmd_type
= LTTNG_CONSUMER_DISCARDED_EVENTS
;
1517 msg
.u
.discarded_events
.session_id
= session_id
;
1518 msg
.u
.discarded_events
.channel_key
= channel_key
;
1522 /* Send command for each consumer */
1524 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1526 uint64_t consumer_discarded
= 0;
1527 pthread_mutex_lock(socket
->lock
);
1528 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1530 pthread_mutex_unlock(socket
->lock
);
1535 * No need for a recv reply status because the answer to the
1536 * command is the reply status message.
1538 ret
= consumer_socket_recv(socket
, &consumer_discarded
,
1539 sizeof(consumer_discarded
));
1541 ERR("get discarded events");
1542 pthread_mutex_unlock(socket
->lock
);
1545 pthread_mutex_unlock(socket
->lock
);
1546 *discarded
+= consumer_discarded
;
1549 DBG("Consumer discarded %" PRIu64
" events in session id %" PRIu64
,
1550 *discarded
, session_id
);
1558 * Ask the consumer the number of lost packets for a channel.
1560 int consumer_get_lost_packets(uint64_t session_id
, uint64_t channel_key
,
1561 struct consumer_output
*consumer
, uint64_t *lost
)
1564 struct consumer_socket
*socket
;
1565 struct lttng_ht_iter iter
;
1566 struct lttcomm_consumer_msg msg
;
1570 DBG3("Consumer lost packets id %" PRIu64
, session_id
);
1572 memset(&msg
, 0, sizeof(msg
));
1573 msg
.cmd_type
= LTTNG_CONSUMER_LOST_PACKETS
;
1574 msg
.u
.lost_packets
.session_id
= session_id
;
1575 msg
.u
.lost_packets
.channel_key
= channel_key
;
1579 /* Send command for each consumer */
1581 cds_lfht_for_each_entry(consumer
->socks
->ht
, &iter
.iter
, socket
,
1583 uint64_t consumer_lost
= 0;
1584 pthread_mutex_lock(socket
->lock
);
1585 ret
= consumer_socket_send(socket
, &msg
, sizeof(msg
));
1587 pthread_mutex_unlock(socket
->lock
);
1592 * No need for a recv reply status because the answer to the
1593 * command is the reply status message.
1595 ret
= consumer_socket_recv(socket
, &consumer_lost
,
1596 sizeof(consumer_lost
));
1598 ERR("get lost packets");
1599 pthread_mutex_unlock(socket
->lock
);
1602 pthread_mutex_unlock(socket
->lock
);
1603 *lost
+= consumer_lost
;
1606 DBG("Consumer lost %" PRIu64
" packets in session id %" PRIu64
,
1615 * Ask the consumer to rotate a channel.
1617 * The new_chunk_id is the session->rotate_count that has been incremented
1618 * when the rotation started. On the relay, this allows to keep track in which
1619 * chunk each stream is currently writing to (for the rotate_pending operation).
1621 int consumer_rotate_channel(struct consumer_socket
*socket
, uint64_t key
,
1622 uid_t uid
, gid_t gid
, struct consumer_output
*output
,
1623 bool is_metadata_channel
)
1626 struct lttcomm_consumer_msg msg
;
1630 DBG("Consumer rotate channel key %" PRIu64
, key
);
1632 pthread_mutex_lock(socket
->lock
);
1633 memset(&msg
, 0, sizeof(msg
));
1634 msg
.cmd_type
= LTTNG_CONSUMER_ROTATE_CHANNEL
;
1635 msg
.u
.rotate_channel
.key
= key
;
1636 msg
.u
.rotate_channel
.metadata
= !!is_metadata_channel
;
1638 if (output
->type
== CONSUMER_DST_NET
) {
1639 msg
.u
.rotate_channel
.relayd_id
= output
->net_seq_index
;
1641 msg
.u
.rotate_channel
.relayd_id
= (uint64_t) -1ULL;
1644 health_code_update();
1645 ret
= consumer_send_msg(socket
, &msg
);
1648 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND
:
1649 ret
= -LTTNG_ERR_CHAN_NOT_FOUND
;
1652 ret
= -LTTNG_ERR_ROTATION_FAIL_CONSUMER
;
1658 pthread_mutex_unlock(socket
->lock
);
1659 health_code_update();
1663 int consumer_init(struct consumer_socket
*socket
,
1664 const lttng_uuid sessiond_uuid
)
1667 struct lttcomm_consumer_msg msg
= {
1668 .cmd_type
= LTTNG_CONSUMER_INIT
,
1673 DBG("Sending consumer initialization command");
1674 lttng_uuid_copy(msg
.u
.init
.sessiond_uuid
, sessiond_uuid
);
1676 health_code_update();
1677 ret
= consumer_send_msg(socket
, &msg
);
1683 health_code_update();
1688 * Ask the consumer to create a new chunk for a given session.
1690 * Called with the consumer socket lock held.
1692 int consumer_create_trace_chunk(struct consumer_socket
*socket
,
1693 uint64_t relayd_id
, uint64_t session_id
,
1694 struct lttng_trace_chunk
*chunk
)
1697 enum lttng_trace_chunk_status chunk_status
;
1698 struct lttng_credentials chunk_credentials
;
1699 const struct lttng_directory_handle
*chunk_directory_handle
;
1701 const char *chunk_name
;
1702 bool chunk_name_overridden
;
1704 time_t creation_timestamp
;
1705 char creation_timestamp_buffer
[ISO8601_STR_LEN
];
1706 const char *creation_timestamp_str
= "(none)";
1707 const bool chunk_has_local_output
= relayd_id
== -1ULL;
1708 struct lttcomm_consumer_msg msg
= {
1709 .cmd_type
= LTTNG_CONSUMER_CREATE_TRACE_CHUNK
,
1710 .u
.create_trace_chunk
.session_id
= session_id
,
1716 if (relayd_id
!= -1ULL) {
1717 LTTNG_OPTIONAL_SET(&msg
.u
.create_trace_chunk
.relayd_id
,
1721 chunk_status
= lttng_trace_chunk_get_name(chunk
, &chunk_name
,
1722 &chunk_name_overridden
);
1723 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
&&
1724 chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_NONE
) {
1725 ERR("Failed to get name of trace chunk");
1726 ret
= -LTTNG_ERR_FATAL
;
1729 if (chunk_name_overridden
) {
1730 ret
= lttng_strncpy(msg
.u
.create_trace_chunk
.override_name
,
1732 sizeof(msg
.u
.create_trace_chunk
.override_name
));
1734 ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1736 ret
= -LTTNG_ERR_FATAL
;
1741 chunk_status
= lttng_trace_chunk_get_creation_timestamp(chunk
,
1742 &creation_timestamp
);
1743 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1744 ret
= -LTTNG_ERR_FATAL
;
1747 msg
.u
.create_trace_chunk
.creation_timestamp
=
1748 (uint64_t) creation_timestamp
;
1749 /* Only used for logging purposes. */
1750 ret
= time_to_iso8601_str(creation_timestamp
,
1751 creation_timestamp_buffer
,
1752 sizeof(creation_timestamp_buffer
));
1753 creation_timestamp_str
= !ret
? creation_timestamp_buffer
:
1754 "(formatting error)";
1756 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1757 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1759 * Anonymous trace chunks should never be transmitted
1760 * to remote peers (consumerd and relayd). They are used
1761 * internally for backward-compatibility purposes.
1763 ret
= -LTTNG_ERR_FATAL
;
1766 msg
.u
.create_trace_chunk
.chunk_id
= chunk_id
;
1768 if (chunk_has_local_output
) {
1769 chunk_status
= lttng_trace_chunk_get_chunk_directory_handle(
1770 chunk
, &chunk_directory_handle
);
1771 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1772 ret
= -LTTNG_ERR_FATAL
;
1777 * This will only compile on platforms that support
1778 * dirfd (POSIX.2008). This is fine as the session daemon
1779 * is only built for such platforms.
1781 * The ownership of the chunk directory handle's is maintained
1782 * by the trace chunk.
1784 chunk_dirfd
= lttng_directory_handle_get_dirfd(
1785 chunk_directory_handle
);
1786 assert(chunk_dirfd
>= 0);
1788 chunk_status
= lttng_trace_chunk_get_credentials(
1789 chunk
, &chunk_credentials
);
1790 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1792 * Not associating credentials to a sessiond chunk is a
1793 * fatal internal error.
1795 ret
= -LTTNG_ERR_FATAL
;
1798 msg
.u
.create_trace_chunk
.credentials
.value
.uid
=
1799 chunk_credentials
.uid
;
1800 msg
.u
.create_trace_chunk
.credentials
.value
.gid
=
1801 chunk_credentials
.gid
;
1802 msg
.u
.create_trace_chunk
.credentials
.is_set
= 1;
1805 DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1806 ", session_id = %" PRIu64
", chunk_id = %" PRIu64
1807 ", creation_timestamp = %s",
1808 relayd_id
, session_id
, chunk_id
,
1809 creation_timestamp_str
);
1810 health_code_update();
1811 ret
= consumer_send_msg(socket
, &msg
);
1812 health_code_update();
1814 ERR("Trace chunk creation error on consumer");
1815 ret
= -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER
;
1819 if (chunk_has_local_output
) {
1820 DBG("Sending trace chunk directory fd to consumer");
1821 health_code_update();
1822 ret
= consumer_send_fds(socket
, &chunk_dirfd
, 1);
1823 health_code_update();
1825 ERR("Trace chunk creation error on consumer");
1826 ret
= -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER
;
1835 * Ask the consumer to close a trace chunk for a given session.
1837 * Called with the consumer socket lock held.
1839 int consumer_close_trace_chunk(struct consumer_socket
*socket
,
1840 uint64_t relayd_id
, uint64_t session_id
,
1841 struct lttng_trace_chunk
*chunk
)
1844 enum lttng_trace_chunk_status chunk_status
;
1845 struct lttcomm_consumer_msg msg
= {
1846 .cmd_type
= LTTNG_CONSUMER_CLOSE_TRACE_CHUNK
,
1847 .u
.close_trace_chunk
.session_id
= session_id
,
1850 time_t close_timestamp
;
1851 enum lttng_trace_chunk_command_type close_command
;
1852 const char *close_command_name
= "none";
1856 if (relayd_id
!= -1ULL) {
1858 &msg
.u
.close_trace_chunk
.relayd_id
, relayd_id
);
1861 chunk_status
= lttng_trace_chunk_get_close_command(
1862 chunk
, &close_command
);
1863 switch (chunk_status
) {
1864 case LTTNG_TRACE_CHUNK_STATUS_OK
:
1865 LTTNG_OPTIONAL_SET(&msg
.u
.close_trace_chunk
.close_command
,
1866 (uint32_t) close_command
);
1868 case LTTNG_TRACE_CHUNK_STATUS_NONE
:
1871 ERR("Failed to get trace chunk close command");
1876 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1878 * Anonymous trace chunks should never be transmitted to remote peers
1879 * (consumerd and relayd). They are used internally for
1880 * backward-compatibility purposes.
1882 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
1883 msg
.u
.close_trace_chunk
.chunk_id
= chunk_id
;
1885 chunk_status
= lttng_trace_chunk_get_close_timestamp(chunk
,
1888 * A trace chunk should be closed locally before being closed remotely.
1889 * Otherwise, the close timestamp would never be transmitted to the
1892 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
1893 msg
.u
.close_trace_chunk
.close_timestamp
= (uint64_t) close_timestamp
;
1895 if (msg
.u
.close_trace_chunk
.close_command
.is_set
) {
1896 close_command_name
= lttng_trace_chunk_command_type_get_name(
1899 DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
1900 ", session_id = %" PRIu64
", chunk_id = %" PRIu64
1901 ", close command = \"%s\"",
1902 relayd_id
, session_id
, chunk_id
, close_command_name
);
1904 health_code_update();
1905 ret
= consumer_send_msg(socket
, &msg
);
1907 ret
= -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER
;
1912 health_code_update();
1917 * Ask the consumer if a trace chunk exists.
1919 * Called with the consumer socket lock held.
1920 * Returns 0 on success, or a negative value on error.
1922 int consumer_trace_chunk_exists(struct consumer_socket
*socket
,
1923 uint64_t relayd_id
, uint64_t session_id
,
1924 struct lttng_trace_chunk
*chunk
,
1925 enum consumer_trace_chunk_exists_status
*result
)
1928 enum lttng_trace_chunk_status chunk_status
;
1929 struct lttcomm_consumer_msg msg
= {
1930 .cmd_type
= LTTNG_CONSUMER_TRACE_CHUNK_EXISTS
,
1931 .u
.trace_chunk_exists
.session_id
= session_id
,
1934 const char *consumer_reply_str
;
1938 if (relayd_id
!= -1ULL) {
1939 LTTNG_OPTIONAL_SET(&msg
.u
.trace_chunk_exists
.relayd_id
,
1943 chunk_status
= lttng_trace_chunk_get_id(chunk
, &chunk_id
);
1944 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1946 * Anonymous trace chunks should never be transmitted
1947 * to remote peers (consumerd and relayd). They are used
1948 * internally for backward-compatibility purposes.
1950 ret
= -LTTNG_ERR_FATAL
;
1953 msg
.u
.trace_chunk_exists
.chunk_id
= chunk_id
;
1955 DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
1956 ", session_id = %" PRIu64
1957 ", chunk_id = %" PRIu64
, relayd_id
, session_id
, chunk_id
);
1959 health_code_update();
1960 ret
= consumer_send_msg(socket
, &msg
);
1962 case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK
:
1963 consumer_reply_str
= "unknown trace chunk";
1964 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
;
1966 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL
:
1967 consumer_reply_str
= "trace chunk exists locally";
1968 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL
;
1970 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE
:
1971 consumer_reply_str
= "trace chunk exists on remote peer";
1972 *result
= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE
;
1975 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
1979 DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
1980 consumer_reply_str
);
1983 health_code_update();