otel: collectors drop their own export-failure lines instead of shipping them

docs/4-implementation/swarm/telemetry.md

@@ -309,6 +309,17 @@ every other exporter still arrive. The container mounts the host's journal
 read-only at `/run/host-journal`, not over its own
 `/var/log/journal`.
 
+Every collector that reads a journal drops its own export failures before it
+exports: the hive collector, the agent and service-container forwarders, and
+the swarm collector's host-journal pipeline. A `filter` processor
+(`nix/collector-own-export-failures.nix`) matches records from the unit
+`opentelemetry-collector.service` whose message contains `Exporting failed` or
+`sending queue is full`. Without it, an ingest outage makes each collector log
+one line per failed batch, its receiver reads the line back and pushes it, and
+the pushes add to the outage. The collector's startup, configuration and
+receiver errors still ship. The failure itself stays visible in the collector's
+`otelcol_exporter_send_failed_*` metrics.
+
 **Both store exporters are unconditional**, and `deploy.victoriametrics.enable`
 doesn't gate them: that option says this host _runs_ the store, while the swarm
 has one either way, reached by its swarm name through the gateway.

nix/agent-modules/otel.nix

@@ -290,9 +290,14 @@ in
           }
           // lib.optionalAttrs (cfg.protocol == "http/json") { encoding = "json"; };
 
+          processors."filter/drop-own-export-failures" = import ../collector-own-export-failures.nix;
+
           service.pipelines.logs = {
             receivers = [ "journald" ];
-            processors = [ "resource" ];
+            processors = [
+              "filter/drop-own-export-failures"
+              "resource"
+            ];
             exporters = [ logExporter ];
           };
 

nix/collector-own-export-failures.nix

@@ -0,0 +1,21 @@
+# The `filter` processor every collector's logs pipeline carries so that a
+# collector never ships its own export-failure lines.
+#
+# A collector that cannot reach its exporter logs one line per failed batch
+# (`Exporting failed. Will retry the request after interval.`, `Exporting
+# failed. Dropping data.`, and the 503 body `sending queue is full`) into the
+# journal its own journald receiver reads. Shipping those lines is more
+# export traffic during the outage that caused them.
+#
+# Only the collector's own unit, and only exporter failures: its startup,
+# config and receiver errors still ship. The failure stays visible as the
+# collector's `otelcol_exporter_send_failed_*` metrics.
+#
+# `body` is the journal entry as a map. A record whose body is not a map
+# raises an OTTL error, which `ignore` turns into "kept".
+{
+  error_mode = "ignore";
+  logs.log_record = [
+    ''body["_SYSTEMD_UNIT"] == "opentelemetry-collector.service" and IsMatch(body["MESSAGE"], "Exporting failed|sending queue is full")''
+  ];
+}

nix/container-modules/swarm-journal-forwarder.nix

@@ -253,9 +253,14 @@ in
             endpoint_params.audience = cfg.clientId;
           };
 
+          processors."filter/drop-own-export-failures" = import ../collector-own-export-failures.nix;
+
           service.pipelines.logs = {
             receivers = [ "journald" ];
-            processors = [ "resource" ];
+            processors = [
+              "filter/drop-own-export-failures"
+              "resource"
+            ];
             exporters = [ "otlphttp" ];
           };
 

nix/host-modules/otel.nix

@@ -682,6 +682,9 @@ in
           # clean, an invalid key is rejected (`otelcol-contrib validate`).
           processors.deltatocumulative = { };
 
+          # `journalUnits` may name this collector's own unit.
+          processors."filter/drop-own-export-failures" = import ../collector-own-export-failures.nix;
+
           service.pipelines.metrics = {
             # EXTENDED, not replaced. Assigning here instead of appending
             # would drop `otlp` — the hive would stop receiving from its
@@ -744,7 +747,10 @@ in
             # failure rather than a no-op. Host identity applies to both
             # signals for the same reason — a line is only comparable
             # across hives once it says which machine collected it.
-            processors = [ "resourcedetection" ];
+            processors = [
+              "filter/drop-own-export-failures"
+              "resourcedetection"
+            ];
             exporters = [ swarmName ];
           };
 

nix/host-modules/swarm-otel.nix

@@ -1507,6 +1507,7 @@ in
                     receivers = [ "journald" ] ++ map (n: "otlp/${n}") serviceProducerNames;
                     processors = [
                       "filter/exclude-user-sessions"
+                      "filter/drop-own-export-failures"
                       "resource/${swarmTierName}"
                     ];
                     exporters = logExporterNames;
@@ -1730,6 +1731,10 @@ in
                     ];
                   };
 
+                  # The host journal carries the host collector's own unit; see
+                  # ../collector-own-export-failures.nix.
+                  "filter/drop-own-export-failures" = import ../collector-own-export-failures.nix;
+
                   # The store exporter's own failure lines cannot reach the
                   # store, and re-feeding them to it repeats the failure.
                   # Failures of any other exporter pass. `body` is the journal

nix/module-eval/agent-otel.nix

@@ -103,6 +103,24 @@ let
       name = "the agent forwarder's journald receiver carries the shared PRIORITY mapping";
       ok = carriesJournaldSeverity (agentSettings agentOtel).receivers.journald;
     }
+    {
+      # The forwarder reads its own journal, which holds its own exporter
+      # failures. Presence control: the resource processor is still listed.
+      name = "the agent forwarder drops its own export-failure lines before exporting";
+      ok =
+        let
+          s = agentSettings agentOtel;
+          p = s.service.pipelines.logs.processors;
+        in
+        p == [
+          "filter/drop-own-export-failures"
+          "resource"
+        ]
+        && s.processors ? "filter/drop-own-export-failures"
+        && lib.hasInfix "opentelemetry-collector.service" (
+          lib.head s.processors."filter/drop-own-export-failures".logs.log_record
+        );
+    }
     {
       # The hop's two ends: what it reads, and where what it reads goes.
       # The endpoint is compared against the value the fixture handed the

nix/module-eval/hive-otel.nix

@@ -117,6 +117,19 @@ let
           ]
         && hiveOtelPipelines.logs.exporters == hiveOtelPipelines.metrics.exporters;
     }
+    {
+      # Presence control: the resource processor is still listed after it.
+      name = "the hive collector drops its own export-failure lines before its logs pipeline exports";
+      ok =
+        hiveOtelPipelines.logs.processors == [
+          "filter/drop-own-export-failures"
+          "resourcedetection"
+        ]
+        && hiveOtelSettings.processors ? "filter/drop-own-export-failures"
+        && lib.hasInfix "opentelemetry-collector.service" (
+          lib.head hiveOtelSettings.processors."filter/drop-own-export-failures".logs.log_record
+        );
+    }
     {
       # `deltatocumulative` is metrics-only: naming it in a logs pipeline
       # kills the collector at startup rather than doing nothing. The second

nix/module-eval/swarm-otel-core.nix

@@ -341,6 +341,34 @@ let
         && lib.hasInfix "Exporting failed" (lib.head f.logs.log_record)
         && lib.hasInfix "otlphttp/victorialogs" (lib.head f.logs.log_record);
     }
+    {
+      # 🩸 The host journal holds the host collector's own failure lines, and
+      # shipping them is more traffic into an outage. The condition is read
+      # from the shared file, so every tier's wiring check below asks about
+      # the same text. Presence controls: the exporter is named, the pipeline
+      # lists the processor before the resource stamp, and the self pipeline
+      # keeps its own store-only filter.
+      name = "the host-journal pipeline drops the collector's own export-failure lines before it exports";
+      ok =
+        let
+          s = otelSettings otelNoStores;
+          host = lib.filter (p: builtins.elem "journald" p.receivers) (lib.attrValues s.service.pipelines);
+          f = s.processors."filter/drop-own-export-failures";
+          cond = lib.head f.logs.log_record;
+          i =
+            lib.lists.findFirstIndex (x: x == "filter/drop-own-export-failures") (-1)
+              (lib.head host).processors;
+        in
+        lib.length host == 1
+        && (lib.head host).exporters != [ ]
+        && i >= 0
+        && i < lib.length (lib.head host).processors - 1
+        && lib.length f.logs.log_record == 1
+        && lib.hasInfix "opentelemetry-collector.service" cond
+        && lib.hasInfix "Exporting failed" cond
+        && lib.hasInfix "sending queue is full" cond
+        && s.processors ? "filter/drop-store-export-failures";
+    }
     {
       # 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