mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Fix panic error in TestTaskQueueStats (#9575)
## What changed?
Fix a panic in `TestTaskQueueStats_{Classic,Pri}_Suite` when tests run
in parallel. The outer `t *testing.T` was passed to the suite and a
panic occurred outside of the testing context.
Sample: https://github.com/temporalio/temporal/actions/runs/23262976802
## Why?
Previously enabled parallel execution which caused this bug more
frequently.
## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [X] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)
## Potential risks
Test only
This commit is contained in:
Sean Kane
@@ -29,7 +29,6 @@ type (
|
||||
// taskQueueStatsSuite encapsulates the test environment and parameters for task queue stats tests.
|
||||
taskQueueStatsSuite struct {
|
||||
testcore.Env
|
||||
t *testing.T
|
||||
usePriMatcher bool
|
||||
minPriority int
|
||||
maxPriority int
|
||||
@@ -47,10 +46,9 @@ type (
|
||||
TaskQueueExpectationsByType map[enumspb.TaskQueueType]TaskQueueExpectations
|
||||
)
|
||||
|
||||
func newTaskQueueStatsSuite(t *testing.T, env testcore.Env, usePriMatcher bool) *taskQueueStatsSuite {
|
||||
func newTaskQueueStatsSuite(env testcore.Env, usePriMatcher bool) *taskQueueStatsSuite {
|
||||
return &taskQueueStatsSuite{
|
||||
Env: env,
|
||||
t: t,
|
||||
usePriMatcher: usePriMatcher,
|
||||
minPriority: 1,
|
||||
maxPriority: 5,
|
||||
@@ -81,7 +79,7 @@ func runTaskQueueStatsTests(t *testing.T, usePriMatcher bool) {
|
||||
// Tests WITHOUT RunTestWithMatchingBehavior
|
||||
t.Run("TestDescribeTaskQueue_NonRoot", func(t *testing.T) {
|
||||
env := testcore.NewEnv(t, baseOpts...)
|
||||
s := newTaskQueueStatsSuite(t, env, usePriMatcher)
|
||||
s := newTaskQueueStatsSuite(env, usePriMatcher)
|
||||
s.testDescribeTaskQueueNonRoot()
|
||||
})
|
||||
|
||||
@@ -93,7 +91,7 @@ func runTaskQueueStatsTests(t *testing.T, usePriMatcher bool) {
|
||||
testcore.WithDynamicConfig(dynamicconfig.TaskQueueInfoByBuildIdTTL, 1*time.Millisecond),
|
||||
)
|
||||
env := testcore.NewEnv(t, opts...)
|
||||
s := newTaskQueueStatsSuite(t, env, usePriMatcher)
|
||||
s := newTaskQueueStatsSuite(env, usePriMatcher)
|
||||
s.publishConsumeWorkflowTasksValidateStats(0, false)
|
||||
})
|
||||
|
||||
@@ -103,7 +101,7 @@ func runTaskQueueStatsTests(t *testing.T, usePriMatcher bool) {
|
||||
testcore.WithDynamicConfig(dynamicconfig.TaskQueueInfoByBuildIdTTL, 1*time.Hour),
|
||||
)
|
||||
env := testcore.NewEnv(t, opts...)
|
||||
s := newTaskQueueStatsSuite(t, env, usePriMatcher)
|
||||
s := newTaskQueueStatsSuite(env, usePriMatcher)
|
||||
s.testAddMultipleTasksValidateStatsCached()
|
||||
})
|
||||
|
||||
@@ -157,7 +155,7 @@ func runSuiteWithMatchingBehaviors(
|
||||
subtest func(s *taskQueueStatsSuite),
|
||||
) {
|
||||
runWithMatchingBehaviors(t, baseOpts, func(env *testcore.TestEnv, behavior testcore.MatchingBehavior) {
|
||||
s := newTaskQueueStatsSuite(t, env, usePriMatcher)
|
||||
s := newTaskQueueStatsSuite(env, usePriMatcher)
|
||||
subtest(s)
|
||||
})
|
||||
}
|
||||
@@ -167,8 +165,8 @@ func (s *taskQueueStatsSuite) testDescribeTaskQueueNonRoot() {
|
||||
Namespace: s.Namespace().String(),
|
||||
TaskQueue: &taskqueuepb.TaskQueue{Name: "/_sys/foo/1", Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
|
||||
})
|
||||
require.NoError(s.t, err)
|
||||
require.NotNil(s.t, resp)
|
||||
require.NoError(s.T(), err)
|
||||
require.NotNil(s.T(), resp)
|
||||
|
||||
_, err = s.FrontendClient().DescribeTaskQueue(context.Background(),
|
||||
&workflowservice.DescribeTaskQueueRequest{
|
||||
@@ -176,7 +174,7 @@ func (s *taskQueueStatsSuite) testDescribeTaskQueueNonRoot() {
|
||||
TaskQueue: &taskqueuepb.TaskQueue{Name: "/_sys/foo/1", Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
|
||||
ReportStats: true,
|
||||
})
|
||||
require.ErrorContains(s.t, err, "DescribeTaskQueue stats are only supported for the root partition")
|
||||
require.ErrorContains(s.T(), err, "DescribeTaskQueue stats are only supported for the root partition")
|
||||
}
|
||||
|
||||
func (s *taskQueueStatsSuite) testAddMultipleTasksValidateStatsCached() {
|
||||
@@ -256,7 +254,7 @@ func (s *taskQueueStatsSuite) currentVersionAbsorbsUnversionedBacklogNoRamping()
|
||||
MaxExtraTasks: 0,
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// DescribeWorkerDeploymentVersion: current version should also show the full backlog for this task queue.
|
||||
@@ -294,7 +292,7 @@ func (s *taskQueueStatsSuite) currentVersionAbsorbsUnversionedBacklogNoRamping()
|
||||
MaxExtraTasks: 0,
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// Since the activity task queue is part of the current version,
|
||||
@@ -369,7 +367,7 @@ func (s *taskQueueStatsSuite) rampingAndCurrentAbsorbsUnversionedBacklog() {
|
||||
// Currently only testing the following API's:
|
||||
// - DescribeWorkerDeploymentVersion for the current and ramping versions.
|
||||
// - DescribeTaskQueue Legacy Mode for the current and ramping versions.
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// DescribeWorkerDeploymentVersion: current version should also show only 70% of the unversioned backlog for this task queue
|
||||
@@ -451,7 +449,7 @@ func (s *taskQueueStatsSuite) rampingAndCurrentAbsorbsUnversionedBacklog() {
|
||||
MaxExtraTasks: 0,
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// Validate current version activity stats
|
||||
@@ -518,7 +516,7 @@ func (s *taskQueueStatsSuite) currentAbsorbsUnversionedBacklogWhenRampingToUnver
|
||||
MaxExtraTasks: 0,
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// There is no way right now for a user to query stats of the "unversioned" version. All we can do in this case
|
||||
@@ -584,7 +582,7 @@ func (s *taskQueueStatsSuite) rampingAbsorbsUnversionedBacklogWhenCurrentIsUnver
|
||||
MaxExtraTasks: 0,
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// We can't query "unversioned" as a WorkerDeploymentVersion, but we can validate that the ramping version
|
||||
@@ -658,7 +656,7 @@ func (s *taskQueueStatsSuite) inactiveVersionDoesNotAbsorbUnversionedBacklog() {
|
||||
// Currently only testing the following API's:
|
||||
// - DescribeWorkerDeploymentVersion
|
||||
// - DescribeTaskQueue Legacy Mode
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// DescribeWorkerDeploymentVersion: current version should should show 100% of the unversioned backlog for this task queue
|
||||
@@ -730,7 +728,7 @@ func (s *taskQueueStatsSuite) inactiveVersionDoesNotAbsorbUnversionedBacklog() {
|
||||
MaxExtraTasks: 0,
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
// The activity task queue of the current version should have the backlog count for the activities that were scheduled
|
||||
@@ -873,7 +871,7 @@ func (s *taskQueueStatsSuite) publishConsumeWorkflowTasksValidateStats(sets int,
|
||||
|
||||
// poll all workflow tasks and enqueue one activity task for each workflow
|
||||
totalAct := s.enqueueActivitiesForEachWorkflow(sets, tqName)
|
||||
require.Equal(s.t, total, totalAct, "should have enqueued the same number of activities as workflows")
|
||||
require.Equal(s.T(), total, totalAct, "should have enqueued the same number of activities as workflows")
|
||||
|
||||
// verify workflow dispatch rate and activity add rate
|
||||
if sets > 0 {
|
||||
@@ -930,7 +928,7 @@ func (s *taskQueueStatsSuite) startUnversionedWorkflows(count int, tqName string
|
||||
for range count {
|
||||
request.WorkflowId = uuid.NewString() // starting "count" different Unversioned workflows.
|
||||
_, err := s.FrontendClient().StartWorkflowExecution(testcore.NewContext(), request)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -961,7 +959,7 @@ func (s *taskQueueStatsSuite) startPinnedWorkflows(count int, tqName string, dep
|
||||
for range count {
|
||||
request.WorkflowId = uuid.NewString() // starting "n" different Pinned workflows.
|
||||
_, err := s.FrontendClient().StartWorkflowExecution(testcore.NewContext(), request)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -992,7 +990,7 @@ func (s *taskQueueStatsSuite) pollWorkflowTasksAndScheduleActivitiesParallel(par
|
||||
wg.Wait()
|
||||
close(errCh)
|
||||
for err := range errCh {
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1070,7 +1068,7 @@ func (s *taskQueueStatsSuite) completeWorkflowTasksAndScheduleActivities(
|
||||
Identity: "current-version-worker",
|
||||
DeploymentOptions: deploymentOpts,
|
||||
})
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
if resp == nil || resp.GetAttempt() < 1 {
|
||||
fmt.Println("Empty poll! Continuing...")
|
||||
continue
|
||||
@@ -1100,7 +1098,7 @@ func (s *taskQueueStatsSuite) completeWorkflowTasksAndScheduleActivities(
|
||||
}
|
||||
|
||||
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(testcore.NewContext(), respondReq)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
i++
|
||||
}
|
||||
}
|
||||
@@ -1115,7 +1113,7 @@ func (s *taskQueueStatsSuite) setCurrentVersion(deploymentName, buildID string)
|
||||
DeploymentName: deploymentName,
|
||||
BuildId: buildID,
|
||||
})
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
}
|
||||
|
||||
// TODO (Shivam): We may have to wait for the propagation status to show completed if we are using async workflows here.
|
||||
@@ -1129,7 +1127,7 @@ func (s *taskQueueStatsSuite) setRampingVersion(deploymentName, buildID string,
|
||||
BuildId: buildID,
|
||||
Percentage: float32(rampPercentage),
|
||||
})
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
}
|
||||
|
||||
func (s *taskQueueStatsSuite) describeWDVTaskQueueStats(
|
||||
@@ -1215,14 +1213,14 @@ func (s *taskQueueStatsSuite) enqueueWorkflows(sets int, tqName string) int {
|
||||
}
|
||||
|
||||
_, err := s.FrontendClient().StartWorkflowExecution(testcore.NewContext(), request)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
|
||||
total++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
s.t.Logf("Enqueued %d workflows", total)
|
||||
s.T().Logf("Enqueued %d workflows", total)
|
||||
return total
|
||||
}
|
||||
|
||||
@@ -1246,7 +1244,7 @@ func (s *taskQueueStatsSuite) createVersionsInTaskQueue(ctx context.Context, tqN
|
||||
}()
|
||||
|
||||
// Wait for the version to be created.
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
@@ -1302,7 +1300,7 @@ func (s *taskQueueStatsSuite) enqueueActivitiesForEachWorkflow(sets int, tqName
|
||||
}
|
||||
|
||||
resp, err := s.FrontendClient().PollWorkflowTaskQueue(testcore.NewContext(), pollReq)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
if resp == nil || resp.GetAttempt() < 1 {
|
||||
continue
|
||||
}
|
||||
@@ -1333,14 +1331,14 @@ func (s *taskQueueStatsSuite) enqueueActivitiesForEachWorkflow(sets int, tqName
|
||||
}
|
||||
|
||||
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(testcore.NewContext(), respondReq)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
|
||||
i++
|
||||
total++
|
||||
}
|
||||
}
|
||||
}
|
||||
s.t.Logf("Enqueued %d activities", total)
|
||||
s.T().Logf("Enqueued %d activities", total)
|
||||
return total
|
||||
}
|
||||
|
||||
@@ -1360,13 +1358,13 @@ func (s *taskQueueStatsSuite) pollActivities(count int, tqName string) {
|
||||
resp, err := s.FrontendClient().PollActivityTaskQueue(
|
||||
testcore.NewContext(), pollReq,
|
||||
)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
if resp == nil || resp.GetAttempt() < 1 {
|
||||
continue // poll again on empty responses
|
||||
}
|
||||
i++
|
||||
}
|
||||
s.t.Logf("Polled %d activities", count)
|
||||
s.T().Logf("Polled %d activities", count)
|
||||
}
|
||||
|
||||
func (s *taskQueueStatsSuite) validateAllTaskQueueStats(
|
||||
@@ -1398,7 +1396,7 @@ func (s *taskQueueStatsSuite) validateRates(
|
||||
ReportStats: true,
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
label := "validateRates[" + tqType.String() + "]"
|
||||
|
||||
@@ -1454,13 +1452,13 @@ func (s *taskQueueStatsSuite) validateDescribeTaskQueueWithDefaultMode(
|
||||
|
||||
// test stats are not reported by default (and therefore also not cached)
|
||||
resp, err := s.FrontendClient().DescribeTaskQueue(ctx, req)
|
||||
require.NoError(s.t, err)
|
||||
require.NotNil(s.t, resp)
|
||||
require.Nil(s.t, resp.Stats, "stats should not be reported by default")
|
||||
require.NoError(s.T(), err)
|
||||
require.NotNil(s.T(), resp)
|
||||
require.Nil(s.T(), resp.Stats, "stats should not be reported by default")
|
||||
//nolint:staticcheck // SA1019 deprecated
|
||||
require.Nil(s.t, resp.TaskQueueStatus, "status should not be reported by default")
|
||||
require.Nil(s.T(), resp.TaskQueueStatus, "status should not be reported by default")
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
label := "DescribeTaskQueue_DefaultMode[" + tqType.String() + "]"
|
||||
|
||||
@@ -1511,14 +1509,14 @@ func (s *taskQueueStatsSuite) validateDescribeTaskQueueWithEnhancedMode(
|
||||
|
||||
if !expectation.CachedEnabled { // skip if testing caching; as this would pin the result to the cache
|
||||
resp, err := s.FrontendClient().DescribeTaskQueue(ctx, req)
|
||||
require.NoError(s.t, err)
|
||||
require.NotNil(s.t, resp)
|
||||
require.Nil(s.t, resp.Stats, "stats should not be reported by default")
|
||||
require.NoError(s.T(), err)
|
||||
require.NotNil(s.T(), resp)
|
||||
require.Nil(s.T(), resp.Stats, "stats should not be reported by default")
|
||||
//nolint:staticcheck // SA1019 deprecated
|
||||
require.Nil(s.t, resp.TaskQueueStatus, "status should not be reported")
|
||||
require.Nil(s.T(), resp.TaskQueueStatus, "status should not be reported")
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
req.ReportStats = true
|
||||
@@ -1564,18 +1562,18 @@ func (s *taskQueueStatsSuite) validateDescribeWorkerDeploymentVersion(
|
||||
|
||||
// test stats are not reported by default (and therefore also not cached)
|
||||
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, req)
|
||||
require.NoError(s.t, err)
|
||||
require.NotNil(s.t, resp)
|
||||
require.NoError(s.T(), err)
|
||||
require.NotNil(s.T(), resp)
|
||||
for _, info := range resp.VersionTaskQueues {
|
||||
require.Nil(s.t, info.Stats, "stats should not be reported by default")
|
||||
require.Nil(s.T(), info.Stats, "stats should not be reported by default")
|
||||
}
|
||||
|
||||
require.EventuallyWithT(s.t, func(c *assert.CollectT) {
|
||||
require.EventuallyWithT(s.T(), func(c *assert.CollectT) {
|
||||
a := require.New(c)
|
||||
|
||||
req.ReportTaskQueueStats = true
|
||||
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, req)
|
||||
require.NoError(s.t, err)
|
||||
require.NoError(s.T(), err)
|
||||
a.Len(resp.VersionTaskQueues, 2, "should be 1 task queue for Workflows and 1 for Activities")
|
||||
|
||||
for _, info := range resp.VersionTaskQueues {
|
||||
|
||||
Reference in New Issue
Block a user