Add multiple service deployer to benchmarks

internal/benchrunner/runners/system/report.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/elastic/elastic-package/internal/benchrunner/reporters"
 	"github.com/elastic/elastic-package/internal/elasticsearch/ingest"
+	"github.com/elastic/elastic-package/internal/packages"
 )
 
 type report struct {
@@ -31,6 +32,7 @@ type report struct {
 	}
 	Parameters struct {
 		PackageVersion      string
+		Deployer            string
 		Input               string
 		Vars                map[string]interface{}
 		DataStream          dataStream
@@ -47,8 +49,8 @@ type report struct {
 	TotalHits           int
 }
 
-func createReport(benchName, corporaFile string, s *scenario, sum *metricsSummary) (reporters.Reportable, error) {
-	r := newReport(benchName, corporaFile, s, sum)
+func createReport(benchName, corporaFile string, s *scenario, sum *metricsSummary, secretVarNames map[string]bool) (reporters.Reportable, error) {
+	r := newReport(benchName, corporaFile, s, sum, secretVarNames)
 	human := reporters.NewReport(s.Package, reportHumanFormat(r))
 
 	jsonBytes, err := reportJSONFormat(r)
@@ -63,7 +65,7 @@ func createReport(benchName, corporaFile string, s *scenario, sum *metricsSummar
 	return mr, nil
 }
 
-func newReport(benchName, corporaFile string, s *scenario, sum *metricsSummary) *report {
+func newReport(benchName, corporaFile string, s *scenario, sum *metricsSummary, secretVarNames map[string]bool) *report {
 	var report report
 	report.Info.Benchmark = benchName
 	report.Info.Description = s.Description
@@ -74,9 +76,13 @@ func newReport(benchName, corporaFile string, s *scenario, sum *metricsSummary)
 	report.Info.Duration = time.Duration(sum.CollectionEndTs-sum.CollectionStartTs) * time.Second
 	report.Info.GeneratedCorporaFile = corporaFile
 	report.Parameters.PackageVersion = s.Version
+	report.Parameters.Deployer = s.Deployer
 	report.Parameters.Input = s.Input
-	report.Parameters.Vars = s.Vars
-	report.Parameters.DataStream = s.DataStream
+	report.Parameters.Vars = maskSecretVars(s.Vars, secretVarNames)
+	report.Parameters.DataStream = dataStream{
+		Name: s.DataStream.Name,
+		Vars: maskSecretVars(s.DataStream.Vars, secretVarNames),
+	}
 	report.Parameters.WarmupTimePeriod = s.WarmupTimePeriod
 	report.Parameters.BenchmarkTimePeriod = s.BenchmarkTimePeriod
 	report.Parameters.WaitForDataTimeout = *s.WaitForDataTimeout
@@ -114,6 +120,7 @@ func reportHumanFormat(r *report) []byte {
 
 	pkvs := []interface{}{
 		"package version", r.Parameters.PackageVersion,
+		"deployer", r.Parameters.Deployer,
 		"input", r.Parameters.Input,
 	}
 
@@ -218,6 +225,52 @@ func reportHumanFormat(r *report) []byte {
 	return []byte(report.String())
 }
 
+func collectSecretVarNames(pkgManifest *packages.PackageManifest, dsManifest *packages.DataStreamManifest) map[string]bool {
+	secrets := make(map[string]bool)
+	for _, v := range pkgManifest.Vars {
+		if v.Secret {
+			secrets[v.Name] = true
+		}
+	}
+	for _, pt := range pkgManifest.PolicyTemplates {
+		for _, input := range pt.Inputs {
+			for _, v := range input.Vars {
+				if v.Secret {
+					secrets[v.Name] = true
+				}
+			}
+		}
+		for _, v := range pt.Vars {
+			if v.Secret {
+				secrets[v.Name] = true
+			}
+		}
+	}
+	for _, stream := range dsManifest.Streams {
+		for _, v := range stream.Vars {
+			if v.Secret {
+				secrets[v.Name] = true
+			}
+		}
+	}
+	return secrets
+}
+
+func maskSecretVars(vars map[string]interface{}, secretNames map[string]bool) map[string]interface{} {
+	if len(vars) == 0 || len(secretNames) == 0 {
+		return vars
+	}
+	masked := make(map[string]interface{}, len(vars))
+	for k, v := range vars {
+		if secretNames[k] {
+			masked[k] = "xxxx"
+		} else {
+			masked[k] = v
+		}
+	}
+	return masked
+}
+
 func renderBenchmarkTable(title string, kv ...interface{}) string {
 	t := table.NewWriter()
 	t.SetStyle(table.StyleRounded)

internal/benchrunner/runners/system/report_test.go

@@ -0,0 +1,78 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package system
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMaskSecretVars(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name        string
+		vars        map[string]interface{}
+		secretNames map[string]bool
+		expected    map[string]interface{}
+	}{
+		{
+			name: "masks secret vars and preserves non-secret vars",
+			vars: map[string]interface{}{
+				"api_key":   "super-secret-value",
+				"host":      "localhost",
+				"retries":   3,
+				"tls":       true,
+				"namespace": "default",
+			},
+			secretNames: map[string]bool{
+				"api_key": true,
+			},
+			expected: map[string]interface{}{
+				"api_key":   "xxxx",
+				"host":      "localhost",
+				"retries":   3,
+				"tls":       true,
+				"namespace": "default",
+			},
+		},
+		{
+			name: "returns vars unchanged when there are no secret names",
+			vars: map[string]interface{}{
+				"host": "localhost",
+			},
+			secretNames: map[string]bool{},
+			expected: map[string]interface{}{
+				"host": "localhost",
+			},
+		},
+		{
+			name: "returns empty vars map when vars are empty",
+			vars: map[string]interface{}{},
+			secretNames: map[string]bool{
+				"api_key": true,
+			},
+			expected: map[string]interface{}{},
+		},
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			original := make(map[string]interface{}, len(tc.vars))
+			for k, v := range tc.vars {
+				original[k] = v
+			}
+
+			actual := maskSecretVars(tc.vars, tc.secretNames)
+
+			assert.Equal(t, tc.expected, actual)
+			assert.Equal(t, original, tc.vars)
+		})
+	}
+}

internal/benchrunner/runners/system/runner.go

@@ -26,7 +26,9 @@ import (
 	"github.com/elastic/elastic-package/internal/benchrunner"
 	"github.com/elastic/elastic-package/internal/benchrunner/reporters"
 	"github.com/elastic/elastic-package/internal/benchrunner/runners/common"
+	pkgcommon "github.com/elastic/elastic-package/internal/common"
 	"github.com/elastic/elastic-package/internal/configuration/locations"
+	"github.com/elastic/elastic-package/internal/environment"
 	"github.com/elastic/elastic-package/internal/kibana"
 	"github.com/elastic/elastic-package/internal/logger"
 	"github.com/elastic/elastic-package/internal/multierror"
@@ -46,6 +48,8 @@ const (
 	defaultNamespace = "ep"
 )
 
+var prefixServiceBenchRunIDEnv = environment.WithElasticPackagePrefix("PREFIX_SERVICE_TEST_RUN_ID")
+
 type runner struct {
 	options  Options
 	scenario *scenario
@@ -58,7 +62,8 @@ type runner struct {
 	mcollector        *collector
 	corporaFile       string
 
-	service servicedeployer.DeployedService
+	service        servicedeployer.DeployedService
+	secretVarNames map[string]bool
 
 	// Execution order of following handlers is defined in runner.TearDown() method.
 	deletePolicyHandler     func(context.Context) error
@@ -146,15 +151,19 @@ func (r *runner) setUp(ctx context.Context) error {
 	serviceLogsDir := locationManager.ServiceLogDir()
 	r.svcInfo.Logs.Folder.Local = serviceLogsDir
 	r.svcInfo.Logs.Folder.Agent = ServiceLogsAgentDir
-	r.svcInfo.Test.RunID = common.NewRunID()
+	prefix := ""
+	if v, found := os.LookupEnv(prefixServiceBenchRunIDEnv); found && v != "" {
+		prefix = v
+	}
+	r.svcInfo.Test.RunID = pkgcommon.CreateTestRunIDWithPrefix(prefix)
 
 	outputDir, err := servicedeployer.CreateOutputDir(locationManager, r.svcInfo.Test.RunID)
 	if err != nil {
 		return fmt.Errorf("could not create output dir for terraform deployer %w", err)
 	}
 	r.svcInfo.OutputDir = outputDir
 
-	serviceName, err := r.serviceDefinedInConfig()
+	serviceName, deployerName, err := r.serviceDefinedInConfig()
 	if err != nil {
 		return fmt.Errorf("failed to determine if service is defined in config: %w", err)
 	}
@@ -163,7 +172,7 @@ func (r *runner) setUp(ctx context.Context) error {
 		// Just in the case service deployer is needed (input_service field), setup the service now so all the
 		// required information is available in r.svcInfo (e.g. hostname, port, etc).
 		// This info may be needed to render the variables in the configuration.
-		s, err := r.setupService(ctx, serviceName)
+		s, err := r.setupService(ctx, serviceName, deployerName)
 		if errors.Is(err, os.ErrNotExist) {
 			logger.Debugf("No service deployer defined for this benchmark")
 		} else if err != nil {
@@ -179,6 +188,15 @@ func (r *runner) setUp(ctx context.Context) error {
 	}
 	r.scenario = scenario
 
+	// If no deployer was explicitly set in the config but a service was used,
+	// resolve the actual deployer name (the only one present in the deploy dir).
+	if serviceName != "" && r.scenario.Deployer == "" {
+		devDeployDir := filepath.Clean(filepath.Join(r.options.PackageRoot, r.options.BenchPath, "deploy"))
+		if deployers, err := servicedeployer.FindAllServiceDeployers(devDeployDir); err == nil && len(deployers) == 1 {
+			r.scenario.Deployer = deployers[0]
+		}
+	}
+
 	if r.scenario.Corpora.Generator != nil {
 		var err error
 		r.generator, err = r.initializeGenerator(ctx)
@@ -221,6 +239,8 @@ func (r *runner) setUp(ctx context.Context) error {
 		return fmt.Errorf("reading data stream manifest failed: %w", err)
 	}
 
+	r.secretVarNames = collectSecretVarNames(pkgManifest, dataStreamManifest)
+
 	r.runtimeDataStream = fmt.Sprintf(
 		"%s-%s.%s-%s",
 		dataStreamManifest.Type,
@@ -263,19 +283,19 @@ func (r *runner) setUp(ctx context.Context) error {
 	return nil
 }
 
-func (r *runner) serviceDefinedInConfig() (string, error) {
+func (r *runner) serviceDefinedInConfig() (string, string, error) {
 	// Read of the configuration to know if a service deployer is needed.
 	// No need to render any template at this point.
 	scenario, err := readRawConfig(r.options.BenchPath, r.options.BenchName)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 
 	if scenario.Corpora.InputService == nil {
-		return "", nil
+		return "", "", nil
 	}
 
-	return scenario.Corpora.InputService.Name, nil
+	return scenario.Corpora.InputService.Name, scenario.Deployer, nil
 }
 
 func (r *runner) run(ctx context.Context) (report reporters.Reportable, err error) {
@@ -319,18 +339,18 @@ func (r *runner) run(ctx context.Context) (report reporters.Reportable, err erro
 		return nil, fmt.Errorf("can't reindex data: %w", err)
 	}
 
-	return createReport(r.options.BenchName, r.corporaFile, r.scenario, msum)
+	return createReport(r.options.BenchName, r.corporaFile, r.scenario, msum, r.secretVarNames)
 }
 
-func (r *runner) setupService(ctx context.Context, serviceName string) (servicedeployer.DeployedService, error) {
+func (r *runner) setupService(ctx context.Context, serviceName string, deployerName string) (servicedeployer.DeployedService, error) {
 	stackVersion, err := r.options.KibanaClient.Version()
 	if err != nil {
 		return nil, fmt.Errorf("cannot request Kibana version: %w", err)
 	}
 
 	// Setup service.
 	logger.Debug("Setting up service...")
-	devDeployDir := filepath.Clean(filepath.Join(r.options.BenchPath, "deploy"))
+	devDeployDir := filepath.Clean(filepath.Join(r.options.PackageRoot, r.options.BenchPath, "deploy"))
 	opts := servicedeployer.FactoryOptions{
 		PackageRoot:            r.options.PackageRoot,
 		DevDeployDir:           devDeployDir,
@@ -339,6 +359,7 @@ func (r *runner) setupService(ctx context.Context, serviceName string) (serviced
 		Type:                   servicedeployer.TypeBench,
 		StackVersion:           stackVersion.Version(),
 		DeployIndependentAgent: false,
+		DeployerName:           deployerName,
 	}
 	serviceDeployer, err := servicedeployer.Factory(opts)
 	if err != nil {

internal/benchrunner/runners/system/scenario.go

@@ -9,6 +9,8 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"slices"
+	"strings"
 	"time"
 
 	"github.com/aymerick/raymond"
@@ -18,6 +20,8 @@ import (
 	"github.com/elastic/elastic-package/internal/servicedeployer"
 )
 
+var allowedDeployerNames = []string{"docker", "k8s", "tf"}
+
 type scenario struct {
 	Package             string                 `config:"package" json:"package"`
 	Description         string                 `config:"description" json:"description"`
@@ -30,6 +34,7 @@ type scenario struct {
 	BenchmarkTimePeriod time.Duration          `config:"benchmark_time_period" json:"benchmark_time_period"`
 	WaitForDataTimeout  *time.Duration         `config:"wait_for_data_timeout" json:"wait_for_data_timeout"`
 	Corpora             corpora                `config:"corpora" json:"corpora"`
+	Deployer            string                 `config:"deployer" json:"deployer"` // Name of the service deployer to use for this scenario.
 }
 
 type dataStream struct {
@@ -113,6 +118,10 @@ func readConfig(benchPath string, scenario string, svcInfo *servicedeployer.Serv
 	if err := cfg.Unpack(c); err != nil {
 		return nil, fmt.Errorf("can't unpack scenario configuration: %s: %w", configPath, err)
 	}
+	if c.Deployer != "" && !slices.Contains(allowedDeployerNames, c.Deployer) {
+		return nil, fmt.Errorf("invalid deployer name %q in system benchmark configuration file %q, allowed values are: %s",
+			c.Deployer, configPath, strings.Join(allowedDeployerNames, ", "))
+	}
 	return c, nil
 }