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 <jdesfossez@efficios.com>
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 1ea14f8ca5991af15f4d8b7d094ab9c85168c163..3ac00cb030bdebc60a3c864a91107873fb683f63 100644 (file)
--- 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;
 }
 
        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.
  *
 /*
  * Check if a stream is ready to be rotated after extracting it.
  *


@@ -4126,6 +4270,68 @@ error:
        return ret;
 }
 
        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)
 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 181c8d8ec2b4e3f5450e5138801740b33a32c813..fb535fba6004d10491c04d0cb10ba95d3c4ec7ae 100644 (file)
--- 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_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,
 };
        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);
 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_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);
 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 7c01bc772a008aa38a05a399c7acadf4849fd0a2..2eb1ebe5d2aa1bb555d10c64e817da53179e84d6 100644 (file)
--- 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;
        }
                }
                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\"",
        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 22d04770ffa1cd9bfc40c41c3cf84831dfbf4884..1b13daec40c813d1745f7f7ec7bed0c11b622596 100644 (file)
--- 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 {
                        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];
                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 6030ca182fd19e5c3078473611ffe1e1ab1c6f8d..6873273a3aa939ac166c97e939745238cce71ee5 100644 (file)
--- 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;
        }
                }
                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",
        case LTTNG_CONSUMER_ROTATE_RENAME:
        {
                DBG("Consumer rename session %" PRIu64 " after rotation",