X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fust-consumer.c;h=fa5f071f07aa6896105e51a092827d298b75daf5;hp=03b0c7d03ecee91f63dff4317dc956c10130f7fc;hb=00e2e675d54dc726a7c8f8887c889cc8ef022003;hpb=b8aa16822f579a6e15b41d2761801a0a65d5f2a5 diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c index 03b0c7d03..fa5f071f0 100644 --- a/src/bin/lttng-sessiond/ust-consumer.c +++ b/src/bin/lttng-sessiond/ust-consumer.c @@ -25,86 +25,96 @@ #include #include #include -#include +#include "consumer.h" #include "ust-consumer.h" /* * Send all stream fds of UST channel to the consumer. */ static int send_channel_streams(int sock, - struct ust_app_channel *uchan, - uid_t uid, gid_t gid) + struct ust_app_channel *uchan, const char *path, + uid_t uid, gid_t gid, struct consumer_output *consumer) { int ret, fd; + char tmp_path[PATH_MAX]; + const char *pathname; struct lttcomm_consumer_msg lum; struct ltt_ust_stream *stream, *tmp; DBG("Sending streams of channel %s to UST consumer", uchan->name); - /* Send channel */ - lum.cmd_type = LTTNG_CONSUMER_ADD_CHANNEL; - - /* - * We need to keep shm_fd open while we transfer the stream file - * descriptors to make sure this key stays unique within the - * session daemon. We can free the channel shm_fd without - * problem after we finished sending stream fds for that - * channel. - */ - lum.u.channel.channel_key = uchan->obj->shm_fd; - lum.u.channel.max_sb_size = uchan->attr.subbuf_size; - lum.u.channel.mmap_len = uchan->obj->memory_map_size; - DBG("Sending channel %d to consumer", lum.u.channel.channel_key); - ret = lttcomm_send_unix_sock(sock, &lum, sizeof(lum)); + consumer_init_channel_comm_msg(&lum, + LTTNG_CONSUMER_ADD_CHANNEL, + uchan->obj->shm_fd, + uchan->attr.subbuf_size, + uchan->obj->memory_map_size, + uchan->name); + + ret = consumer_send_channel(sock, &lum); if (ret < 0) { - PERROR("send consumer channel"); goto error; } + fd = uchan->obj->shm_fd; - ret = lttcomm_send_fds_unix_sock(sock, &fd, 1); + ret = consumer_send_fds(sock, &fd, 1); if (ret < 0) { - PERROR("send consumer channel ancillary data"); goto error; } + /* 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, path); + if (ret < 0) { + PERROR("snprintf stream path"); + goto error; + } + pathname = tmp_path; + DBG3("UST local consumer tracefile path: %s", pathname); + } else { + ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s", + consumer->subdir, path); + if (ret < 0) { + PERROR("snprintf stream path"); + goto error; + } + pathname = tmp_path; + DBG3("UST network consumer subdir path: %s", pathname); + } + cds_list_for_each_entry_safe(stream, tmp, &uchan->streams.head, list) { int fds[2]; if (!stream->obj->shm_fd) { continue; } - lum.cmd_type = LTTNG_CONSUMER_ADD_STREAM; - lum.u.stream.channel_key = uchan->obj->shm_fd; - lum.u.stream.stream_key = stream->obj->shm_fd; - lum.u.stream.state = LTTNG_CONSUMER_ACTIVE_STREAM; - /* - * FIXME Hack alert! we force MMAP for now. Mixup - * between EVENT and UST enums elsewhere. - */ - lum.u.stream.output = DEFAULT_UST_CHANNEL_OUTPUT; - lum.u.stream.mmap_len = stream->obj->memory_map_size; - lum.u.stream.uid = uid; - lum.u.stream.gid = gid; - strncpy(lum.u.stream.path_name, stream->pathname, PATH_MAX - 1); - lum.u.stream.path_name[PATH_MAX - 1] = '\0'; - DBG("Sending stream %d to consumer", lum.u.stream.stream_key); - ret = lttcomm_send_unix_sock(sock, &lum, sizeof(lum)); - if (ret < 0) { - PERROR("send consumer stream"); - goto error; - } + consumer_init_stream_comm_msg(&lum, + LTTNG_CONSUMER_ADD_STREAM, + uchan->obj->shm_fd, + stream->obj->shm_fd, + LTTNG_CONSUMER_ACTIVE_STREAM, + DEFAULT_UST_CHANNEL_OUTPUT, + stream->obj->memory_map_size, + uid, + gid, + consumer->net_seq_index, + 0, /* Metadata flag unset */ + stream->name, + pathname); + + /* Send stream and file descriptor */ fds[0] = stream->obj->shm_fd; fds[1] = stream->obj->wait_fd; - ret = lttcomm_send_fds_unix_sock(sock, fds, 2); + ret = consumer_send_stream(sock, consumer, &lum, fds, 2); if (ret < 0) { - PERROR("send consumer stream ancillary data"); goto error; } } - DBG("consumer channel streams sent"); + DBG("UST consumer channel streams sent"); return 0; @@ -115,10 +125,13 @@ error: /* * Send all stream fds of the UST session to the consumer. */ -int ust_consumer_send_session(int consumer_fd, struct ust_app_session *usess) +int ust_consumer_send_session(int consumer_fd, struct ust_app_session *usess, + struct consumer_output *consumer) { int ret = 0; int sock = consumer_fd; + char tmp_path[PATH_MAX]; + const char *pathname; struct lttng_ht_iter iter; struct lttcomm_consumer_msg lum; struct ust_app_channel *ua_chan; @@ -134,46 +147,73 @@ int ust_consumer_send_session(int consumer_fd, struct ust_app_session *usess) int fd; int fds[2]; - /* Send metadata channel fd */ - lum.cmd_type = LTTNG_CONSUMER_ADD_CHANNEL; - lum.u.channel.channel_key = usess->metadata->obj->shm_fd; - lum.u.channel.max_sb_size = usess->metadata->attr.subbuf_size; - lum.u.channel.mmap_len = usess->metadata->obj->memory_map_size; - DBG("Sending metadata channel %d to consumer", lum.u.channel.channel_key); - ret = lttcomm_send_unix_sock(sock, &lum, sizeof(lum)); + consumer_init_channel_comm_msg(&lum, + LTTNG_CONSUMER_ADD_CHANNEL, + usess->metadata->obj->shm_fd, + usess->metadata->attr.subbuf_size, + usess->metadata->obj->memory_map_size, + "metadata"); + + ret = consumer_send_channel(sock, &lum); if (ret < 0) { - PERROR("send consumer channel"); goto error; } + fd = usess->metadata->obj->shm_fd; - ret = lttcomm_send_fds_unix_sock(sock, &fd, 1); + ret = consumer_send_fds(sock, &fd, 1); if (ret < 0) { - PERROR("send consumer metadata channel"); goto error; } - /* Send metadata stream fd */ - lum.cmd_type = LTTNG_CONSUMER_ADD_STREAM; - lum.u.stream.channel_key = usess->metadata->obj->shm_fd; - lum.u.stream.stream_key = usess->metadata->stream_obj->shm_fd; - lum.u.stream.state = LTTNG_CONSUMER_ACTIVE_STREAM; - lum.u.stream.output = DEFAULT_UST_CHANNEL_OUTPUT; - lum.u.stream.mmap_len = usess->metadata->stream_obj->memory_map_size; - lum.u.stream.uid = usess->uid; - lum.u.stream.gid = usess->gid; - strncpy(lum.u.stream.path_name, usess->metadata->pathname, PATH_MAX - 1); - lum.u.stream.path_name[PATH_MAX - 1] = '\0'; - DBG("Sending metadata stream %d to consumer", lum.u.stream.stream_key); - ret = lttcomm_send_unix_sock(sock, &lum, sizeof(lum)); - if (ret < 0) { - PERROR("send consumer metadata stream"); - goto error; + /* Get correct 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, usess->path); + if (ret < 0) { + PERROR("snprintf stream path"); + goto error; + } + pathname = tmp_path; + + /* Create directory */ + ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, + usess->uid, usess->gid); + if (ret < 0) { + if (ret != -EEXIST) { + ERR("Trace directory creation error"); + goto error; + } + } + } else { + ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s", + consumer->subdir, usess->path); + if (ret < 0) { + PERROR("snprintf metadata path"); + goto error; + } + pathname = tmp_path; } + + consumer_init_stream_comm_msg(&lum, + LTTNG_CONSUMER_ADD_STREAM, + usess->metadata->obj->shm_fd, + usess->metadata->stream_obj->shm_fd, + LTTNG_CONSUMER_ACTIVE_STREAM, + DEFAULT_UST_CHANNEL_OUTPUT, + usess->metadata->stream_obj->memory_map_size, + usess->uid, + usess->gid, + consumer->net_seq_index, + 1, /* Flag metadata set */ + "metadata", + pathname); + + /* Send stream and file descriptor */ fds[0] = usess->metadata->stream_obj->shm_fd; fds[1] = usess->metadata->stream_obj->wait_fd; - ret = lttcomm_send_fds_unix_sock(sock, fds, 2); + ret = consumer_send_stream(sock, consumer, &lum, fds, 2); if (ret < 0) { - PERROR("send consumer stream"); goto error; } } @@ -190,7 +230,8 @@ int ust_consumer_send_session(int consumer_fd, struct ust_app_session *usess) continue; } - ret = send_channel_streams(sock, ua_chan, usess->uid, usess->gid); + ret = send_channel_streams(sock, ua_chan, usess->path, usess->uid, + usess->gid, consumer); if (ret < 0) { rcu_read_unlock(); goto error; @@ -205,3 +246,42 @@ int ust_consumer_send_session(int consumer_fd, struct ust_app_session *usess) error: return ret; } + +/* + * Send relayd socket to consumer associated with a session name. + * + * On success return positive value. On error, negative value. + */ +int ust_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); + + msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET; + 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("UST consumer relayd socket sent"); + +error: + return ret; +}