fix: Update fluent-bit to cap pushes and refused request log

pilot/managers/fluentbit/__init__.py

@@ -44,6 +44,10 @@
 ExecStart={fluent_bit} -c {conf_dir}/fluent-bit.conf
 Restart=on-failure
 RestartSec=5s
+# Fluent Bit logs every refused request. Cap what reaches the journal, and
+# through it syslog, so a refusing endpoint cannot fill the disk.
+LogRateLimitIntervalSec=30s
+LogRateLimitBurst=200
 
 [Install]
 WantedBy=default.target
@@ -95,15 +99,16 @@ def is_running(self) -> bool:
         return result.returncode == 0
 
     def install(self, config: TelemetryConfig) -> None:
+        """Write the config and unit. A running shipper is restarted, as `enable --now` would not."""
         self._write_configs(config)
-        if user_service_installed(self.unit_name):
-            self._restart_service()
-            return
+        was_installed = user_service_installed(self.unit_name)
         install_user_service(
             unit_dir=self.conf_dir,
             unit_name=self.unit_name,
             unit_text=self._render_unit(),
         )
+        if was_installed:
+            self._restart_service()
 
     def remove(self) -> None:
         env = self._systemctl_env()
@@ -135,10 +140,11 @@ def _render_files(self, config: TelemetryConfig) -> dict[str, str]:
             "parsers.conf": _PARSERS_CONF,
             "pilot.lua": _PILOT_LUA,
             "token.env": f"DATUM_LOG_TOKEN={config.token}",
+            self.unit_name: self._render_unit(),
         }
 
     def _read_files(self) -> dict[str, str]:
-        names = ("fluent-bit.conf", "parsers.conf", "pilot.lua", "token.env")
+        names = ("fluent-bit.conf", "parsers.conf", "pilot.lua", "token.env", self.unit_name)
         return {name: path.read_text() for name in names if (path := self.conf_dir / name).exists()}
 
     def _render_unit(self) -> str:

pilot/managers/fluentbit/templates/fluent-bit.conf.template

@@ -1,5 +1,11 @@
 [SERVICE]
-    Flush                     1
+    # Datum allows 12 requests a minute per token. Every line reaches the
+    # output through one emitter (see rewrite_tag below), so a flush sends one
+    # chunk: 2 a minute, each sent at most 1 + Retry_Limit (3) times, is at
+    # most 8 a minute even while Datum refuses every one.
+    Flush                     30
+    scheduler.base            10
+    scheduler.cap             60
     Log_Level                 info
     Parsers_File              {{ conf_dir }}/parsers.conf
 
@@ -192,13 +198,30 @@
     Protected_Mode  On
 
 
+# ============================================================
+# Funnel every input into one stream
+#
+# Each input keeps its own chunks, and the output sends each
+# chunk as its own request. Re-emitting every normalized line
+# under one tag gives the output a single chunk stream.
+# ============================================================
+
+[FILTER]
+    Name                  rewrite_tag
+    Match                 pilot.*
+    Rule                  $product ^pilot$ datum.logs false
+    Emitter_Name          datum
+    Emitter_Storage.type  filesystem
+    Emitter_Mem_Buf_Limit 10M
+
+
 # ============================================================
 # Datum
 # ============================================================
 
 [OUTPUT]
     Name          http
-    Match         pilot.*
+    Match         datum.logs
 
     Host          {{ host }}
     Port          {{ port }}
@@ -210,8 +233,12 @@
 
     Header        Authorization Bearer ${DATUM_LOG_TOKEN}
 
-    Body_Key      body
-    Headers_Key   headers
+    # One JSON array of Datum log lines per chunk.
+    Format        json
+    json_date_key false
 
-    Workers       2
-    Retry_Limit   False
+    Workers       1
+    # A refused chunk is dropped after a few tries rather than retried
+    # forever, and the backlog on disk is bounded.
+    Retry_Limit   3
+    storage.total_limit_size 100M

pilot/managers/fluentbit/templates/pilot.lua

@@ -1,43 +1,5 @@
 -- Pilot -> Datum normalizer
 
-local function json_escape(value)
-    if value == nil then
-        return ""
-    end
-
-    value = tostring(value)
-
-    value = string.gsub(value, "\\", "\\\\")
-    value = string.gsub(value, "\"", "\\\"")
-    value = string.gsub(value, "\b", "\\b")
-    value = string.gsub(value, "\f", "\\f")
-    value = string.gsub(value, "\n", "\\n")
-    value = string.gsub(value, "\r", "\\r")
-    value = string.gsub(value, "\t", "\\t")
-
-    return value
-end
-
-
-local function json_string(value)
-    return "\"" .. json_escape(value) .. "\""
-end
-
-
-local function serialize_attributes(attributes)
-    local parts = {}
-
-    for key, value in pairs(attributes) do
-        table.insert(
-            parts,
-            json_string(key) .. ":" .. json_string(value)
-        )
-    end
-
-    return "{" .. table.concat(parts, ",") .. "}"
-end
-
-
 -- Extract filename from Path_Key
 
 local function get_source(filename)
@@ -330,29 +292,66 @@ local function add_attribute(attributes, key, value)
 end
 
 
--- Datum request body
-
-local function build_body(
-    ts,
-    level,
-    message,
-    service,
-    source,
-    attributes
-)
-
-    local line =
-        "{"
-        .. "\"ts\":" .. json_string(ts)
-        .. ",\"level\":" .. json_string(level)
-        .. ",\"message\":" .. json_string(message)
-        .. ",\"service\":" .. json_string(service)
-        .. ",\"source\":" .. json_string(source)
-        .. ",\"product\":\"pilot\""
-        .. ",\"attributes\":" .. serialize_attributes(attributes)
-        .. "}"
-
-    return "{\"lines\":[" .. line .. "]}"
+-- Datum caps one message at 8 KiB. Cut on a byte boundary, then drop the
+-- last character only if the cut left it short of the bytes its lead byte
+-- announces, so the line stays valid UTF-8 without losing a whole one.
+
+local MAX_MESSAGE = 8192
+
+local function utf8_width(byte)
+    if byte >= 240 then
+        return 4
+    elseif byte >= 224 then
+        return 3
+    elseif byte >= 192 then
+        return 2
+    end
+    return 1
+end
+
+local function cap_message(message)
+    if #message <= MAX_MESSAGE then
+        return message
+    end
+
+    local cut = string.sub(message, 1, MAX_MESSAGE)
+
+    local lead = MAX_MESSAGE
+    while lead > MAX_MESSAGE - 3 do
+        local byte = string.byte(cut, lead)
+        if byte < 128 or byte >= 192 then
+            break
+        end
+        lead = lead - 1
+    end
+
+    if MAX_MESSAGE - lead + 1 < utf8_width(string.byte(cut, lead)) then
+        return string.sub(cut, 1, lead - 1)
+    end
+
+    return cut
+end
+
+
+-- Datum log line. Fluent Bit's http output batches these into one JSON
+-- array per chunk, which Datum accepts as a batch. An empty Lua table
+-- encodes as an array, so attributes are left out when there are none.
+
+local function datum_line(ts, level, message, service, source, attributes)
+    local line = {
+        ts = ts,
+        level = level,
+        message = cap_message(message),
+        service = service,
+        source = source,
+        product = "pilot",
+    }
+
+    if next(attributes) ~= nil then
+        line["attributes"] = attributes
+    end
+
+    return line
 end
 
 
@@ -470,9 +469,7 @@ function normalize(tag, timestamp, record)
                 and key ~= "product"
                 and key ~= "service"
                 and key ~= "source"
-                and key ~= "attributes"
-                and key ~= "body"
-                and key ~= "headers" then
+                and key ~= "attributes" then
 
                 add_attribute(
                     attributes,
@@ -807,22 +804,12 @@ function normalize(tag, timestamp, record)
     end
 
 
-    -- Build Datum request
-
-    record["body"] = build_body(
+    return 1, timestamp, datum_line(
         ts,
         level,
         message,
         service,
         source,
         attributes
     )
-
-
-    record["headers"] = {
-        ["Content-Type"] = "application/json"
-    }
-
-
-    return 1, timestamp, record
-end
+end

tests/pilot/managers/test_fluentbit.py

@@ -48,7 +48,7 @@ def test_logs_install_writes_config_parsers_lua_and_unit(tmp_path: Path) -> None
 
     lua = (tmp_path / "system" / "fluent-bit" / "pilot.lua").read_text()
     assert "function normalize" in lua
-    assert "\\\"product\\\":\\\"pilot\\\"" in lua
+    assert 'product = "pilot"' in lua
 
     token_file = tmp_path / "system" / "fluent-bit" / "token.env"
     assert "DATUM_LOG_TOKEN=secret" in token_file.read_text()
@@ -57,6 +57,44 @@ def test_logs_install_writes_config_parsers_lua_and_unit(tmp_path: Path) -> None
     unit_text = install.call_args.kwargs["unit_text"]
     assert "fluent-bit" in unit_text
     assert str(tmp_path / "system" / "fluent-bit" / "fluent-bit.conf") in unit_text
+    assert "LogRateLimitIntervalSec=30s" in unit_text
+
+
+def test_logs_output_batches_within_datum_limits(tmp_path: Path) -> None:
+    """One JSON array per chunk, flushed rarely enough for Datum's 12 requests a minute."""
+    configurator = _configurator(tmp_path)
+    config = TelemetryConfig(endpoint="https://datum.internal", token="secret")
+
+    with (
+        patch("pilot.managers.fluentbit.cli_root", return_value=tmp_path),
+        patch("pilot.managers.fluentbit.shutil.which", return_value="/usr/bin/fluent-bit"),
+    ):
+        conf = configurator._render_files(config)["fluent-bit.conf"]
+
+    assert "Flush                     30" in conf
+    assert "Format        json" in conf
+    assert "json_date_key false" in conf
+    assert "Retry_Limit   3" in conf
+    assert "storage.total_limit_size 100M" in conf
+    assert "Body_Key" not in conf
+    assert "Retry_Limit   False" not in conf
+
+
+def test_logs_output_reads_one_stream(tmp_path: Path) -> None:
+    """Each input chunks on its own, so every line is funneled through one emitter the output reads."""
+    configurator = _configurator(tmp_path)
+    config = TelemetryConfig(endpoint="https://datum.internal", token="secret")
+
+    with (
+        patch("pilot.managers.fluentbit.cli_root", return_value=tmp_path),
+        patch("pilot.managers.fluentbit.shutil.which", return_value="/usr/bin/fluent-bit"),
+    ):
+        conf = configurator._render_files(config)["fluent-bit.conf"]
+
+    output = conf[conf.index("[OUTPUT]") :]
+    assert "Match         datum.logs" in output
+    assert "Rule                  $product ^pilot$ datum.logs false" in conf
+    assert conf.index("Name            lua") < conf.index("Name                  rewrite_tag")
 
 
 def test_python_input_tails_bench_dir_only(tmp_path: Path) -> None:
@@ -97,6 +135,7 @@ def test_logs_install_disables_tls_for_http_endpoint(tmp_path: Path) -> None:
 
 
 def test_logs_install_restarts_when_already_installed(tmp_path: Path) -> None:
+    """The unit is rewritten too, so a changed unit reaches hosts that already run the shipper."""
     configurator = _configurator(tmp_path)
     config = TelemetryConfig(endpoint="https://datum.internal", token="test-token")
 
@@ -109,7 +148,7 @@ def test_logs_install_restarts_when_already_installed(tmp_path: Path) -> None:
     ):
         configurator.install(config)
 
-    install.assert_not_called()
+    install.assert_called_once()
     restart_calls = [call for call in run.call_args_list if "restart" in call.args[0]]
     assert len(restart_calls) == 1
 
@@ -297,3 +336,14 @@ def test_logs_sync_restarts_the_shipper_for_a_new_token(tmp_path: Path) -> None:
     install = _sync(tmp_path, config, binary="/usr/bin/fluent-bit", active=True)
 
     install.assert_called_once()
+
+
+def test_logs_sync_reinstalls_the_shipper_for_a_changed_unit(tmp_path: Path) -> None:
+    config = TelemetryConfig(endpoint="https://datum.internal", token="jwt")
+    _sync(tmp_path, config, binary="/usr/bin/fluent-bit", active=False)
+    unit = tmp_path / "system" / "fluent-bit" / LogsConfigurator.unit_name
+    unit.write_text(unit.read_text().replace("LogRateLimitBurst=200\n", ""))
+
+    install = _sync(tmp_path, config, binary="/usr/bin/fluent-bit", active=True)
+
+    install.assert_called_once()