relayd: add payload logging to session rotation commands

[lttng-tools.git] / src / bin / lttng-relayd / main.c

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index c84ebaac1c2c66b3d6f282aa6e1d69d161e16f23..cbe1a607c36add6e8b320f8f57c13a816a9b3dc5 100644 (file)
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1746,18 +1746,27 @@ static
 int try_rotate_stream(struct relay_stream *stream)
 {
        int ret = 0;
+       uint64_t trace_seq;
 
        /* No rotation expected. */
        if (stream->rotate_at_seq_num == -1ULL) {
                goto end;
        }
 
-       if (stream->prev_seq < stream->rotate_at_seq_num ||
-                       stream->prev_seq == -1ULL) {
-               DBG("Stream %" PRIu64 " no yet ready for rotation",
-                               stream->stream_handle);
+       trace_seq = min(stream->prev_data_seq, stream->prev_index_seq);
+       if (stream->prev_data_seq == -1ULL || stream->prev_index_seq == -1ULL ||
+                       trace_seq < stream->rotate_at_seq_num) {
+               DBG("Stream %" PRIu64 " not yet ready for rotation (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+                               stream->stream_handle,
+                               stream->rotate_at_seq_num,
+                               stream->prev_data_seq,
+                               stream->prev_index_seq);
                goto end;
-       } else if (stream->prev_seq > stream->rotate_at_seq_num) {
+       } else if (stream->prev_data_seq > stream->rotate_at_seq_num) {
+               /*
+                * prev_data_seq is checked here since indexes and rotation
+                * commands are serialized with respect to each other.
+                */
                DBG("Rotation after too much data has been written in tracefile "
                                "for stream %" PRIu64 ", need to truncate before "
                                "rotating", stream->stream_handle);
@@ -1767,7 +1776,20 @@ int try_rotate_stream(struct relay_stream *stream)
                        goto end;
                }
        } else {
-               /* stream->prev_seq == stream->rotate_at_seq_num */
+               if (trace_seq != stream->rotate_at_seq_num) {
+                       /*
+                        * Unexpected, protocol error/bug.
+                        * It could mean that we received a rotation position
+                        * that is in the past.
+                        */
+                       ERR("Stream %" PRIu64 " is in an inconsistent state (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+                               stream->stream_handle,
+                               stream->rotate_at_seq_num,
+                               stream->prev_data_seq,
+                               stream->prev_index_seq);
+                       ret = -1;
+                       goto end;
+               }
                DBG("Stream %" PRIu64 " ready for rotation",
                                stream->stream_handle);
                ret = do_rotate_stream(stream);
@@ -1969,14 +1991,14 @@ static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
                 * Ensure that both the index and stream data have been
                 * flushed up to the requested point.
                 */
-               stream_seq = min(stream->prev_seq, stream->prev_index_seq);
+               stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
        } else {
-               stream_seq = stream->prev_seq;
+               stream_seq = stream->prev_data_seq;
        }
-       DBG("Data pending for stream id %" PRIu64 ": prev_seq %" PRIu64
+       DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
                        ", prev_index_seq %" PRIu64
                        ", and last_seq %" PRIu64, msg.stream_id,
-                       stream->prev_seq, stream->prev_index_seq,
+                       stream->prev_data_seq, stream->prev_index_seq,
                        msg.last_net_seq_num);
 
        /* Avoid wrapping issue */
@@ -2207,9 +2229,9 @@ static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
                                 * Ensure that both the index and stream data have been
                                 * flushed up to the requested point.
                                 */
-                               stream_seq = min(stream->prev_seq, stream->prev_index_seq);
+                               stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
                        } else {
-                               stream_seq = stream->prev_seq;
+                               stream_seq = stream->prev_data_seq;
                        }
                        if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
                                is_data_inflight = 1;
@@ -2628,6 +2650,7 @@ static int relay_mkdir(const struct lttcomm_relayd_hdr *recv_hdr,
                goto end;
        }
 
+       DBG("MKDIR command has path \"%s\", changed to \"%s\"", path_view.data, path);
        ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, -1, -1);
        if (ret < 0) {
                ERR("relay creating output directory");
@@ -2751,6 +2774,8 @@ static int relay_rotate_rename(const struct lttcomm_relayd_hdr *recv_hdr,
                goto end;
        }
 
+       DBG("ROTATE_RENAME command has argument old path = \"%s\", new_path = \"%s\"",
+                       old_path_view.data, new_path_view.data);
        complete_old_path = create_output_path(old_path_view.data);
        if (!complete_old_path) {
                ERR("Failed to build old output path in rotate_rename command");
@@ -2764,6 +2789,8 @@ static int relay_rotate_rename(const struct lttcomm_relayd_hdr *recv_hdr,
                ret = -1;
                goto end;
        }
+       DBG("Expanded ROTATE_RENAME arguments to old path = \"%s\", new_path = \"%s\"",
+                       complete_old_path, complete_new_path);
 
        ret = utils_mkdir_recursive(complete_new_path, S_IRWXU | S_IRWXG,
                        -1, -1);
@@ -3505,7 +3532,7 @@ static enum relay_connection_status relay_process_data_receive_payload(
        stream->tracefile_size_current += state->header.data_size +
                        state->header.padding_size;
 
-       if (stream->prev_seq == -1ULL) {
+       if (stream->prev_data_seq == -1ULL) {
                new_stream = true;
        }
        if (index_flushed) {
@@ -3514,7 +3541,7 @@ static enum relay_connection_status relay_process_data_receive_payload(
                stream->prev_index_seq = state->header.net_seq_num;
        }
 
-       stream->prev_seq = state->header.net_seq_num;
+       stream->prev_data_seq = state->header.net_seq_num;
 
        /*
         * Resetting the protocol state (to RECEIVE_HEADER) will trash the