From eac7f2a7c198ff5c56f7950fee029b187c03a0fa Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 26 Oct 2026 22:28:23 +0100 Subject: [PATCH] queue collector exports on disk and retry without a time limit --- docs/scheduler/observability.md | 12 ++++++++++++ nix/host-modules/otel.nix | 1 + nix/host-modules/swarm-otel.nix | 2 ++ nix/module-eval/hive-otel.nix | 17 +++++++++++++++++ nix/module-eval/swarm-otel-core.nix | 26 ++++++++++++++++++++++++++ nix/otel-export-queue.nix | 27 +++++++++++++++++++++++++++ 6 files changed, 85 insertions(+) create mode 100644 nix/otel-export-queue.nix diff --git a/docs/scheduler/observability.md b/docs/scheduler/observability.md index a58d63a9..cbe8150b 100644 --- a/docs/scheduler/observability.md +++ b/docs/scheduler/observability.md @@ -64,6 +64,18 @@ ordinary DNS. runs on the same host as the agents and restarts on failure. Telemetry isn't the control plane, so degraded telemetry isn't degraded operation. +Both collector tiers queue outgoing batches on disk (`file_storage`) and retry +a failing batch until the receiving end accepts it, so a restart or a short +outage of the next hop delays telemetry instead of dropping it. The queue holds +at most 5000 batches. Two limits remain: + +- A full queue makes the exporter refuse new batches and log the drop. The + journald receiver has already moved its cursor past a refused batch, so an + outage longer than the queue can hold still loses data, later and a bounded + amount. +- Retry has no time limit, so the exporter keeps resending a batch that the + store rejects with a retryable error, and enough of those fill the queue. + ### Why two tiers **Agents export only to the hive tier.** Exporting straight to `upstream.endpoint` would mean diff --git a/nix/host-modules/otel.nix b/nix/host-modules/otel.nix index ddec8d59..6d57db0f 100644 --- a/nix/host-modules/otel.nix +++ b/nix/host-modules/otel.nix @@ -526,6 +526,7 @@ in # samples get labelled as. endpoint = "https://${swarmOtelCfg.domain}" + lib.optionalString senderAuth "/${toString hiveName}"; } + // import ../otel-export-queue.nix // lib.optionalAttrs senderAuth { auth.authenticator = authName; }; # ⚠️ An extension not listed here is INERT: the collector starts diff --git a/nix/host-modules/swarm-otel.nix b/nix/host-modules/swarm-otel.nix index b3affaf1..c4beb06d 100644 --- a/nix/host-modules/swarm-otel.nix +++ b/nix/host-modules/swarm-otel.nix @@ -1264,6 +1264,7 @@ in ${if upstream.protocol == "grpc" then "otlp" else "otlphttp"} = { endpoint = upstream.endpoint; } + // import ../otel-export-queue.nix // lib.optionalAttrs (upstream.headersCredential != null) { # Interpolated by the collector from its environment at # runtime, never by nix: `EnvironmentFile` below is what @@ -1296,6 +1297,7 @@ in logs_endpoint = logsPushUrl + "?_msg_field=MESSAGE&_stream_fields=_HOSTNAME,_MACHINE_ID,_SYSTEMD_UNIT"; } + // import ../otel-export-queue.nix // lib.optionalAttrs haveCollectorSecret { auth.authenticator = pushAuthenticator "victorialogs"; }; diff --git a/nix/module-eval/hive-otel.nix b/nix/module-eval/hive-otel.nix index c858da87..08a6e7be 100644 --- a/nix/module-eval/hive-otel.nix +++ b/nix/module-eval/hive-otel.nix @@ -404,6 +404,23 @@ let && hiveOtelSettings.extensions."oauth2client/swarm".endpoint_params.audience == "hive-h1" && hiveOtelSettings.exporters."otlphttp/swarm".endpoint == "https://${otel.domain}/h1"; } + { + # Same reason as the swarm tier's case in ./swarm-otel-core.nix: the + # queue in front of the swarm exporter has to be on `file_storage` or a + # restart drops what the journald receiver already counted as consumed. + name = "the hive collector queues on disk and retries without a time limit"; + ok = + let + e = hiveOtelSettings.exporters."otlphttp/swarm"; + in + e.sending_queue.enabled + && e.sending_queue.storage == "file_storage" + && e.sending_queue.queue_size > 0 + && e.retry_on_failure.max_elapsed_time == "0s" + && e.endpoint == "https://${hiveOtel.services.hyperhive.swarm.otel.domain}/h1" + && hiveOtelSettings.extensions ? file_storage + && builtins.elem "file_storage" hiveOtelSettings.service.extensions; + } ]; in runGroup "hive-otel" cases diff --git a/nix/module-eval/swarm-otel-core.nix b/nix/module-eval/swarm-otel-core.nix index 697950bb..5662d5a3 100644 --- a/nix/module-eval/swarm-otel-core.nix +++ b/nix/module-eval/swarm-otel-core.nix @@ -338,6 +338,32 @@ let && lib.hasInfix "Exporting failed" (lib.head f.logs.log_record) && lib.hasInfix "otlphttp/victorialogs" (lib.head f.logs.log_record); } + { + # The journald receiver saves its cursor before an entry reaches the + # exporter, so an entry in an in-memory queue when the store is down or + # the collector restarts is gone for good. The queue has to sit on the + # `file_storage` extension, and the retry has to outlast a switch. + # Presence controls: the extension is defined and started, the queue is + # bounded, and the exporters carry their endpoints. + name = "the log exporters queue on disk and retry without a time limit"; + ok = + let + s = otelSettings otelNoStores; + queued = + e: + e.sending_queue.enabled + && e.sending_queue.storage == "file_storage" + && e.sending_queue.queue_size > 0 + && e.retry_on_failure.enabled + && e.retry_on_failure.max_elapsed_time == "0s"; + in + queued s.exporters."otlphttp/victorialogs" + && queued s.exporters.otlphttp + && s.exporters."otlphttp/victorialogs" ? logs_endpoint + && s.exporters.otlphttp.endpoint == "https://upstream.example/otel" + && s.extensions ? file_storage + && builtins.elem "file_storage" s.service.extensions; + } { # Presence control: the default sends both signals to the upstream, so # the negative arms below cannot pass on a misspelled exporter name. diff --git a/nix/otel-export-queue.nix b/nix/otel-export-queue.nix new file mode 100644 index 00000000..e45cd15e --- /dev/null +++ b/nix/otel-export-queue.nix @@ -0,0 +1,27 @@ +# Export settings every forwarding exporter in this tree attaches, imported by +# both collector tiers (`host-modules/otel.nix`, `host-modules/swarm-otel.nix`). +# +# `sending_queue.storage` puts the queue on the `file_storage` extension, so +# batches waiting for an exporter survive a collector restart. The journald +# receiver saves its cursor before it hands an entry on: an entry sitting in an +# in-memory queue when the exporter's target is down or the collector restarts +# counts as consumed and is not read again. +# +# `queue_size` counts requests (batches), not records. When the queue is full +# the exporter refuses new data and logs the drop; the oldest queued batches +# are kept. +# +# `max_elapsed_time = "0s"` retries a failing batch until it is accepted. +# The queue bound is what limits the disk used; the default of five minutes +# gives up on a backlog that a store restart outlasts. +{ + sending_queue = { + enabled = true; + storage = "file_storage"; + queue_size = 5000; + }; + retry_on_failure = { + enabled = true; + max_elapsed_time = "0s"; + }; +}