in_systemd: support capturing with namespace

CMakeLists.txt

@@ -1045,6 +1045,24 @@ if(JOURNALD_FOUND)
   if(FLB_HAVE_SYSTEMD_SDBUS)
     FLB_DEFINITION(FLB_HAVE_SYSTEMD_SDBUS)
   endif()
+
+  set(FLB_CMAKE_REQUIRED_INCLUDES ${CMAKE_REQUIRED_INCLUDES})
+  set(FLB_CMAKE_REQUIRED_LIBRARIES ${CMAKE_REQUIRED_LIBRARIES})
+  list(APPEND CMAKE_REQUIRED_INCLUDES ${JOURNALD_INCLUDE_DIR})
+  list(APPEND CMAKE_REQUIRED_LIBRARIES ${JOURNALD_LIBRARIES})
+
+  include(CheckSymbolExists)
+  check_symbol_exists(sd_journal_open_namespace "systemd/sd-journal.h"
+                      FLB_HAVE_SYSTEMD_JOURNAL_NAMESPACE)
+
+  set(CMAKE_REQUIRED_INCLUDES ${FLB_CMAKE_REQUIRED_INCLUDES})
+  set(CMAKE_REQUIRED_LIBRARIES ${FLB_CMAKE_REQUIRED_LIBRARIES})
+  unset(FLB_CMAKE_REQUIRED_INCLUDES)
+  unset(FLB_CMAKE_REQUIRED_LIBRARIES)
+
+  if(FLB_HAVE_SYSTEMD_JOURNAL_NAMESPACE)
+    FLB_DEFINITION(FLB_HAVE_SYSTEMD_JOURNAL_NAMESPACE)
+  endif()
 else()
   FLB_OPTION(FLB_IN_SYSTEMD OFF)
 endif()

plugins/in_systemd/systemd.c

@@ -728,6 +728,11 @@ static struct flb_config_map config_map[] = {
       0, FLB_TRUE, offsetof(struct flb_systemd_config, path),
       "Set the systemd journal path"
     },
+    {
+      FLB_CONFIG_MAP_STR, "namespace", (char *)NULL,
+      0, FLB_TRUE, offsetof(struct flb_systemd_config, journal_namespace),
+      "Set the systemd journal namespace (requires systemd >= 245)"
+    },
     {
       FLB_CONFIG_MAP_INT, "max_fields", FLB_SYSTEMD_MAX_FIELDS,
       0, FLB_TRUE, offsetof(struct flb_systemd_config, max_fields),

plugins/in_systemd/systemd_config.c

@@ -25,6 +25,7 @@
 
 #include <sys/types.h>
 #include <sys/stat.h>
+#include <string.h>
 #include <unistd.h>
 
 #ifdef FLB_HAVE_SQLDB
@@ -74,6 +75,12 @@ struct flb_systemd_config *flb_systemd_config_create(struct flb_input_instance *
         return NULL;
     }
 
+    if (ctx->path && ctx->journal_namespace) {
+        flb_plg_error(ctx->ins, "path and namespace are mutually exclusive");
+        flb_systemd_config_destroy(ctx);
+        return NULL;
+    }
+
     /* Config: path */
     if (ctx->path) {
         ret = stat(ctx->path, &st);
@@ -99,12 +106,32 @@ struct flb_systemd_config *flb_systemd_config_create(struct flb_input_instance *
     if (ctx->path) {
         ret = sd_journal_open_directory(&ctx->j, ctx->path, 0);
     }
+    else if (ctx->journal_namespace) {
+#ifdef FLB_HAVE_SYSTEMD_JOURNAL_NAMESPACE
+        ret = sd_journal_open_namespace(&ctx->j, ctx->journal_namespace,
+                                        SD_JOURNAL_LOCAL_ONLY);
+#else
+        flb_plg_error(ctx->ins,
+                      "namespace requires libsystemd >= 245; this binary was built "
+                      "against an older version; use path instead");
+        flb_systemd_config_destroy(ctx);
+        return NULL;
+#endif
+    }
     else {
         ret = sd_journal_open(&ctx->j, SD_JOURNAL_LOCAL_ONLY);
     }
-    if (ret != 0) {
-        flb_plg_error(ctx->ins, "could not open the Journal");
-        flb_free(ctx);
+    if (ret < 0) {
+        if (ctx->journal_namespace) {
+            flb_plg_error(ctx->ins,
+                          "could not open journal namespace '%s': %s (error=%d)",
+                          ctx->journal_namespace, strerror(-ret), ret);
+        }
+        else {
+            flb_plg_error(ctx->ins, "could not open the Journal: %s (error=%d)",
+                          strerror(-ret), ret);
+        }
+        flb_systemd_config_destroy(ctx);
         return NULL;
     }
     ctx->fd = sd_journal_get_fd(ctx->j);
@@ -254,7 +281,15 @@ struct flb_systemd_config *flb_systemd_config_create(struct flb_input_instance *
                 sd_journal_next_skip(ctx->j, 1);
             }
             else {
-                flb_plg_warn(ctx->ins, "seek_cursor failed");
+                if (ctx->journal_namespace) {
+                    flb_plg_warn(ctx->ins,
+                                 "seek_cursor failed for journal namespace '%s'; "
+                                 "use a distinct database file per namespace",
+                                 ctx->journal_namespace);
+                }
+                else {
+                    flb_plg_warn(ctx->ins, "seek_cursor failed");
+                }
             }
             flb_free(cursor);
         }

plugins/in_systemd/systemd_config.h

@@ -48,6 +48,7 @@ struct flb_systemd_config {
     sd_journal *j;       /* Journal context */
     char *cursor;
     flb_sds_t path;
+    flb_sds_t journal_namespace;
     flb_sds_t filter_type; /* sysytemd filter type: and|or */
     struct mk_list *systemd_filters;
     int pending_records;

tests/runtime/in_systemd.c

@@ -107,8 +107,86 @@ void flb_test_duplicated_keys()
     flb_destroy(ctx);
 }
 
+void flb_test_namespace()
+{
+    int ret;
+    int in_ffd;
+    int out_ffd;
+    flb_ctx_t *ctx;
+
+    ctx = flb_create();
+    TEST_ASSERT(ctx != NULL);
+
+    TEST_CHECK(flb_service_set(ctx,
+                               "flush", "2",
+                               "grace", "1",
+                               "log_level", "error",
+                               NULL) == 0);
+
+    in_ffd = flb_input(ctx, (char *) "systemd", NULL);
+    TEST_ASSERT(in_ffd >= 0);
+    TEST_CHECK(flb_input_set(ctx, in_ffd,
+                             "tag", "test",
+                             "namespace", "fluent-bit-test-nonexistent",
+                             "read_from_tail", "on",
+                             NULL) == 0);
+
+    out_ffd = flb_output(ctx, (char *) "null", NULL);
+    TEST_ASSERT(out_ffd >= 0);
+    TEST_CHECK(flb_output_set(ctx, out_ffd,
+                              "match", "test",
+                              NULL) == 0);
+
+    ret = flb_start(ctx);
+#ifdef FLB_HAVE_SYSTEMD_JOURNAL_NAMESPACE
+    TEST_CHECK(ret == 0);
+    if (ret == 0) {
+        flb_stop(ctx);
+    }
+#else
+    TEST_CHECK(ret == -1);
+#endif
+
+    flb_destroy(ctx);
+}
+
+void flb_test_namespace_path_conflict()
+{
+    int in_ffd;
+    int out_ffd;
+    flb_ctx_t *ctx;
+
+    ctx = flb_create();
+    TEST_ASSERT(ctx != NULL);
+
+    TEST_CHECK(flb_service_set(ctx,
+                               "flush", "2",
+                               "grace", "1",
+                               "log_level", "error",
+                               NULL) == 0);
+
+    in_ffd = flb_input(ctx, (char *) "systemd", NULL);
+    TEST_ASSERT(in_ffd >= 0);
+    TEST_CHECK(flb_input_set(ctx, in_ffd,
+                             "tag", "test",
+                             "path", "/",
+                             "namespace", "fluent-bit-test",
+                             NULL) == 0);
+
+    out_ffd = flb_output(ctx, (char *) "null", NULL);
+    TEST_ASSERT(out_ffd >= 0);
+    TEST_CHECK(flb_output_set(ctx, out_ffd,
+                              "match", "test",
+                              NULL) == 0);
+
+    TEST_CHECK(flb_start(ctx) == -1);
+    flb_destroy(ctx);
+}
+
 /* Test list */
 TEST_LIST = {
     { "duplicated_keys", flb_test_duplicated_keys },
+    { "namespace", flb_test_namespace },
+    { "namespace_path_conflict", flb_test_namespace_path_conflict },
     { NULL, NULL}
 };