relayd: Add backward compatibility for --group-output-by-session

[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 5f72c32a906f700ec14009766d056ead3e49b375..325be9cf124df7d495fd3a27863195bebb9c55c4 100644 (file)
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -62,22 +62,23 @@
 #include <common/buffer-view.h>
 #include <common/string-utils/format.h>
 
-#include "version.h"
+#include "backward-compatibility-group-by.h"
 #include "cmd.h"
+#include "connection.h"
 #include "ctf-trace.h"
+#include "health-relayd.h"
 #include "index.h"
-#include "utils.h"
-#include "lttng-relayd.h"
 #include "live.h"
-#include "health-relayd.h"
-#include "testpoint.h"
-#include "viewer-stream.h"
+#include "lttng-relayd.h"
 #include "session.h"
+#include "sessiond-trace-chunks.h"
 #include "stream.h"
-#include "connection.h"
-#include "tracefile-array.h"
 #include "tcp_keep_alive.h"
-#include "sessiond-trace-chunks.h"
+#include "testpoint.h"
+#include "tracefile-array.h"
+#include "utils.h"
+#include "version.h"
+#include "viewer-stream.h"
 
 static const char *help_msg =
 #ifdef LTTNG_EMBED_HELP
@@ -98,6 +99,7 @@ enum relay_connection_status {
 /* command line options */
 char *opt_output_path, *opt_working_directory;
 static int opt_daemon, opt_background, opt_print_version;
+enum relay_group_output_by opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_UNKNOWN;
 
 /*
  * We need to wait for listener and live listener threads, as well as
@@ -185,6 +187,8 @@ static struct option long_options[] = {
        { "config", 1, 0, 'f' },
        { "version", 0, 0, 'V' },
        { "working-directory", 1, 0, 'w', },
+       { "group-output-by-session", 0, 0, 's', },
+       { "group-output-by-host", 0, 0, 'p', },
        { NULL, 0, 0, 0, },
 };
 
@@ -337,6 +341,20 @@ static int set_option(int opt, const char *arg, const char *optname)
                        }
                }
                break;
+       case 's':
+               if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
+                       ERR("Cannot set --group-output-by-session, another --group-output-by argument is present");
+                       exit(EXIT_FAILURE);
+               }
+               opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_SESSION;
+               break;
+       case 'p':
+               if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
+                       ERR("Cannot set --group-output-by-host, another --group-output-by argument is present");
+                       exit(EXIT_FAILURE);
+               }
+               opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
+               break;
        default:
                /* Unknown option or other error.
                 * Error is printed by getopt, just return */
@@ -541,6 +559,10 @@ static int set_options(int argc, char **argv)
                }
        }
 
+       if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
+               opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
+       }
+
 exit:
        free(optstring);
        return retval;
@@ -1349,12 +1371,57 @@ static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
                goto send_reply;
        }
 
+       /*
+        * Backward compatibility for --group-output-by-session.
+        * Prior to lttng 2.11, the complete path is passed by the stream.
+        * Starting at 2.11, lttng-relayd uses chunk. When dealing with producer
+        * >=2.11 the chunk is responsible for the output path. When dealing
+        * with producer < 2.11 the chunk output_path is the root output path
+        * and the stream carries the complete path (path_name).
+        * To support --group-output-by-session with older producer (<2.11), we
+        * need to craft the path based on the stream path.
+        */
+       if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_SESSION) {
+               if (conn->minor < 4) {
+                       /*
+                        * From 2.1 to 2.3, the session_name is not passed on
+                        * the RELAYD_CREATE_SESSION command. The session name
+                        * is necessary to detect the presence of a base_path
+                        * inside the stream path. Without it we cannot perform
+                        * a valid group-output-by-session transformation.
+                        */
+                       WARN("Unable to perform a --group-by-session transformation for session %" PRIu64
+                            " for stream with path \"%s\" as it is produced by a peer using a protocol older than v2.4",
+                                       session->id, path_name);
+               } else if (conn->minor >= 4 && conn->minor < 11) {
+                       char *group_by_session_path_name;
+
+                       assert(session->session_name[0] != '\0');
+
+                       group_by_session_path_name =
+                                       backward_compat_group_by_session(
+                                                       path_name,
+                                                       session->session_name);
+                       if (!group_by_session_path_name) {
+                               ERR("Failed to apply group by session to stream of session %" PRIu64,
+                                               session->id);
+                               goto send_reply;
+                       }
+
+                       DBG("Transformed session path from \"%s\" to \"%s\" to honor per-session name grouping",
+                                       path_name, group_by_session_path_name);
+
+                       free(path_name);
+                       path_name = group_by_session_path_name;
+               }
+       }
+
        trace = ctf_trace_get_by_path_or_create(session, path_name);
        if (!trace) {
                goto send_reply;
        }
-       /* This stream here has one reference on the trace. */
 
+       /* This stream here has one reference on the trace. */
        pthread_mutex_lock(&last_relay_stream_id_lock);
        stream_handle = ++last_relay_stream_id;
        pthread_mutex_unlock(&last_relay_stream_id_lock);