queue collector exports on disk and retry without a time limit

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

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

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";
                 };

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

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.

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";
+  };
+}