X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=4a19c93475802af9e4e5f98a37325b1db207a7c4;hp=8727ce3776a1d85d80a9bdef7ae09110cedb256e;hb=ac2f30af91cd6d9c382698038b525b54845dbe11;hpb=00fb02ace5151a6546f4e97e5439512913a50e68 diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 8727ce377..4a19c9347 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -735,7 +735,8 @@ error: * * The consumer socket lock must be held by the caller. */ -int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd) +int consumer_send_fds(struct consumer_socket *sock, const int *fds, + size_t nb_fd) { int ret; @@ -973,7 +974,7 @@ void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg, */ int consumer_send_stream(struct consumer_socket *sock, struct consumer_output *dst, struct lttcomm_consumer_msg *msg, - int *fds, size_t nb_fd) + const int *fds, size_t nb_fd) { int ret; @@ -1064,37 +1065,70 @@ error: return ret; } -int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock, - int pipe) +static +int consumer_send_pipe(struct consumer_socket *consumer_sock, + enum lttng_consumer_command cmd, int pipe) { int ret; struct lttcomm_consumer_msg msg; + const char *pipe_name; + const char *command_name; + + switch (cmd) { + case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE: + pipe_name = "channel monitor"; + command_name = "SET_CHANNEL_MONITOR_PIPE"; + break; + case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE: + pipe_name = "channel rotate"; + command_name = "SET_CHANNEL_ROTATE_PIPE"; + break; + default: + ERR("Unexpected command received in %s (cmd = %d)", __func__, + (int) cmd); + abort(); + } /* Code flow error. Safety net. */ memset(&msg, 0, sizeof(msg)); - msg.cmd_type = LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE; + msg.cmd_type = cmd; pthread_mutex_lock(consumer_sock->lock); - DBG3("Sending set_channel_monitor_pipe command to consumer"); + DBG3("Sending %s command to consumer", command_name); ret = consumer_send_msg(consumer_sock, &msg); if (ret < 0) { goto error; } - DBG3("Sending channel monitoring pipe %d to consumer on socket %d", + DBG3("Sending %s pipe %d to consumer on socket %d", + pipe_name, pipe, *consumer_sock->fd_ptr); ret = consumer_send_fds(consumer_sock, &pipe, 1); if (ret < 0) { goto error; } - DBG2("Channel monitoring pipe successfully sent"); + DBG2("%s pipe successfully sent", pipe_name); error: pthread_mutex_unlock(consumer_sock->lock); return ret; } +int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock, + int pipe) +{ + return consumer_send_pipe(consumer_sock, + LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe); +} + +int consumer_send_channel_rotate_pipe(struct consumer_socket *consumer_sock, + int pipe) +{ + return consumer_send_pipe(consumer_sock, + LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE, pipe); +} + /* * Set consumer subdirectory using the session name and a generated datetime if * needed. This is appended to the current subdirectory. @@ -1427,6 +1461,16 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, if (ret < 0) { ret = -LTTNG_ERR_NOMEM; goto error; + } else if (ret >= sizeof(msg.u.snapshot_channel.pathname)) { + ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%i bytes required) with path \"%s/%s/%s-%s-%" PRIu64 "%s\"", + sizeof(msg.u.snapshot_channel.pathname), + ret, output->consumer->dst.net.base_dir, + output->consumer->subdir, + output->name, output->datetime, + output->nb_snapshot, + session_path); + ret = -LTTNG_ERR_SNAPSHOT_FAIL; + goto error; } } else { ret = snprintf(msg.u.snapshot_channel.pathname, @@ -1439,7 +1483,16 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, if (ret < 0) { ret = -LTTNG_ERR_NOMEM; goto error; + } else if (ret >= sizeof(msg.u.snapshot_channel.pathname)) { + ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%i bytes required) with path \"%s/%s-%s-%" PRIu64 "%s\"", + sizeof(msg.u.snapshot_channel.pathname), + ret, output->consumer->dst.session_root_path, + output->name, output->datetime, output->nb_snapshot, + session_path); + ret = -LTTNG_ERR_SNAPSHOT_FAIL; + goto error; } + msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL; /* Create directory. Ignore if exist. */ @@ -1580,6 +1633,72 @@ end: return ret; } +/* + * Ask the consumer to rotate a channel. + * domain_path contains "/kernel" for kernel or the complete path for UST + * (ex: /ust/uid/1000/64-bit); + * + * The new_chunk_id is the session->rotate_count that has been incremented + * when the rotation started. On the relay, this allows to keep track in which + * chunk each stream is currently writing to (for the rotate_pending operation). + */ +int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key, + uid_t uid, gid_t gid, struct consumer_output *output, + char *domain_path, bool is_metadata_channel, + uint64_t new_chunk_id, + bool *rotate_pending_relay) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + + DBG("Consumer rotate channel key %" PRIu64, key); + + pthread_mutex_lock(socket->lock); + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL; + msg.u.rotate_channel.key = key; + msg.u.rotate_channel.metadata = !!is_metadata_channel; + msg.u.rotate_channel.new_chunk_id = new_chunk_id; + + if (output->type == CONSUMER_DST_NET) { + msg.u.rotate_channel.relayd_id = output->net_seq_index; + ret = snprintf(msg.u.rotate_channel.pathname, + sizeof(msg.u.rotate_channel.pathname), "%s%s%s", + output->dst.net.base_dir, + output->chunk_path, domain_path); + if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) { + ERR("Failed to format channel path name when asking consumer to rotate channel"); + ret = -1; + goto error; + } + *rotate_pending_relay = true; + } else { + msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL; + ret = snprintf(msg.u.rotate_channel.pathname, + sizeof(msg.u.rotate_channel.pathname), "%s%s%s", + output->dst.session_root_path, + output->chunk_path, domain_path); + if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) { + ERR("Failed to format channel path name when asking consumer to rotate channel"); + ret = -1; + goto error; + } + } + + health_code_update(); + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error; + } + +error: + pthread_mutex_unlock(socket->lock); + health_code_update(); + return ret; +} + int consumer_rotate_rename(struct consumer_socket *socket, uint64_t session_id, const struct consumer_output *output, const char *old_path, const char *new_path, uid_t uid, gid_t gid) @@ -1636,6 +1755,51 @@ error: return ret; } +/* + * Ask the relay if a rotation is still pending. Must be called with the socket + * lock held. + * + * Return 1 if the rotation is still pending, 0 if finished, a negative value + * on error. + */ +int consumer_rotate_pending_relay(struct consumer_socket *socket, + struct consumer_output *output, uint64_t session_id, + uint64_t chunk_id) +{ + int ret; + struct lttcomm_consumer_msg msg; + uint32_t pending = 0; + + assert(socket); + + DBG("Consumer rotate pending on relay for session %" PRIu64 ", chunk id %" PRIu64, + session_id, chunk_id); + assert(output->type == CONSUMER_DST_NET); + + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_ROTATE_PENDING_RELAY; + msg.u.rotate_pending_relay.session_id = session_id; + msg.u.rotate_pending_relay.relayd_id = output->net_seq_index; + msg.u.rotate_pending_relay.chunk_id = chunk_id; + + health_code_update(); + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error; + } + + ret = consumer_socket_recv(socket, &pending, sizeof(pending)); + if (ret < 0) { + goto error; + } + + ret = pending; + +error: + health_code_update(); + return ret; +} + /* * Ask the consumer to create a directory. *