feat: disable pipeline optimization

cmd/manager/main.go

@@ -102,7 +102,7 @@ func main() {
 	flag.BoolVar(&enableReconciliationInvalidPipelines, "enable-reconciliation-invalid-pipelines", false,
 		"Enable the reconciliation process for pipelines with invalid configurations")
 	flag.DurationVar(&reconciliationRetryDelay, "reconciliation-retry-delay", 30*time.Second, "Specify the delay before retrying the reconciliation process for pipelines")
-	flag.BoolVar(&enableConfigOptimization, "enable-config-optimization", false, "Collapse kubernetes_logs sources with identical settings into one source per group in generated agent configs. Vector CRs can opt out with the vector-operator.kaasops.io/config-optimization=disabled annotation")
+	flag.BoolVar(&enableConfigOptimization, "enable-config-optimization", false, "Collapse kubernetes_logs sources with identical settings into one source per group in generated agent configs. A Vector CR (whole agent) or an individual (Cluster)VectorPipeline (just its source) can opt out with the vector-operator.kaasops.io/config-optimization=disabled annotation")
 	flag.BoolVar(&enableCheckpointMigration, "enable-checkpoint-migration", false, "Migrate vector file checkpoints when the config optimization renames kubernetes_logs sources: the agent config secret name is bound to the optimization mode (switching it rolls the DaemonSet) and a checkpoint-merger init container consolidates checkpoints before vector starts")
 	flag.StringVar(&checkpointMergerImage, "checkpoint-merger-image", "", "Override the checkpoint-merger init container image (default kaasops/checkpoint-merger:<operator version>)")
 

docs/config-optimization.md

@@ -20,6 +20,16 @@ metadata:
     vector-operator.kaasops.io/config-optimization: disabled
 ```
 
+A single pipeline can be excluded with the same annotation on the
+(Cluster)VectorPipeline; its source stays standalone while the rest of the group
+is still collapsed (see [Backpressure isolation](#backpressure-isolation)):
+
+```yaml
+metadata:
+  annotations:
+    vector-operator.kaasops.io/config-optimization: disabled
+```
+
 The flag is expected to become the default behavior in a future release and
 the gate to be removed eventually.
 
@@ -45,6 +55,30 @@ Measured on a 1000-pipeline single-node cluster (vector 0.48): watch requests to
 - **Enabling (or disabling) the optimization on a running cluster renames the sources, so vector re-reads the log files currently retained on the nodes: expect a one-time redelivery of recent logs** — unless checkpoint migration is enabled, see below.
 - Vector logs a warning about unconsumed `<router>._unmatched` outputs of the generated route transforms; it is harmless — events not matching any pipeline namespace are dropped there.
 
+## Backpressure isolation
+
+In vector a source emits events only as fast as the slowest sink that applies backpressure (the default `buffer.when_full: block`). Without optimization each pipeline has its own source, so a stuck sink (an unavailable destination retrying forever) backpressures only its own pipeline. After collapsing, the pipelines of a signature group share one source, so a single stuck sink backpressures that shared source and stalls every pipeline of the group (head-of-line blocking). Pipelines in other groups and standalone sources are unaffected.
+
+This is an inherent trade-off of a shared source: the savings come from collapsing the per-pipeline watchers and readers, while in vector backpressure isolation only comes from a non-blocking buffer. Options, cheapest first:
+
+- Set `buffer.when_full: drop_newest` (or a disk buffer with `overflow`) and/or a bounded `request.retry_max_duration_secs` on the unreliable sink. This breaks the backpressure back to the shared source, keeps the optimization's savings, and affects only that sink.
+- Exclude the pipeline from optimization with the per-pipeline annotation above. It keeps its own source (and thus its own backpressure domain) at the cost of that source's watch. If another, non-excluded pipeline shares the excluded pipeline's namespace, that namespace is then watched by both the dedicated and the collapsed source, and its pod logs are read twice. Toggling this annotation renames that pipeline's source, so its retained logs are re-read once (a bounded one-time redelivery). Unlike the whole-agent mode switch this is a live config reload, not covered by checkpoint migration even when `--enable-checkpoint-migration` is on (see below). The re-read is scoped to the toggled pipeline's namespace; the rest of the group keeps flowing.
+
+The chart ships an optional `PrometheusRule` (`prometheusRule.enabled`, off by default) with two alerts: `VectorOptimizedSourceStalled` (the collapsed source stopped emitting while it was flowing, i.e. the group stalled) and `VectorSinkFailing`, which names the offending sink (`component_id`) when an HTTP-based sink fails most of its requests, counting both transport failures (connection refused, DNS, TLS, timeout) and 5xx/429 responses. A sink that hangs without erroring, or a non-HTTP sink, is caught only by the first alert (throughput collapse); then check which sink destination is unreachable. The `component_id` is `<namespace>-<pipeline>-<sinkKey>`; the parts are dash-joined and not uniquely splittable, so to map it to the exact pipeline (e.g. to apply the opt-out annotation), grep the id in the agent config Secret. Both alerts need the agent to expose `vector_*` metrics: set `spec.agent.internalMetrics: true` on the Vector CR (off by default) so it runs an `internal_metrics` source feeding a `prometheus_exporter` sink that the operator's agent PodMonitor scrapes. Without it the alerts have no data.
+
+Both fire for the same incident, so inhibit the symptom under the cause in Alertmanager to page once:
+
+```yaml
+inhibit_rules:
+  - source_matchers: [alertname="VectorSinkFailing"]
+    target_matchers: [alertname="VectorOptimizedSourceStalled"]
+    equal: [pod]
+```
+
+### Before enabling on a large cluster
+
+Optimization changes the blast radius of an already-failing sink: without it, a sink stuck with `when_full: block` stalls only its own pipeline (easy to miss among thousands); after collapsing, it stalls its whole signature group, which can stop logs broadly. So surface failing sinks first. `VectorSinkFailing` is independent of optimization and fires on the legacy config too, so set `spec.agent.internalMetrics: true` (so the alerts have data) and enable `prometheusRule.enabled` ahead of `--enable-config-optimization`, let it flag the failing sinks, and for each one fix the destination, set `buffer.when_full: drop_newest` / a bounded `request.retry_max_duration_secs`, or pre-exclude its pipeline with the annotation. Then enable optimization.
+
 ## Checkpoint migration
 
 Vector keys file checkpoints by a fingerprint of the file content, not by the source name, so the positions saved under the old source names stay valid after the rename and can be carried over. `--enable-checkpoint-migration` does that:
@@ -61,6 +95,7 @@ args:
 - **Restricted image sets / air-gapped clusters: mirror the merger image before enabling the flag.** It runs as an init container, so if its image cannot be pulled the agent pod is stuck in `Init:ImagePullBackOff` and vector does not start on that node — fail-open does not cover an unpullable image. The blast radius is contained (DaemonSet `maxUnavailable=1`: one node's agent is down and the rollout stalls there; other nodes keep their previous pod). Recover by making the image available, or by turning `--enable-checkpoint-migration` off (the init container is removed and agents restart, falling back to the one-time re-read).
 - Rolling back to the legacy config restores the saved per-source positions; only files that appeared while the optimization was active are re-read.
 - A mode switch is a rolling restart of the agents: on large clusters expect it to take a while, and the re-created watch connections to arrive gradually (which is what you want). Enabling both flags in one change gives a single migrated rollout.
+- **The per-pipeline opt-out is not covered by migration.** Migration hinges on the config-Secret name changing (which rolls the DaemonSet so the merger init container runs). The whole-agent flag and the per-CR annotation change that name; the per-(Cluster)VectorPipeline annotation does not — it is applied as a live config reload with no pod restart, so the merger does not run. Toggling it therefore re-reads the retained logs of that one pipeline's namespace once, even with `--enable-checkpoint-migration` on. This is intentional: a full rolling restart to isolate a single pipeline (typically during a backpressure incident, where that pipeline's sink is already misbehaving) would be far more disruptive than a bounded one-time redelivery.
 - The first migrated rollout pulls the merger image on each node before that node's agent restarts; with `maxUnavailable=1` this is sequential, so the per-node restart includes the image pull (and, on a chart upgrade, the operator's own new image pull happens first). The image is small (distroless + a small static binary) so the pull is quick, but on large clusters pre-pulling it shortens the window.
 
 ### Observability

helm/charts/vector-operator/templates/prometheusrule.yaml

@@ -0,0 +1,70 @@
+{{- if .Values.prometheusRule.enabled }}
+apiVersion: monitoring.coreos.com/v1
+kind: PrometheusRule
+metadata:
+  name: {{ include "chart.fullname" . }}-agent
+  namespace: {{ default .Release.Namespace .Values.prometheusRule.namespace }}
+  labels:
+    {{- include "chart.labels" . | nindent 4 }}
+  {{- with .Values.prometheusRule.additionalLabels }}
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  groups:
+    - name: vector-agent-config-optimization
+      rules:
+        # Symptom: a sink stuck with the default buffer.when_full=block backpressures the
+        # shared optimized source and stalls every pipeline of the group (head-of-line
+        # blocking), so the source stops emitting. Heuristic (zero now, flowed within
+        # lookback): may also fire for an idle agent; blind for the first `window` after
+        # start; groups by `pod`, which the scrape (PodMonitor) must add.
+        - alert: VectorOptimizedSourceStalled
+          expr: |-
+            sum by (pod) (rate(vector_component_sent_events_total{component_id=~"optimizedSource.*"}[{{ .Values.prometheusRule.window }}])) == 0
+            and
+            sum by (pod) (max_over_time(rate(vector_component_sent_events_total{component_id=~"optimizedSource.*"}[{{ .Values.prometheusRule.window }}])[{{ .Values.prometheusRule.lookback }}:{{ .Values.prometheusRule.window }}])) > 0
+          for: {{ .Values.prometheusRule.window }}
+          labels:
+            severity: warning
+          annotations:
+            summary: Vector optimized source stalled (likely head-of-line blocking)
+            description: >-
+              The collapsed kubernetes_logs source on pod {{`{{ $labels.pod }}`}} stopped
+              emitting events while it was flowing earlier. Under config optimization every
+              pipeline of a signature group shares this source, so one sink stuck with
+              buffer.when_full=block stalls the whole group; with a single large group this
+              stops all logs. The VectorSinkFailing alert names the offending sink when it
+              fails (transport errors or 5xx/429); a non-HTTP or silently hanging sink will
+              not show there, then check which sink destination is unreachable. Remedy: set
+              buffer.when_full=drop_newest or a
+              bounded request.retry_max_duration_secs on that sink, or exclude its pipeline
+              with the vector-operator.kaasops.io/config-optimization=disabled annotation.
+              Heuristic: may also fire for an agent that has simply gone idle.
+        # Cause: names the failing sink directly (component_id). Counts both transport
+        # failures (errors_total: refused/DNS/TLS/timeout) and error responses
+        # (responses_total 5xx/429, which are retried, not counted as errors). HTTP-based
+        # sinks only; component_id also covers any HTTP-based source (kubernetes_logs is not).
+        - alert: VectorSinkFailing
+          expr: |-
+            (
+              sum by (pod, component_id) (rate(vector_http_client_errors_total[{{ .Values.prometheusRule.window }}]))
+              + sum by (pod, component_id) (rate(vector_http_client_responses_total{status=~"5..|429"}[{{ .Values.prometheusRule.window }}]))
+            )
+            / sum by (pod, component_id) (rate(vector_http_client_requests_sent_total[{{ .Values.prometheusRule.window }}]))
+            > 0.5
+          for: {{ .Values.prometheusRule.window }}
+          labels:
+            severity: warning
+          annotations:
+            summary: Vector sink {{`{{ $labels.component_id }}`}} on pod {{`{{ $labels.pod }}`}} is failing most requests
+            description: >-
+              The HTTP sink {{`{{ $labels.component_id }}`}} on pod {{`{{ $labels.pod }}`}} is
+              failing the majority of its requests (transport errors or 5xx/429 responses).
+              Under config optimization a stuck sink backpressures its whole signature group
+              and can stall all of it (see VectorOptimizedSourceStalled). The component_id is
+              <namespace>-<pipeline>-<sinkKey> (dash-joined, not uniquely splittable; to map it
+              to the exact pipeline, grep the id in the agent config Secret).
+              Remedy: fix the destination, set buffer.when_full=drop_newest or a bounded
+              request.retry_max_duration_secs, or exclude its pipeline with the
+              vector-operator.kaasops.io/config-optimization=disabled annotation.
+{{- end }}

helm/charts/vector-operator/values.yaml

@@ -72,7 +72,7 @@ args:
 #  - "-watch-name=vector-operator" # Filter the list of watched objects by checking the app.kubernetes.io/managed-by label
 #  - "-enable-reconciliation-invalid-pipelines=true" # Enable the reconciliation process for pipelines with invalid configurations
 #  - "-reconciliation-retry-delay=120s" # Specify the delay before retrying the reconciliation process for pipelines
-#  - "-enable-config-optimization" # Collapse kubernetes_logs sources with identical settings into one source per group (opt out per Vector CR with the vector-operator.kaasops.io/config-optimization=disabled annotation)
+#  - "-enable-config-optimization" # Collapse kubernetes_logs sources with identical settings into one source per group (opt out per Vector CR or per (Cluster)VectorPipeline with the vector-operator.kaasops.io/config-optimization=disabled annotation)
 #  - "-enable-checkpoint-migration" # Migrate vector file checkpoints when the config optimization renames sources: mode switches roll the agent DaemonSet and a checkpoint-merger init container consolidates checkpoints, avoiding a one-time re-read of retained logs
 
 vector:
@@ -103,3 +103,22 @@ clustervectorpipeline: {}
 
 openshift:
   enabled: false
+
+# PrometheusRule with two alerts: VectorOptimizedSourceStalled (the collapsed source
+# stalled, i.e. head-of-line blocking) and VectorSinkFailing (names the failing HTTP sink).
+# Requires the Prometheus Operator CRDs AND the agent exposing vector_* metrics: set
+# spec.agent.internalMetrics: true on the Vector CR (off by default) so the agent runs an
+# internal_metrics -> prometheus_exporter pipeline; the operator's agent PodMonitor then
+# scrapes it. Without those metrics these alerts have no data. Disabled by default.
+prometheusRule:
+  enabled: false
+  # -- extra labels for the PrometheusRule, e.g. to match the Prometheus ruleSelector
+  additionalLabels: {}
+  # -- namespace for the PrometheusRule; defaults to the release namespace
+  namespace: ""
+  # -- rate window for the stall alert (also its `for:` duration)
+  window: 10m
+  # -- how far back to confirm the optimized source was flowing. The stall alert is a
+  # -- heuristic: it can false-positive on an idle low-volume agent that flowed within
+  # -- lookback, and is blind for the first `window` after an agent starts.
+  lookback: 1h

internal/common/annotations.go

@@ -3,7 +3,13 @@ package common
 const (
 	AnnotationServiceName = "observability.kaasops.io/service-name"
 	AnnotationRestartedAt = "vector-operator.kaasops.io/restartedAt"
-	// AnnotationConfigOptimization set to "disabled" on a Vector CR opts the agent
-	// out of the config optimization enabled by --enable-config-optimization.
+	// AnnotationConfigOptimization set to AnnotationValueDisabled opts out of the
+	// config optimization enabled by --enable-config-optimization. On a Vector CR it
+	// opts the whole agent out; on a (Cluster)VectorPipeline it keeps just that
+	// pipeline's kubernetes_logs source standalone while the rest of the group still
+	// collapses.
 	AnnotationConfigOptimization = "vector-operator.kaasops.io/config-optimization"
+
+	// AnnotationValueDisabled is the opt-out value for AnnotationConfigOptimization.
+	AnnotationValueDisabled = "disabled"
 )

internal/config/agent.go

@@ -8,6 +8,7 @@ import (
 	goyaml "sigs.k8s.io/yaml"
 
 	vectorv1alpha1 "github.com/kaasops/vector-operator/api/v1alpha1"
+	"github.com/kaasops/vector-operator/internal/common"
 	"github.com/kaasops/vector-operator/internal/pipeline"
 	"github.com/kaasops/vector-operator/internal/utils/k8s"
 )
@@ -35,11 +36,16 @@ func BuildAgentConfig(p VectorConfigParams, pipelines ...pipeline.Pipeline) (*Ve
 func buildAgentConfig(params VectorConfigParams, pipelines ...pipeline.Pipeline) (*VectorConfig, error) {
 	cfg := newVectorConfig(params)
 
+	// sources of pipelines that opted out of config optimization via annotation; kept
+	// out of source collapsing so they keep a dedicated source (backpressure isolation)
+	var optOutSources map[string]struct{}
+
 	for _, pipeline := range pipelines {
 		p := &PipelineConfig{}
 		if err := UnmarshalJson(pipeline.GetSpec(), p); err != nil {
 			return nil, fmt.Errorf("failed to unmarshal pipeline %s: %w", pipeline.GetName(), err)
 		}
+		optedOut := pipeline.GetAnnotations()[common.AnnotationConfigOptimization] == common.AnnotationValueDisabled
 		for k, v := range p.Sources {
 			// Validate source
 			if _, ok := pipeline.(*vectorv1alpha1.VectorPipeline); ok {
@@ -66,6 +72,12 @@ func buildAgentConfig(params VectorConfigParams, pipelines ...pipeline.Pipeline)
 			}
 			v.Name = addPrefix(pipeline.GetNamespace(), pipeline.GetName(), k)
 			cfg.Sources[v.Name] = v
+			if optedOut {
+				if optOutSources == nil {
+					optOutSources = make(map[string]struct{})
+				}
+				optOutSources[v.Name] = struct{}{}
+			}
 		}
 		for k, v := range p.Transforms {
 			v.Name = addPrefix(pipeline.GetNamespace(), pipeline.GetName(), k)
@@ -84,7 +96,7 @@ func buildAgentConfig(params VectorConfigParams, pipelines ...pipeline.Pipeline)
 	}
 
 	if params.OptimizeSources {
-		optimizeAgentSources(cfg)
+		optimizeAgentSources(cfg, optOutSources)
 	}
 
 	// Add exporter pipeline

internal/config/agent_optimize.go

@@ -47,13 +47,18 @@ const (
 // identical settings. Per-pipeline event streams are restored with route transforms
 // matching on the pod namespace, so inputs of downstream transforms and sinks are
 // rewritten to the corresponding route output. Sources with any other namespace
-// selector or with unique settings are left as is.
-func optimizeAgentSources(cfg *VectorConfig) {
+// selector, with unique settings, or named in optOut are left as is.
+func optimizeAgentSources(cfg *VectorConfig, optOut map[string]struct{}) {
 	groups := make(map[string][]*Source)
 	for _, src := range cfg.Sources {
 		if src.Type != KubernetesLogsType {
 			continue
 		}
+		// pipelines that opted out via annotation keep their own dedicated source,
+		// preserving backpressure isolation from the rest of the group
+		if _, ok := optOut[src.Name]; ok {
+			continue
+		}
 		if _, ok := singleNamespaceOf(src); !ok {
 			continue
 		}