out_file: Mitigated the File output cardinality gap

plugins/out_file/file.c

@@ -24,6 +24,7 @@
 #include <fluent-bit/flb_time.h>
 #include <fluent-bit/flb_metrics.h>
 #include <fluent-bit/flb_hash_table.h>
+#include <fluent-bit/flb_lock.h>
 #include <fluent-bit/flb_log_event_decoder.h>
 #include <fluent-bit/flb_record_accessor.h>
 #include <msgpack.h>
@@ -88,12 +89,14 @@ struct flb_file_conf {
     int missing_field_action;
     int limit_reached_action;
     int dynamic_destination;
+    int limit_tag_files;
     int enable_strftime;
     int time_path;
     int time_file;
     struct flb_record_accessor *ra_path;
     struct flb_record_accessor *ra_file;
     struct flb_hash_table *dynamic_files;
+    flb_lock_t destination_lock;
     struct flb_output_instance *ins;
 };
 
@@ -225,6 +228,7 @@ static void file_conf_destroy(struct flb_file_conf *ctx)
     }
 
     file_rotate_destroy(ctx->rotation);
+    flb_lock_destroy(&ctx->destination_lock);
 
     flb_free(ctx);
 }
@@ -384,18 +388,57 @@ static int apply_destination_action(struct flb_file_conf *ctx,
     return -1;
 }
 
+/* Reserve destinations before rotation or filesystem changes, including failed opens. */
+static int reserve_destination(struct flb_file_conf *ctx,
+                               char *output, size_t output_size)
+{
+    int ret;
+    int limit_reached = FLB_FALSE;
+    void *stored_value;
+    size_t stored_size;
+
+    if (ctx->dynamic_files == NULL) {
+        return 0;
+    }
+
+    if (flb_lock_acquire(&ctx->destination_lock, FLB_LOCK_DEFAULT_RETRY_LIMIT,
+                         FLB_LOCK_DEFAULT_RETRY_DELAY) != 0) {
+        flb_plg_error(ctx->ins, "could not acquire destination lock");
+        return -1;
+    }
+
+    ret = flb_hash_table_get(ctx->dynamic_files, output, strlen(output),
+                             &stored_value, &stored_size);
+    if (ret < 0) {
+        if (ctx->dynamic_files->total_count >= ctx->max_dynamic_files) {
+            limit_reached = FLB_TRUE;
+        }
+        else {
+            ret = flb_hash_table_add(ctx->dynamic_files, output, strlen(output), "", 0);
+        }
+    }
+
+    flb_lock_release(&ctx->destination_lock, FLB_LOCK_DEFAULT_RETRY_LIMIT,
+                     FLB_LOCK_DEFAULT_RETRY_DELAY);
+
+    if (limit_reached == FLB_TRUE) {
+        return apply_destination_action(ctx, ctx->limit_reached_action,
+                                        "max_dynamic_files limit reached",
+                                        output, output_size);
+    }
+
+    return ret < 0 ? -1 : 0;
+}
+
 static int resolve_dynamic_destination(struct flb_file_conf *ctx,
                                        const char *tag,
                                        msgpack_object map,
                                        const struct flb_time *timestamp,
                                        char *output,
-                                       size_t output_size,
-                                       int *new_destination)
+                                       size_t output_size)
 {
     int ret;
     char sanitized_tag[PATH_MAX];
-    void *stored_value;
-    size_t stored_size;
     flb_sds_t dynamic_path = NULL;
     flb_sds_t dynamic_file = NULL;
     flb_sds_t timestamp_path = NULL;
@@ -407,7 +450,6 @@ static int resolve_dynamic_destination(struct flb_file_conf *ctx,
     const char *path;
     const char *file;
 
-    *new_destination = FLB_FALSE;
     path = ctx->out_path;
     file = ctx->out_file;
 
@@ -522,24 +564,7 @@ static int resolve_dynamic_destination(struct flb_file_conf *ctx,
         return -1;
     }
 
-    if (ctx->dynamic_files == NULL) {
-        return 0;
-    }
-
-    ret = flb_hash_table_get(ctx->dynamic_files, output, strlen(output),
-                             &stored_value, &stored_size);
-    if (ret >= 0) {
-        return 0;
-    }
-
-    if (ctx->dynamic_files->total_count >= ctx->max_dynamic_files) {
-        return apply_destination_action(ctx, ctx->limit_reached_action,
-                                        "max_dynamic_files limit reached",
-                                        output, output_size);
-    }
-
-    *new_destination = FLB_TRUE;
-    return 0;
+    return reserve_destination(ctx, output, output_size);
 }
 
 
@@ -562,6 +587,10 @@ static int cb_file_init(struct flb_output_instance *ins,
         return -1;
     }
     ctx->ins = ins;
+    if (flb_lock_init(&ctx->destination_lock) != 0) {
+        flb_free(ctx);
+        return -1;
+    }
     ctx->format = FLB_OUT_FILE_FMT_JSON; /* default */
     ctx->delimiter = NULL;
     ctx->label_delimiter = NULL;
@@ -574,6 +603,9 @@ static int cb_file_init(struct flb_output_instance *ins,
         return -1;
     }
 
+    /* Preserve existing tag behavior unless the limit was explicitly configured. */
+    ctx->limit_tag_files = flb_output_get_property("max_dynamic_files", ins) != NULL;
+
     ctx->missing_field_action = parse_dynamic_action(ctx, "on_missing_field",
                                                      ctx->on_missing_field);
     ctx->limit_reached_action = parse_dynamic_action(ctx, "on_limit_reached",
@@ -639,7 +671,9 @@ static int cb_file_init(struct flb_output_instance *ins,
         return -1;
     }
 
-    if (ctx->dynamic_destination == FLB_TRUE && ctx->max_dynamic_files > 0) {
+    if ((ctx->dynamic_destination == FLB_TRUE ||
+         (ctx->limit_tag_files == FLB_TRUE && ctx->out_file == NULL)) &&
+        ctx->max_dynamic_files > 0) {
         table_size = ctx->max_dynamic_files;
         if (table_size > 128) {
             table_size = 128;
@@ -1136,7 +1170,6 @@ static int flush_dynamic_logs(struct flb_event_chunk *event_chunk,
                               struct flb_log_event *log_event)
 {
     int ret;
-    int new_destination;
     FILE *fp;
     long file_pos;
     struct file_rotate_entry *rot_entry = NULL;
@@ -1153,7 +1186,7 @@ static int flush_dynamic_logs(struct flb_event_chunk *event_chunk,
                                               log_event)) == FLB_EVENT_DECODER_SUCCESS) {
         ret = resolve_dynamic_destination(ctx, event_chunk->tag, *log_event->body,
                                           &log_event->timestamp,
-                                          output, output_size, &new_destination);
+                                          output, output_size);
         if (ret == 1) {
             continue;
         }
@@ -1176,15 +1209,6 @@ static int flush_dynamic_logs(struct flb_event_chunk *event_chunk,
             return FLB_ERROR;
         }
 
-        if (new_destination == FLB_TRUE) {
-            ret = flb_hash_table_add(ctx->dynamic_files, output, strlen(output), "", 0);
-            if (ret < 0) {
-                close_output_file(fp, rot_entry);
-                flb_log_event_decoder_destroy(log_decoder);
-                return FLB_ERROR;
-            }
-        }
-
         file_pos = ftell(fp);
         if (ctx->format == FLB_OUT_FILE_FMT_MSGPACK) {
             if (fwrite(log_decoder->record_base, 1, log_decoder->record_length, fp) !=
@@ -1284,6 +1308,14 @@ static void cb_file_flush(struct flb_event_chunk *event_chunk,
         }
     }
 
+    if (ctx->limit_tag_files == FLB_TRUE && ctx->dynamic_destination == FLB_FALSE &&
+        ctx->out_file == NULL) {
+        ret = reserve_destination(ctx, out_file, sizeof(out_file));
+        if (ret != 0) {
+            FLB_OUTPUT_RETURN(ret == 1 ? FLB_OK : FLB_ERROR);
+        }
+    }
+
     if (file_rotate_acquire(ctx->rotation, out_file, &rot_entry) != 0) {
         FLB_OUTPUT_RETURN(FLB_ERROR);
     }
@@ -1465,7 +1497,9 @@ static struct flb_config_map config_map[] = {
     {
      FLB_CONFIG_MAP_INT, "max_dynamic_files", "1024",
      0, FLB_TRUE, offsetof(struct flb_file_conf, max_dynamic_files),
-     "Maximum number of distinct record-accessor destinations. Set to 0 for unlimited"
+     "Maximum number of distinct dynamic destinations. Also limits tag-derived "
+     "filenames when explicitly configured. Failed open attempts also count. "
+     "Set to 0 for unlimited"
     },
 
     {

tests/integration/scenarios/out_file/tests/test_destination_limit.py

@@ -0,0 +1,118 @@
+"""Destination admission is bounded across workers, including failed opens."""
+
+import json
+from pathlib import Path
+import time
+
+import pytest
+import yaml
+
+from utils.fluent_bit_manager import FluentBitManager
+
+
+DESTINATIONS = 40
+LIMIT = 8
+
+
+def create_file_output(tmp_path, mode, limit, action, failed_opens=False):
+    output_dir = tmp_path / "output"
+    output_dir.mkdir()
+    inputs = []
+    for index in range(DESTINATIONS):
+        destination = f"dest{index}"
+        if failed_opens:
+            destination += "x" * 300
+        elif mode == "tag":
+            destination += "/events"
+        inputs.append({
+            "name": "dummy", "tag": destination,
+            "samples": 1 if failed_opens or action != "fallback" else 2,
+            "dummy": json.dumps({"id": index, "dest": destination}),
+        })
+    output = {
+        "name": "file", "match": "*", "path": str(output_dir),
+        "format": "plain", "mkdir": True, "rotate": True, "workers": 4,
+        "on_limit_reached": action, "fallback_path": str(output_dir),
+        "fallback_file": "overflow.log",
+    }
+    if limit is not None:
+        output["max_dynamic_files"] = limit
+    if mode == "accessor":
+        output["file"] = "$dest"
+    config = tmp_path / "fluent-bit.yaml"
+    config.write_text(yaml.safe_dump({
+        "service": {"flush": 0.2, "grace": 1, "http_server": True,
+                    "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}"},
+        "pipeline": {"inputs": inputs, "outputs": [output]},
+    }))
+    manager = FluentBitManager(str(config))
+    return manager, output_dir
+
+
+def wait_until(predicate, manager):
+    deadline = time.monotonic() + 30
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(0.1)
+    pytest.fail(Path(manager.log_file).read_text())
+
+
+def record_count(directory):
+    return sum(path.read_bytes().count(b"\n") for path in directory.rglob("*") if path.is_file())
+
+
+@pytest.mark.parametrize("mode,limit,action", [
+    ("tag", LIMIT, "fallback"),
+    ("tag", LIMIT, "error"),
+    ("tag", LIMIT, "drop"),
+    ("tag", 0, "fallback"),
+    ("tag", None, "fallback"),
+    ("accessor", LIMIT, "fallback"),
+])
+def test_destination_limit_workers(tmp_path, mode, limit, action):
+    manager, output_dir = create_file_output(tmp_path, mode, limit, action)
+    capped = limit == LIMIT
+    destination_count = LIMIT if capped else DESTINATIONS
+    samples = 2 if action == "fallback" else 1
+    expected_records = samples * (destination_count if action != "fallback" else DESTINATIONS)
+    try:
+        manager.start()
+        wait_until(lambda: record_count(output_dir) == expected_records, manager)
+        if capped and action != "fallback":
+            wait_until(lambda: Path(manager.log_file).read_text().count(
+                "max_dynamic_files limit reached") == DESTINATIONS - LIMIT, manager)
+    finally:
+        manager.stop()
+
+    files = [path for path in output_dir.rglob("*") if path.is_file()]
+    destinations = [path for path in files if path.name != "overflow.log"]
+    assert len(destinations) == destination_count
+    # Reusing admitted destinations remains possible after the cap is reached.
+    assert all(len(path.read_text().splitlines()) == samples for path in destinations)
+    if mode == "tag":
+        assert len(list(output_dir.iterdir())) == len(files)
+    overflow = output_dir / "overflow.log"
+    if capped and action == "fallback":
+        assert len(overflow.read_text().splitlines()) == 2 * (DESTINATIONS - LIMIT)
+    else:
+        assert not overflow.exists()
+    if action == "fallback":
+        ids = sorted(json.loads(line)["id"] for path in files for line in path.read_text().splitlines())
+        assert ids == sorted(list(range(DESTINATIONS)) * 2)
+
+
+@pytest.mark.parametrize("mode", ["tag", "accessor"])
+def test_failed_opens_consume_destination_slots(tmp_path, mode):
+    manager, output_dir = create_file_output(tmp_path, mode, LIMIT, "error", failed_opens=True)
+    try:
+        manager.start()
+        wait_until(lambda: Path(manager.log_file).read_text().count("error opening:") +
+                   Path(manager.log_file).read_text().count("max_dynamic_files limit reached")
+                   == DESTINATIONS, manager)
+    finally:
+        manager.stop()
+    log = Path(manager.log_file).read_text()
+    assert log.count("error opening:") == LIMIT
+    assert log.count("max_dynamic_files limit reached") == DESTINATIONS - LIMIT
+    assert not list(output_dir.iterdir())