diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index 8ae3e8f5..150b61bd 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -69,6 +69,20 @@ const ( // PayloadDistributionJsonFlag is a JSON string (or @file) configuring a weighted // activity payload-size distribution. See loadgen.PayloadConfig for details. PayloadDistributionJsonFlag = "payload-distribution-json" + // HealthyHeartbeatActivitiesFlag is how many healthy heartbeating remote + // activities each internal iteration runs. + HealthyHeartbeatActivitiesFlag = "healthy-heartbeat-activities" + // HealthyHeartbeatDurationFlag is how long each healthy heartbeat activity runs. + HealthyHeartbeatDurationFlag = "healthy-heartbeat-duration" + // HealthyHeartbeatTimeoutFlag is the heartbeat timeout of each healthy heartbeat activity. + HealthyHeartbeatTimeoutFlag = "healthy-heartbeat-timeout" +) + +const ( + // DefaultHealthyHeartbeatActivities disables healthy heartbeat activities. + DefaultHealthyHeartbeatActivities = 0 + DefaultHealthyHeartbeatDuration = 30 * time.Second + DefaultHealthyHeartbeatTimeout = 10 * time.Second ) type tpsState struct { @@ -117,6 +131,9 @@ type tpsConfig struct { IncludeNexusStandaloneActivity bool IncludeNexusWorkflowActions bool Payload *loadgen.PayloadConfig + HealthyHeartbeatActivities int + HealthyHeartbeatDuration time.Duration + HealthyHeartbeatTimeout time.Duration } type tpsExecutor struct { @@ -155,6 +172,9 @@ func init() { o.Bool(IncludeNexusStandaloneActivityFlag, false, "Include a Nexus operation that starts a standalone activity (Go worker only).") o.Bool(IncludeNexusWorkflowActionsFlag, false, "Include Nexus operations that signal, update, and query a workflow (Go worker only).") o.String(PayloadDistributionJsonFlag, "", "JSON payload-size distribution; use @ to read from a file.") + o.Int(HealthyHeartbeatActivitiesFlag, DefaultHealthyHeartbeatActivities, "Healthy heartbeating activities per internal iteration (Go worker only).") + o.Duration(HealthyHeartbeatDurationFlag, DefaultHealthyHeartbeatDuration, "How long each healthy heartbeat activity runs.") + o.Duration(HealthyHeartbeatTimeoutFlag, DefaultHealthyHeartbeatTimeout, "Heartbeat timeout of each healthy heartbeat activity.") }, ExecutorFn: func() loadgen.Executor { return newThroughputStressExecutor() }, }) @@ -270,6 +290,23 @@ func (t *tpsExecutor) Configure(info loadgen.ScenarioInfo) error { } } + config.HealthyHeartbeatActivities = info.OptionInt(HealthyHeartbeatActivitiesFlag) + if config.HealthyHeartbeatActivities < 0 { + return fmt.Errorf("%s must be non-negative, got %d", HealthyHeartbeatActivitiesFlag, config.HealthyHeartbeatActivities) + } + config.HealthyHeartbeatDuration = info.OptionDuration(HealthyHeartbeatDurationFlag) + if config.HealthyHeartbeatDuration <= 0 { + return fmt.Errorf("%s must be positive, got %v", HealthyHeartbeatDurationFlag, config.HealthyHeartbeatDuration) + } + config.HealthyHeartbeatTimeout = info.OptionDuration(HealthyHeartbeatTimeoutFlag) + if config.HealthyHeartbeatTimeout <= 0 { + return fmt.Errorf("%s must be positive, got %v", HealthyHeartbeatTimeoutFlag, config.HealthyHeartbeatTimeout) + } + + if config.HealthyHeartbeatDuration > time.Duration(1<<63-1)-config.HealthyHeartbeatTimeout { + return fmt.Errorf("%s plus %s exceeds the maximum duration", HealthyHeartbeatDurationFlag, HealthyHeartbeatTimeoutFlag) + } + t.config = config t.rng = rand.New(rand.NewSource(config.RngSeed)) return nil @@ -709,6 +746,11 @@ func (t *tpsExecutor) createActionsChunk( ) } + for range t.config.HealthyHeartbeatActivities { + asyncActions = append(asyncActions, + healthyHeartbeatActivity(t.config.HealthyHeartbeatDuration, t.config.HealthyHeartbeatTimeout)) + } + // Add sleep activities, if configured. // It simulates custom traffic patterns by generating activities that sleep // for a specified duration, with optional priority and fairness keys. @@ -1178,6 +1220,26 @@ func (t *tpsExecutor) createStandaloneNexusOperationAction(input *NexusOperation }), DefaultRemoteActivity) } +// Keep heartbeat timeouts observable rather than masking them with activity retries. +func healthyHeartbeatActivity(duration, heartbeatTimeout time.Duration) *Action { + return &Action{ + Variant: &Action_ExecActivity{ + ExecActivity: &ExecuteActivityAction{ + ActivityType: &ExecuteActivityAction_Heartbeat{ + Heartbeat: &ExecuteActivityAction_HeartbeatTimeoutActivity{ + SuccessDuration: durationpb.New(duration), + HeartbeatInterval: durationpb.New(heartbeatTimeout / 4), + }, + }, + StartToCloseTimeout: durationpb.New(duration + heartbeatTimeout), + HeartbeatTimeout: durationpb.New(heartbeatTimeout), + RetryPolicy: &common.RetryPolicy{MaximumAttempts: 1}, + Locality: &ExecuteActivityAction_Remote{Remote: &RemoteActivityOptions{}}, + }, + }, + } +} + func (t *tpsExecutor) createStandaloneActivityAction(taskQueue string, rng *rand.Rand) *Action { size := int32(t.samplePayloadSize(rng)) return ClientActivity(ClientActions(&ClientAction{ diff --git a/scenarios/throughput_stress_heartbeat_test.go b/scenarios/throughput_stress_heartbeat_test.go new file mode 100644 index 00000000..39aa3577 --- /dev/null +++ b/scenarios/throughput_stress_heartbeat_test.go @@ -0,0 +1,129 @@ +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" + ks "github.com/temporalio/omes/loadgen/kitchensink" + "go.uber.org/zap" +) + +func healthyHeartbeatActions(t *testing.T, options map[string]string) []*ks.ExecuteActivityAction { + t.Helper() + executor := newThroughputStressExecutor() + info := loadgen.ScenarioInfo{ + RunID: "tps-healthy-heartbeat", + ExecutionID: "exec", + Logger: zap.NewNop().Sugar(), + Options: loadgen.MustResolveScenarioOptions("throughput_stress", options), + } + require.NoError(t, executor.Configure(info)) + + var activities []*ks.ExecuteActivityAction + walkActions(executor.createActions(info.NewRun(1))[0].GetActions(), func(action *ks.Action) { + activity := action.GetExecActivity() + if heartbeat := activity.GetHeartbeat(); heartbeat != nil && heartbeat.GetFailAttempts() == 0 { + activities = append(activities, activity) + } + }) + return activities +} + +func TestThroughputStressHealthyHeartbeatActivities(t *testing.T) { + t.Parallel() + + t.Run("off by default", func(t *testing.T) { + require.Empty(t, healthyHeartbeatActions(t, map[string]string{NexusEnabledFlag: "false"})) + }) + + t.Run("configured per internal iteration", func(t *testing.T) { + activities := healthyHeartbeatActions(t, map[string]string{ + NexusEnabledFlag: "false", + IterFlag: "3", + HealthyHeartbeatActivitiesFlag: "2", + HealthyHeartbeatDurationFlag: "20s", + HealthyHeartbeatTimeoutFlag: "8s", + }) + require.Len(t, activities, 6) + for _, activity := range activities { + require.Equal(t, 20*time.Second, activity.GetHeartbeat().GetSuccessDuration().AsDuration()) + require.Equal(t, 2*time.Second, activity.GetHeartbeat().GetHeartbeatInterval().AsDuration()) + require.Equal(t, 8*time.Second, activity.GetHeartbeatTimeout().AsDuration()) + require.Equal(t, 28*time.Second, activity.GetStartToCloseTimeout().AsDuration()) + require.Equal(t, int32(1), activity.GetRetryPolicy().GetMaximumAttempts()) + require.NotNil(t, activity.GetRemote()) + } + }) +} + +func TestThroughputStressConfigureInvalidHealthyHeartbeat(t *testing.T) { + t.Parallel() + + for flag, value := range map[string]string{ + HealthyHeartbeatActivitiesFlag: "-1", + HealthyHeartbeatDurationFlag: "0s", + HealthyHeartbeatTimeoutFlag: "0s", + } { + t.Run(flag, func(t *testing.T) { + err := newThroughputStressExecutor().Configure(loadgen.ScenarioInfo{ + RunID: "tps-invalid-healthy-heartbeat", + Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{flag: value}), + }) + require.ErrorContains(t, err, flag) + }) + } +} + +// The activity outlives its heartbeat timeout and is not retried, so the run +// only succeeds if the Go worker keeps heartbeating it. +func TestThroughputStressHealthyHeartbeatKeepsActivityAlive(t *testing.T) { + t.Parallel() + + runID := fmt.Sprintf("tps-healthy-heartbeat-%d", time.Now().UnixNano()) + env := workertest.SetupTestEnvironment(t, workertest.WithExecutorTimeout(time.Minute)) + _, err := env.RunExecutorTest(t, newThroughputStressExecutor(), loadgen.ScenarioInfo{ + RunID: runID, + Configuration: loadgen.RunConfiguration{Iterations: 1}, + Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{ + IterFlag: "1", + NexusEnabledFlag: "false", + SleepTimeFlag: "1ms", + VisibilityVerificationTimeoutFlag: "10s", + HealthyHeartbeatActivitiesFlag: "1", + HealthyHeartbeatDurationFlag: "6s", + HealthyHeartbeatTimeoutFlag: "3s", + }), + }, clioptions.LangGo) + require.NoError(t, err) +} + +func TestThroughputStressHealthyHeartbeatDurationOverflow(t *testing.T) { + for _, tc := range []struct { + name string + duration time.Duration + invalid bool + }{ + {name: "largest representable sum", duration: time.Duration(1<<63-1) - time.Second}, + {name: "sum overflows", duration: time.Duration(1<<63 - 1), invalid: true}, + } { + t.Run(tc.name, func(t *testing.T) { + info := loadgen.ScenarioInfo{Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{ + NexusEnabledFlag: "false", + HealthyHeartbeatDurationFlag: tc.duration.String(), + HealthyHeartbeatTimeoutFlag: time.Second.String(), + })} + err := newThroughputStressExecutor().Configure(info) + if tc.invalid { + require.ErrorContains(t, err, HealthyHeartbeatDurationFlag) + } else { + require.NoError(t, err) + require.Equal(t, time.Duration(1<<63-1), healthyHeartbeatActivity(tc.duration, time.Second).GetExecActivity().GetStartToCloseTimeout().AsDuration()) + } + }) + } +}