X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fkernel-consumer.c;h=c86d52803b80ac19c254eb3a1e267f3223d6c5d9;hp=bbf50d57b594544f62ffb26f9d7c8a3f23f5a832;hb=ca22feea083301934d1c8511851c86fb008c0697;hpb=97b6ffd01e16cbb0871722ee605bf02794031859 diff --git a/src/bin/lttng-sessiond/kernel-consumer.c b/src/bin/lttng-sessiond/kernel-consumer.c index bbf50d57b..c86d52803 100644 --- a/src/bin/lttng-sessiond/kernel-consumer.c +++ b/src/bin/lttng-sessiond/kernel-consumer.c @@ -24,147 +24,290 @@ #include #include -#include +#include "consumer.h" #include "kernel-consumer.h" /* - * Send all stream fds of kernel channel to the consumer. + * Sending a single channel to the consumer with command ADD_CHANNEL. */ -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_add_channel(int sock, struct ltt_kernel_channel *channel) { - int ret, count = 0, consumer_sock; - struct ltt_kernel_stream *stream; + int ret; struct lttcomm_consumer_msg lkm; - DBG("Sending streams of channel %s to kernel consumer", + /* Safety net */ + assert(channel); + + DBG("Kernel consumer adding channel %s to kernel consumer", channel->channel->name); - consumer_sock = consumer_data->cmd_sock; + /* 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, + channel->stream_count); - /* 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 = consumer_send_channel(sock, &lkm); if (ret < 0) { - PERROR("send consumer channel"); goto error; } - /* Send streams */ - cds_list_for_each_entry(stream, &channel->stream_list.head, list) { - 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)); +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; + char tmp_path[PATH_MAX]; + const char *pathname; + struct lttcomm_consumer_msg lkm; + struct consumer_output *consumer; + + /* Safety net */ + assert(session); + assert(session->consumer); + + DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd); + + /* Get consumer output pointer */ + consumer = session->consumer; + + /* Get the right path name destination */ + if (consumer->type == CONSUMER_DST_LOCAL) { + /* Set application path to the destination path */ + ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s", + consumer->dst.trace_path, consumer->subdir); if (ret < 0) { - PERROR("send consumer stream"); + PERROR("snprintf metadata path"); goto error; } - ret = lttcomm_send_fds_unix_sock(consumer_sock, &stream->fd, 1); + pathname = tmp_path; + + /* Create directory */ + ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, + session->uid, session->gid); + if (ret < 0) { + if (ret != -EEXIST) { + ERR("Trace directory creation error"); + goto error; + } + } + DBG3("Kernel local consumer tracefile path: %s", pathname); + } else { + ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir); if (ret < 0) { - PERROR("send consumer stream ancillary data"); + PERROR("snprintf metadata path"); goto error; } + pathname = tmp_path; + DBG3("Kernel network consumer subdir path: %s", pathname); } - DBG("consumer channel streams sent"); + /* 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", + 1); - return 0; + 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, + consumer->net_seq_index, + 1, /* Metadata flag set */ + "metadata", + pathname, + session->id); + + /* Send stream and file descriptor */ + ret = consumer_send_stream(sock, consumer, &lkm, + &session->metadata_stream_fd, 1); + if (ret < 0) { + goto error; + } error: return ret; } /* - * Send all stream fds of the kernel session to the consumer. + * Sending a single stream to the consumer with command ADD_STREAM. */ -int kernel_consumer_send_session(struct consumer_data *consumer_data, - struct ltt_kernel_session *session) +int kernel_consumer_add_stream(int sock, struct ltt_kernel_channel *channel, + struct ltt_kernel_stream *stream, struct ltt_kernel_session *session) { int ret; - struct ltt_kernel_channel *chan; + char tmp_path[PATH_MAX]; + const char *pathname; struct lttcomm_consumer_msg lkm; - int sock = session->consumer_fd; + struct consumer_output *consumer; - DBG("Sending metadata stream fd"); + assert(channel); + assert(stream); + 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; - } + DBG("Sending stream %d of channel %s to kernel consumer", + stream->fd, channel->channel->name); - 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)); + /* Get consumer output pointer */ + consumer = session->consumer; + + /* Get the right path name destination */ + if (consumer->type == CONSUMER_DST_LOCAL) { + /* Set application path to the destination path */ + ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s", + consumer->dst.trace_path, consumer->subdir); + if (ret < 0) { + PERROR("snprintf stream path"); + goto error; + } + pathname = tmp_path; + DBG3("Kernel local consumer tracefile path: %s", pathname); + } else { + ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir); if (ret < 0) { - PERROR("send consumer channel"); + PERROR("snprintf stream path"); goto error; } + pathname = tmp_path; + DBG3("Kernel network consumer subdir path: %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, + consumer->net_seq_index, + 0, /* Metadata flag unset */ + stream->name, + pathname, + session->id); + + /* Send stream and file descriptor */ + ret = consumer_send_stream(sock, consumer, &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(int sock, + struct ltt_kernel_channel *channel, struct ltt_kernel_session *session) +{ + int ret; + struct ltt_kernel_stream *stream; + + /* Safety net */ + assert(channel); + assert(session); + assert(session->consumer); + + /* Bail out if consumer is disabled */ + if (!session->consumer->enabled) { + ret = LTTNG_OK; + goto error; + } + + DBG("Sending streams of channel %s to kernel consumer", + channel->channel->name); + + ret = kernel_consumer_add_channel(sock, channel); + if (ret < 0) { + goto error; + } + + /* Send streams */ + cds_list_for_each_entry(stream, &channel->stream_list.head, list) { + if (!stream->fd) { + continue; + } - /* 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)); + /* Add stream on the kernel consumer side. */ + ret = kernel_consumer_add_stream(sock, channel, stream, session); if (ret < 0) { - PERROR("send consumer stream"); goto error; } - ret = lttcomm_send_fds_unix_sock(sock, &session->metadata_stream_fd, 1); + } + +error: + return ret; +} + +/* + * Send all stream fds of the kernel session to the consumer. + */ +int kernel_consumer_send_session(int sock, struct ltt_kernel_session *session) +{ + int ret; + struct ltt_kernel_channel *chan; + + /* Safety net */ + assert(session); + assert(session->consumer); + + /* Bail out if consumer is disabled */ + if (!session->consumer->enabled) { + ret = LTTNG_OK; + goto error; + } + + DBG("Sending session stream to kernel consumer"); + + if (session->metadata_stream_fd >= 0) { + ret = kernel_consumer_add_metadata(sock, session); 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; error: return ret; } -