Add var generation benchmark (#6028)

Author: swiatekm

internal/pkg/composable/benchmark_test.go

@@ -0,0 +1,77 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package composable
+
+import (
+	"maps"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"gopkg.in/yaml.v3"
+	"k8s.io/apimachinery/pkg/util/uuid"
+
+	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+
+	"github.com/stretchr/testify/require"
+)
+
+// BenchmarkGenerateVars100Pods checks the cost of generating vars with the kubernetes provider tracking 100 Pods.
+// This scenario does come up in reality, in particular in our internal Serverless clusters, and we've historically
+// had bad performance in it. Test data is taken almost directly from a real cluster.
+func BenchmarkGenerateVars100Pods(b *testing.B) {
+	log, err := logger.New("", false)
+	require.NoError(b, err)
+	c := controller{
+		contextProviders: make(map[string]*contextProviderState),
+		dynamicProviders: make(map[string]*dynamicProviderState),
+		logger:           log,
+	}
+	podCount := 100
+
+	providerDataFiles, err := os.ReadDir("./testdata")
+	require.NoError(b, err)
+
+	providerData := make(map[string]map[string]interface{}, len(providerDataFiles))
+	for _, providerDataFile := range providerDataFiles {
+		fileName := providerDataFile.Name()
+		providerName := strings.TrimSuffix(fileName, filepath.Ext(fileName))
+		rawData, err := os.ReadFile(filepath.Join("./testdata", fileName))
+		require.NoError(b, err)
+		var data map[string]interface{}
+		err = yaml.Unmarshal(rawData, &data)
+		require.NoError(b, err)
+		providerData[providerName] = data
+	}
+
+	for providerName, providerMapping := range providerData {
+		if providerName == "kubernetes" {
+			providerState := &dynamicProviderState{
+				mappings: make(map[string]dynamicProviderMapping),
+			}
+			for i := 0; i < podCount; i++ {
+				podData := maps.Clone(providerMapping)
+				podUID := uuid.NewUUID()
+				podMapping := dynamicProviderMapping{
+					mapping: podData,
+				}
+				providerState.mappings[string(podUID)] = podMapping
+			}
+			c.dynamicProviders[providerName] = providerState
+		} else {
+			providerState := &contextProviderState{
+				mapping: providerData[providerName],
+			}
+			c.contextProviders[providerName] = providerState
+		}
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		c.generateVars(mapstr.M{})
+	}
+}

internal/pkg/composable/controller.go

@@ -212,26 +212,7 @@ func (c *controller) Run(ctx context.Context) error {
 
 		c.logger.Debugf("Computing new variable state for composable inputs")
 
-		// build the vars list of mappings
-		vars := make([]*transpiler.Vars, 1)
-		mapping := map[string]interface{}{}
-		for name, state := range c.contextProviders {
-			mapping[name] = state.Current()
-		}
-		// this is ensured not to error, by how the mappings states are verified
-		vars[0], _ = transpiler.NewVars("", mapping, fetchContextProviders)
-
-		// add to the vars list for each dynamic providers mappings
-		for name, state := range c.dynamicProviders {
-			for _, mappings := range state.Mappings() {
-				local, _ := cloneMap(mapping) // will not fail; already been successfully cloned once
-				local[name] = mappings.mapping
-				id := fmt.Sprintf("%s-%s", name, mappings.id)
-				// this is ensured not to error, by how the mappings states are verified
-				v, _ := transpiler.NewVarsWithProcessors(id, local, name, mappings.processors, fetchContextProviders)
-				vars = append(vars, v)
-			}
-		}
+		vars := c.generateVars(fetchContextProviders)
 
 	UPDATEVARS:
 		for {
@@ -291,6 +272,30 @@ func (c *controller) Close() {
 	}
 }
 
+func (c *controller) generateVars(fetchContextProviders mapstr.M) []*transpiler.Vars {
+	// build the vars list of mappings
+	vars := make([]*transpiler.Vars, 1)
+	mapping := map[string]interface{}{}
+	for name, state := range c.contextProviders {
+		mapping[name] = state.Current()
+	}
+	// this is ensured not to error, by how the mappings states are verified
+	vars[0], _ = transpiler.NewVars("", mapping, fetchContextProviders)
+
+	// add to the vars list for each dynamic providers mappings
+	for name, state := range c.dynamicProviders {
+		for _, mappings := range state.Mappings() {
+			local, _ := cloneMap(mapping) // will not fail; already been successfully cloned once
+			local[name] = mappings.mapping
+			id := fmt.Sprintf("%s-%s", name, mappings.id)
+			// this is ensured not to error, by how the mappings states are verified
+			v, _ := transpiler.NewVarsWithProcessors(id, local, name, mappings.processors, fetchContextProviders)
+			vars = append(vars, v)
+		}
+	}
+	return vars
+}
+
 type contextProviderState struct {
 	context.Context
 

internal/pkg/composable/testdata/agent.yaml

@@ -0,0 +1,7 @@
+id: 36806a7e-1981-4a55-a3b6-ec9dd1ad1a4c
+unprivileged: false
+version:
+  build_time: 2024-09-12 21:13:27 +0000 UTC
+  commit: d99b09b0769f6f34428321eedb00c0b4339c202b
+  snapshot: true
+  version: 9.0.0

internal/pkg/composable/testdata/env.yaml

@@ -0,0 +1,50 @@
+API_KEY: ""
+BEAT_SETUID_AS: elastic-agent
+ELASTIC_CONTAINER: "true"
+ELASTIC_NETINFO: "false"
+ES_HOST: https://localhost:9243
+ES_PASSWORD:
+ES_USERNAME: elastic
+GODEBUG: madvdontneed=1
+HOME: /root
+HOSTNAME: kind
+KUBE_DNS_PORT: udp://10.96.0.10:53
+KUBE_DNS_PORT_53_TCP: tcp://10.96.0.10:53
+KUBE_DNS_PORT_53_TCP_ADDR: 10.96.0.10
+KUBE_DNS_PORT_53_TCP_PORT: "53"
+KUBE_DNS_PORT_53_TCP_PROTO: tcp
+KUBE_DNS_PORT_53_UDP: udp://10.96.0.10:53
+KUBE_DNS_PORT_53_UDP_ADDR: 10.96.0.10
+KUBE_DNS_PORT_53_UDP_PORT: "53"
+KUBE_DNS_PORT_53_UDP_PROTO: udp
+KUBE_DNS_PORT_9153_TCP: tcp://10.96.0.10:9153
+KUBE_DNS_PORT_9153_TCP_ADDR: 10.96.0.10
+KUBE_DNS_PORT_9153_TCP_PORT: "9153"
+KUBE_DNS_PORT_9153_TCP_PROTO: tcp
+KUBE_DNS_SERVICE_HOST: 10.96.0.10
+KUBE_DNS_SERVICE_PORT: "53"
+KUBE_DNS_SERVICE_PORT_DNS: "53"
+KUBE_DNS_SERVICE_PORT_DNS_TCP: "53"
+KUBE_DNS_SERVICE_PORT_METRICS: "9153"
+KUBERNETES_PORT: tcp://10.96.0.1:443
+KUBERNETES_PORT_443_TCP: tcp://10.96.0.1:443
+KUBERNETES_PORT_443_TCP_ADDR: 10.96.0.1
+KUBERNETES_PORT_443_TCP_PORT: "443"
+KUBERNETES_PORT_443_TCP_PROTO: tcp
+KUBERNETES_SERVICE_HOST: 10.96.0.1
+KUBERNETES_SERVICE_PORT: "443"
+KUBERNETES_SERVICE_PORT_HTTPS: "443"
+LIBBEAT_MONITORING_CGROUPS_HIERARCHY_OVERRIDE: /
+METRICS_SERVER_PORT: tcp://10.96.195.62:443
+METRICS_SERVER_PORT_443_TCP: tcp://10.96.195.62:443
+METRICS_SERVER_PORT_443_TCP_ADDR: 10.96.195.62
+METRICS_SERVER_PORT_443_TCP_PORT: "443"
+METRICS_SERVER_PORT_443_TCP_PROTO: tcp
+METRICS_SERVER_SERVICE_HOST: 10.96.195.62
+METRICS_SERVER_SERVICE_PORT: "443"
+METRICS_SERVER_SERVICE_PORT_HTTPS: "443"
+NODE_NAME: kind
+PATH: /usr/share/elastic-agent:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+POD_NAME: elastic-agent-standalone-99pcr
+PWD: /usr/share/elastic-agent
+SHLVL: "0"