diff --git a/pilot/managers/fluentbit/__init__.py b/pilot/managers/fluentbit/__init__.py index c992cecec..66d82d492 100644 --- a/pilot/managers/fluentbit/__init__.py +++ b/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: diff --git a/pilot/managers/fluentbit/templates/fluent-bit.conf.template b/pilot/managers/fluentbit/templates/fluent-bit.conf.template index 5b33227e4..c219a44b6 100644 --- a/pilot/managers/fluentbit/templates/fluent-bit.conf.template +++ b/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 diff --git a/pilot/managers/fluentbit/templates/pilot.lua b/pilot/managers/fluentbit/templates/pilot.lua index 886486341..46f3bf479 100644 --- a/pilot/managers/fluentbit/templates/pilot.lua +++ b/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,9 +804,7 @@ function normalize(tag, timestamp, record) end - -- Build Datum request - - record["body"] = build_body( + return 1, timestamp, datum_line( ts, level, message, @@ -817,12 +812,4 @@ function normalize(tag, timestamp, record) source, attributes ) - - - record["headers"] = { - ["Content-Type"] = "application/json" - } - - - return 1, timestamp, record -end \ No newline at end of file +end diff --git a/tests/pilot/managers/test_fluentbit.py b/tests/pilot/managers/test_fluentbit.py index 4c5b92ee6..26f88ad38 100644 --- a/tests/pilot/managers/test_fluentbit.py +++ b/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()