diff --git a/cmd/manager/main.go b/cmd/manager/main.go index aade900d..f2d32666 100644 --- a/cmd/manager/main.go +++ b/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:)") diff --git a/docs/config-optimization.md b/docs/config-optimization.md index b6399453..91f824e7 100644 --- a/docs/config-optimization.md +++ b/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 `._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 `--`; 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 diff --git a/helm/charts/vector-operator/templates/prometheusrule.yaml b/helm/charts/vector-operator/templates/prometheusrule.yaml new file mode 100644 index 00000000..6b453709 --- /dev/null +++ b/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 + -- (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 }} \ No newline at end of file diff --git a/helm/charts/vector-operator/values.yaml b/helm/charts/vector-operator/values.yaml index 71dfc000..b4bc5de1 100644 --- a/helm/charts/vector-operator/values.yaml +++ b/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 diff --git a/internal/common/annotations.go b/internal/common/annotations.go index 5d7cb33b..ad9e532d 100644 --- a/internal/common/annotations.go +++ b/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" ) diff --git a/internal/config/agent.go b/internal/config/agent.go index a0656608..790096da 100644 --- a/internal/config/agent.go +++ b/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 diff --git a/internal/config/agent_optimize.go b/internal/config/agent_optimize.go index ddc29b31..0db282e6 100644 --- a/internal/config/agent_optimize.go +++ b/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 } diff --git a/internal/config/agent_optimize_test.go b/internal/config/agent_optimize_test.go index 390ba8ea..b65d2093 100644 --- a/internal/config/agent_optimize_test.go +++ b/internal/config/agent_optimize_test.go @@ -27,6 +27,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" vectorv1alpha1 "github.com/kaasops/vector-operator/api/v1alpha1" + "github.com/kaasops/vector-operator/internal/common" "github.com/kaasops/vector-operator/internal/pipeline" ) @@ -86,6 +87,131 @@ func TestOptimizeSourcesDisabledKeepsSources(t *testing.T) { assert.Equal(t, []string{"ns-a-pipe-logs"}, cfg.Sinks["ns-a-pipe-out"].Inputs) } +func withConfigOptAnnotation(p pipeline.Pipeline, value string) pipeline.Pipeline { + p.(*vectorv1alpha1.VectorPipeline).Annotations = map[string]string{common.AnnotationConfigOptimization: value} + return p +} + +func testLogPipelineOptOut(namespace string) pipeline.Pipeline { + return withConfigOptAnnotation(testLogPipeline(namespace), common.AnnotationValueDisabled) +} + +func TestOptimizeSourcesPerPipelineOptOut(t *testing.T) { + cfg, _, err := BuildAgentConfig(VectorConfigParams{OptimizeSources: true}, + testLogPipeline("ns-a"), testLogPipeline("ns-b"), testLogPipelineOptOut("ns-c")) + require.NoError(t, err) + + // ns-a + ns-b collapse; the opted-out pipeline keeps its own dedicated source + require.Len(t, cfg.Sources, 2) + standalone := cfg.Sources["ns-c-pipe-logs"] + require.NotNil(t, standalone) + assert.Equal(t, "kubernetes.io/metadata.name=ns-c", standalone.ExtraNamespaceLabelSelector) + assert.Equal(t, []string{"ns-c-pipe-logs"}, cfg.Sinks["ns-c-pipe-out"].Inputs) + + // the collapsed source covers only the non-opted-out namespaces + var collapsed *Source + for name, s := range cfg.Sources { + if name != "ns-c-pipe-logs" { + collapsed = s + } + } + require.NotNil(t, collapsed) + assert.Equal(t, "kubernetes.io/metadata.name in (ns-a,ns-b)", collapsed.ExtraNamespaceLabelSelector) +} + +func TestOptimizeSourcesOptOutOnlyOnDisabledValue(t *testing.T) { + // any value other than "disabled" must NOT opt out + enabled := withConfigOptAnnotation(testLogPipeline("ns-c"), "enabled") + cfg, _, err := BuildAgentConfig(VectorConfigParams{OptimizeSources: true}, + testLogPipeline("ns-a"), testLogPipeline("ns-b"), enabled) + require.NoError(t, err) + + require.Len(t, cfg.Sources, 1) + assert.Nil(t, cfg.Sources["ns-c-pipe-logs"]) +} + +func TestOptimizeSourcesOptOutMultipleSources(t *testing.T) { + twin := withConfigOptAnnotation(testPipeline("ns-c", "pipe", + `{"logs": {"type": "kubernetes_logs"}, "logs2": {"type": "kubernetes_logs"}}`, + `{"out": {"type": "blackhole", "inputs": ["logs", "logs2"]}}`), common.AnnotationValueDisabled) + cfg, _, err := BuildAgentConfig(VectorConfigParams{OptimizeSources: true}, + testLogPipeline("ns-a"), testLogPipeline("ns-b"), twin) + require.NoError(t, err) + + // both sources of the opted-out pipeline stay standalone and keep the sink wiring + require.NotNil(t, cfg.Sources["ns-c-pipe-logs"]) + require.NotNil(t, cfg.Sources["ns-c-pipe-logs2"]) + assert.Equal(t, []string{"ns-c-pipe-logs", "ns-c-pipe-logs2"}, cfg.Sinks["ns-c-pipe-out"].Inputs) +} + +func TestOptimizeSourcesOptOutSharedNamespace(t *testing.T) { + // two pipelines in ns-a, one opted out, plus ns-b: documents the double-read of ns-a + optedB := withConfigOptAnnotation(testPipeline("ns-a", "second", + `{"logs": {"type": "kubernetes_logs"}}`, + `{"out": {"type": "blackhole", "inputs": ["logs"]}}`), common.AnnotationValueDisabled) + cfg, _, err := BuildAgentConfig(VectorConfigParams{OptimizeSources: true}, + testLogPipeline("ns-a"), optedB, testLogPipeline("ns-b")) + require.NoError(t, err) + + standalone := cfg.Sources["ns-a-second-logs"] + require.NotNil(t, standalone) + assert.Equal(t, "kubernetes.io/metadata.name=ns-a", standalone.ExtraNamespaceLabelSelector) + assert.Equal(t, []string{"ns-a-second-logs"}, cfg.Sinks["ns-a-second-out"].Inputs) + + // the collapsed source still covers ns-a (from the non-opted pipeline), so ns-a is read twice + var collapsed *Source + for name, s := range cfg.Sources { + if name != "ns-a-second-logs" { + collapsed = s + } + } + require.NotNil(t, collapsed) + assert.Equal(t, "kubernetes.io/metadata.name in (ns-a,ns-b)", collapsed.ExtraNamespaceLabelSelector) +} + +func TestOptimizeSourcesReCollapseAfterOptOutRemoved(t *testing.T) { + // opted out -> standalone source + opted, _, err := BuildAgentConfig(VectorConfigParams{OptimizeSources: true}, + testLogPipeline("ns-a"), testLogPipeline("ns-b"), testLogPipelineOptOut("ns-c")) + require.NoError(t, err) + require.NotNil(t, opted.Sources["ns-c-pipe-logs"]) + + // annotation removed -> collapses back into the shared source + re, _, err := BuildAgentConfig(VectorConfigParams{OptimizeSources: true}, + testLogPipeline("ns-a"), testLogPipeline("ns-b"), testLogPipeline("ns-c")) + require.NoError(t, err) + require.Len(t, re.Sources, 1) + assert.Nil(t, re.Sources["ns-c-pipe-logs"]) +} + +func TestOptimizeSourcesOptOutClusterVectorPipeline(t *testing.T) { + cvp := &vectorv1alpha1.ClusterVectorPipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cvp", + Annotations: map[string]string{common.AnnotationConfigOptimization: common.AnnotationValueDisabled}, + }, + Spec: vectorv1alpha1.VectorPipelineSpec{ + Sources: &runtime.RawExtension{Raw: []byte(`{"logs": {"type": "kubernetes_logs", "extra_namespace_label_selector": "kubernetes.io/metadata.name=ns-x"}}`)}, + Sinks: &runtime.RawExtension{Raw: []byte(`{"out": {"type": "blackhole", "inputs": ["logs"]}}`)}, + }, + } + cfg, _, err := BuildAgentConfig(VectorConfigParams{OptimizeSources: true}, + testLogPipeline("ns-a"), testLogPipeline("ns-b"), cvp) + require.NoError(t, err) + + // only ns-a + ns-b collapse; the opted-out CVP keeps its own ns-x source + collapsed, groups := cfg.OptimizationSummary() + assert.Equal(t, 2, collapsed) + assert.Equal(t, 1, groups) + standalone := false + for _, s := range cfg.Sources { + if s.ExtraNamespaceLabelSelector == "kubernetes.io/metadata.name=ns-x" { + standalone = true + } + } + assert.True(t, standalone) +} + func TestOptimizeSourcesGroupsBySignature(t *testing.T) { withSelector := testPipeline("ns-c", "pipe", `{"logs": {"type": "kubernetes_logs", "extra_label_selector": "app=foo"}}`, @@ -189,7 +315,7 @@ func TestOptimizeSourcesChunksLargeGroups(t *testing.T) { ExtraNamespaceLabelSelector: fmt.Sprintf("kubernetes.io/metadata.name=ns-%04d", i), } } - optimizeAgentSources(cfg) + optimizeAgentSources(cfg, nil) // 2200 namespaces are split into 3 sources of at most maxNamespacesPerSource require.Len(t, cfg.Sources, 3) diff --git a/internal/controller/vector_controller.go b/internal/controller/vector_controller.go index 74ebd3f2..cb43f6fe 100644 --- a/internal/controller/vector_controller.go +++ b/internal/controller/vector_controller.go @@ -71,7 +71,7 @@ type VectorReconciler struct { // built with the sources optimization: the controller-level feature flag is on and // the Vector CR is not opted out with the config-optimization=disabled annotation. func optimizeSources(enabled bool, v *v1alpha1.Vector) bool { - return enabled && v.Annotations[common.AnnotationConfigOptimization] != "disabled" + return enabled && v.Annotations[common.AnnotationConfigOptimization] != common.AnnotationValueDisabled } //+kubebuilder:rbac:groups=observability.kaasops.io,resources=vectors,verbs=get;list;watch;create;update;patch;delete diff --git a/internal/pipeline/hash.go b/internal/pipeline/hash.go index 0219849a..9af783c2 100644 --- a/internal/pipeline/hash.go +++ b/internal/pipeline/hash.go @@ -28,13 +28,16 @@ type tmp struct { Spec v1alpha1.VectorPipelineSpec Labels map[string]string ServiceName string + // omitempty so pipelines without the annotation keep a stable hash (no re-hash on upgrade) + ConfigOptimization string `json:",omitempty"` } func GetPipelineHash(pipeline Pipeline) (*uint32, error) { a, err := json.Marshal(tmp{ - Spec: pipeline.GetSpec(), - Labels: pipeline.GetLabels(), - ServiceName: pipeline.GetAnnotations()[common.AnnotationServiceName], + Spec: pipeline.GetSpec(), + Labels: pipeline.GetLabels(), + ServiceName: pipeline.GetAnnotations()[common.AnnotationServiceName], + ConfigOptimization: pipeline.GetAnnotations()[common.AnnotationConfigOptimization], }) if err != nil { return nil, err diff --git a/internal/pipeline/hash_test.go b/internal/pipeline/hash_test.go new file mode 100644 index 00000000..09fde380 --- /dev/null +++ b/internal/pipeline/hash_test.go @@ -0,0 +1,105 @@ +/* +Copyright 2022. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package pipeline + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + + "github.com/kaasops/vector-operator/api/v1alpha1" + "github.com/kaasops/vector-operator/internal/common" + "github.com/kaasops/vector-operator/internal/utils/hash" +) + +// Toggling the per-pipeline config-optimization opt-out annotation must change the +// pipeline hash so the reconcile propagates it to an agent config rebuild. +func TestGetPipelineHashTracksConfigOptimizationAnnotation(t *testing.T) { + base := &v1alpha1.VectorPipeline{ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "ns"}} + h1, err := GetPipelineHash(base) + require.NoError(t, err) + + opted := &v1alpha1.VectorPipeline{ObjectMeta: metav1.ObjectMeta{ + Name: "p", + Namespace: "ns", + Annotations: map[string]string{common.AnnotationConfigOptimization: "disabled"}, + }} + h2, err := GetPipelineHash(opted) + require.NoError(t, err) + + assert.NotEqual(t, *h1, *h2) +} + +// The same for a ClusterVectorPipeline (cluster-scoped opt-out path). +func TestGetPipelineHashTracksConfigOptimizationAnnotationCVP(t *testing.T) { + base := &v1alpha1.ClusterVectorPipeline{ObjectMeta: metav1.ObjectMeta{Name: "p"}} + h1, err := GetPipelineHash(base) + require.NoError(t, err) + + opted := &v1alpha1.ClusterVectorPipeline{ObjectMeta: metav1.ObjectMeta{ + Name: "p", + Annotations: map[string]string{common.AnnotationConfigOptimization: common.AnnotationValueDisabled}, + }} + h2, err := GetPipelineHash(opted) + require.NoError(t, err) + + assert.NotEqual(t, *h1, *h2) +} + +// Without the annotation the hash must match the pre-field struct, so an operator +// upgrade does not re-hash and re-validate every pipeline. +func TestGetPipelineHashStableWithoutAnnotation(t *testing.T) { + p := &v1alpha1.VectorPipeline{ + ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "ns"}, + Spec: v1alpha1.VectorPipelineSpec{ + Sources: &runtime.RawExtension{Raw: []byte(`{"logs":{"type":"kubernetes_logs"}}`)}, + }, + } + h, err := GetPipelineHash(p) + require.NoError(t, err) + + legacy, err := json.Marshal(struct { + Spec v1alpha1.VectorPipelineSpec + Labels map[string]string + ServiceName string + }{Spec: p.Spec}) + require.NoError(t, err) + assert.Equal(t, hash.Get(legacy), *h) +} + +// Toggling the annotation must read as "changed" (IsPipelineChanged == false) so the +// pipeline reconcile propagates to an agent rebuild. +func TestIsPipelineChangedDetectsConfigOptimizationToggle(t *testing.T) { + base := &v1alpha1.VectorPipeline{ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "ns"}} + h, err := GetPipelineHash(base) + require.NoError(t, err) + + annotated := &v1alpha1.VectorPipeline{ObjectMeta: metav1.ObjectMeta{ + Name: "p", + Namespace: "ns", + Annotations: map[string]string{common.AnnotationConfigOptimization: common.AnnotationValueDisabled}, + }} + annotated.SetLastAppliedPipeline(h) // the no-annotation hash was last applied + + unchanged, err := IsPipelineChanged(annotated) + require.NoError(t, err) + assert.False(t, unchanged) +} diff --git a/test/e2e/config_optimization_e2e_test.go b/test/e2e/config_optimization_e2e_test.go index 074c17a3..74f7bae4 100644 --- a/test/e2e/config_optimization_e2e_test.go +++ b/test/e2e/config_optimization_e2e_test.go @@ -170,6 +170,35 @@ var _ = Describe("Config Optimization", Label(config.LabelSmoke, config.LabelFas }, config.ServiceCreateTimeout, config.DefaultPollInterval).Should(Succeed()) }) + It("should opt out a single pipeline via its annotation", func() { + source := fmt.Sprintf("%s-opt-pipeline-3-logs", f.Namespace()) + + By("annotating one pipeline with config-optimization=disabled") + cmd := exec.Command("kubectl", "-n", f.Namespace(), "annotate", "vectorpipeline", "opt-pipeline-3", + "vector-operator.kaasops.io/config-optimization=disabled", "--overwrite") + _, err := utils.Run(cmd) + Expect(err).NotTo(HaveOccurred()) + + By("verifying that pipeline keeps its own source while the rest stay collapsed") + collapsed := fmt.Sprintf("%s-opt-pipeline-1-logs", f.Namespace()) + Eventually(func() error { + if err := f.VerifyAgentConfigContains("optimized-agent", source, `"optimizedSource-`); err != nil { + return err + } + // opt-pipeline-1 shares the namespace and stays collapsed, so its own source is gone + return f.VerifyAgentConfigNotContains("optimized-agent", collapsed) + }, config.ServiceCreateTimeout, config.DefaultPollInterval).Should(Succeed()) + + By("removing the annotation re-collapses the pipeline source") + cmd = exec.Command("kubectl", "-n", f.Namespace(), "annotate", "vectorpipeline", "opt-pipeline-3", + "vector-operator.kaasops.io/config-optimization-") + _, err = utils.Run(cmd) + Expect(err).NotTo(HaveOccurred()) + Eventually(func() error { + return f.VerifyAgentConfigNotContains("optimized-agent", source) + }, config.ServiceCreateTimeout, config.DefaultPollInterval).Should(Succeed()) + }) + It("should keep sources with different settings standalone", func() { By("creating a pipeline with an extra_label_selector") f.ApplyTestData("config-optimization/pipeline-with-selector.yaml")