in_tail: Implement long line truncation

For avoiding to skip long line consumption,
it sometimes needs to consume until the limit of buffers.
This could provide different approach of mitigation for consuming long lines.

Signed-off-by: Hiroshi Hatake <[email protected]>

Author: cosmo0920

plugins/in_tail/tail.c

@@ -719,6 +719,12 @@ static struct flb_config_map config_map[] = {
      0, FLB_TRUE, offsetof(struct flb_tail_config, skip_empty_lines),
      "Allows to skip empty lines."
     },
+
+    {
+      FLB_CONFIG_MAP_BOOL, "truncate_long_lines", "false",
+      0, FLB_TRUE, offsetof(struct flb_tail_config, truncate_long_lines),
+      "Truncate overlong lines after input encoding to UTF-8"
+    },
 #ifdef __linux__
     {
      FLB_CONFIG_MAP_BOOL, "file_cache_advise", "true",

plugins/in_tail/tail_config.c

@@ -485,6 +485,13 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
                                "multiline_truncated_total",
                                "Total number of truncated occurences for multilines",
                                1, (char *[]) {"name"});
+    ctx->cmt_long_line_truncated = \
+            cmt_counter_create(ins->cmt,
+                               "fluentbit", "input",
+                               "long_line_truncated_total",
+                               "Total number of truncated occurences for long lines",
+                               1, (char *[]) {"name"});
+
     /* OLD metrics */
     flb_metrics_add(FLB_TAIL_METRIC_F_OPENED,
                     "files_opened", ctx->ins->metrics);
@@ -494,6 +501,8 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
                     "files_rotated", ctx->ins->metrics);
     flb_metrics_add(FLB_TAIL_METRIC_M_TRUNCATED,
                     "multiline_truncated", ctx->ins->metrics);
+    flb_metrics_add(FLB_TAIL_METRIC_L_TRUNCATED,
+                    "long_line_truncated", ctx->ins->metrics);
 #endif
 
     return ctx;

plugins/in_tail/tail_config.h

@@ -42,6 +42,7 @@
 #define FLB_TAIL_METRIC_F_CLOSED  101  /* number of closed files  */
 #define FLB_TAIL_METRIC_F_ROTATED 102  /* number of rotated files */
 #define FLB_TAIL_METRIC_M_TRUNCATED 103  /* number of truncated occurrences of multiline */
+#define FLB_TAIL_METRIC_L_TRUNCATED 104  /* number of truncated occurrences of long lines */
 #endif
 
 struct flb_tail_config {
@@ -54,6 +55,7 @@ struct flb_tail_config {
     /* Buffer Config */
     size_t buf_chunk_size;     /* allocation chunks        */
     size_t buf_max_size;       /* max size of a buffer     */
+    int    truncate_long_lines; /* truncate long lines after re-encode */
 
     /* Static files processor */
     size_t static_batch_size;
@@ -169,6 +171,7 @@ struct flb_tail_config {
     struct cmt_counter *cmt_files_closed;
     struct cmt_counter *cmt_files_rotated;
     struct cmt_counter *cmt_multiline_truncated;
+    struct cmt_counter *cmt_long_line_truncated;
 
     /* Hash: hash tables for quick acess to registered files */
     struct flb_hash_table *static_hash;

plugins/in_tail/tail_file.c

@@ -457,6 +457,24 @@ static FLB_INLINE const char *flb_skip_leading_zeros_simd(const char *data, cons
     return data;
 }
 
+/* Return a UTF-8 safe cut position <= max */
+static size_t utf8_safe_truncate_pos(const char *s, size_t len, size_t max)
+{
+    size_t cut = 0;
+
+    cut = (len <= max) ? len : max;
+    if (cut == len) {
+        return cut;
+    }
+
+    /* backtrack over continuation bytes 10xxxxxx */
+    while (cut > 0 && ((unsigned char)s[cut] & 0xC0) == 0x80) {
+        cut--;
+    }
+
+    return cut;
+}
+
 static int process_content(struct flb_tail_file *file, size_t *bytes)
 {
     size_t len;
@@ -481,6 +499,12 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
 #ifdef FLB_HAVE_UNICODE_ENCODER
     size_t decoded_len;
 #endif
+    size_t cut = 0;
+    size_t dec_len = 0;
+    size_t window = 0;
+    int truncation_happened = FLB_FALSE;
+    size_t bytes_override = 0;
+    void *nl = NULL;
 #ifdef FLB_HAVE_METRICS
     uint64_t ts;
     char *name;
@@ -542,6 +566,38 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
         data = (char *)flb_skip_leading_zeros_simd(data, end, &processed_bytes);
     }
 
+    if (ctx->truncate_long_lines == FLB_TRUE) {
+        dec_len = (size_t)(end - data);
+        window = ctx->buf_max_size + 1;
+        if (window > dec_len) {
+            window = dec_len;
+        }
+
+        nl = memchr(data, '\n', window);
+        if (nl == NULL && dec_len > ctx->buf_max_size) {
+            cut = utf8_safe_truncate_pos(data, dec_len, ctx->buf_max_size);
+
+            if (cut > 0) {
+                if (ctx->multiline == FLB_TRUE) {
+                    flb_tail_mult_flush(file, ctx);
+                }
+
+                flb_tail_file_pack_line(NULL, data, cut, file, processed_bytes);
+
+    #ifdef FLB_HAVE_METRICS
+                cmt_counter_inc(ctx->cmt_long_line_truncated,
+                                cfl_time_now(), 1,
+                                (char*[]){ (char*) flb_input_name(ctx->ins) });
+    #endif
+                bytes_override = (original_len > 0) ? original_len : file->buf_len;
+                truncation_happened = FLB_TRUE;
+
+                lines++;
+                goto truncation_end;
+            }
+        }
+    }
+
     while (data < end && (p = memchr(data, '\n', end - data))) {
         len = (p - data);
         crlf = 0;
@@ -700,6 +756,7 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
         file->last_processed_bytes = processed_bytes;
     }
 
+truncation_end:
     if (decoded) {
         flb_free(decoded);
         decoded = NULL;
@@ -709,9 +766,13 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
 
     if (lines > 0) {
         /* Append buffer content to a chunk */
-        if (original_len > 0) {
+        if (truncation_happened) {
+            *bytes = bytes_override;
+        }
+        else if (original_len > 0) {
             *bytes = original_len;
-        } else {
+        }
+        else {
             *bytes = processed_bytes;
         }