diff --git a/README.md b/README.md index 79ba04eb..a23f9857 100644 --- a/README.md +++ b/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. diff --git a/docs/bandwidth-stress.md b/docs/bandwidth-stress.md new file mode 100644 index 00000000..b151136c --- /dev/null +++ b/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 \ + --tls-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 +``` diff --git a/scenarios/bandwidth_stress.go b/scenarios/bandwidth_stress.go new file mode 100644 index 00000000..92f8a192 --- /dev/null +++ b/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 @ 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}}, + }, + } +} diff --git a/scenarios/bandwidth_stress_test.go b/scenarios/bandwidth_stress_test.go new file mode 100644 index 00000000..4e895149 --- /dev/null +++ b/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, + }, + } +}