Feat/config optimization sources

cmd/manager/main.go

@@ -82,6 +82,7 @@ func main() {
 	var configCheckTimeout time.Duration
 	var enableReconciliationInvalidPipelines bool
 	var reconciliationRetryDelay time.Duration
+	var enableConfigOptimization bool
 
 	flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
 		"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
@@ -99,6 +100,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")
 
 	opts := zap.Options{
 		Development: true,
@@ -201,12 +203,13 @@ func main() {
 	defer close(vectorAgentEventCh)
 
 	if err = (&controller.VectorReconciler{
-		Client:             mgr.GetClient(),
-		Scheme:             mgr.GetScheme(),
-		Clientset:          clientset,
-		ConfigCheckTimeout: configCheckTimeout,
-		DiscoveryClient:    dc,
-		EventChan:          vectorAgentEventCh,
+		Client:                   mgr.GetClient(),
+		Scheme:                   mgr.GetScheme(),
+		Clientset:                clientset,
+		ConfigCheckTimeout:       configCheckTimeout,
+		EnableConfigOptimization: enableConfigOptimization,
+		DiscoveryClient:          dc,
+		EventChan:                vectorAgentEventCh,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Vector")
 		os.Exit(1)
@@ -224,6 +227,7 @@ func main() {
 		Scheme:                                   mgr.GetScheme(),
 		Clientset:                                clientset,
 		ConfigCheckTimeout:                       configCheckTimeout,
+		EnableConfigOptimization:                 enableConfigOptimization,
 		VectorAgentEventCh:                       vectorAgentsPipelineEventCh,
 		VectorAggregatorsEventCh:                 vectorAggregatorsPipelineEventCh,
 		ClusterVectorAggregatorsEventCh:          clusterVectorAggregatorsPipelineEventCh,

docs/config-optimization.md

@@ -0,0 +1,46 @@
+# Config optimization
+
+With many namespaced VectorPipelines the generated agent config contains one `kubernetes_logs` source per pipeline source. Inside vector every `kubernetes_logs` source runs its own kube-apiserver clients: three watch streams (Pods, Namespaces, Nodes) and its own pod metadata cache, plus a separate log file scanner. With N pipelines every agent pod keeps 3×N watch connections and N copies of the node's pod metadata. On large clusters this dominates kube-apiserver traffic and agent memory.
+
+Sources optimization collapses such sources into one. It is a controller-level
+feature flag (off by default):
+
+```yaml
+# helm values
+args:
+  - "-enable-config-optimization"
+```
+
+A particular Vector CR can be opted out (e.g. for a staged rollout or an agent
+image too old to compile the generated routing):
+
+```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.
+
+## What it does
+
+Sources which differ **only in the watched namespace** (same type, label/field selectors and other options — the form generated for namespaced VectorPipelines) are grouped, and every group is replaced with:
+
+- a single `kubernetes_logs` source watching the union of the namespaces (`kubernetes.io/metadata.name in (ns1,ns2,...)`);
+- route transforms which split the stream back per namespace, so every pipeline receives exactly the events it received before. For large groups the routing is two-level (a remap computes an md5-based bucket of the namespace, a first-level route selects the bucket, a second-level route selects the namespace) to keep the number of conditions evaluated per event near `2*sqrt(N)` instead of `N`.
+
+Inputs of pipeline transforms and sinks are rewired automatically. Sources with unique settings (for example a custom `extra_label_selector`) are left untouched. Groups larger than 1000 namespaces are split to keep the generated namespace selector short.
+
+An event matching several pipelines is still delivered to all of them. The log file of such a pod is now read from disk once instead of once per pipeline.
+
+## Effect
+
+Measured on a 1000-pipeline single-node cluster (vector 0.48): watch requests to the kube-apiserver drop by three orders of magnitude (the periodic reconnect waves of thousands of watch streams disappear), agent memory drops ~20×, agent CPU and delivery throughput stay at parity under nominal load.
+
+## Requirements and notes
+
+- Namespace selection relies on the `kubernetes.io/metadata.name` label, which is set automatically on every namespace since Kubernetes 1.21.
+- The optimized source names are derived from a hash of the group settings and do not depend on the namespace list: pipelines can be added and removed without renaming the source, so vector file checkpoints survive such changes.
+- **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.** Plan the rollout accordingly (e.g. enable it together with deduplication on the storage side, or accept the duplicates).
+- 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.

helm/charts/vector-operator/values.yaml

@@ -72,6 +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)
 
 vector:
   enable: false

internal/common/annotations.go

@@ -3,4 +3,7 @@ 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 = "vector-operator.kaasops.io/config-optimization"
 )

internal/config/agent.go

@@ -83,6 +83,10 @@ func buildAgentConfig(params VectorConfigParams, pipelines ...pipeline.Pipeline)
 		}
 	}
 
+	if params.OptimizeSources {
+		optimizeAgentSources(cfg)
+	}
+
 	// Add exporter pipeline
 	if params.InternalMetrics && !isExporterSinkExists(cfg.Sinks) {
 		cfg.Sources[DefaultInternalMetricsSourceName] = defaultInternalMetricsSource

internal/config/agent_optimize.go

@@ -0,0 +1,273 @@
+/*
+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 config
+
+import (
+	"crypto/md5"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/kaasops/vector-operator/internal/utils/hash"
+	"github.com/kaasops/vector-operator/internal/utils/k8s"
+)
+
+const (
+	optimizedSourcePrefix   = "optimizedSource"
+	optimizedBucketerPrefix = "optimizedBucketer"
+	optimizedRouterPrefix   = "optimizedRouter"
+
+	// For small groups a single flat route transform is cheaper per event
+	// than the two-level (bucketed) routing.
+	flatRoutingThreshold = 16
+	maxRouteBuckets      = 256
+
+	// Larger groups are split to keep the generated namespace selector
+	// (one name per namespace) well below any request-URI limits.
+	maxNamespacesPerSource = 1000
+)
+
+// optimizeAgentSources collapses kubernetes_logs sources scoped to a single namespace
+// (the form generated for namespaced VectorPipelines) into one source per group of
+// 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) {
+	groups := make(map[string][]*Source)
+	for _, src := range cfg.Sources {
+		if src.Type != KubernetesLogsType {
+			continue
+		}
+		if _, ok := singleNamespaceOf(src); !ok {
+			continue
+		}
+		sig := sourceSignature(src)
+		groups[sig] = append(groups[sig], src)
+	}
+
+	signatures := make([]string, 0, len(groups))
+	for sig := range groups {
+		signatures = append(signatures, sig)
+	}
+	sort.Strings(signatures)
+
+	for _, sig := range signatures {
+		sources := groups[sig]
+		if len(sources) < 2 {
+			continue
+		}
+		sort.Slice(sources, func(i, j int) bool { return sources[i].Name < sources[j].Name })
+		namespaces := sortedNamespacesOf(sources)
+		gid := fmt.Sprintf("%08x", hash.Get([]byte(sig)))
+		// hash collision of two group signatures: extremely unlikely, but a silent
+		// component-name clash would corrupt the config; chunked groups occupy
+		// "-<chunk>" suffixed names, so the first chunk is checked too
+		for cfg.Sources[fmt.Sprintf("%s-%s", optimizedSourcePrefix, gid)] != nil ||
+			cfg.Sources[fmt.Sprintf("%s-%s-0", optimizedSourcePrefix, gid)] != nil {
+			gid += "x"
+		}
+		chunks := (len(namespaces) + maxNamespacesPerSource - 1) / maxNamespacesPerSource
+		if chunks == 1 {
+			collapseSourceGroup(cfg, gid, namespaces, sources)
+		} else {
+			nsChunk := make(map[string]int, len(namespaces))
+			for i, ns := range namespaces {
+				nsChunk[ns] = i / maxNamespacesPerSource
+			}
+			for c := 0; c < chunks; c++ {
+				chunkNamespaces := namespaces[c*maxNamespacesPerSource : min((c+1)*maxNamespacesPerSource, len(namespaces))]
+				chunkSources := make([]*Source, 0, len(chunkNamespaces))
+				for _, src := range sources {
+					ns, _ := singleNamespaceOf(src)
+					if nsChunk[ns] == c {
+						chunkSources = append(chunkSources, src)
+					}
+				}
+				collapseSourceGroup(cfg, fmt.Sprintf("%s-%d", gid, c), chunkNamespaces, chunkSources)
+			}
+		}
+		cfg.internal.optimizedSources += len(sources)
+		cfg.internal.sourceGroups += chunks
+	}
+}
+
+// collapseSourceGroup replaces the group sources with one source watching the union
+// of their namespaces and routes its events back to the original per-source streams.
+// Routes are keyed by namespace: consumers of all pipelines of a namespace share one
+// route output, which keeps the number of conditions evaluated per event equal to
+// the number of namespaces, not sources.
+func collapseSourceGroup(cfg *VectorConfig, gid string, namespaces []string, sources []*Source) {
+	collapsed := *sources[0]
+	collapsed.Name = fmt.Sprintf("%s-%s", optimizedSourcePrefix, gid)
+	collapsed.ExtraNamespaceLabelSelector = fmt.Sprintf("kubernetes.io/metadata.name in (%s)", strings.Join(namespaces, ","))
+
+	nsOutput := make(map[string]string, len(namespaces))
+	if len(namespaces) <= flatRoutingThreshold {
+		router := fmt.Sprintf("%s-%s", optimizedRouterPrefix, gid)
+		cfg.Transforms[router] = &Transform{
+			Name:    router,
+			Type:    RouteTransformType,
+			Inputs:  []string{collapsed.Name},
+			Options: map[string]interface{}{"route": namespaceRoutes(namespaces)},
+		}
+		for _, ns := range namespaces {
+			nsOutput[ns] = fmt.Sprintf("%s.%s", router, ns)
+		}
+	} else {
+		buckets := bucketCount(len(namespaces))
+		bucketer := fmt.Sprintf("%s-%s", optimizedBucketerPrefix, gid)
+		cfg.Transforms[bucketer] = &Transform{
+			Name:   bucketer,
+			Type:   RemapTransformType,
+			Inputs: []string{collapsed.Name},
+			Options: map[string]interface{}{
+				// NB: the mod() function, not the % operator: after an expression
+				// VRL parses `%` as the start of a metadata query.
+				"source": fmt.Sprintf("%%bucket = mod(parse_int!(slice!(md5(string!(.kubernetes.pod_namespace)), 0, 2), base: 16), %d)", buckets),
+			},
+		}
+		bucketNamespaces := make(map[int][]string)
+		for _, ns := range namespaces {
+			b := nsBucket(ns, buckets)
+			bucketNamespaces[b] = append(bucketNamespaces[b], ns)
+		}
+		l1 := fmt.Sprintf("%s-%s-l1", optimizedRouterPrefix, gid)
+		l1Routes := make(map[string]string, len(bucketNamespaces))
+		for b, members := range bucketNamespaces {
+			key := fmt.Sprintf("%d", b)
+			l1Routes[key] = fmt.Sprintf("%%bucket == %d", b)
+			l2 := fmt.Sprintf("%s-%s-%s", optimizedRouterPrefix, gid, key)
+			cfg.Transforms[l2] = &Transform{
+				Name:    l2,
+				Type:    RouteTransformType,
+				Inputs:  []string{fmt.Sprintf("%s.%s", l1, key)},
+				Options: map[string]interface{}{"route": namespaceRoutes(members)},
+			}
+			for _, ns := range members {
+				nsOutput[ns] = fmt.Sprintf("%s.%s", l2, ns)
+			}
+		}
+		cfg.Transforms[l1] = &Transform{
+			Name:    l1,
+			Type:    RouteTransformType,
+			Inputs:  []string{bucketer},
+			Options: map[string]interface{}{"route": l1Routes},
+		}
+	}
+
+	routeOutput := make(map[string]string, len(sources))
+	for _, src := range sources {
+		ns, _ := singleNamespaceOf(src)
+		routeOutput[src.Name] = nsOutput[ns]
+		delete(cfg.Sources, src.Name)
+	}
+	cfg.Sources[collapsed.Name] = &collapsed
+
+	for _, t := range cfg.Transforms {
+		t.Inputs = rewriteInputs(t.Inputs, routeOutput)
+	}
+	for _, s := range cfg.Sinks {
+		s.Inputs = rewriteInputs(s.Inputs, routeOutput)
+	}
+}
+
+// namespaceRoutes returns a route condition per namespace. An event is sent to
+// every consumer of its namespace route, which preserves the fan-out semantics
+// of the original per-pipeline sources.
+func namespaceRoutes(namespaces []string) map[string]string {
+	routes := make(map[string]string, len(namespaces))
+	for _, ns := range namespaces {
+		routes[ns] = fmt.Sprintf(".kubernetes.pod_namespace == %q", ns)
+	}
+	return routes
+}
+
+// singleNamespaceOf reports the namespace if the source is scoped to exactly one
+// namespace by name, i.e. has the selector generated for namespaced VectorPipelines.
+func singleNamespaceOf(src *Source) (string, bool) {
+	ns := strings.TrimPrefix(src.ExtraNamespaceLabelSelector, k8s.NamespaceNameToLabel(""))
+	if ns == "" || ns == src.ExtraNamespaceLabelSelector || strings.ContainsAny(ns, ",=!() ") {
+		return "", false
+	}
+	return ns, true
+}
+
+func sortedNamespacesOf(sources []*Source) []string {
+	seen := make(map[string]bool, len(sources))
+	namespaces := make([]string, 0, len(sources))
+	for _, src := range sources {
+		ns, _ := singleNamespaceOf(src)
+		if !seen[ns] {
+			seen[ns] = true
+			namespaces = append(namespaces, ns)
+		}
+	}
+	sort.Strings(namespaces)
+	return namespaces
+}
+
+// sourceSignature identifies sources which differ only in the watched namespace.
+func sourceSignature(src *Source) string {
+	b, _ := json.Marshal(struct {
+		Type               string
+		ExtraLabelSelector string
+		ExtraFieldSelector string
+		UseApiServerCache  bool
+		Options            map[string]any
+	}{src.Type, src.ExtraLabelSelector, src.ExtraFieldSelector, src.UseApiServerCache, src.Options})
+	return string(b)
+}
+
+// bucketCount returns the number of routing buckets: the power of two near
+// sqrt(n), which minimizes route conditions evaluated per event (n/buckets + buckets).
+func bucketCount(n int) int {
+	buckets := 1
+	for buckets*buckets < n && buckets < maxRouteBuckets {
+		buckets *= 2
+	}
+	if half := buckets / 2; half > 0 && half+(n+half-1)/half <= buckets+(n+buckets-1)/buckets {
+		return half
+	}
+	return buckets
+}
+
+// nsBucket mirrors the bucket expression generated for the remap transform:
+// the first byte of the md5 hex digest modulo the bucket count.
+func nsBucket(ns string, buckets int) int {
+	return int(md5.Sum([]byte(ns))[0]) % buckets
+}
+
+// rewriteInputs replaces inputs referring to collapsed sources with the route
+// outputs and dedupes the result: several identical sources of one pipeline map
+// to the same route output (legacy delivered such events once per source; the
+// collapsed config delivers them once).
+func rewriteInputs(inputs []string, routeOutput map[string]string) []string {
+	seen := make(map[string]bool, len(inputs))
+	result := inputs[:0]
+	for _, input := range inputs {
+		if out, ok := routeOutput[input]; ok {
+			input = out
+		}
+		if !seen[input] {
+			seen[input] = true
+			result = append(result, input)
+		}
+	}
+	return result
+}