Add multiple namespace support.

Author: robholland

README.md

@@ -22,16 +22,53 @@ The table below lists the environment variables available and the relevant Tempo
 
 | Environment Variable | Relevant Client or Worker option | Description |
 | --- | --- | --- |
-| TEMPORAL_GRPC_ENDPOINT | [ClientOptions.HostPort](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ClientOptions) | The Temporal Frontend GRPC endpoint |
-| TEMPORAL_TLS_KEY | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Key file |
-| TEMPORAL_TLS_CERT | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Cert file |
-| TEMPORAL_TLS_CA | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS CA Cert file |
-| TEMPORAL_NAMESPACE | [ClientOptions.Namespace](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ClientOptions) | The Temporal Namespace |
+| TEMPORAL_GRPC_ENDPOINT | [ClientOptions.HostPort](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ClientOptions) | The Temporal Frontend GRPC endpoint (supports comma-separated values for multiple namespaces) |
+| TEMPORAL_TLS_KEY | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Key file (supports comma-separated values for multiple namespaces) |
+| TEMPORAL_TLS_CERT | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Cert file (supports comma-separated values for multiple namespaces) |
+| TEMPORAL_TLS_CA | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS CA Cert file (supports comma-separated values for multiple namespaces) |
+| TEMPORAL_NAMESPACE | [ClientOptions.Namespace](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ClientOptions) | The Temporal Namespace (supports comma-separated values for multiple namespaces) |
 | TEMPORAL_TASK_QUEUE | [TaskQueue](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/worker#New) | The Temporal Task Queue |
 | TEMPORAL_WORKFLOW_TASK_POLLERS | [WorkerOptions.MaxConcurrentWorkflowTaskPollers](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#WorkerOptions) | Number of workflow task pollers |
 | TEMPORAL_ACTIVITY_TASK_POLLERS | [WorkerOptions.MaxConcurrentActivityTaskPollers](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#WorkerOptions) | Number of activity task pollers |
 | PROMETHEUS_ENDPOINT | n/a | The address to serve prometheus metrics on |
 
+#### Multi-Namespace Support
+
+The worker supports working with multiple namespaces simultaneously. This allows you to spread load across multiple namespaces with a single worker deployment, providing a more realistic load pattern.
+
+**Configuration Options:**
+
+1. **Same configuration for all namespaces:**
+   ```bash
+   export TEMPORAL_NAMESPACE="ns1,ns2,ns3"
+   export TEMPORAL_GRPC_ENDPOINT="temporal.example.com:7233"
+   export TEMPORAL_TLS_CERT="/path/to/cert.pem"
+   export TEMPORAL_TLS_KEY="/path/to/key.pem"
+   ```
+
+2. **Different configurations per namespace:**
+   ```bash
+   export TEMPORAL_NAMESPACE="ns1,ns2,ns3"
+   export TEMPORAL_GRPC_ENDPOINT="temporal1.example.com:7233,temporal2.example.com:7233,temporal3.example.com:7233"
+   export TEMPORAL_TLS_CERT="/certs/ns1.pem,/certs/ns2.pem,/certs/ns3.pem"
+   export TEMPORAL_TLS_KEY="/keys/ns1.key,/keys/ns2.key,/keys/ns3.key"
+   ```
+
+3. **Mixed configuration (some shared, some different):**
+   ```bash
+   export TEMPORAL_NAMESPACE="ns1,ns2,ns3"
+   export TEMPORAL_GRPC_ENDPOINT="temporal.example.com:7233"  # Same endpoint for all
+   export TEMPORAL_TLS_CERT="/certs/ns1.pem,/certs/ns2.pem"  # ns3 will reuse ns2's cert
+   export TEMPORAL_TLS_KEY="/keys/shared.key"                # Same key for all
+   ```
+
+**How it works:**
+- If you provide exactly as many values as namespaces, each namespace uses its corresponding value
+- If you provide only one value but multiple namespaces, that single value is reused for all namespaces
+- If you provide fewer values than namespaces, the last value is repeated for the remaining namespaces
+
+The worker will create a separate worker instance for each namespace, all running concurrently within the same process.
+
 #### Kubernetes Deployment
 
 There are several ways to deploy the worker in Kubernetes:
@@ -48,11 +85,23 @@ kubectl run benchmark-worker --image ghcr.io/temporalio/benchmark-workers:main \
     --env "TEMPORAL_ACTIVITY_TASK_POLLERS=8"
 ```
 
-2. **Using the example deployment YAML**:
+2. **Multi-namespace deployment example**:
+
+```
+kubectl run benchmark-worker --image ghcr.io/temporalio/benchmark-workers:main \
+    --image-pull-policy Always \
+    --env "TEMPORAL_GRPC_ENDPOINT=temporal-frontend.temporal:7233" \
+    --env "TEMPORAL_NAMESPACE=namespace1,namespace2,namespace3" \
+    --env "TEMPORAL_TASK_QUEUE=benchmark" \
+    --env "TEMPORAL_WORKFLOW_TASK_POLLERS=16" \
+    --env "TEMPORAL_ACTIVITY_TASK_POLLERS=8"
+```
+
+3. **Using the example deployment YAML**:
 
 We provide an [example deployment spec](./deployment.yaml) for you to customize to your requirements. Once you have edited the environment variables in the deployment.yaml you can create the deployment with `kubectl apply -f ./deployment.yaml`.
 
-3. **Using the Helm chart (Recommended)**:
+4. **Using the Helm chart (Recommended)**:
 
 We provide a Helm chart that can be installed from the GitHub Container Registry:
 
@@ -106,10 +155,10 @@ The table below lists the environment variables available and the relevant Tempo
 
 | Environment Variable | Relevant Client or Worker option | Description |
 | --- | --- | --- |
-| TEMPORAL_GRPC_ENDPOINT | [ClientOptions.HostPort](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ClientOptions) | The Temporal Frontend GRPC endpoint |
-| TEMPORAL_TLS_KEY | [ClientOptions.ConnectionOptions.TLS.Certificates](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Key file |
-| TEMPORAL_TLS_CERT | [ClientOptions.ConnectionOptions.TLS.Certificates](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Cert file |
-| TEMPORAL_TLS_CA | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS CA Cert file |
+| TEMPORAL_GRPC_ENDPOINT | [ClientOptions.HostPort](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ClientOptions) | The Temporal Frontend GRPC endpoint (supports comma-separated values for multiple namespaces) |
+| TEMPORAL_TLS_KEY | [ClientOptions.ConnectionOptions.TLS.Certificates](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Key file (supports comma-separated values for multiple namespaces) |
+| TEMPORAL_TLS_CERT | [ClientOptions.ConnectionOptions.TLS.Certificates](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS Cert file (supports comma-separated values for multiple namespaces) |
+| TEMPORAL_TLS_CA | [ClientOptions.ConnectionOptions.TLS](https://pkg.go.dev/go.temporal.io/sdk@v1.15.0/internal#ConnectionOptions) | Path to TLS CA Cert file (supports comma-separated values for multiple namespaces) |
 | PROMETHEUS_ENDPOINT | n/a | The address to serve prometheus metrics on |
 
 The runner is also configured via command line options:
@@ -119,20 +168,52 @@ Usage: runner [flags] [workflow input] ...
   -c int
     	concurrent workflows (default 10)
   -n string
-    	namespace (default "default")
+    	namespace (comma-separated list supported) (default "default")
+  -s string
+    	signal type
   -t string
     	workflow type
   -tq string
     	task queue (default "benchmark")
   -w	wait for workflows to complete (default true)
 ```
 
+#### Multi-Namespace Support in Runner
+
+The runner supports distributing workflow executions across multiple namespaces. This provides a more realistic load pattern by spreading the load across different namespaces.
+
+**Configuration Examples:**
+
+1. **Command line flag for multiple namespaces:**
+   ```bash
+   runner -n "namespace1,namespace2,namespace3" -t ExecuteActivity '{"Count":1,"Activity":"Sleep","Input":{"SleepTimeInSeconds":3}}'
+   ```
+
+2. **Environment variable for multiple namespaces:**
+   ```bash
+   export TEMPORAL_NAMESPACE="namespace1,namespace2,namespace3"
+   runner -t ExecuteActivity '{"Count":1,"Activity":"Sleep","Input":{"SleepTimeInSeconds":3}}'
+   ```
+
+3. **Different GRPC endpoints per namespace:**
+   ```bash
+   export TEMPORAL_NAMESPACE="ns1,ns2,ns3"
+   export TEMPORAL_GRPC_ENDPOINT="temporal1.example.com:7233,temporal2.example.com:7233,temporal3.example.com:7233"
+   runner -t ExecuteActivity '{"Count":1,"Activity":"Sleep","Input":{"SleepTimeInSeconds":3}}'
+   ```
+
+**How it works:**
+- The runner creates a separate client for each namespace
+- Workflow executions are distributed across namespaces using round-robin rotation
+- Each client can have different connection settings (endpoints, TLS certificates, etc.)
+
 To use the runner in a Kubernetes cluster you could use:
 
 ```
 kubectl run benchmark-runner --image ghcr.io/temporalio/benchmark-workers:main \
     --image-pull-policy Always \
     --env "TEMPORAL_GRPC_ENDPOINT=temporal-frontend.temporal:7233" \
+    --env "TEMPORAL_NAMESPACE=namespace1,namespace2,namespace3" \
     --command -- runner -t ExecuteActivity '{ "Count": 3, "Activity": "Echo", "Input": { "Message": "test" } }'
 ```
 
@@ -146,6 +227,12 @@ The worker provides the following workflows for you to use during benchmarking:
 
 This workflow takes a count, an activity name and an activity input. The activity `Activity` will be run `Count` times with the given `input`. If the activity returns an error the workflow will fail with that error.
 
+### ReceiveSignal
+
+`ReceiveSignal()`
+
+This workflow waits to receive a signal. It can be used with the runner's signal functionality to test signal-based workflows.
+
 ### DSLWorkflow
 
 `DSLWorkflow([]DSLStep)`

cmd/runner/main.go

@@ -9,6 +9,8 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/alitto/pond"
@@ -24,9 +26,33 @@ var nWorfklows = flag.Int("c", 10, "concurrent workflows")
 var sWorkflow = flag.String("t", "", "workflow type")
 var sSignalType = flag.String("s", "", "signal type")
 var bWait = flag.Bool("w", true, "wait for workflows to complete")
-var sNamespace = flag.String("n", "default", "namespace")
+var sNamespace = flag.String("n", "default", "namespace (comma-separated list supported)")
 var sTaskQueue = flag.String("tq", "benchmark", "task queue")
 
+// parseCommaSeparatedEnv parses a comma-separated environment variable and returns a slice
+// If there's only one value but multiple namespaces are needed, it reuses that value
+func parseCommaSeparatedEnv(envVar string, numNamespaces int) []string {
+	value := os.Getenv(envVar)
+	if value == "" {
+		return make([]string, numNamespaces)
+	}
+
+	values := strings.Split(value, ",")
+	for i, v := range values {
+		values[i] = strings.TrimSpace(v)
+	}
+
+	// If we have fewer values than namespaces, repeat the last value
+	if len(values) < numNamespaces {
+		lastValue := values[len(values)-1]
+		for len(values) < numNamespaces {
+			values = append(values, lastValue)
+		}
+	}
+
+	return values
+}
+
 func main() {
 	flag.Usage = func() {
 		fmt.Fprintf(flag.CommandLine.Output(), "Usage: %s [flags] [workflow input] ...\n", os.Args[0])
@@ -39,58 +65,90 @@ func main() {
 		log.Printf("WARNING: failed to set GOMAXPROCS: %v.\n", err)
 	}
 
-	namespace := *sNamespace
+	namespaces := *sNamespace
 	envNamespace := os.Getenv("TEMPORAL_NAMESPACE")
 	if envNamespace != "" && envNamespace != "default" {
-		namespace = envNamespace
+		namespaces = envNamespace
 	}
 
-	clientOptions := client.Options{
-		HostPort:  os.Getenv("TEMPORAL_GRPC_ENDPOINT"),
-		Namespace: namespace,
-		Logger:    NewNopLogger(),
+	// Parse comma-separated namespaces
+	namespaceList := strings.Split(namespaces, ",")
+	for i, ns := range namespaceList {
+		namespaceList[i] = strings.TrimSpace(ns)
 	}
 
-	tlsKeyPath := os.Getenv("TEMPORAL_TLS_KEY")
-	tlsCertPath := os.Getenv("TEMPORAL_TLS_CERT")
-	tlsCaPath := os.Getenv("TEMPORAL_TLS_CA")
+	log.Printf("Using namespaces: %v", namespaceList)
+
+	// Parse comma-separated configuration values
+	grpcEndpoints := parseCommaSeparatedEnv("TEMPORAL_GRPC_ENDPOINT", len(namespaceList))
+	tlsKeyPaths := parseCommaSeparatedEnv("TEMPORAL_TLS_KEY", len(namespaceList))
+	tlsCertPaths := parseCommaSeparatedEnv("TEMPORAL_TLS_CERT", len(namespaceList))
+	tlsCaPaths := parseCommaSeparatedEnv("TEMPORAL_TLS_CA", len(namespaceList))
+
+	// Create clients for each namespace
+	clients := make([]client.Client, len(namespaceList))
+	for i, namespace := range namespaceList {
+		clientOptions := client.Options{
+			HostPort:  grpcEndpoints[i],
+			Namespace: namespace,
+			Logger:    NewNopLogger(),
+		}
 
-	if tlsKeyPath != "" && tlsCertPath != "" {
-		tlsConfig := tls.Config{}
+		tlsKeyPath := tlsKeyPaths[i]
+		tlsCertPath := tlsCertPaths[i]
+		tlsCaPath := tlsCaPaths[i]
 
-		cert, err := tls.LoadX509KeyPair(tlsCertPath, tlsKeyPath)
-		if err != nil {
-			log.Fatalln("Unable to create key pair for TLS", err)
-		}
+		if tlsKeyPath != "" && tlsCertPath != "" {
+			tlsConfig := tls.Config{}
 
-		var tlsCaPool *x509.CertPool
-		if tlsCaPath != "" {
-			tlsCaPool = x509.NewCertPool()
-			b, err := os.ReadFile(tlsCaPath)
+			cert, err := tls.LoadX509KeyPair(tlsCertPath, tlsKeyPath)
 			if err != nil {
-				log.Fatalln("Failed reading server CA: %w", err)
-			} else if !tlsCaPool.AppendCertsFromPEM(b) {
-				log.Fatalln("Server CA PEM file invalid")
+				log.Fatalf("Unable to create key pair for TLS for namespace %s: %v", namespace, err)
+			}
+
+			var tlsCaPool *x509.CertPool
+			if tlsCaPath != "" {
+				tlsCaPool = x509.NewCertPool()
+				b, err := os.ReadFile(tlsCaPath)
+				if err != nil {
+					log.Fatalf("Failed reading server CA for namespace %s: %v", namespace, err)
+				} else if !tlsCaPool.AppendCertsFromPEM(b) {
+					log.Fatalf("Server CA PEM file invalid for namespace %s", namespace)
+				}
 			}
-		}
 
-		tlsConfig.Certificates = []tls.Certificate{cert}
-		tlsConfig.RootCAs = tlsCaPool
+			tlsConfig.Certificates = []tls.Certificate{cert}
+			tlsConfig.RootCAs = tlsCaPool
+
+			if os.Getenv("TEMPORAL_TLS_DISABLE_HOST_VERIFICATION") != "" {
+				tlsConfig.InsecureSkipVerify = true
+			}
 
-		if os.Getenv("TEMPORAL_TLS_DISABLE_HOST_VERIFICATION") != "" {
-			tlsConfig.InsecureSkipVerify = true
+			clientOptions.ConnectionOptions.TLS = &tlsConfig
 		}
 
-		clientOptions.ConnectionOptions.TLS = &tlsConfig
-	}
+		if os.Getenv("PROMETHEUS_ENDPOINT") != "" {
+			clientOptions.MetricsHandler = sdktally.NewMetricsHandler(newPrometheusScope(prometheus.Configuration{
+				ListenAddress: os.Getenv("PROMETHEUS_ENDPOINT"),
+				TimerType:     "histogram",
+			}))
+		}
 
-	if os.Getenv("PROMETHEUS_ENDPOINT") != "" {
-		clientOptions.MetricsHandler = sdktally.NewMetricsHandler(newPrometheusScope(prometheus.Configuration{
-			ListenAddress: os.Getenv("PROMETHEUS_ENDPOINT"),
-			TimerType:     "histogram",
-		}))
+		c, err := client.Dial(clientOptions)
+		if err != nil {
+			log.Fatalf("Unable to create client for namespace %s (endpoint: %s): %v", namespace, grpcEndpoints[i], err)
+		}
+		clients[i] = c
+		log.Printf("Created client for namespace: %s (endpoint: %s)", namespace, grpcEndpoints[i])
 	}
 
+	// Ensure all clients are closed on exit
+	defer func() {
+		for _, c := range clients {
+			c.Close()
+		}
+	}()
+
 	var input []interface{}
 	for _, a := range flag.Args() {
 		var i interface{}
@@ -101,18 +159,19 @@ func main() {
 		input = append(input, i)
 	}
 
-	c, err := client.Dial(clientOptions)
-	if err != nil {
-		log.Fatalln("Unable to create client", err)
-	}
-	defer c.Close()
-
 	pool := pond.New(*nWorfklows, 0)
 
+	// Counter for rotating among clients
+	var clientCounter uint64
+
 	var starter func() (client.WorkflowRun, error)
 
 	if *sSignalType != "" {
 		starter = func() (client.WorkflowRun, error) {
+			// Rotate among clients
+			clientIndex := atomic.AddUint64(&clientCounter, 1) % uint64(len(clients))
+			c := clients[clientIndex]
+
 			wID := uuid.New()
 			return c.SignalWithStartWorkflow(
 				context.Background(),
@@ -129,6 +188,10 @@ func main() {
 		}
 	} else {
 		starter = func() (client.WorkflowRun, error) {
+			// Rotate among clients
+			clientIndex := atomic.AddUint64(&clientCounter, 1) % uint64(len(clients))
+			c := clients[clientIndex]
+
 			return c.ExecuteWorkflow(
 				context.Background(),
 				client.StartWorkflowOptions{