diff --git a/docs/4-implementation/swarm/telemetry.md b/docs/4-implementation/swarm/telemetry.md index 786478fa4..7709c1ce6 100644 --- a/docs/4-implementation/swarm/telemetry.md +++ b/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. diff --git a/nix/agent-modules/otel.nix b/nix/agent-modules/otel.nix index f913ffce5..edd884f7a 100644 --- a/nix/agent-modules/otel.nix +++ b/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 ]; }; diff --git a/nix/collector-own-export-failures.nix b/nix/collector-own-export-failures.nix new file mode 100644 index 000000000..644ede347 --- /dev/null +++ b/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")'' + ]; +} diff --git a/nix/container-modules/swarm-journal-forwarder.nix b/nix/container-modules/swarm-journal-forwarder.nix index 6aa2f4cd3..35240a862 100644 --- a/nix/container-modules/swarm-journal-forwarder.nix +++ b/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" ]; }; diff --git a/nix/host-modules/otel.nix b/nix/host-modules/otel.nix index e902b48d0..c6ab8bfd7 100644 --- a/nix/host-modules/otel.nix +++ b/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 ]; }; diff --git a/nix/host-modules/swarm-otel.nix b/nix/host-modules/swarm-otel.nix index 255b64c2a..ffc2a8168 100644 --- a/nix/host-modules/swarm-otel.nix +++ b/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 diff --git a/nix/module-eval/agent-otel.nix b/nix/module-eval/agent-otel.nix index c7f3b0955..1459238a7 100644 --- a/nix/module-eval/agent-otel.nix +++ b/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 diff --git a/nix/module-eval/hive-otel.nix b/nix/module-eval/hive-otel.nix index 3d0c36825..d693cd958 100644 --- a/nix/module-eval/hive-otel.nix +++ b/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 diff --git a/nix/module-eval/swarm-otel-core.nix b/nix/module-eval/swarm-otel-core.nix index d119875b9..12026a17f 100644 --- a/nix/module-eval/swarm-otel-core.nix +++ b/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