in_systemd: fix endless loop while reading a rotated journal file

It fixes fluent-bit issue fluent#899. Signed-off-by: Alexander Kabakaev <[email protected]>
c445 · Apr 17, 2019 · 7d30108 · 7d30108
1 parent 8cc3a18
commit 7d30108
Show file tree

Hide file tree

Showing 3 changed files with 38 additions and 7 deletions.
diff --git a/plugins/in_systemd/systemd.c b/plugins/in_systemd/systemd.c
@@ -113,6 +113,9 @@ static int in_systemd_collect(struct flb_input_instance *i_ins,
      */
     if (ctx->pending_records == FLB_FALSE) {
         ret = sd_journal_process(ctx->j);
+        if (ret == SD_JOURNAL_INVALIDATE) {
+            flb_debug("[in_systemd] received event on added or removed journal file");
+        }
         if (ret != SD_JOURNAL_APPEND && ret != SD_JOURNAL_NOP) {
             return FLB_SYSTEMD_NONE;
         }
@@ -148,6 +151,8 @@ static int in_systemd_collect(struct flb_input_instance *i_ins,
         ret = sd_journal_get_realtime_usec(ctx->j, &usec);
         if (ret != 0) {
             flb_error("[in_systemd] error reading from systemd journal. sd_journal_get_realtime_usec() return value '%s'", ret);
+            /* It seems the journal file was deleted (rotated). */
+            ret_j = -1;
             break;
         }
         sec = usec / 1000000;
@@ -195,7 +200,7 @@ static int in_systemd_collect(struct flb_input_instance *i_ins,
                entries < ctx->max_fields) {
             key = (char *) data;
             if (ctx->strip_underscores == FLB_TRUE && key[0] == '_') {
-                key++; 
+                key++;
                 length--;
             }
             sep = strchr(key, '=');
@@ -205,12 +210,15 @@ static int in_systemd_collect(struct flb_input_instance *i_ins,
 
             val = sep + 1;
             len = length - (sep - key) - 1;
-            msgpack_pack_str(&mp_pck,  len);
+            msgpack_pack_str(&mp_pck, len);
             msgpack_pack_str_body(&mp_pck, val, len);
 
             entries++;
         }
         rows++;
+        if (entries == ctx->max_fields) {
+            flb_debug("[in_systemd] max number of fields is reached: %i; all other fields are discarded", ctx->max_fields);
+        }
 
         /*
          * The fields were packed, now we need to adjust the msgpack map size
@@ -243,12 +251,10 @@ static int in_systemd_collect(struct flb_input_instance *i_ins,
             msgpack_sbuffer_init(&mp_sbuf);
             strncpy(last_tag, tag, tag_len);
             last_tag_len = tag_len;
-            ret_j = -1;
             break;
         }
 
         if (rows >= ctx->max_entries) {
-            ret_j = -1;
             break;
         }
     }
@@ -275,15 +281,28 @@ static int in_systemd_collect(struct flb_input_instance *i_ins,
     if (ret_j == 0) {
         ctx->pending_records = FLB_FALSE;
         return FLB_SYSTEMD_OK;
-    }
-
+    } else if (ret_j > 0) {
     /*
-     * ret_j == -1, the loop was broken due to some special condition like
+     * ret_j == 1, but the loop was broken due to some special condition like
      * buffer size limit or it reach the max number of rows that it supposed to
      * process on this call. Assume there are pending records.
      */
     ctx->pending_records = FLB_TRUE;
     return FLB_SYSTEMD_MORE;
+    } else {
+        ctx->pending_records = FLB_FALSE;
+        sd_journal_get_cursor(ctx->j, &cursor);
+        tmp = flb_input_get_property("read_from_tail", ctx->i_ins);
+        if (tmp != NULL && flb_utils_bool(tmp)) {
+            sd_journal_seek_tail(ctx->j);
+        } else {
+            sd_journal_seek_head(ctx->j);
+        }
+        sd_journal_get_cursor(ctx->j, &tmp);
+        flb_error("[in_systemd] sd_journal_next() returned error %i; "
+            "journal is re-opened, unread logs are lost", ret_j);
+        return FLB_SYSTEMD_ERROR;
+    }
 }
 
 static int in_systemd_collect_archive(struct flb_input_instance *i_ins,
@@ -332,6 +351,12 @@ static int in_systemd_collect_archive(struct flb_input_instance *i_ins,
         return 0;
     }
 
+    if (ret == FLB_SYSTEMD_ERROR) {
+        flb_error("[in_systemd] error reading from journal file,"
+                  "re-opening journal because it was probably rotated.");
+        flb_systemd_config_destroy(ctx);
+        return -1;
+    }
     /* If FLB_SYSTEMD_NONE or FLB_SYSTEMD_MORE, keep trying */
     write(ctx->ch_manager[1], &val, sizeof(uint64_t));
 

diff --git a/plugins/in_systemd/systemd_config.c b/plugins/in_systemd/systemd_config.c
@@ -39,6 +39,7 @@ struct flb_systemd_config *flb_systemd_config_create(struct flb_input_instance *
     struct flb_config_prop *prop;
     struct flb_systemd_config *ctx;
     int journal_filter_is_and;
+    size_t size;
 
     /* Allocate space for the configuration */
     ctx = flb_calloc(1, sizeof(struct flb_systemd_config));
@@ -200,6 +201,10 @@ struct flb_systemd_config *flb_systemd_config_create(struct flb_input_instance *
         ctx->strip_underscores = FLB_FALSE;
     }
 
+    sd_journal_get_data_threshold(ctx->j, &size);
+    flb_debug("[in_systemd] sd_journal library may truncate values "
+        "to sd_journal_get_data_threshold() bytes: %i", size);
+
     return ctx;
 }
 

diff --git a/plugins/in_systemd/systemd_config.h b/plugins/in_systemd/systemd_config.h
@@ -27,6 +27,7 @@
 #include <systemd/sd-journal.h>
 
 /* return values */
+#define FLB_SYSTEMD_ERROR   -1 /* Systemd journal file read error. */
 #define FLB_SYSTEMD_NONE     0
 #define FLB_SYSTEMD_OK       1
 #define FLB_SYSTEMD_MORE     2