relayd: implement support for clear feature

Implement support for the clear feature on relayd receiving end.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Change-Id: Ice56dea0b7f59a4881f28fef84bcb0406e0afca6
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>

diff --git a/src/bin/lttng-relayd/stream.c b/src/bin/lttng-relayd/stream.c
index 62b426360ac1830def68230f3d40d860d968520c..4238d3b53ff217e05613e5ed7aab0aef72f11921 100644 (file)
--- a/src/bin/lttng-relayd/stream.c
+++ b/src/bin/lttng-relayd/stream.c
@@ -72,6 +72,11 @@ end:
 static void stream_complete_rotation(struct relay_stream *stream)
 {
        DBG("Rotation completed for stream %" PRIu64, stream->stream_handle);
+       if (stream->ongoing_rotation.value.next_trace_chunk) {
+               tracefile_array_reset(stream->tfa);
+               tracefile_array_commit_seq(stream->tfa,
+                               stream->index_received_seqcount);
+       }
        lttng_trace_chunk_put(stream->trace_chunk);
        stream->trace_chunk = stream->ongoing_rotation.value.next_trace_chunk;
        stream->ongoing_rotation = (typeof(stream->ongoing_rotation)) {};
@@ -525,8 +530,10 @@ static int try_rotate_stream_index(struct relay_stream *stream)
                                stream->ongoing_rotation.value.packet_seq_num);
                DBG("Rotating stream %" PRIu64 " index file",
                                stream->stream_handle);
-               ret = create_index_file(stream,
-                               stream->ongoing_rotation.value.next_trace_chunk);
+               if (stream->index_file) {
+                       lttng_index_file_put(stream->index_file);
+                       stream->index_file = NULL;
+               }
                stream->ongoing_rotation.value.index_rotated = true;
 
                /*
@@ -868,6 +875,12 @@ int stream_set_pending_rotation(struct relay_stream *stream,
                 * A metadata stream has no index; consider it already rotated.
                 */
                stream->ongoing_rotation.value.index_rotated = true;
+               if (next_trace_chunk) {
+                       /*
+                        * The metadata will be received again in the new chunk.
+                        */
+                       stream->metadata_received = 0;
+               }
                ret = stream_rotate_data_file(stream);
        } else {
                ret = try_rotate_stream_index(stream);