X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fkernel-consumer.c;h=5d690548c4e1153492246d485ee69f214088d75a;hp=bbf50d57b594544f62ffb26f9d7c8a3f23f5a832;hb=00e2e675d54dc726a7c8f8887c889cc8ef022003;hpb=b8aa16822f579a6e15b41d2761801a0a65d5f2a5 diff --git a/src/bin/lttng-sessiond/kernel-consumer.c b/src/bin/lttng-sessiond/kernel-consumer.c index bbf50d57b..5d690548c 100644 --- a/src/bin/lttng-sessiond/kernel-consumer.c +++ b/src/bin/lttng-sessiond/kernel-consumer.c @@ -24,34 +24,186 @@ #include #include -#include +#include "consumer.h" #include "kernel-consumer.h" +/* + * Sending a single channel to the consumer with command ADD_CHANNEL. + */ +int kernel_consumer_add_channel(int sock, struct ltt_kernel_channel *channel) +{ + int ret; + struct lttcomm_consumer_msg lkm; + + /* Safety net */ + assert(channel); + + DBG("Kernel consumer adding channel %s to kernel consumer", + channel->channel->name); + + /* Prep channel message structure */ + consumer_init_channel_comm_msg(&lkm, + LTTNG_CONSUMER_ADD_CHANNEL, + channel->fd, + channel->channel->attr.subbuf_size, + 0, /* Kernel */ + channel->channel->name); + + ret = consumer_send_channel(sock, &lkm); + if (ret < 0) { + goto error; + } + +error: + return ret; +} + +/* + * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM. + */ +int kernel_consumer_add_metadata(int sock, struct ltt_kernel_session *session) +{ + int ret; + const char *pathname; + struct lttcomm_consumer_msg lkm; + struct consumer_output *output; + + /* Safety net */ + assert(session); + assert(session->consumer); + + DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd); + + /* Get consumer output pointer */ + output = session->consumer; + + /* Get correct path name destination */ + if (output->type == CONSUMER_DST_LOCAL) { + pathname = output->dst.trace_path; + } else { + pathname = output->subdir; + } + + /* Prep channel message structure */ + consumer_init_channel_comm_msg(&lkm, + LTTNG_CONSUMER_ADD_CHANNEL, + session->metadata->fd, + session->metadata->conf->attr.subbuf_size, + 0, /* for kernel */ + "metadata"); + + ret = consumer_send_channel(sock, &lkm); + if (ret < 0) { + goto error; + } + + /* Prep stream message structure */ + consumer_init_stream_comm_msg(&lkm, + LTTNG_CONSUMER_ADD_STREAM, + session->metadata->fd, + session->metadata_stream_fd, + LTTNG_CONSUMER_ACTIVE_STREAM, + DEFAULT_KERNEL_CHANNEL_OUTPUT, + 0, /* Kernel */ + session->uid, + session->gid, + output->net_seq_index, + 1, /* Metadata flag set */ + "metadata", + pathname); + + /* Send stream and file descriptor */ + ret = consumer_send_stream(sock, output, &lkm, + &session->metadata_stream_fd, 1); + if (ret < 0) { + goto error; + } + +error: + return ret; +} + +/* + * Sending a single stream to the consumer with command ADD_STREAM. + */ +int kernel_consumer_add_stream(int sock, struct ltt_kernel_channel *channel, + struct ltt_kernel_stream *stream, struct ltt_kernel_session *session) +{ + int ret; + const char *pathname; + struct lttcomm_consumer_msg lkm; + struct consumer_output *output; + + assert(channel); + assert(stream); + assert(session); + assert(session->consumer); + + DBG("Sending stream %d of channel %s to kernel consumer", + stream->fd, channel->channel->name); + + /* Get consumer output pointer */ + output = session->consumer; + + /* Get correct path name destination */ + if (output->type == CONSUMER_DST_LOCAL) { + pathname = output->dst.trace_path; + DBG3("Consumer is local to %s", pathname); + } else { + pathname = output->subdir; + DBG3("Consumer is network to subdir %s", pathname); + } + + /* Prep stream consumer message */ + consumer_init_stream_comm_msg(&lkm, LTTNG_CONSUMER_ADD_STREAM, + channel->fd, + stream->fd, + stream->state, + channel->channel->attr.output, + 0, /* Kernel */ + session->uid, + session->gid, + output->net_seq_index, + 0, /* Metadata flag unset */ + stream->name, + pathname); + + /* Send stream and file descriptor */ + ret = consumer_send_stream(sock, output, &lkm, &stream->fd, 1); + if (ret < 0) { + goto error; + } + +error: + return ret; +} + /* * Send all stream fds of kernel channel to the consumer. */ -int kernel_consumer_send_channel_stream(struct consumer_data *consumer_data, - struct ltt_kernel_channel *channel, uid_t uid, gid_t gid) +int kernel_consumer_send_channel_stream(int sock, + struct ltt_kernel_channel *channel, struct ltt_kernel_session *session) { - int ret, count = 0, consumer_sock; + int ret; struct ltt_kernel_stream *stream; - struct lttcomm_consumer_msg lkm; + + /* Safety net */ + assert(channel); + assert(session); + assert(session->consumer); + + /* Bail out if consumer is disabled */ + if (!session->consumer->enabled) { + ret = LTTCOMM_OK; + goto error; + } DBG("Sending streams of channel %s to kernel consumer", channel->channel->name); - consumer_sock = consumer_data->cmd_sock; - - /* Send channel */ - lkm.cmd_type = LTTNG_CONSUMER_ADD_CHANNEL; - lkm.u.channel.channel_key = channel->fd; - lkm.u.channel.max_sb_size = channel->channel->attr.subbuf_size; - lkm.u.channel.mmap_len = 0; /* for kernel */ - DBG("Sending channel %d to consumer", lkm.u.channel.channel_key); - ret = lttcomm_send_unix_sock(consumer_sock, &lkm, sizeof(lkm)); + ret = kernel_consumer_add_channel(sock, channel); if (ret < 0) { - PERROR("send consumer channel"); goto error; } @@ -60,37 +212,14 @@ int kernel_consumer_send_channel_stream(struct consumer_data *consumer_data, if (!stream->fd) { continue; } - /* Reset consumer message structure */ - memset(&lkm, 0, sizeof(lkm)); - lkm.cmd_type = LTTNG_CONSUMER_ADD_STREAM; - lkm.u.stream.channel_key = channel->fd; - lkm.u.stream.stream_key = stream->fd; - lkm.u.stream.state = stream->state; - lkm.u.stream.output = channel->channel->attr.output; - lkm.u.stream.mmap_len = 0; /* for kernel */ - lkm.u.stream.uid = uid; - lkm.u.stream.gid = gid; - strncpy(lkm.u.stream.path_name, stream->pathname, PATH_MAX - 1); - lkm.u.stream.path_name[PATH_MAX - 1] = '\0'; - count++; - - DBG("Sending stream %d to consumer", lkm.u.stream.stream_key); - ret = lttcomm_send_unix_sock(consumer_sock, &lkm, sizeof(lkm)); - if (ret < 0) { - PERROR("send consumer stream"); - goto error; - } - ret = lttcomm_send_fds_unix_sock(consumer_sock, &stream->fd, 1); + + /* Add stream on the kernel consumer side. */ + ret = kernel_consumer_add_stream(sock, channel, stream, session); if (ret < 0) { - PERROR("send consumer stream ancillary data"); goto error; } } - DBG("consumer channel streams sent"); - - return 0; - error: return ret; } @@ -98,69 +227,42 @@ error: /* * Send all stream fds of the kernel session to the consumer. */ -int kernel_consumer_send_session(struct consumer_data *consumer_data, - struct ltt_kernel_session *session) +int kernel_consumer_send_session(int sock, struct ltt_kernel_session *session) { int ret; struct ltt_kernel_channel *chan; - struct lttcomm_consumer_msg lkm; - int sock = session->consumer_fd; - DBG("Sending metadata stream fd"); + /* Safety net */ + assert(session); + assert(session->consumer); - /* Extra protection. It's NOT supposed to be set to -1 at this point */ - if (session->consumer_fd < 0) { - session->consumer_fd = consumer_data->cmd_sock; + /* Bail out if consumer is disabled */ + if (!session->consumer->enabled) { + ret = LTTCOMM_OK; + goto error; } + DBG("Sending session stream to kernel consumer"); + if (session->metadata_stream_fd >= 0) { - /* Send metadata channel fd */ - lkm.cmd_type = LTTNG_CONSUMER_ADD_CHANNEL; - lkm.u.channel.channel_key = session->metadata->fd; - lkm.u.channel.max_sb_size = session->metadata->conf->attr.subbuf_size; - lkm.u.channel.mmap_len = 0; /* for kernel */ - DBG("Sending metadata channel %d to consumer", lkm.u.channel.channel_key); - ret = lttcomm_send_unix_sock(sock, &lkm, sizeof(lkm)); + ret = kernel_consumer_add_metadata(sock, session); if (ret < 0) { - PERROR("send consumer channel"); goto error; } - /* Send metadata stream fd */ - lkm.cmd_type = LTTNG_CONSUMER_ADD_STREAM; - lkm.u.stream.channel_key = session->metadata->fd; - lkm.u.stream.stream_key = session->metadata_stream_fd; - lkm.u.stream.state = LTTNG_CONSUMER_ACTIVE_STREAM; - lkm.u.stream.output = DEFAULT_KERNEL_CHANNEL_OUTPUT; - lkm.u.stream.mmap_len = 0; /* for kernel */ - lkm.u.stream.uid = session->uid; - lkm.u.stream.gid = session->gid; - strncpy(lkm.u.stream.path_name, session->metadata->pathname, - PATH_MAX - 1); - lkm.u.stream.path_name[PATH_MAX - 1] = '\0'; - - DBG("Sending metadata stream %d to consumer", lkm.u.stream.stream_key); - ret = lttcomm_send_unix_sock(sock, &lkm, sizeof(lkm)); - if (ret < 0) { - PERROR("send consumer stream"); - goto error; - } - ret = lttcomm_send_fds_unix_sock(sock, &session->metadata_stream_fd, 1); - if (ret < 0) { - PERROR("send consumer stream"); - goto error; - } + /* Flag that at least the metadata has been sent to the consumer. */ + session->consumer_fds_sent = 1; } + /* Send channel and streams of it */ cds_list_for_each_entry(chan, &session->channel_list.head, list) { - ret = kernel_consumer_send_channel_stream(consumer_data, chan, - session->uid, session->gid); + ret = kernel_consumer_send_channel_stream(sock, chan, session); if (ret < 0) { goto error; } } - DBG("consumer fds (metadata and channel streams) sent"); + DBG("Kernel consumer FDs of metadata and channel streams sent"); return 0; @@ -168,3 +270,53 @@ error: return ret; } +/* + * Send relayd socket to consumer associated with a session name. + * + * On success return positive value. On error, negative value. + */ +int kernel_consumer_send_relayd_socket(int consumer_sock, + struct lttcomm_sock *sock, struct consumer_output *consumer, + enum lttng_stream_type type) +{ + int ret; + struct lttcomm_consumer_msg msg; + + /* Code flow error. Safety net. */ + assert(sock); + assert(consumer); + + /* Bail out if consumer is disabled */ + if (!consumer->enabled) { + ret = LTTCOMM_OK; + goto error; + } + + msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET; + /* + * Assign network consumer output index using the temporary consumer since + * this call should only be made from within a set_consumer_uri() function + * call in the session daemon. + */ + msg.u.relayd_sock.net_index = consumer->net_seq_index; + msg.u.relayd_sock.type = type; + memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock)); + + DBG2("Sending relayd sock info to consumer"); + ret = lttcomm_send_unix_sock(consumer_sock, &msg, sizeof(msg)); + if (ret < 0) { + PERROR("send consumer relayd socket info"); + goto error; + } + + DBG2("Sending relayd socket file descriptor to consumer"); + ret = consumer_send_fds(consumer_sock, &sock->fd, 1); + if (ret < 0) { + goto error; + } + + DBG("Kernel consumer relayd socket sent"); + +error: + return ret; +}