From: Julien Desfossez Date: Thu, 14 Dec 2017 19:29:52 +0000 (-0500) Subject: Consumer rotate a channel X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=commitdiff_plain;h=b99a8d4211e26a847ca8916884354f159c999ad2 Consumer rotate a channel This command is sent from the session daemon for each channel in a session. When the consumer receives the command, it stores the position at which we need to rotate the stream. If a stream is ready to be rotated, we perform the rotation immediately. Signed-off-by: Julien Desfossez Signed-off-by: Jérémie Galarneau --- diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index 1ea14f8ca..3ac00cb03 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -3913,6 +3913,150 @@ unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos, return start_pos; } +static +int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_active) +{ + int ret = 0; + + switch (consumer_data.type) { + case LTTNG_CONSUMER_KERNEL: + ret = kernctl_buffer_flush(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } + break; + case LTTNG_CONSUMER32_UST: + case LTTNG_CONSUMER64_UST: + lttng_ustctl_flush_buffer(stream, producer_active); + break; + default: + ERR("Unknown consumer_data type"); + abort(); + } + +end: + return ret; +} + +/* + * Sample the rotate position for all the streams of a channel. If a stream + * is already at the rotate position (produced == consumed), we flag it as + * ready for rotation. The rotation of ready streams occurs after we have + * replied to the session daemon that we have finished sampling the positions. + * + * Returns 0 on success, < 0 on error + */ +int lttng_consumer_rotate_channel(uint64_t key, const char *path, + uint64_t relayd_id, uint32_t metadata, uint64_t new_chunk_id, + struct lttng_consumer_local_data *ctx) +{ + int ret; + struct lttng_consumer_channel *channel; + struct lttng_consumer_stream *stream; + struct lttng_ht_iter iter; + struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht; + + DBG("Consumer sample rotate position for channel %" PRIu64, key); + + rcu_read_lock(); + + channel = consumer_find_channel(key); + if (!channel) { + ERR("No channel found for key %" PRIu64, key); + ret = -1; + goto end; + } + + pthread_mutex_lock(&channel->lock); + channel->current_chunk_id = new_chunk_id; + + ret = lttng_strncpy(channel->pathname, path, sizeof(channel->pathname)); + if (ret) { + ERR("Failed to copy new path to channel during channel rotation"); + ret = -1; + goto end_unlock_channel; + } + + if (relayd_id == -1ULL) { + /* + * The domain path (/ust or /kernel) has been created before, we + * now need to create the last part of the path: the application/user + * specific section (uid/1000/64-bit). + */ + ret = utils_mkdir_recursive(channel->pathname, S_IRWXU | S_IRWXG, + channel->uid, channel->gid); + if (ret < 0) { + ERR("Failed to create trace directory at %s during rotation", + channel->pathname); + ret = -1; + goto end_unlock_channel; + } + } + + cds_lfht_for_each_entry_duplicate(ht->ht, + ht->hash_fct(&channel->key, lttng_ht_seed), + ht->match_fct, &channel->key, &iter.iter, + stream, node_channel_id.node) { + unsigned long consumed_pos; + + health_code_update(); + + /* + * Lock stream because we are about to change its state. + */ + pthread_mutex_lock(&stream->lock); + + ret = lttng_strncpy(stream->channel_read_only_attributes.path, + channel->pathname, + sizeof(stream->channel_read_only_attributes.path)); + if (ret) { + ERR("Failed to sample channel path name during channel rotation"); + goto end_unlock_stream; + } + ret = lttng_consumer_sample_snapshot_positions(stream); + if (ret < 0) { + ERR("Failed to sample snapshot position during channel rotation"); + goto end_unlock_stream; + } + + ret = lttng_consumer_get_produced_snapshot(stream, + &stream->rotate_position); + if (ret < 0) { + ERR("Failed to sample produced position during channel rotation"); + goto end_unlock_stream; + } + + lttng_consumer_get_consumed_snapshot(stream, + &consumed_pos); + if (consumed_pos == stream->rotate_position) { + stream->rotate_ready = true; + } + channel->nr_stream_rotate_pending++; + + ret = consumer_flush_buffer(stream, 1); + if (ret < 0) { + ERR("Failed to flush stream %" PRIu64 " during channel rotation", + stream->key); + goto end_unlock_stream; + } + + pthread_mutex_unlock(&stream->lock); + } + pthread_mutex_unlock(&channel->lock); + + ret = 0; + goto end; + +end_unlock_stream: + pthread_mutex_unlock(&stream->lock); +end_unlock_channel: + pthread_mutex_unlock(&channel->lock); +end: + rcu_read_unlock(); + return ret; +} + /* * Check if a stream is ready to be rotated after extracting it. * @@ -4126,6 +4270,68 @@ error: return ret; } +/* + * Rotate all the ready streams now. + * + * This is especially important for low throughput streams that have already + * been consumed, we cannot wait for their next packet to perform the + * rotation. + * + * Returns 0 on success, < 0 on error + */ +int lttng_consumer_rotate_ready_streams(uint64_t key, + struct lttng_consumer_local_data *ctx) +{ + int ret; + struct lttng_consumer_channel *channel; + struct lttng_consumer_stream *stream; + struct lttng_ht_iter iter; + struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht; + + rcu_read_lock(); + + DBG("Consumer rotate ready streams in channel %" PRIu64, key); + + channel = consumer_find_channel(key); + if (!channel) { + ERR("No channel found for key %" PRIu64, key); + ret = -1; + goto end; + } + + cds_lfht_for_each_entry_duplicate(ht->ht, + ht->hash_fct(&channel->key, lttng_ht_seed), + ht->match_fct, &channel->key, &iter.iter, + stream, node_channel_id.node) { + health_code_update(); + + pthread_mutex_lock(&stream->lock); + + if (!stream->rotate_ready) { + pthread_mutex_unlock(&stream->lock); + continue; + } + DBG("Consumer rotate ready stream %" PRIu64, stream->key); + + ret = lttng_consumer_rotate_stream(ctx, stream, NULL); + pthread_mutex_unlock(&stream->lock); + if (ret) { + goto end; + } + + ret = consumer_post_rotation(stream, ctx); + if (ret) { + goto end; + } + } + + ret = 0; + +end: + rcu_read_unlock(); + return ret; +} + static int rotate_rename_local(const char *old_path, const char *new_path, uid_t uid, gid_t gid) diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 181c8d8ec..fb535fba6 100644 --- a/src/common/consumer/consumer.h +++ b/src/common/consumer/consumer.h @@ -63,6 +63,7 @@ enum lttng_consumer_command { LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL, LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE, + LTTNG_CONSUMER_ROTATE_CHANNEL, LTTNG_CONSUMER_ROTATE_RENAME, LTTNG_CONSUMER_MKDIR, }; @@ -817,9 +818,14 @@ void consumer_del_stream_for_data(struct lttng_consumer_stream *stream); void consumer_add_metadata_stream(struct lttng_consumer_stream *stream); void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream); int consumer_create_index_file(struct lttng_consumer_stream *stream); +int lttng_consumer_rotate_channel(uint64_t key, const char *path, + uint64_t relayd_id, uint32_t metadata, + uint64_t new_chunk_id, struct lttng_consumer_local_data *ctx); int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream); int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *stream, bool *rotated); +int lttng_consumer_rotate_ready_streams(uint64_t key, + struct lttng_consumer_local_data *ctx); int lttng_consumer_rotate_rename(const char *current_path, const char *new_path, uid_t uid, gid_t gid, uint64_t relayd_id); void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream); diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 7c01bc772..2eb1ebe5d 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -1119,6 +1119,42 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } + case LTTNG_CONSUMER_ROTATE_CHANNEL: + { + DBG("Consumer rotate channel %" PRIu64, msg.u.rotate_channel.key); + + /* + * Sample the rotate position of all the streams in this channel. + */ + ret = lttng_consumer_rotate_channel(msg.u.rotate_channel.key, + msg.u.rotate_channel.pathname, + msg.u.rotate_channel.relayd_id, + msg.u.rotate_channel.metadata, + msg.u.rotate_channel.new_chunk_id, + ctx); + if (ret < 0) { + ERR("Rotate channel failed"); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + + health_code_update(); + + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + + /* Rotate the streams that are ready right now. */ + ret = lttng_consumer_rotate_ready_streams( + msg.u.rotate_channel.key, ctx); + if (ret < 0) { + ERR("Rotate ready streams failed"); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + + break; + } case LTTNG_CONSUMER_ROTATE_RENAME: { DBG("Consumer rename session %" PRIu64 " after rotation, old path = \"%s\", new path = \"%s\"", diff --git a/src/common/sessiond-comm/sessiond-comm.h b/src/common/sessiond-comm/sessiond-comm.h index 22d04770f..1b13daec4 100644 --- a/src/common/sessiond-comm/sessiond-comm.h +++ b/src/common/sessiond-comm/sessiond-comm.h @@ -542,6 +542,13 @@ struct lttcomm_consumer_msg { struct { uint64_t session_id; } LTTNG_PACKED regenerate_metadata; + struct { + char pathname[PATH_MAX]; + uint32_t metadata; /* This is a metadata channel. */ + uint64_t relayd_id; /* Relayd id if apply. */ + uint64_t key; + uint64_t new_chunk_id; + } LTTNG_PACKED rotate_channel; struct { char old_path[LTTNG_PATH_MAX]; char new_path[LTTNG_PATH_MAX]; diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 6030ca182..6873273a3 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -1960,6 +1960,45 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } + case LTTNG_CONSUMER_ROTATE_CHANNEL: + { + /* + * Sample the rotate position of all the streams in this channel. + */ + ret = lttng_consumer_rotate_channel(msg.u.rotate_channel.key, + msg.u.rotate_channel.pathname, + msg.u.rotate_channel.relayd_id, + msg.u.rotate_channel.metadata, + msg.u.rotate_channel.new_chunk_id, + ctx); + if (ret < 0) { + ERR("Rotate channel failed"); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + + health_code_update(); + + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + + /* + * Rotate the streams that are ready right now. + * FIXME: this is a second consecutive iteration over the + * streams in a channel, there is probably a better way to + * handle this, but it needs to be after the + * consumer_send_status_msg() call. + */ + ret = lttng_consumer_rotate_ready_streams( + msg.u.rotate_channel.key, ctx); + if (ret < 0) { + ERR("Rotate channel failed"); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + break; + } case LTTNG_CONSUMER_ROTATE_RENAME: { DBG("Consumer rename session %" PRIu64 " after rotation",