mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
26f2023316
## What changed? Emit task dispatch latency metric in matching_engine with the following improvements: - latency includes history calls - latency is not reset in case of sync-match forwards - metric is not lost for backlogged task in new matcher - origin partition is preserved during forward and used at the partition tag in the metric (keeping existing behavior) - task and poll forwarding do not cause duplicate emits, keeping existing behavior Also, fixed the following unrelated bugs that surfaced while testing the metric: - Query priority is not lost when forwarded. ## Why? Fixes bugs. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [x] added new functional test(s) ## Potential risks None.
78 lines
2.4 KiB
Go
78 lines
2.4 KiB
Go
package tests
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
deploymentspb "go.temporal.io/server/api/deployment/v1"
|
|
"go.temporal.io/server/api/matchingservice/v1"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/primitives/timestamp"
|
|
"go.temporal.io/server/common/testing/testvars"
|
|
"go.temporal.io/server/tests/testcore"
|
|
)
|
|
|
|
// runWithMatchingBehaviors runs a test with all combinations of matching behaviors.
|
|
func runWithMatchingBehaviors(
|
|
t *testing.T,
|
|
baseOpts []testcore.TestOption,
|
|
subtest func(s *testcore.TestEnv, behavior testcore.MatchingBehavior),
|
|
) {
|
|
for _, behavior := range testcore.AllMatchingBehaviors() {
|
|
t.Run(behavior.Name(), func(t *testing.T) {
|
|
opts := append([]testcore.TestOption{}, baseOpts...)
|
|
opts = append(opts, behavior.Options()...)
|
|
|
|
env := testcore.NewEnv(t, opts...)
|
|
behavior.InjectHooks(env)
|
|
|
|
subtest(env, behavior)
|
|
})
|
|
}
|
|
}
|
|
|
|
// syncDeploymentVersionToTaskQueues sends a SyncDeploymentUserData request to the matching service
|
|
// to register a deployment version as current for the specified task queue types, then waits for
|
|
// the data to propagate to all partitions using CheckTaskQueueUserDataPropagation.
|
|
func syncDeploymentVersionToTaskQueues(
|
|
t testing.TB,
|
|
matchingClient matchingservice.MatchingServiceClient,
|
|
namespaceID namespace.ID,
|
|
tv *testvars.TestVars,
|
|
tqTypes ...enumspb.TaskQueueType,
|
|
) {
|
|
t.Helper()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
now := timestamp.TimePtr(time.Now())
|
|
resp, err := matchingClient.SyncDeploymentUserData(
|
|
ctx, &matchingservice.SyncDeploymentUserDataRequest{
|
|
NamespaceId: namespaceID.String(),
|
|
TaskQueue: tv.TaskQueue().GetName(),
|
|
TaskQueueTypes: tqTypes,
|
|
Operation: &matchingservice.SyncDeploymentUserDataRequest_UpdateVersionData{
|
|
UpdateVersionData: &deploymentspb.DeploymentVersionData{
|
|
Version: tv.DeploymentVersion(),
|
|
RoutingUpdateTime: now,
|
|
CurrentSinceTime: now,
|
|
},
|
|
},
|
|
},
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
// Wait for the data to propagate to all partitions.
|
|
_, err = matchingClient.CheckTaskQueueUserDataPropagation(
|
|
ctx, &matchingservice.CheckTaskQueueUserDataPropagationRequest{
|
|
NamespaceId: namespaceID.String(),
|
|
TaskQueue: tv.TaskQueue().GetName(),
|
|
Version: resp.GetVersion(),
|
|
},
|
|
)
|
|
require.NoError(t, err)
|
|
}
|