Merge branch 'main' into dependabot/cargo/loadgen/kitchen-sink-gen/anstream-0.6.20

Author: chaptersix

.github/workflows/all-docker-images.yml

@@ -1,4 +1,8 @@
 name: Build all language docker images
+
+permissions:
+  contents: read
+
 on:
   workflow_call:
     # TODO: Can eventually support repo refs too rather than just versions if/when we need that.

.github/workflows/ci.yml

@@ -1,7 +1,13 @@
 name: Continuous Integration
 
+permissions:
+  contents: read
+  actions: write
+  checks: write
+
 env:
-  MISE_VERSION: v2025.8.1
+  MISE_VERSION: v2025.12.1
+  MISE_NODE_VERIFY: no
 
 on: # rebuild any PRs and main branch changes
   pull_request:

.github/workflows/docker-images.yml

@@ -1,4 +1,8 @@
 name: Build docker images
+
+permissions:
+  contents: read
+
 on:
   workflow_call:
     inputs:

cmd/cli/cleanup_scenario.go

@@ -49,9 +49,9 @@ type scenarioCleaner struct {
 func (c *scenarioCleaner) addCLIFlags(fs *pflag.FlagSet) {
 	c.scenario.AddCLIFlags(fs)
 	fs.DurationVar(&c.pollInterval, "poll-interval", time.Second, "Interval for polling completion of job")
-	c.clientOptions.AddCLIFlags(fs)
-	c.metricsOptions.AddCLIFlags(fs, "")
-	c.loggingOptions.AddCLIFlags(fs)
+	fs.AddFlagSet(c.clientOptions.FlagSet())
+	fs.AddFlagSet(c.metricsOptions.FlagSet(""))
+	fs.AddFlagSet(c.loggingOptions.FlagSet())
 }
 
 func (c *scenarioCleaner) run(ctx context.Context) error {
@@ -62,8 +62,8 @@ func (c *scenarioCleaner) run(ctx context.Context) error {
 	} else if c.scenario.RunID == "" {
 		return fmt.Errorf("run ID not found")
 	}
-	metrics := c.metricsOptions.MustCreateMetrics(c.logger)
-	defer metrics.Shutdown(ctx)
+	metrics := c.metricsOptions.MustCreateMetrics(ctx, c.logger)
+	defer metrics.Shutdown(ctx, c.logger)
 	client := c.clientOptions.MustDial(metrics, c.logger)
 	defer client.Close()
 	taskQueue := loadgen.TaskQueueForRun(c.scenario.RunID)

cmd/cli/prepare_worker.go

@@ -42,7 +42,7 @@ type workerBuilder struct {
 func (b *workerBuilder) addCLIFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&b.DirName, "dir-name", "", "Directory name for prepared worker")
 	b.SdkOptions.AddCLIFlags(fs)
-	b.loggingOptions.AddCLIFlags(fs)
+	fs.AddFlagSet(b.loggingOptions.FlagSet())
 }
 
 func (b *workerBuilder) preRun() {

cmd/cli/run_scenario.go

@@ -57,15 +57,17 @@ type scenarioRunConfig struct {
 	timeout                       time.Duration
 	doNotRegisterSearchAttributes bool
 	ignoreAlreadyStarted          bool
+	exportHistoriesDir            string
+	exportHistoriesFilter         string
 }
 
 func (r *scenarioRunner) addCLIFlags(fs *pflag.FlagSet) {
 	r.scenario.AddCLIFlags(fs)
 	r.scenarioRunConfig.addCLIFlags(fs)
 	fs.DurationVar(&r.connectTimeout, "connect-timeout", 0, "Duration to try to connect to server before failing")
-	r.clientOptions.AddCLIFlags(fs)
-	r.metricsOptions.AddCLIFlags(fs, "")
-	r.loggingOptions.AddCLIFlags(fs)
+	fs.AddFlagSet(r.clientOptions.FlagSet())
+	fs.AddFlagSet(r.metricsOptions.FlagSet(""))
+	fs.AddFlagSet(r.loggingOptions.FlagSet())
 }
 
 func (r *scenarioRunConfig) addCLIFlags(fs *pflag.FlagSet) {
@@ -84,6 +86,8 @@ func (r *scenarioRunConfig) addCLIFlags(fs *pflag.FlagSet) {
 			"If the search attributes are not registed by the scenario they must be registered through some other method")
 	fs.BoolVar(&r.ignoreAlreadyStarted, "ignore-already-started", false,
 		"Ignore if a workflow with the same ID already exists. A Scenario may choose to override this behavior.")
+	fs.StringVar(&r.exportHistoriesDir, "export-histories-dir", "", "Export workflow histories to this directory")
+	fs.StringVar(&r.exportHistoriesFilter, "export-histories-filter", "all", "Filter which workflows are exported by execution status (options: 'failed', 'terminated', 'failed,terminated', 'all'). Default is 'all'")
 }
 
 func (r *scenarioRunner) preRun() {
@@ -121,8 +125,8 @@ func (r *scenarioRunner) run(ctx context.Context) error {
 		scenarioOptions[key] = value
 	}
 
-	metrics := r.metricsOptions.MustCreateMetrics(r.logger)
-	defer metrics.Shutdown(ctx)
+	metrics := r.metricsOptions.MustCreateMetrics(ctx, r.logger)
+	defer metrics.Shutdown(ctx, r.logger)
 	start := time.Now()
 	var client client.Client
 	var err error
@@ -145,9 +149,16 @@ func (r *scenarioRunner) run(ctx context.Context) error {
 		return fmt.Errorf("failed to get root directory: %w", err)
 	}
 
+	// Generate a random execution ID to ensure no two executions with the same RunID collide
+	executionID, err := generateExecutionID()
+	if err != nil {
+		return fmt.Errorf("failed to generate execution ID: %w", err)
+	}
+
 	scenarioInfo := loadgen.ScenarioInfo{
 		ScenarioName:   r.scenario.Scenario,
 		RunID:          r.scenario.RunID,
+		ExecutionID:    executionID,
 		Logger:         r.logger,
 		MetricsHandler: metrics.NewHandler(),
 		Client:         client,
@@ -164,11 +175,19 @@ func (r *scenarioRunner) run(ctx context.Context) error {
 		ScenarioOptions: scenarioOptions,
 		Namespace:       r.clientOptions.Namespace,
 		RootPath:        repoDir,
+		ExportOptions: loadgen.ExportOptions{
+			ExportHistoriesDir:    r.exportHistoriesDir,
+			ExportHistoriesFilter: r.exportHistoriesFilter,
+		},
 	}
 	executor := scenario.ExecutorFn()
 	err = executor.Run(ctx, scenarioInfo)
 	if err != nil {
 		return fmt.Errorf("failed scenario: %w", err)
 	}
+	err = loadgen.ExportWorkflowHistories(ctx, scenarioInfo)
+	if err != nil {
+		scenarioInfo.Logger.Errorf("Error exporting workflow histories:\n %v", err)
+	}
 	return nil
 }

cmd/cli/run_scenario_with_worker.go

@@ -41,7 +41,7 @@ type workerWithScenarioRunner struct {
 func (r *workerWithScenarioRunner) addCLIFlags(fs *pflag.FlagSet) {
 	r.workerRunner.addCLIFlags(fs)
 	r.scenarioRunConfig.addCLIFlags(fs)
-	r.metricsOptions.AddCLIFlags(fs, "")
+	fs.AddFlagSet(r.metricsOptions.FlagSet(""))
 }
 
 func (r *workerWithScenarioRunner) preRun() {
@@ -71,19 +71,11 @@ func (r *workerWithScenarioRunner) run(ctx context.Context) error {
 
 	// Run scenario
 	scenarioRunner := scenarioRunner{
-		logger:   r.Logger,
-		scenario: r.ScenarioID,
-		scenarioRunConfig: scenarioRunConfig{
-			iterations:                    r.iterations,
-			duration:                      r.duration,
-			maxConcurrent:                 r.maxConcurrent,
-			maxIterationsPerSecond:        r.maxIterationsPerSecond,
-			scenarioOptions:               r.scenarioOptions,
-			timeout:                       r.timeout,
-			doNotRegisterSearchAttributes: r.doNotRegisterSearchAttributes,
-		},
-		clientOptions:  r.ClientOptions,
-		metricsOptions: r.metricsOptions,
+		logger:            r.Logger,
+		scenario:          r.ScenarioID,
+		scenarioRunConfig: r.scenarioRunConfig,
+		clientOptions:     r.ClientOptions,
+		metricsOptions:    r.metricsOptions,
 	}
 	scenarioErr := scenarioRunner.run(ctx)
 	cancel()

cmd/cli/run_worker.go

@@ -51,9 +51,9 @@ func (r *workerRunner) addCLIFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&r.EmbeddedServerAddress, "embedded-server-address", "", "Address to bind local embedded server to")
 	fs.IntVar(&r.TaskQueueIndexSuffixStart, "task-queue-suffix-index-start", 0, "Inclusive start for task queue suffix range")
 	fs.IntVar(&r.TaskQueueIndexSuffixEnd, "task-queue-suffix-index-end", 0, "Inclusive end for task queue suffix range")
-	r.ClientOptions.AddCLIFlags(fs)
-	r.MetricsOptions.AddCLIFlags(fs, "worker-")
-	r.WorkerOptions.AddCLIFlags(fs, "worker-")
+	fs.AddFlagSet(r.ClientOptions.FlagSet())
+	fs.AddFlagSet(r.MetricsOptions.FlagSet("worker-"))
+	fs.AddFlagSet(r.WorkerOptions.FlagSet())
 }
 
 func (r *workerRunner) preRun() {

cmd/cli/util.go

@@ -1,6 +1,8 @@
 package cli
 
 import (
+	"crypto/rand"
+	"encoding/hex"
 	"fmt"
 	"path/filepath"
 	"runtime"
@@ -16,3 +18,13 @@ func getRepoDir() (string, error) {
 	repoDir := filepath.Dir(cmdDir)  // project root
 	return repoDir, nil
 }
+
+// generateExecutionID generates a random execution ID to uniquely identify this particular
+// execution of a scenario. This ensures no two executions with the same RunID collide.
+func generateExecutionID() (string, error) {
+	bytes := make([]byte, 8) // 8 bytes = 16 hex characters
+	if _, err := rand.Read(bytes); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(bytes), nil
+}