Add bandwidth stress scenario

README.md

@@ -9,6 +9,8 @@ benchmark features and situations. Backwards compatibility may not be maintained
   profiles, and running against a specific SDK version.
 - **[Authoring scenarios](./docs/authoring-scenarios.md)** — writing a new scenario, choosing an
   executor, exposing options, and authoring conventions.
+- **[The bandwidth_stress scenario](./docs/bandwidth-stress.md)** — configurable payload traffic for
+  bandwidth limiter and storage calibration.
 - **[The throughput_stress scenario](./docs/throughput-stress.md)** — sleep activities, Nexus,
   standalone activities, and standalone Nexus operations.
 

docs/bandwidth-stress.md

@@ -0,0 +1,128 @@
+# Bandwidth stress scenario
+
+`bandwidth_stress` generates sustained, incompressible activity input and result payloads. It is intended
+for calibrating namespace bandwidth limits against downstream storage traffic.
+
+Each iteration starts one workflow and runs one or more remote payload activities. The default is one
+activity with a 100 KiB input and a 100 KiB result. Use standard run flags to control workflow rate,
+concurrency, and duration.
+
+The default shape produces two billable Actions: the workflow start and the activity schedule. It also
+writes two user payloads: the activity input and activity result. At the default size, this is 200 KiB
+of user payload per workflow and 100 KiB per Action. This is the initial replay hypothesis, not a claim
+that the original OMES workflow had this exact shape.
+
+The following command models the initial INC-1725 workload delta. The cell rose from approximately
+8,000 to 8,500 Actions per second before the incident to approximately 11,500 to 12,000 Actions per
+second during the incident. This scenario produces two Actions per workflow, so 1,500 workflow starts
+per second represents the lower bound of the observed increase. Each workflow carries 200 KiB of user
+payload across the activity input and result.
+
+```sh
+go run ./cmd/omes run-scenario-with-worker \
+  --scenario bandwidth_stress \
+  --language go \
+  --server-address s-server-test-bwrl-bandwidth.e2e.tmprl-test.cloud:7233 \
+  --namespace s-server-test-bwrl-bandwidth.e2e \
+  --tls \
+  --tls-server-name s-server-test-bwrl-bandwidth.e2e.tmprl-test.cloud \
+  --tls-cert-path <client-certificate-path> \
+  --tls-key-path <client-private-key-path> \
+  --run-id bandwidth-inc-1725 \
+  --duration 1h \
+  --max-iterations-per-second 1500 \
+  --max-concurrent 3000 \
+  --option activities-per-workflow=1 \
+  --option 'payload-distribution-json={"size":{"type":"fixed","value":"102400"}}'
+```
+
+Do not begin calibration at the incident rate. Use separate runs for each plateau so every stage has a
+distinct run ID and metric interval. Wait at least five minutes between stages, or until Cassandra query
+latency and errors return to the pretest baseline.
+
+| Stage | Iterations/s | Expected Actions/s | User payload MiB/s | Duration |
+| --- | ---: | ---: | ---: | ---: |
+| 25% | 375 | 750 | 73 | 10m |
+| 50% | 750 | 1,500 | 146 | 10m |
+| 65% | 975 | 1,950 | 190 | 10m |
+| 75% | 1,125 | 2,250 | 220 | 10m |
+| 85% | 1,275 | 2,550 | 249 | 10m |
+| 100% | 1,500 | 3,000 | 293 | 15m |
+
+For each stage, change `--max-iterations-per-second`, `--duration`, and `--run-id`. Keep the scenario
+options and worker configuration unchanged. The expected values assume one activity per workflow and
+two 100 KiB payloads. Confirm the Action rate from metrics because retries and failed starts can change
+the observed rate.
+
+The Cassandra transmitted write rate rose from approximately 10 to 20 MB/s to approximately 1.2 to
+1.4 GB/s. At 1,500 workflows per second, this scenario supplies approximately 293 MiB/s of serialized
+user payload. The replay must measure whether persistence amplifies that payload to the observed
+Cassandra delta. Do not derive the offered workflow rate directly from Cassandra transmitted bytes.
+
+Do not use the incident replay profile on a small scaffold cell. The initial `s-server-test-bwrl` cell has
+a 1,500 RPS cell limit and the namespace inherits a 500 APS limit. Without a deliberate capacity
+change, keep the offered rate at or below 250 workflows per second. A small-cell functional ramp can
+use 62, 125, 187, and 250 workflows per second. These stages validate payload charging and throttling,
+but they do not reproduce the incident's storage pressure.
+
+Stop the ramp when any signal approaches the incident condition instead of continuing automatically:
+
+| Signal | Incident condition |
+| --- | ---: |
+| Cassandra transmitted write bytes | 1.2 to 1.4 GB/s |
+| Cassandra query p99 | 100 to 120 ms |
+| Storage latency p99 | 48 to 55 ms |
+| Cassandra query errors | 15 to 35/s |
+| StartWorkflowExecution p99 | 175 to 185 ms |
+| RespondActivityTaskCompleted p99 | 80 to 95 ms |
+| RespondWorkflowTaskCompleted p99 | 50 to 60 ms |
+
+The first sustained latency or error increase identifies the pressure knee. Repeat the stage immediately
+below that knee for at least 30 minutes before selecting a limit. The complete incident profile is only
+needed after lower stages establish a safe operating range.
+
+The historical cell used one 14 CU C40i Astra database. Modeled capacity was 69K STPS for the cell,
+84K for Astra, 84K for History, 100K for WAL, and 120K for shards. History had 28 replicas. Storage
+grew from approximately 3.6 TB to 7.4 TB during the incident. Prefer a test cell with the same storage
+shape. If an exact match is unavailable, record every component capacity and compare pressure as a
+fraction of Astra capacity rather than comparing raw throughput alone.
+
+## Options
+
+`activities-per-workflow` controls how many remote payload activities each workflow runs sequentially.
+Its default is `1`.
+
+`payload-distribution-json` controls the payload size of each activity input and result. It accepts the
+shared OMES distribution format. The default is a fixed 100 KiB payload:
+
+```json
+{"size":{"type":"fixed","value":"102400"}}
+```
+
+A mixed distribution can represent a workload with several payload sizes:
+
+```json
+{"size":{"type":"discrete","weights":{"10240":1,"102400":9}}}
+```
+
+Measure the replay using the same stable interval for each ratio:
+
+```text
+charged History request bytes / Actions
+Cassandra transmitted bytes / charged History request bytes
+Cassandra transmitted bytes / Actions
+```
+
+Record these values for every plateau:
+
+```text
+run ID and exact start/end time
+offered iterations/s and observed Actions/s
+charged History request bytes/s
+Cassandra transmitted write bytes/s
+Astra storage growth/s
+Cassandra and storage p50, p95, and p99 latency
+Cassandra query errors/s
+frontend p50, p95, and p99 latency for the three payload RPCs
+bandwidth throttled requests/s
+```

scenarios/bandwidth_stress.go

@@ -0,0 +1,104 @@
+package scenarios
+
+import (
+	"context"
+	"fmt"
+	"hash/fnv"
+	"math/rand"
+
+	"github.com/temporalio/omes/loadgen"
+	. "github.com/temporalio/omes/loadgen/kitchensink"
+	"go.temporal.io/sdk/temporal"
+)
+
+const (
+	bandwidthActivitiesPerWorkflowFlag = "activities-per-workflow"
+	bandwidthPayloadDistributionFlag   = "payload-distribution-json"
+	bandwidthDefaultPayloadSize        = 100 * 1024
+	bandwidthDefaultPayloadConfig      = `{"size":{"type":"fixed","value":"102400"}}`
+)
+
+type bandwidthStressExecutor struct {
+	activitiesPerWorkflow int
+	payload               *loadgen.PayloadConfig
+	rngSeed               int64
+}
+
+var _ loadgen.Configurable = (*bandwidthStressExecutor)(nil)
+
+func init() {
+	loadgen.MustRegisterScenario(loadgen.Scenario{
+		Description: "Replays sustained variable payload writes for bandwidth limit and storage calibration.",
+		Options: func(o *loadgen.OptionSet) {
+			o.Int(bandwidthActivitiesPerWorkflowFlag, 1, "Remote payload activities per workflow.")
+			o.String(
+				bandwidthPayloadDistributionFlag,
+				bandwidthDefaultPayloadConfig,
+				"JSON activity payload size distribution; use @<file> to read from a file.",
+			)
+		},
+		ExecutorFn: func() loadgen.Executor { return &bandwidthStressExecutor{} },
+	})
+}
+
+func (b *bandwidthStressExecutor) Configure(info loadgen.ScenarioInfo) error {
+	b.activitiesPerWorkflow = info.OptionInt(bandwidthActivitiesPerWorkflowFlag)
+	if b.activitiesPerWorkflow <= 0 {
+		return fmt.Errorf(
+			"%s must be positive, got %d",
+			bandwidthActivitiesPerWorkflowFlag,
+			b.activitiesPerWorkflow,
+		)
+	}
+
+	payload, err := loadgen.ParseAndValidatePayloadConfig(info.OptionString(bandwidthPayloadDistributionFlag))
+	if err != nil {
+		return fmt.Errorf("invalid %s: %w", bandwidthPayloadDistributionFlag, err)
+	}
+	if payload == nil || payload.Size == nil {
+		return fmt.Errorf("%s must configure a size distribution", bandwidthPayloadDistributionFlag)
+	}
+	b.payload = payload
+
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(info.RunID))
+	b.rngSeed = int64(h.Sum64())
+	return nil
+}
+
+func (b *bandwidthStressExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error {
+	if err := b.Configure(info); err != nil {
+		return fmt.Errorf("failed to parse scenario configuration: %w", err)
+	}
+	info.Configuration.DoNotRegisterSearchAttributes = true
+
+	executor := loadgen.KitchenSinkExecutor{
+		TestInput: &TestInput{},
+		UpdateWorkflowOptions: func(
+			_ context.Context,
+			run *loadgen.Run,
+			options *loadgen.KitchenSinkWorkflowOptions,
+		) error {
+			options.StartOptions.TypedSearchAttributes = temporal.NewSearchAttributes()
+			options.Params = b.testInput(run.Iteration)
+			return nil
+		},
+	}
+	return executor.Run(ctx, info)
+}
+
+func (b *bandwidthStressExecutor) testInput(iteration int) *TestInput {
+	rng := rand.New(rand.NewSource(b.rngSeed + int64(iteration)))
+	actions := make([]*Action, 0, b.activitiesPerWorkflow+1)
+	for range b.activitiesPerWorkflow {
+		size := int(b.payload.SamplePayloadSize(rng, bandwidthDefaultPayloadSize))
+		actions = append(actions, PayloadActivity(size, size, DefaultRemoteActivity))
+	}
+	actions = append(actions, NewEmptyReturnResultAction())
+
+	return &TestInput{
+		WorkflowInput: &WorkflowInput{
+			InitialActions: []*ActionSet{{Actions: actions}},
+		},
+	}
+}

scenarios/bandwidth_stress_test.go

@@ -0,0 +1,106 @@
+package scenarios
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/temporalio/omes/clioptions"
+	"github.com/temporalio/omes/internal/workertest"
+	"github.com/temporalio/omes/loadgen"
+)
+
+func TestBandwidthStressConfiguration(t *testing.T) {
+	t.Parallel()
+
+	t.Run("default incident payload", func(t *testing.T) {
+		executor := newBandwidthStressExecutor(t, nil)
+		actions := executor.testInput(1).GetWorkflowInput().GetInitialActions()[0].GetActions()
+
+		require.Len(t, actions, 2)
+		payload := actions[0].GetExecActivity().GetPayload()
+		require.EqualValues(t, bandwidthDefaultPayloadSize, payload.GetBytesToReceive())
+		require.EqualValues(t, bandwidthDefaultPayloadSize, payload.GetBytesToReturn())
+		require.NotNil(t, actions[0].GetExecActivity().GetRemote())
+		require.NotNil(t, actions[1].GetReturnResult())
+	})
+
+	t.Run("custom distribution and activity count", func(t *testing.T) {
+		executor := newBandwidthStressExecutor(t, map[string]string{
+			bandwidthActivitiesPerWorkflowFlag: "3",
+			bandwidthPayloadDistributionFlag:   `{"size":{"type":"discrete","weights":{"1024":1,"2048":1}}}`,
+		})
+		actions := executor.testInput(10).GetWorkflowInput().GetInitialActions()[0].GetActions()
+
+		require.Len(t, actions, 4)
+		for _, action := range actions[:3] {
+			payload := action.GetExecActivity().GetPayload()
+			require.Contains(t, []int32{1024, 2048}, payload.GetBytesToReceive())
+			require.Equal(t, payload.GetBytesToReceive(), payload.GetBytesToReturn())
+		}
+	})
+
+	t.Run("invalid activity count", func(t *testing.T) {
+		executor := &bandwidthStressExecutor{}
+		err := executor.Configure(bandwidthStressScenarioInfo(map[string]string{
+			bandwidthActivitiesPerWorkflowFlag: "0",
+		}))
+		require.ErrorContains(t, err, "activities-per-workflow must be positive")
+	})
+
+	t.Run("missing size distribution", func(t *testing.T) {
+		executor := &bandwidthStressExecutor{}
+		err := executor.Configure(bandwidthStressScenarioInfo(map[string]string{
+			bandwidthPayloadDistributionFlag: `{}`,
+		}))
+		require.ErrorContains(t, err, "must configure a size distribution")
+	})
+
+	t.Run("run validates configuration", func(t *testing.T) {
+		executor := &bandwidthStressExecutor{}
+		err := executor.Run(t.Context(), bandwidthStressScenarioInfo(map[string]string{
+			bandwidthActivitiesPerWorkflowFlag: "0",
+		}))
+		require.ErrorContains(t, err, "failed to parse scenario configuration")
+		require.ErrorContains(t, err, "activities-per-workflow must be positive")
+	})
+}
+
+func TestBandwidthStress(t *testing.T) {
+	t.Parallel()
+
+	env := workertest.SetupTestEnvironment(t, workertest.WithExecutorTimeout(time.Minute))
+	executor := loadgen.GetScenario("bandwidth_stress").ExecutorFn()
+	info := loadgen.ScenarioInfo{
+		ScenarioName: "bandwidth_stress",
+		RunID:        fmt.Sprintf("bandwidth-%d", time.Now().UnixNano()),
+		Configuration: loadgen.RunConfiguration{
+			Iterations:    2,
+			MaxConcurrent: 2,
+		},
+		Options: loadgen.MustResolveScenarioOptions("bandwidth_stress", map[string]string{
+			bandwidthPayloadDistributionFlag: `{"size":{"type":"fixed","value":"1024"}}`,
+		}),
+	}
+	_, err := env.RunExecutorTest(t, executor, info, clioptions.LangGo)
+	require.NoError(t, err)
+}
+
+func newBandwidthStressExecutor(t *testing.T, provided map[string]string) *bandwidthStressExecutor {
+	t.Helper()
+	executor := &bandwidthStressExecutor{}
+	require.NoError(t, executor.Configure(bandwidthStressScenarioInfo(provided)))
+	return executor
+}
+
+func bandwidthStressScenarioInfo(provided map[string]string) loadgen.ScenarioInfo {
+	return loadgen.ScenarioInfo{
+		ScenarioName: "bandwidth_stress",
+		RunID:        "bandwidth-test",
+		Options:      loadgen.MustResolveScenarioOptions("bandwidth_stress", provided),
+		Configuration: loadgen.RunConfiguration{
+			Iterations: 1,
+		},
+	}
+}