mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
- When a deactivated (drained/draining) version gets rollbacked to becoming active again, the drainage information that was associated with it will get cleared. - This is required because when this version were to get deactivated again, we need the code to start the drainage process on it again. Right now, it was not doing that since we were not clearing the drainage information which did not start a fresh drainage tracker for this version! *NOTE*: This change requires a patch to be present. This is because when this scenario is replicated in production, it will result in two new history calls (syncSummary as well as a CAN of the version workflow). This will specifically happen when the rolled-back version gets deactivated eventually, over [here](https://github.com/temporalio/temporal/blob/d85e67120957a16e0c14278bfb00e80412f16f5c/service/worker/workerdeployment/version_workflow.go#L546) - note how previously, since the drainage information was not cleared, [this](https://github.com/temporalio/temporal/blob/d85e67120957a16e0c14278bfb00e80412f16f5c/service/worker/workerdeployment/version_workflow.go#L261) was never executed!) *How did I verify that this requires a patch*: - Ran the replay test script before the fix introduced in this PR and noticed: <img width="793" height="79" alt="image" src="https://github.com/user-attachments/assets/bb975d26-4ad1-4847-b477-b74c7d2f55e4" /> - Ran the replay test script after the fix introduced in this PR and noticed: <img width="911" height="99" alt="image" src="https://github.com/user-attachments/assets/ad3ff3e1-666a-49b4-bda4-e05f3412d984" /> In other words, the total number of our entity workflows went up after introducing this change and thus requires a patch gate! - we found a bug in the versioning entity workflows and this is an effort to fix this! - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [x] added new functional test(s) - yes, there is a risk of NDE's happening in production but I mitigated that risk by placing the fix against a version gate.
3065 lines
126 KiB
Go
3065 lines
126 KiB
Go
package tests
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/dgryski/go-farm"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/stretchr/testify/suite"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
deploymentpb "go.temporal.io/api/deployment/v1"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
taskqueuepb "go.temporal.io/api/taskqueue/v1"
|
|
"go.temporal.io/api/workflowservice/v1"
|
|
sdkworker "go.temporal.io/sdk/worker"
|
|
"go.temporal.io/server/common/dynamicconfig"
|
|
"go.temporal.io/server/common/testing/testhooks"
|
|
"go.temporal.io/server/common/testing/testvars"
|
|
"go.temporal.io/server/common/worker_versioning"
|
|
"go.temporal.io/server/service/worker/workerdeployment"
|
|
"go.temporal.io/server/tests/testcore"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
type (
|
|
WorkerDeploymentSuite struct {
|
|
testcore.FunctionalTestBase
|
|
}
|
|
)
|
|
|
|
func TestWorkerDeploymentSuite(t *testing.T) {
|
|
t.Parallel()
|
|
suite.Run(t, new(WorkerDeploymentSuite))
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) SetupSuite() {
|
|
s.FunctionalTestBase.SetupSuiteWithCluster(testcore.WithDynamicConfigOverrides(map[dynamicconfig.Key]any{
|
|
dynamicconfig.EnableDeploymentVersions.Key(): true,
|
|
dynamicconfig.FrontendEnableWorkerVersioningDataAPIs.Key(): true, // [wv-cleanup-pre-release]
|
|
dynamicconfig.FrontendEnableWorkerVersioningWorkflowAPIs.Key(): true, // [wv-cleanup-pre-release]
|
|
dynamicconfig.FrontendEnableWorkerVersioningRuleAPIs.Key(): true, // [wv-cleanup-pre-release]
|
|
dynamicconfig.FrontendEnableExecuteMultiOperation.Key(): true,
|
|
|
|
// Make sure we don't hit the rate limiter in tests
|
|
dynamicconfig.FrontendGlobalNamespaceNamespaceReplicationInducingAPIsRPS.Key(): 1000,
|
|
dynamicconfig.FrontendMaxNamespaceNamespaceReplicationInducingAPIsBurstRatioPerInstance.Key(): 1,
|
|
dynamicconfig.FrontendNamespaceReplicationInducingAPIsRPS.Key(): 1000,
|
|
|
|
// Make drainage happen sooner
|
|
dynamicconfig.VersionDrainageStatusRefreshInterval.Key(): testVersionDrainageRefreshInterval,
|
|
dynamicconfig.VersionDrainageStatusVisibilityGracePeriod.Key(): testVersionDrainageVisibilityGracePeriod,
|
|
|
|
// To increase the rate at which the per-ns worker can consume tasks from a task queue. Required since
|
|
// tests in this suite create a lot of tasks and expect them to be consumed quickly.
|
|
dynamicconfig.WorkerPerNamespaceWorkerOptions.Key(): sdkworker.Options{
|
|
MaxConcurrentWorkflowTaskPollers: 100,
|
|
MaxConcurrentActivityTaskPollers: 100,
|
|
},
|
|
|
|
dynamicconfig.MatchingMaxTaskQueuesInDeploymentVersion.Key(): 1000,
|
|
dynamicconfig.VisibilityPersistenceSlowQueryThreshold.Key(): 60 * time.Second,
|
|
dynamicconfig.WorkflowExecutionMaxInFlightUpdates.Key(): 1000,
|
|
}))
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) SetupTest() {
|
|
s.FunctionalTestBase.SetupTest()
|
|
}
|
|
|
|
// pollFromDeployment calls PollWorkflowTaskQueue to start deployment related workflows
|
|
func (s *WorkerDeploymentSuite) pollFromDeployment(ctx context.Context, tv *testvars.TestVars) {
|
|
_, _ = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
|
Namespace: s.Namespace().String(),
|
|
TaskQueue: tv.TaskQueue(),
|
|
Identity: "random",
|
|
DeploymentOptions: tv.WorkerDeploymentOptions(true),
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) pollFromDeploymentWithTaskQueueNumber(ctx context.Context, tv *testvars.TestVars, taskQueueNumber int) {
|
|
_, _ = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
|
Namespace: s.Namespace().String(),
|
|
TaskQueue: tv.WithTaskQueueNumber(taskQueueNumber).TaskQueue(),
|
|
Identity: "random",
|
|
DeploymentOptions: tv.WorkerDeploymentOptions(true),
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) pollFromDeploymentExpectFail(ctx context.Context, tv *testvars.TestVars, expectedError string) {
|
|
_, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
|
Namespace: s.Namespace().String(),
|
|
TaskQueue: tv.TaskQueue(),
|
|
Identity: "random",
|
|
DeploymentOptions: tv.WorkerDeploymentOptions(true),
|
|
})
|
|
s.Error(err)
|
|
s.Equal(expectedError, err.Error())
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) ensureCreateVersionWithExpectedTaskQueues(ctx context.Context, tv *testvars.TestVars, expectedTaskQueues int) {
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
respV, _ := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv.DeploymentVersionString(),
|
|
})
|
|
|
|
a.Equal(expectedTaskQueues, len(respV.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos()))
|
|
}, 5*time.Minute, 500*time.Millisecond)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) ensureCreateVersionInDeployment(
|
|
tv *testvars.TestVars,
|
|
) {
|
|
v := tv.DeploymentVersionString()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
res, _ := s.FrontendClient().DescribeWorkerDeployment(ctx,
|
|
&workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
|
|
found := false
|
|
if res != nil {
|
|
for _, vs := range res.GetWorkerDeploymentInfo().GetVersionSummaries() {
|
|
if vs.GetVersion() == v {
|
|
found = true
|
|
}
|
|
}
|
|
}
|
|
a.True(found)
|
|
}, 1*time.Minute, 100*time.Millisecond)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) ensureCreateDeployment(
|
|
tv *testvars.TestVars,
|
|
) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
res, _ := s.FrontendClient().DescribeWorkerDeployment(ctx,
|
|
&workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NotNil(res)
|
|
}, 5*time.Second, 100*time.Millisecond)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) startVersionWorkflow(ctx context.Context, tv *testvars.TestVars) {
|
|
go s.pollFromDeployment(ctx, tv)
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv.DeploymentVersionString(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion())
|
|
}, time.Minute, time.Second)
|
|
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestForceCAN_NoOpenWFS() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
// Start a version workflow
|
|
s.startVersionWorkflow(ctx, tv)
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
|
|
// Set the version as current
|
|
_, err := s.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, &workflowservice.SetWorkerDeploymentCurrentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
Version: tv.DeploymentVersionString(),
|
|
})
|
|
s.NoError(err)
|
|
|
|
// ForceCAN
|
|
workflowID := worker_versioning.GenerateDeploymentWorkflowID(tv.DeploymentSeries())
|
|
workflowExecution := &commonpb.WorkflowExecution{
|
|
WorkflowId: workflowID,
|
|
}
|
|
|
|
err = s.SendSignal(s.Namespace().String(), workflowExecution, workerdeployment.ForceCANSignalName, nil, tv.ClientIdentity())
|
|
s.NoError(err)
|
|
|
|
// Verify if the state is intact even after a CAN
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestDeploymentVersionLimits() {
|
|
s.OverrideDynamicConfig(dynamicconfig.MatchingMaxVersionsInDeployment, 1)
|
|
s.OverrideDynamicConfig(dynamicconfig.MatchingMaxTaskQueuesInDeploymentVersion, 1)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*120)
|
|
defer cancel()
|
|
|
|
firstDeployment := testvars.New(s).WithDeploymentSeriesNumber(1)
|
|
secondDeployment := testvars.New(s).WithDeploymentSeriesNumber(2)
|
|
|
|
firstDeploymentVersionOne := firstDeployment.WithBuildIDNumber(1)
|
|
firstDeploymentVersionTwo := firstDeployment.WithBuildIDNumber(2)
|
|
|
|
secondDeploymentVersionOne := secondDeployment.WithBuildIDNumber(1)
|
|
|
|
expectedErrorMaxVersions := fmt.Sprintf("cannot add version %v since maximum number of versions (1) have been registered in the deployment", firstDeploymentVersionTwo.DeploymentVersionString())
|
|
expectedErrorMaxTaskQueues := fmt.Sprintf("cannot add task queue %v since maximum number of task queues (1) have been registered in deployment", secondDeploymentVersionOne.WithTaskQueueNumber(2).TaskQueue().GetName())
|
|
|
|
// First deployment version should be fine
|
|
go s.pollFromDeployment(ctx, firstDeploymentVersionOne)
|
|
s.ensureCreateVersionInDeployment(firstDeploymentVersionOne)
|
|
|
|
// pollers of second version in the same deployment should be rejected
|
|
s.pollFromDeploymentExpectFail(ctx, firstDeploymentVersionTwo, expectedErrorMaxVersions)
|
|
|
|
// But first version of another deployment fine
|
|
go s.pollFromDeployment(ctx, secondDeploymentVersionOne)
|
|
s.ensureCreateVersionInDeployment(secondDeploymentVersionOne)
|
|
|
|
// pollers of the second TQ in the same deployment version should be rejected
|
|
s.pollFromDeploymentExpectFail(ctx, secondDeploymentVersionOne.WithTaskQueueNumber(2), expectedErrorMaxTaskQueues)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestNamespaceDeploymentsLimit() {
|
|
// TODO (carly): check the error messages that poller receives in each case and make sense they are informative and appropriate (e.g. do not expose internal stuff)
|
|
s.T().Skip() // Need to separate this test so other tests do not create deployment in the same NS
|
|
|
|
s.OverrideDynamicConfig(dynamicconfig.MatchingMaxDeployments, 1)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
|
|
tv := testvars.New(s)
|
|
|
|
// First deployment version should be fine
|
|
go s.pollFromDeployment(ctx, tv)
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
|
|
// pollers of the second deployment version should be rejected
|
|
s.pollFromDeploymentExpectFail(ctx, tv.WithDeploymentSeriesNumber(2), "reached maximum deployments in namespace (1)")
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestDescribeWorkerDeployment_TwoVersions_Sorted() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
// Starting two versions of the deployment
|
|
firstVersion := tv.WithBuildIDNumber(1)
|
|
secondVersion := tv.WithBuildIDNumber(2)
|
|
|
|
go s.pollFromDeployment(ctx, firstVersion)
|
|
|
|
// waiting for 1ms to start the second version later.
|
|
startTime := time.Now()
|
|
waitTime := 1 * time.Millisecond
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
a.Greater(time.Since(startTime), waitTime)
|
|
}, 10*time.Second, 1000*time.Millisecond)
|
|
|
|
go s.pollFromDeployment(ctx, secondVersion)
|
|
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.NotNil(resp.GetWorkerDeploymentInfo())
|
|
a.Equal(tv.DeploymentSeries(), resp.GetWorkerDeploymentInfo().GetName())
|
|
|
|
a.NotNil(resp.GetWorkerDeploymentInfo().GetVersionSummaries())
|
|
a.Equal(2, len(resp.GetWorkerDeploymentInfo().GetVersionSummaries()))
|
|
|
|
// Verify that the version summaries are non-nil and sorted.
|
|
versionSummaries := resp.GetWorkerDeploymentInfo().GetVersionSummaries()
|
|
|
|
a.NotNil(versionSummaries[0].GetVersion())
|
|
a.NotNil(versionSummaries[1].GetVersion())
|
|
a.Equal(versionSummaries[0].GetVersion(), secondVersion.DeploymentVersionString())
|
|
a.Equal(versionSummaries[1].GetVersion(), firstVersion.DeploymentVersionString())
|
|
|
|
a.NotNil(resp.GetWorkerDeploymentInfo().GetVersionSummaries()[0].GetCreateTime())
|
|
a.NotNil(resp.GetWorkerDeploymentInfo().GetVersionSummaries()[1].GetCreateTime())
|
|
|
|
a.NotNil(resp.GetWorkerDeploymentInfo().GetCreateTime())
|
|
|
|
a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, resp.GetWorkerDeploymentInfo().GetVersionSummaries()[0].GetStatus())
|
|
a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, resp.GetWorkerDeploymentInfo().GetVersionSummaries()[1].GetStatus())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestDescribeWorkerDeployment_MultipleVersions_Sorted() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
numVersions := 10
|
|
|
|
for i := 0; i < numVersions; i++ {
|
|
go s.pollFromDeployment(ctx, tv.WithBuildIDNumber(i))
|
|
|
|
// waiting for 1ms to start the next version later.
|
|
startTime := time.Now()
|
|
waitTime := 1 * time.Millisecond
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
require.Greater(t, time.Since(startTime), waitTime)
|
|
}, 10*time.Second, 1000*time.Millisecond)
|
|
}
|
|
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
|
|
a.NotNil(resp.GetWorkerDeploymentInfo().GetVersionSummaries())
|
|
a.Equal(numVersions, len(resp.GetWorkerDeploymentInfo().GetVersionSummaries()))
|
|
|
|
// Verify that the version summaries are sorted.
|
|
versionSummaries := resp.GetWorkerDeploymentInfo().GetVersionSummaries()
|
|
for i := 0; i < numVersions-1; i++ {
|
|
a.Less(versionSummaries[i+1].GetCreateTime().AsTime(), versionSummaries[i].GetCreateTime().AsTime())
|
|
}
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
// Testing ConflictToken
|
|
func (s *WorkerDeploymentSuite) TestConflictToken_Describe_SetCurrent_SetRamping() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
firstVersion := tv.WithBuildIDNumber(1)
|
|
secondVersion := tv.WithBuildIDNumber(2)
|
|
|
|
// Start deployment version workflow + worker-deployment workflow.
|
|
go s.pollFromDeployment(ctx, firstVersion)
|
|
go s.pollFromDeployment(ctx, secondVersion)
|
|
|
|
var cT []byte
|
|
// No current deployment version set.
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(worker_versioning.UnversionedVersionId, resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion())
|
|
cT = resp.GetConflictToken()
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
s.ensureCreateVersionInDeployment(firstVersion)
|
|
// Set first version as current version
|
|
_, err := s.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, &workflowservice.SetWorkerDeploymentCurrentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
Version: firstVersion.DeploymentVersionString(),
|
|
ConflictToken: cT,
|
|
})
|
|
s.Nil(err)
|
|
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(firstVersion.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion())
|
|
cT = resp.GetConflictToken()
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Set a new second version and set it as the current version
|
|
go s.pollFromDeployment(ctx, secondVersion)
|
|
_, _ = s.FrontendClient().SetWorkerDeploymentRampingVersion(ctx, &workflowservice.SetWorkerDeploymentRampingVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
Version: secondVersion.DeploymentVersionString(),
|
|
Percentage: 5,
|
|
ConflictToken: cT,
|
|
IgnoreMissingTaskQueues: true, // here until we have 'has version started' safeguard in place
|
|
})
|
|
s.Nil(err)
|
|
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(secondVersion.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingVersion())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestConflictToken_SetCurrent_SetRamping_Wrong() {
|
|
s.OverrideDynamicConfig(dynamicconfig.FrontendMaskInternalErrorDetails, true)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
expectedError := "conflict token mismatch"
|
|
|
|
firstVersion := tv.WithBuildIDNumber(1)
|
|
|
|
// Start deployment version workflow + worker-deployment workflow.
|
|
go s.pollFromDeployment(ctx, firstVersion)
|
|
|
|
cTWrong, _ := time.Now().MarshalBinary() // wrong token
|
|
// Wait until deployment exists
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(worker_versioning.UnversionedVersionId, resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Set first version as current version with wrong token
|
|
_, err := s.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, &workflowservice.SetWorkerDeploymentCurrentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
Version: firstVersion.DeploymentVersionString(),
|
|
ConflictToken: cTWrong,
|
|
})
|
|
s.Equal(err.Error(), expectedError)
|
|
|
|
// Set first version as ramping version with wrong token
|
|
_, err = s.FrontendClient().SetWorkerDeploymentRampingVersion(ctx, &workflowservice.SetWorkerDeploymentRampingVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
Version: firstVersion.DeploymentVersionString(),
|
|
Percentage: 5,
|
|
ConflictToken: cTWrong,
|
|
})
|
|
s.Equal(err.Error(), expectedError)
|
|
}
|
|
|
|
// Testing ListWorkerDeployments
|
|
|
|
func (s *WorkerDeploymentSuite) TestListWorkerDeployments_OneVersion_OneDeployment() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
s.startVersionWorkflow(ctx, tv)
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
|
|
latestVersionSummary := &deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
Version: tv.DeploymentVersionString(),
|
|
CreateTime: timestamppb.Now(),
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: nil,
|
|
CurrentSinceTime: nil,
|
|
RoutingUpdateTime: nil,
|
|
FirstActivationTime: nil,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE,
|
|
}
|
|
|
|
expectedDeploymentSummaries := s.buildWorkerDeploymentSummary(
|
|
tv.DeploymentSeries(),
|
|
timestamppb.Now(),
|
|
&deploymentpb.RoutingConfig{
|
|
CurrentVersion: worker_versioning.UnversionedVersionId, // default current version is __unversioned__
|
|
},
|
|
latestVersionSummary,
|
|
nil,
|
|
nil,
|
|
)
|
|
|
|
s.startAndValidateWorkerDeployments(ctx, &workflowservice.ListWorkerDeploymentsRequest{
|
|
Namespace: s.Namespace().String(),
|
|
}, []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary{expectedDeploymentSummaries})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestListWorkerDeployments_TwoVersions_SameDeployment_OneCurrent_NoRamping() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
firstVersion := tv.WithBuildIDNumber(1)
|
|
secondVersion := tv.WithBuildIDNumber(2)
|
|
|
|
routingInfo := &deploymentpb.RoutingConfig{
|
|
CurrentVersion: firstVersion.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: timestamppb.Now(),
|
|
}
|
|
|
|
s.startVersionWorkflow(ctx, firstVersion)
|
|
s.ensureCreateVersionInDeployment(firstVersion)
|
|
|
|
s.startVersionWorkflow(ctx, secondVersion)
|
|
s.ensureCreateVersionInDeployment(secondVersion)
|
|
|
|
s.setCurrentVersion(ctx, firstVersion, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
latestVersionSummary := &deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
Version: secondVersion.DeploymentVersionString(),
|
|
CreateTime: timestamppb.Now(),
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: nil,
|
|
CurrentSinceTime: nil,
|
|
RoutingUpdateTime: nil,
|
|
FirstActivationTime: nil,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE,
|
|
}
|
|
currentVersionSummary := &deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
Version: firstVersion.DeploymentVersionString(),
|
|
CreateTime: timestamppb.Now(),
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: nil,
|
|
CurrentSinceTime: timestamppb.Now(),
|
|
RoutingUpdateTime: timestamppb.Now(),
|
|
FirstActivationTime: timestamppb.Now(),
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
}
|
|
|
|
expectedDeploymentSummary := s.buildWorkerDeploymentSummary(
|
|
tv.DeploymentSeries(),
|
|
timestamppb.Now(),
|
|
routingInfo,
|
|
latestVersionSummary,
|
|
currentVersionSummary,
|
|
nil,
|
|
)
|
|
|
|
s.startAndValidateWorkerDeployments(ctx, &workflowservice.ListWorkerDeploymentsRequest{
|
|
Namespace: s.Namespace().String(),
|
|
}, []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary{
|
|
expectedDeploymentSummary,
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestListWorkerDeployments_TwoVersions_SameDeployment_OneCurrent_OneRamping() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
currentVersionVars := tv.WithBuildIDNumber(1)
|
|
rampingVersionVars := tv.WithBuildIDNumber(2)
|
|
|
|
routingInfo := &deploymentpb.RoutingConfig{
|
|
CurrentVersion: currentVersionVars.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: timestamppb.Now(),
|
|
RampingVersion: rampingVersionVars.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: timestamppb.Now(),
|
|
}
|
|
|
|
s.startVersionWorkflow(ctx, currentVersionVars)
|
|
s.ensureCreateVersionInDeployment(currentVersionVars)
|
|
|
|
s.startVersionWorkflow(ctx, rampingVersionVars)
|
|
s.ensureCreateVersionInDeployment(rampingVersionVars)
|
|
|
|
s.setCurrentVersion(ctx, currentVersionVars, worker_versioning.UnversionedVersionId, true, "") // starts first version's version workflow + set it to current
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: "",
|
|
PreviousPercentage: 0,
|
|
})
|
|
|
|
latestVersionSummary := &deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: timestamppb.Now(),
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: timestamppb.Now(),
|
|
CurrentSinceTime: nil,
|
|
RoutingUpdateTime: timestamppb.Now(),
|
|
FirstActivationTime: timestamppb.Now(),
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
}
|
|
currentVersionSummary := &deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
Version: currentVersionVars.DeploymentVersionString(),
|
|
CreateTime: timestamppb.Now(),
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: nil,
|
|
CurrentSinceTime: timestamppb.Now(),
|
|
RoutingUpdateTime: timestamppb.Now(),
|
|
FirstActivationTime: timestamppb.Now(),
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
}
|
|
|
|
expectedDeploymentSummary := s.buildWorkerDeploymentSummary(
|
|
tv.DeploymentSeries(),
|
|
timestamppb.Now(),
|
|
routingInfo,
|
|
latestVersionSummary,
|
|
currentVersionSummary,
|
|
latestVersionSummary, // latest version added is the ramping version
|
|
)
|
|
|
|
s.startAndValidateWorkerDeployments(ctx, &workflowservice.ListWorkerDeploymentsRequest{
|
|
Namespace: s.Namespace().String(),
|
|
}, []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary{
|
|
expectedDeploymentSummary,
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestListWorkerDeployments_RampingVersionPercentageChange_RampingChangedTime() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
s.startVersionWorkflow(ctx, tv)
|
|
s.setAndVerifyRampingVersion(ctx, tv, false, 50, true, "", nil) // set version as ramping
|
|
rampingVersionChangedTime := timestamppb.Now()
|
|
|
|
routingInfo := &deploymentpb.RoutingConfig{
|
|
CurrentVersion: worker_versioning.UnversionedVersionId,
|
|
CurrentVersionChangedTime: nil,
|
|
RampingVersion: tv.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: rampingVersionChangedTime,
|
|
}
|
|
|
|
// to simulate time passing before the next ramping version update
|
|
//nolint:forbidigo
|
|
time.Sleep(2 * time.Second)
|
|
|
|
// modify ramping version percentage
|
|
s.setAndVerifyRampingVersion(ctx, tv, false, 75, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: tv.DeploymentVersionString(),
|
|
PreviousPercentage: 50,
|
|
})
|
|
|
|
// only the ramping version percentage should be updated, not the ramping version update time
|
|
// since we are not changing the ramping version
|
|
routingInfo.RampingVersionPercentage = 75
|
|
|
|
rampingVersionSummary := &deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
Version: tv.DeploymentVersionString(),
|
|
CreateTime: rampingVersionChangedTime,
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: rampingVersionChangedTime,
|
|
CurrentSinceTime: nil,
|
|
RoutingUpdateTime: timestamppb.Now(), // since the ramp percentage changed, the routing update time is updated
|
|
FirstActivationTime: rampingVersionChangedTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
}
|
|
|
|
expectedDeploymentSummary := s.buildWorkerDeploymentSummary(
|
|
tv.DeploymentSeries(),
|
|
rampingVersionChangedTime,
|
|
routingInfo,
|
|
rampingVersionSummary, // latest version added is the ramping version
|
|
nil,
|
|
rampingVersionSummary,
|
|
)
|
|
|
|
s.startAndValidateWorkerDeployments(ctx, &workflowservice.ListWorkerDeploymentsRequest{
|
|
Namespace: s.Namespace().String(),
|
|
}, []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary{
|
|
expectedDeploymentSummary,
|
|
})
|
|
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestListWorkerDeployments_MultipleVersions_MultipleDeployments_OnePage() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
expectedDeploymentSummaries := s.createVersionsInDeployments(ctx, tv, 2)
|
|
|
|
s.startAndValidateWorkerDeployments(ctx, &workflowservice.ListWorkerDeploymentsRequest{
|
|
Namespace: s.Namespace().String(),
|
|
}, expectedDeploymentSummaries)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestListWorkerDeployments_MultipleVersions_MultipleDeployments_MultiplePages() {
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
expectedDeploymentSummaries := s.createVersionsInDeployments(ctx, tv, 5)
|
|
|
|
s.startAndValidateWorkerDeployments(ctx, &workflowservice.ListWorkerDeploymentsRequest{
|
|
Namespace: s.Namespace().String(),
|
|
PageSize: 1,
|
|
}, expectedDeploymentSummaries)
|
|
}
|
|
|
|
// Testing SetWorkerDeploymentRampingVersion
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_Ramping_With_Current() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
rampingVersionVars := tv.WithBuildIDNumber(1)
|
|
currentVersionVars := tv.WithBuildIDNumber(2)
|
|
|
|
versionCreateTime := timestamppb.Now()
|
|
s.startVersionWorkflow(ctx, rampingVersionVars)
|
|
s.startVersionWorkflow(ctx, currentVersionVars)
|
|
|
|
// set version as ramping
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: "",
|
|
PreviousPercentage: 0,
|
|
})
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: rampingVersionVars.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: worker_versioning.UnversionedVersionId,
|
|
CurrentVersionChangedTime: nil,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// set current version
|
|
setCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, currentVersionVars, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
// fresh DescribeWorkerDeployment call
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: rampingVersionVars.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: currentVersionVars.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: currentVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_DuplicateRamp() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
rampingVersionVars := testvars.New(s).WithBuildIDNumber(1)
|
|
|
|
versionCreateTime := timestamppb.Now()
|
|
s.startVersionWorkflow(ctx, rampingVersionVars)
|
|
|
|
// set version as ramping
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: "",
|
|
PreviousPercentage: 0,
|
|
})
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: rampingVersionVars.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: rampingVersionVars.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: rampingVersionVars.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: worker_versioning.UnversionedVersionId,
|
|
CurrentVersionChangedTime: nil,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
},
|
|
LastModifierIdentity: rampingVersionVars.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// setting version as ramping again
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: rampingVersionVars.DeploymentVersionString(),
|
|
PreviousPercentage: 50,
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_Invalid_SetCurrent_To_Ramping() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
versionCreateTime := timestamppb.Now()
|
|
currentVersionVars := testvars.New(s).WithBuildIDNumber(1)
|
|
s.startVersionWorkflow(ctx, currentVersionVars)
|
|
|
|
setCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, currentVersionVars, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: currentVersionVars.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: currentVersionVars.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "", // no ramping info should be set
|
|
RampingVersionPercentage: 0, // no ramping info should be set
|
|
RampingVersionChangedTime: nil, // no ramping info should be set
|
|
RampingVersionPercentageChangedTime: nil, // no ramping info should be set
|
|
CurrentVersion: currentVersionVars.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: currentVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: currentVersionVars.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
expectedError := fmt.Errorf("ramping version %s is already current", currentVersionVars.DeploymentVersionString())
|
|
s.setAndVerifyRampingVersion(ctx, currentVersionVars, false, 50, true, expectedError.Error(), nil) // setting current version to ramping should fail
|
|
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: currentVersionVars.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: currentVersionVars.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "", // no ramping info should be set
|
|
RampingVersionPercentage: 0, // no ramping info should be set
|
|
RampingVersionChangedTime: nil, // no ramping info should be set
|
|
RampingVersionPercentageChangedTime: nil, // no ramping info should be set
|
|
CurrentVersion: currentVersionVars.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: timestamppb.Now(),
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: currentVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: currentVersionVars.ClientIdentity(),
|
|
},
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_ModifyExistingRampVersionPercentage() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
versionCreateTime := timestamppb.Now()
|
|
rampingVersionVars := tv.WithBuildIDNumber(1)
|
|
s.startVersionWorkflow(ctx, rampingVersionVars)
|
|
|
|
// set version as ramping
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: "",
|
|
PreviousPercentage: 0,
|
|
})
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: rampingVersionVars.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: worker_versioning.UnversionedVersionId,
|
|
CurrentVersionChangedTime: nil,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// modify ramping version percentage
|
|
modifyRampingPercentageTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 75, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: rampingVersionVars.DeploymentVersionString(),
|
|
PreviousPercentage: 50,
|
|
})
|
|
|
|
// RampingVersionPercentage and RampingVersionPercentageChangedTime should be updated
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: rampingVersionVars.DeploymentVersionString(),
|
|
RampingVersionPercentage: 75, // ramping version percentage is updated to 75
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: modifyRampingPercentageTime, // timestamp is updated as the ramp percentage changed from 50 -> 75
|
|
CurrentVersion: worker_versioning.UnversionedVersionId,
|
|
CurrentVersionChangedTime: nil,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_WithCurrent_Unset_Ramp() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
versionCreateTime := timestamppb.Now()
|
|
rampingVersionVars := tv.WithBuildIDNumber(1)
|
|
currentVersionVars := tv.WithBuildIDNumber(2)
|
|
|
|
s.startVersionWorkflow(ctx, rampingVersionVars)
|
|
s.startVersionWorkflow(ctx, currentVersionVars)
|
|
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", nil) // set version as ramping
|
|
|
|
setCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, currentVersionVars, worker_versioning.UnversionedVersionId, true, "") // set version as curent
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: rampingVersionVars.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: currentVersionVars.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
{
|
|
Version: currentVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// unset ramping version
|
|
unsetRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, true, 0, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: rampingVersionVars.DeploymentVersionString(),
|
|
PreviousPercentage: 50,
|
|
})
|
|
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: unsetRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: unsetRampingUpdateTime,
|
|
CurrentVersion: currentVersionVars.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING},
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: unsetRampingUpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING,
|
|
},
|
|
{
|
|
Version: currentVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_SetRampingAsCurrent() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
versionCreateTime := timestamppb.Now()
|
|
rampingVersionVars := tv.WithBuildIDNumber(1)
|
|
s.startVersionWorkflow(ctx, rampingVersionVars)
|
|
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", nil)
|
|
|
|
// set ramping version as current
|
|
setCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, rampingVersionVars, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "", // no ramping info should be set
|
|
RampingVersionPercentage: 0, // no ramping info should be set
|
|
RampingVersionChangedTime: setCurrentUpdateTime, // ramping version got updated to ""
|
|
RampingVersionPercentageChangedTime: setCurrentUpdateTime, // ramping version got updated to ""
|
|
CurrentVersion: rampingVersionVars.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: rampingVersionVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_NoCurrent_Unset_Ramp() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
rampingVersionVars := tv.WithBuildIDNumber(1)
|
|
s.startVersionWorkflow(ctx, rampingVersionVars)
|
|
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, false, 50, true, "", nil)
|
|
s.setAndVerifyRampingVersion(ctx, rampingVersionVars, true, 0, true, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
PreviousVersion: rampingVersionVars.DeploymentVersionString(),
|
|
PreviousPercentage: 50,
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_Batching() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
s.InjectHook(testhooks.TaskQueuesInDeploymentSyncBatchSize, 1)
|
|
|
|
// registering 5 task-queues in the version which would result in the creation of 5 batches, each with 1 task-queue, during the SyncState call.
|
|
versionCreateTime := timestamppb.Now()
|
|
taskQueues := 5
|
|
for i := 0; i < taskQueues; i++ {
|
|
go s.pollFromDeploymentWithTaskQueueNumber(ctx, tv, i)
|
|
}
|
|
|
|
// ensure the version has been created in the deployment with the right number of task-queues
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
s.ensureCreateVersionWithExpectedTaskQueues(ctx, tv, taskQueues)
|
|
|
|
// verify that all the registered task-queues have "" set as their ramping version
|
|
for i := 0; i < taskQueues; i++ {
|
|
s.verifyTaskQueueVersioningInfo(ctx, tv.WithTaskQueueNumber(i).TaskQueue(), worker_versioning.UnversionedVersionId, "", 0)
|
|
}
|
|
|
|
// set ramping version to 50%
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, tv, false, 50, true, "", nil)
|
|
|
|
// verify the task queues have new ramping version
|
|
for i := 0; i < taskQueues; i++ {
|
|
s.verifyTaskQueueVersioningInfo(ctx, tv.WithTaskQueueNumber(i).TaskQueue(), worker_versioning.UnversionedVersionId, tv.DeploymentVersionString(), 50)
|
|
}
|
|
|
|
// verify if the worker-deployment has the right ramping version set
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.Nil(err)
|
|
s.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingVersion())
|
|
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: tv.DeploymentVersionString(),
|
|
RampingVersionPercentage: 50,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: worker_versioning.UnversionedVersionId,
|
|
CurrentVersionChangedTime: nil,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
FirstActivationTime: setRampingUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
}
|
|
|
|
// TestSetWorkerDeploymentRampingVersion_UnversionedRamp_Batching verifies that the batching functionality works
|
|
// when ramping unversioned.
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_UnversionedRamp_Batching() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*60)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
s.InjectHook(testhooks.TaskQueuesInDeploymentSyncBatchSize, 1)
|
|
|
|
// registering 5 task-queues in the version which would result in the creation of 5 batches, each with 1 task-queue, during the SyncState call.
|
|
versionCreateTime := timestamppb.Now()
|
|
taskQueues := 5
|
|
for i := 0; i < taskQueues; i++ {
|
|
go s.pollFromDeploymentWithTaskQueueNumber(ctx, tv, i)
|
|
}
|
|
|
|
// ensure the version has been created in the deployment with the right number of task-queues
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
s.ensureCreateVersionWithExpectedTaskQueues(ctx, tv, taskQueues)
|
|
|
|
// make the current version versioned, so that we can set ramp to unversioned later
|
|
setCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, tv, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
// set ramp to unversioned which should trigger a batch of SyncDeploymentVersionUserData requests.
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersionUnversionedOption(ctx, tv, true, false, 75, true, "", nil)
|
|
|
|
// check that the current version's task queues have ramping version == __unversioned__
|
|
for i := 0; i < taskQueues; i++ {
|
|
s.verifyTaskQueueVersioningInfo(ctx, tv.WithTaskQueueNumber(i).TaskQueue(), tv.DeploymentVersionString(), worker_versioning.UnversionedVersionId, 75)
|
|
}
|
|
|
|
// verify if the worker-deployment has the right ramping version set
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.Nil(err)
|
|
// nolint:staticcheck // SA1019: old worker versioning
|
|
s.Equal(worker_versioning.UnversionedVersionId, resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingVersion())
|
|
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: worker_versioning.UnversionedVersionId,
|
|
RampingVersionPercentage: 75,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: tv.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
}
|
|
|
|
// SetCurrent tests
|
|
|
|
func (s *WorkerDeploymentSuite) TestDescribeWorkerDeployment_SetCurrentVersion() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
versionsCreateTime := timestamppb.Now()
|
|
firstVersion := tv.WithBuildIDNumber(1)
|
|
secondVersion := tv.WithBuildIDNumber(2)
|
|
|
|
go s.pollFromDeployment(ctx, firstVersion)
|
|
|
|
// No current deployment version set.
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(worker_versioning.UnversionedVersionId, resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion()) //nolint:staticcheck // SA1019: old worker versioning
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Set first version as current version
|
|
firstVersionCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, firstVersion, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionsCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
CurrentVersion: firstVersion.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: firstVersionCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: firstVersion.DeploymentVersionString(),
|
|
CreateTime: versionsCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: firstVersionCurrentUpdateTime,
|
|
CurrentSinceTime: firstVersionCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: firstVersionCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// Set a new second version and set it as the current version
|
|
go s.pollFromDeployment(ctx, secondVersion)
|
|
secondVersionCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, secondVersion, firstVersion.DeploymentVersionString(), true, "")
|
|
|
|
// Verify that the first version is draining, has an updated last deactivation time + second version is current, has an updated first activation time.
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionsCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
CurrentVersion: secondVersion.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: secondVersionCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: firstVersion.DeploymentVersionString(),
|
|
CreateTime: versionsCreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING},
|
|
RoutingUpdateTime: firstVersionCurrentUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: firstVersionCurrentUpdateTime,
|
|
LastDeactivationTime: secondVersionCurrentUpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING,
|
|
},
|
|
{
|
|
Version: secondVersion.DeploymentVersionString(),
|
|
CreateTime: versionsCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: secondVersionCurrentUpdateTime,
|
|
CurrentSinceTime: secondVersionCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: secondVersionCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetCurrentVersion_Batching() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
|
|
s.InjectHook(testhooks.TaskQueuesInDeploymentSyncBatchSize, 1)
|
|
|
|
// registering 5 task-queues in the version which would result in the creation of 5 batches, each with 1 task-queue, during the SyncState call.
|
|
versionCreateTime := timestamppb.Now()
|
|
taskQueues := 5
|
|
for i := 0; i < taskQueues; i++ {
|
|
go s.pollFromDeploymentWithTaskQueueNumber(ctx, tv, i)
|
|
}
|
|
|
|
// ensure the version has been created in the deployment with the right number of task-queues
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
s.ensureCreateVersionWithExpectedTaskQueues(ctx, tv, taskQueues)
|
|
|
|
// verify that all the registered task-queues have "__unversioned__" as their current version
|
|
for i := 0; i < taskQueues; i++ {
|
|
s.verifyTaskQueueVersioningInfo(ctx, tv.WithTaskQueueNumber(i).TaskQueue(), worker_versioning.UnversionedVersionId, "", 0)
|
|
}
|
|
|
|
// set current and check that the current version's task queues have new current version
|
|
setCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, tv, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
// verify the current version has propogated to all the registered task-queues userData
|
|
for i := 0; i < taskQueues; i++ {
|
|
s.verifyTaskQueueVersioningInfo(ctx, tv.WithTaskQueueNumber(i).TaskQueue(), tv.DeploymentVersionString(), "", 0)
|
|
}
|
|
|
|
// verify if the worker-deployment has the right current version set
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.Nil(err)
|
|
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: nil,
|
|
RampingVersionPercentageChangedTime: nil,
|
|
CurrentVersion: tv.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentUpdateTime,
|
|
CurrentSinceTime: setCurrentUpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentUpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
}
|
|
|
|
// Should see that the current version of the task queues becomes unversioned
|
|
func (s *WorkerDeploymentSuite) TestSetCurrentVersion_Unversioned_NoRamp() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
currentVars := testvars.New(s).WithBuildIDNumber(1)
|
|
|
|
versionCreateTime := timestamppb.Now()
|
|
go s.pollFromDeployment(ctx, currentVars)
|
|
s.ensureCreateVersionInDeployment(currentVars)
|
|
|
|
// check that the current version's task queues have current version unversioned to start
|
|
s.verifyTaskQueueVersioningInfo(ctx, currentVars.TaskQueue(), worker_versioning.UnversionedVersionId, "", 0)
|
|
|
|
// set current and check that the current version's task queues have new current version
|
|
firstCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, currentVars, worker_versioning.UnversionedVersionId, true, "")
|
|
s.verifyTaskQueueVersioningInfo(ctx, currentVars.TaskQueue(), currentVars.DeploymentVersionString(), "", 0)
|
|
|
|
// set current unversioned and check that the current version's task queues have current version unversioned again
|
|
secondCurrentUpdateTime := timestamppb.Now()
|
|
s.setCurrentVersionUnversionedOption(ctx, currentVars, true, currentVars.DeploymentVersionString(), true, "")
|
|
s.verifyTaskQueueVersioningInfo(ctx, currentVars.TaskQueue(), worker_versioning.UnversionedVersionId, "", 0)
|
|
|
|
// check that deployment has current version == __unversioned__
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: currentVars.DeploymentSeries(),
|
|
})
|
|
s.Nil(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: currentVars.DeploymentSeries(),
|
|
CreateTime: versionCreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
CurrentVersion: worker_versioning.UnversionedVersionId,
|
|
CurrentVersionChangedTime: secondCurrentUpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: currentVars.DeploymentVersionString(),
|
|
CreateTime: versionCreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING},
|
|
RoutingUpdateTime: secondCurrentUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: firstCurrentUpdateTime,
|
|
LastDeactivationTime: secondCurrentUpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING,
|
|
},
|
|
},
|
|
LastModifierIdentity: currentVars.ClientIdentity(),
|
|
},
|
|
})
|
|
}
|
|
|
|
// Should see that the current version of the task queue becomes unversioned, and the unversioned ramping version of the task queue is removed
|
|
func (s *WorkerDeploymentSuite) TestSetCurrentVersion_Unversioned_PromoteUnversionedRamp() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*60)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
currentVars := tv.WithBuildIDNumber(1)
|
|
|
|
go s.pollFromDeployment(ctx, currentVars)
|
|
s.ensureCreateVersionInDeployment(currentVars)
|
|
|
|
// make the current version versioned, so that we can set ramp to unversioned
|
|
s.setCurrentVersion(ctx, currentVars, worker_versioning.UnversionedVersionId, true, "")
|
|
// set ramp to unversioned
|
|
s.setAndVerifyRampingVersionUnversionedOption(ctx, tv, true, false, 75, true, "", nil)
|
|
// check that the current version's task queues have ramping version == __unversioned__
|
|
s.verifyTaskQueueVersioningInfo(ctx, currentVars.TaskQueue(), currentVars.DeploymentVersionString(), worker_versioning.UnversionedVersionId, 75)
|
|
|
|
// set current to unversioned
|
|
s.setCurrentVersionUnversionedOption(ctx, tv, true, currentVars.DeploymentVersionString(), true, "")
|
|
|
|
// check that the current version's task queues have ramping version == "" and current version == "__unversioned__"
|
|
s.verifyTaskQueueVersioningInfo(ctx, currentVars.TaskQueue(), worker_versioning.UnversionedVersionId, "", 0)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetCurrentVersion_Concurrent_DifferentVersions_NoUnexpectedErrors() {
|
|
s.OverrideDynamicConfig(dynamicconfig.WorkflowExecutionMaxInFlightUpdates, 10) // this is the default
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
tv := testvars.New(s)
|
|
errChan := make(chan error)
|
|
|
|
versions := 10
|
|
for i := 0; i < versions; i++ {
|
|
s.startVersionWorkflow(ctx, tv.WithBuildIDNumber(i))
|
|
}
|
|
|
|
// Concurrently set 10 different versions as current version
|
|
for i := 0; i < versions; i++ {
|
|
go func() {
|
|
_, err := s.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, &workflowservice.SetWorkerDeploymentCurrentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: tv.WithBuildIDNumber(i).DeploymentVersionString(),
|
|
IgnoreMissingTaskQueues: true,
|
|
Identity: tv.ClientIdentity(),
|
|
})
|
|
errChan <- err
|
|
}()
|
|
}
|
|
|
|
for i := 0; i < versions; i++ {
|
|
err := <-errChan
|
|
if err != nil {
|
|
switch err.(type) {
|
|
// DeadlineExceeded and ResourceExhausted are expected errors since there could be more
|
|
// in-flight updates than WorkflowExecutionMaxInFlightUpdates or we could get a timeout error.
|
|
case *serviceerror.DeadlineExceeded, *serviceerror.ResourceExhausted:
|
|
continue
|
|
default:
|
|
s.FailNow("Unexpected error: ", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Verify that the current version is set.
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
})
|
|
s.NoError(err)
|
|
s.NotEqual(worker_versioning.UnversionedVersionId, resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion())
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetCurrentVersion_Concurrent_SameVersion_NoUnexpectedErrors() {
|
|
s.OverrideDynamicConfig(dynamicconfig.WorkflowExecutionMaxInFlightUpdates, 10) // this is the default
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
tv := testvars.New(s)
|
|
errChan := make(chan error)
|
|
|
|
s.startVersionWorkflow(ctx, tv) // create version
|
|
|
|
// Concurrently set the same version as current version 10 times.
|
|
for i := 0; i < 10; i++ {
|
|
go func() {
|
|
_, err := s.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, &workflowservice.SetWorkerDeploymentCurrentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: tv.DeploymentVersionString(),
|
|
IgnoreMissingTaskQueues: true,
|
|
Identity: tv.ClientIdentity(),
|
|
})
|
|
errChan <- err
|
|
}()
|
|
}
|
|
|
|
for i := 0; i < 10; i++ {
|
|
err := <-errChan
|
|
if err != nil {
|
|
switch err.(type) {
|
|
// DeadlineExceeded and ResourceExhausted are expected errors since there could be more
|
|
// in-flight updates than WorkflowExecutionMaxInFlightUpdates or we could get a timeout error.
|
|
case *serviceerror.DeadlineExceeded, *serviceerror.ResourceExhausted:
|
|
continue
|
|
default:
|
|
s.FailNow("Unexpected error: ", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Verify that the current version is set.
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
})
|
|
s.NoError(err)
|
|
s.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion())
|
|
}
|
|
|
|
// TestConcurrentPollers_DifferentTaskQueues_SameVersion_SetCurrentVersion aims to test that when there are multiple pollers polling on different task queues,
|
|
// all belonging to the same version, a setCurrentVersion call succeeds with all the task queues eventually having this version as the current version in their versioning info.
|
|
func (s *WorkerDeploymentSuite) TestConcurrentPollers_DifferentTaskQueues_SameVersion_SetCurrentVersion() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
|
|
defer cancel()
|
|
|
|
// start 10 different pollers each polling on a different task queue but belonging to the same version
|
|
tv := testvars.New(s)
|
|
|
|
versions := 10
|
|
for i := 0; i < versions; i++ {
|
|
go s.startVersionWorkflow(ctx, tv.WithTaskQueueNumber(i))
|
|
}
|
|
|
|
// set this version as current version
|
|
s.setCurrentVersion(ctx, tv, worker_versioning.UnversionedVersionId, false, "")
|
|
|
|
// verify that the task queues, eventually, have this version as the current version in their versioning info
|
|
for i := 0; i < versions; i++ {
|
|
s.verifyTaskQueueVersioningInfo(ctx, tv.WithTaskQueueNumber(i).TaskQueue(), tv.DeploymentVersionString(), "", 0)
|
|
}
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetRampingVersion_Concurrent_DifferentVersions_NoUnexpectedErrors() {
|
|
s.OverrideDynamicConfig(dynamicconfig.WorkflowExecutionMaxInFlightUpdates, 10) // this is the default
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
tv := testvars.New(s)
|
|
errChan := make(chan error)
|
|
|
|
versions := 10
|
|
for i := 0; i < versions; i++ {
|
|
s.startVersionWorkflow(ctx, tv.WithBuildIDNumber(i))
|
|
}
|
|
|
|
// Concurrently set 10 different versions as ramping version
|
|
for i := 0; i < versions; i++ {
|
|
go func() {
|
|
_, err := s.FrontendClient().SetWorkerDeploymentRampingVersion(ctx, &workflowservice.SetWorkerDeploymentRampingVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: tv.WithBuildIDNumber(i).DeploymentVersionString(),
|
|
IgnoreMissingTaskQueues: true,
|
|
Identity: tv.ClientIdentity(),
|
|
Percentage: 50,
|
|
})
|
|
errChan <- err
|
|
}()
|
|
}
|
|
|
|
for i := 0; i < versions; i++ {
|
|
err := <-errChan
|
|
if err != nil {
|
|
switch err.(type) {
|
|
// DeadlineExceeded and ResourceExhausted are expected errors since there could be more
|
|
// in-flight updates than WorkflowExecutionMaxInFlightUpdates or we could get a timeout error.
|
|
case *serviceerror.DeadlineExceeded, *serviceerror.ResourceExhausted:
|
|
continue
|
|
default:
|
|
s.FailNow("Unexpected error: ", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Verify that the ramping version is set.
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
})
|
|
s.NoError(err)
|
|
s.NotNil(resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingVersion())
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestSetRampingVersion_Concurrent_SameVersion_NoUnexpectedErrors() {
|
|
s.OverrideDynamicConfig(dynamicconfig.WorkflowExecutionMaxInFlightUpdates, 10) // this is the default
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
tv := testvars.New(s)
|
|
errChan := make(chan error)
|
|
|
|
s.startVersionWorkflow(ctx, tv) // create version
|
|
|
|
// Concurrently set the same version as ramping version 10 times.
|
|
for i := 0; i < 10; i++ {
|
|
go func() {
|
|
_, err := s.FrontendClient().SetWorkerDeploymentRampingVersion(ctx, &workflowservice.SetWorkerDeploymentRampingVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: tv.DeploymentVersionString(),
|
|
IgnoreMissingTaskQueues: true,
|
|
Identity: tv.ClientIdentity(),
|
|
Percentage: 50,
|
|
})
|
|
errChan <- err
|
|
}()
|
|
}
|
|
|
|
for i := 0; i < 10; i++ {
|
|
err := <-errChan
|
|
if err != nil {
|
|
switch err.(type) {
|
|
// DeadlineExceeded and ResourceExhausted are expected errors since there could be more
|
|
// in-flight updates than WorkflowExecutionMaxInFlightUpdates or we could get a timeout error.
|
|
case *serviceerror.DeadlineExceeded, *serviceerror.ResourceExhausted:
|
|
continue
|
|
default:
|
|
s.FailNow("Unexpected error: ", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Verify that the ramping version is set.
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
})
|
|
s.NoError(err)
|
|
s.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingVersion())
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestResourceExhaustedErrors_Converted_To_ReadableMessage() {
|
|
s.OverrideDynamicConfig(dynamicconfig.WorkflowExecutionMaxInFlightUpdates, 2) // Lowering the limit to encounter ResourceExhausted errors
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
tv := testvars.New(s)
|
|
versions := 5
|
|
errChan := make(chan error, versions)
|
|
|
|
// Start all version workflows first
|
|
for i := 0; i < versions; i++ {
|
|
s.startVersionWorkflow(ctx, tv.WithBuildIDNumber(i))
|
|
}
|
|
|
|
// Test SetRampingVersion
|
|
s.testConcurrentRequestsResourceExhausted(ctx, tv, versions, errChan, "SetWorkerDeploymentRampingVersion", func(i int) error {
|
|
_, err := s.FrontendClient().SetWorkerDeploymentRampingVersion(ctx, &workflowservice.SetWorkerDeploymentRampingVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: tv.WithBuildIDNumber(i).DeploymentVersionString(),
|
|
IgnoreMissingTaskQueues: true,
|
|
Identity: tv.ClientIdentity(),
|
|
Percentage: 50,
|
|
})
|
|
return err
|
|
})
|
|
|
|
// Test SetCurrentVersion
|
|
s.testConcurrentRequestsResourceExhausted(ctx, tv, versions, errChan, "SetWorkerDeploymentCurrentVersion", func(i int) error {
|
|
_, err := s.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, &workflowservice.SetWorkerDeploymentCurrentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: tv.WithBuildIDNumber(i).DeploymentVersionString(),
|
|
IgnoreMissingTaskQueues: true,
|
|
Identity: tv.ClientIdentity(),
|
|
})
|
|
return err
|
|
})
|
|
|
|
// Test UpdateVersionMetadata
|
|
metadata := map[string]*commonpb.Payload{
|
|
"key1": {Data: testRandomMetadataValue},
|
|
"key2": {Data: testRandomMetadataValue},
|
|
}
|
|
s.testConcurrentRequestsResourceExhausted(ctx, tv, versions, errChan, "UpdateWorkerDeploymentVersionMetadata", func(i int) error {
|
|
_, err := s.FrontendClient().UpdateWorkerDeploymentVersionMetadata(ctx, &workflowservice.UpdateWorkerDeploymentVersionMetadataRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv.WithBuildIDNumber(i).DeploymentVersionString(),
|
|
UpsertEntries: metadata,
|
|
})
|
|
return err
|
|
})
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) testConcurrentRequestsResourceExhausted(
|
|
ctx context.Context,
|
|
tv *testvars.TestVars,
|
|
versions int,
|
|
errChan chan error,
|
|
apiName string,
|
|
requestFn func(int) error,
|
|
) {
|
|
// Launch concurrent requests
|
|
for i := 0; i < versions; i++ {
|
|
go func(i int) {
|
|
errChan <- requestFn(i)
|
|
}(i)
|
|
}
|
|
|
|
// Expect ResourceExhausted errors to be converted to Internal errors with the appropriate message
|
|
for i := 0; i < versions; i++ {
|
|
err := <-errChan
|
|
if err != nil {
|
|
switch err.(type) {
|
|
case *serviceerror.ResourceExhausted:
|
|
s.Equal(err.Error(), fmt.Sprintf("Too many %s requests have been issued in rapid succession. Please throttle the request rate to avoid exceeding Worker Deployment resource limits.", apiName))
|
|
continue
|
|
default:
|
|
s.FailNow("Unexpected error: ", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Should see it fail because unversioned is already current
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_Unversioned_UnversionedCurrent() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
rampingVars := tv.WithBuildIDNumber(1)
|
|
s.startVersionWorkflow(ctx, rampingVars)
|
|
s.setAndVerifyRampingVersionUnversionedOption(ctx, rampingVars, true, false, 50, true, "ramping version __unversioned__ is already current", nil)
|
|
}
|
|
|
|
// Should see that the ramping version of the task queues in the current version is unversioned
|
|
func (s *WorkerDeploymentSuite) TestSetWorkerDeploymentRampingVersion_Unversioned_VersionedCurrent() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer cancel()
|
|
tv := testvars.New(s)
|
|
currentVars := tv.WithBuildIDNumber(1)
|
|
|
|
go s.pollFromDeployment(ctx, currentVars)
|
|
s.ensureCreateVersionInDeployment(currentVars)
|
|
|
|
// check that the current version's task queues have ramping version == ""
|
|
s.setCurrentVersion(ctx, currentVars, worker_versioning.UnversionedVersionId, true, "")
|
|
s.verifyTaskQueueVersioningInfo(ctx, currentVars.TaskQueue(), currentVars.DeploymentVersionString(), "", 0)
|
|
|
|
// set ramp to unversioned
|
|
s.setAndVerifyRampingVersionUnversionedOption(ctx, tv, true, false, 75, true, "", nil)
|
|
|
|
// check that deployment has ramping version == __unversioned__
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentSeries(),
|
|
})
|
|
s.Nil(err)
|
|
s.Equal(worker_versioning.UnversionedVersionId, resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingVersion())
|
|
|
|
// check that the current version's task queues have ramping version == __unversioned__
|
|
s.verifyTaskQueueVersioningInfo(ctx, currentVars.TaskQueue(), currentVars.DeploymentVersionString(), worker_versioning.UnversionedVersionId, 75)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestTwoPollers_EnsureCreateVersion() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
|
|
defer func() {
|
|
cancel()
|
|
}()
|
|
tv := testvars.New(s)
|
|
tv1 := tv.WithBuildIDNumber(1)
|
|
tv2 := tv.WithBuildIDNumber(2)
|
|
|
|
go s.pollFromDeployment(ctx, tv1)
|
|
go s.pollFromDeployment(ctx, tv2)
|
|
s.ensureCreateVersionWithExpectedTaskQueues(ctx, tv1, 1)
|
|
s.ensureCreateVersionWithExpectedTaskQueues(ctx, tv2, 1)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) verifyTaskQueueVersioningInfo(ctx context.Context, tq *taskqueuepb.TaskQueue, expectedCurrentVersion, expectedRampingVersion string, expectedPercentage float32) {
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
tqDesc, err := s.FrontendClient().DescribeTaskQueue(ctx, &workflowservice.DescribeTaskQueueRequest{
|
|
Namespace: s.Namespace().String(),
|
|
TaskQueue: tq,
|
|
})
|
|
a := require.New(t)
|
|
a.Nil(err)
|
|
a.Equal(expectedCurrentVersion, tqDesc.GetVersioningInfo().GetCurrentVersion()) //nolint:staticcheck // SA1019: old worker versioning
|
|
a.Equal(expectedRampingVersion, tqDesc.GetVersioningInfo().GetRampingVersion()) //nolint:staticcheck // SA1019: old worker versioning
|
|
a.Equal(expectedPercentage, tqDesc.GetVersioningInfo().GetRampingVersionPercentage())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
// This test shall first rollback a drained version to a current version. After that, it shall deploy a new version
|
|
// which shall further drain this current version.
|
|
// Note: This test reproduces a bug we saw in production where the drainage status was not being properly cleared when a draining version
|
|
// is reactivated and then re-deactivated
|
|
func (s *WorkerDeploymentSuite) TestDrainRollbackedVersion() {
|
|
s.OverrideDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv1 := testvars.New(s).WithBuildIDNumber(1)
|
|
tv2 := testvars.New(s).WithBuildIDNumber(2)
|
|
tv3 := testvars.New(s).WithBuildIDNumber(3)
|
|
|
|
// Start deployment workflow 1 and wait for the deployment version to exist
|
|
v1CreateTime := timestamppb.New(time.Now())
|
|
s.startVersionWorkflow(ctx, tv1)
|
|
|
|
// Set v1 as current version
|
|
setCurrentV1UpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, tv1, worker_versioning.UnversionedVersionId, true, "")
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv1.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
CurrentVersion: tv1.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV1UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV1UpdateTime,
|
|
CurrentSinceTime: setCurrentV1UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv1.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// Start deployment workflow 2 and set v2 to current so that v1 can start draining
|
|
v2CreateTime := timestamppb.New(time.Now())
|
|
s.startVersionWorkflow(ctx, tv2)
|
|
|
|
setCurrentV2UpdateTime := timestamppb.New(time.Now())
|
|
s.setCurrentVersion(ctx, tv2, tv1.DeploymentVersionString(), true, "")
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv2.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv2.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: nil,
|
|
RampingVersionPercentageChangedTime: nil,
|
|
CurrentVersion: tv2.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV2UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING},
|
|
RoutingUpdateTime: setCurrentV1UpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime,
|
|
LastDeactivationTime: setCurrentV2UpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: setCurrentV2UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// wait for v1 to be drained
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv1.DeploymentVersionString(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(enumspb.VERSION_DRAINAGE_STATUS_DRAINED, resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo().GetStatus())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Verify that the drainageStatus of v1 has been updated in the VersionSummaries
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv2.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv2.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: nil,
|
|
RampingVersionPercentageChangedTime: nil,
|
|
CurrentVersion: tv2.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV2UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED},
|
|
RoutingUpdateTime: setCurrentV1UpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime,
|
|
LastDeactivationTime: setCurrentV2UpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: setCurrentV2UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// start ramping traffic back to v1
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, tv1, false, 10, false, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
ConflictToken: nil,
|
|
PreviousVersion: "",
|
|
PreviousPercentage: 0,
|
|
})
|
|
|
|
// verify if the right information is set in the DescribeWorkerDeployment response
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv2.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv2.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: tv1.DeploymentVersionString(),
|
|
RampingVersionPercentage: 10,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: tv2.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV2UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
FirstActivationTime: setCurrentV1UpdateTime, // note: this is setCurrentV1UpdateTime since the version was initially activated when it was current.
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: setCurrentV2UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Set version v1 as the current version; this shall drain out v2
|
|
newCurrentV1UpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, tv1, tv2.DeploymentVersionString(), true, "")
|
|
|
|
// Verify that v2 is drained
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv2.DeploymentVersionString(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(enumspb.VERSION_DRAINAGE_STATUS_DRAINED, resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo().GetStatus())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// verify if the right information is set in the DescribeWorkerDeployment response
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv1.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
CurrentVersion: tv1.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: newCurrentV1UpdateTime,
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: newCurrentV1UpdateTime,
|
|
RampingVersionPercentageChangedTime: newCurrentV1UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: newCurrentV1UpdateTime,
|
|
CurrentSinceTime: newCurrentV1UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime, // note: this is setCurrentV1UpdateTime since the version was initially activated when it was current.
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED},
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: newCurrentV1UpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Roll out a new version v3 and set it to current
|
|
v3CreateTime := timestamppb.Now()
|
|
s.startVersionWorkflow(ctx, tv3)
|
|
|
|
newCurrentV3UpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, tv3, tv1.DeploymentVersionString(), true, "")
|
|
|
|
// Verify that v1 is drained eventually
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv1.DeploymentVersionString(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(enumspb.VERSION_DRAINAGE_STATUS_DRAINED, resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo().GetStatus())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Verify that v1, which was rolled back to being current previously, is drained with it's information present
|
|
// in the deployment workflow.
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv1.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
CurrentVersion: tv3.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: newCurrentV3UpdateTime,
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: newCurrentV1UpdateTime,
|
|
RampingVersionPercentageChangedTime: newCurrentV1UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingUpdateTime: newCurrentV1UpdateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED},
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime, // note: this is setCurrentV1UpdateTime since the version was initially activated when it was current.
|
|
LastDeactivationTime: newCurrentV3UpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED},
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: newCurrentV1UpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED,
|
|
},
|
|
{
|
|
Version: tv3.DeploymentVersionString(),
|
|
CreateTime: v3CreateTime,
|
|
RoutingUpdateTime: newCurrentV3UpdateTime,
|
|
DrainageInfo: nil,
|
|
CurrentSinceTime: newCurrentV3UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: newCurrentV3UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
// Test that rolling back to a drained version works
|
|
func (s *WorkerDeploymentSuite) TestSetRampingVersion_AfterDrained() {
|
|
s.OverrideDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv1 := testvars.New(s).WithBuildIDNumber(1)
|
|
tv2 := testvars.New(s).WithBuildIDNumber(2)
|
|
|
|
// Start deployment workflow 1 and wait for the deployment version to exist
|
|
v1CreateTime := timestamppb.New(time.Now())
|
|
s.startVersionWorkflow(ctx, tv1)
|
|
|
|
// Set v1 as current version
|
|
setCurrentV1UpdateTime := timestamppb.Now()
|
|
s.setCurrentVersion(ctx, tv1, worker_versioning.UnversionedVersionId, true, "")
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv1.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
CurrentVersion: tv1.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV1UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV1UpdateTime,
|
|
CurrentSinceTime: setCurrentV1UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv1.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// Start deployment workflow 2 and set v2 to current so that v1 can start draining
|
|
v2CreateTime := timestamppb.New(time.Now())
|
|
s.startVersionWorkflow(ctx, tv2)
|
|
|
|
setCurrentV2UpdateTime := timestamppb.New(time.Now())
|
|
s.setCurrentVersion(ctx, tv2, tv1.DeploymentVersionString(), true, "")
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv2.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv2.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: nil,
|
|
RampingVersionPercentageChangedTime: nil,
|
|
CurrentVersion: tv2.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV2UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING},
|
|
RoutingUpdateTime: setCurrentV1UpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime,
|
|
LastDeactivationTime: setCurrentV2UpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: setCurrentV2UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
|
|
// wait for v1 to be drained
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv1.DeploymentVersionString(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(enumspb.VERSION_DRAINAGE_STATUS_DRAINED, resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo().GetStatus())
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// Verify that the drainageStatus of v1 has been updated in the VersionSummaries
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv2.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv2.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: "",
|
|
RampingVersionPercentage: 0,
|
|
RampingVersionChangedTime: nil,
|
|
RampingVersionPercentageChangedTime: nil,
|
|
CurrentVersion: tv2.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV2UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: &deploymentpb.VersionDrainageInfo{Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED},
|
|
RoutingUpdateTime: setCurrentV1UpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV1UpdateTime,
|
|
LastDeactivationTime: setCurrentV2UpdateTime,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: setCurrentV2UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
|
|
// start ramping traffic back to v1
|
|
setRampingUpdateTime := timestamppb.Now()
|
|
s.setAndVerifyRampingVersion(ctx, tv1, false, 10, false, "", &workflowservice.SetWorkerDeploymentRampingVersionResponse{
|
|
ConflictToken: nil,
|
|
PreviousVersion: "",
|
|
PreviousPercentage: 0,
|
|
})
|
|
|
|
// verify if the right information is set in the DescribeWorkerDeployment response
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err = s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv2.DeploymentSeries(),
|
|
})
|
|
s.NoError(err)
|
|
s.verifyDescribeWorkerDeployment(resp, &workflowservice.DescribeWorkerDeploymentResponse{
|
|
WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
|
|
Name: tv2.DeploymentSeries(),
|
|
CreateTime: v1CreateTime,
|
|
RoutingConfig: &deploymentpb.RoutingConfig{
|
|
RampingVersion: tv1.DeploymentVersionString(),
|
|
RampingVersionPercentage: 10,
|
|
RampingVersionChangedTime: setRampingUpdateTime,
|
|
RampingVersionPercentageChangedTime: setRampingUpdateTime,
|
|
CurrentVersion: tv2.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: setCurrentV2UpdateTime,
|
|
},
|
|
VersionSummaries: []*deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
{
|
|
Version: tv1.DeploymentVersionString(),
|
|
CreateTime: v1CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setRampingUpdateTime,
|
|
CurrentSinceTime: nil,
|
|
RampingSinceTime: setRampingUpdateTime,
|
|
FirstActivationTime: setCurrentV1UpdateTime, // note: this is setCurrentV1UpdateTime since the version was initially activated when it was current.
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING,
|
|
},
|
|
{
|
|
Version: tv2.DeploymentVersionString(),
|
|
CreateTime: v2CreateTime,
|
|
DrainageInfo: nil,
|
|
RoutingUpdateTime: setCurrentV2UpdateTime,
|
|
CurrentSinceTime: setCurrentV2UpdateTime,
|
|
RampingSinceTime: nil,
|
|
FirstActivationTime: setCurrentV2UpdateTime,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
},
|
|
},
|
|
LastModifierIdentity: tv2.ClientIdentity(),
|
|
},
|
|
})
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestDeleteWorkerDeployment_ValidDelete() {
|
|
s.OverrideDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
tv1 := testvars.New(s).WithBuildIDNumber(1)
|
|
|
|
// Start deployment workflow 1 and wait for the deployment version to exist
|
|
s.startVersionWorkflow(ctx, tv1)
|
|
|
|
// Signal the first version to be drained. Only do this in tests.
|
|
versionWorkflowID := worker_versioning.GenerateVersionWorkflowID(tv1.DeploymentSeries(), tv1.BuildID())
|
|
workflowExecution := &commonpb.WorkflowExecution{
|
|
WorkflowId: versionWorkflowID,
|
|
}
|
|
input := &deploymentpb.VersionDrainageInfo{
|
|
Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED,
|
|
LastChangedTime: timestamppb.New(time.Now()),
|
|
LastCheckedTime: timestamppb.New(time.Now()),
|
|
}
|
|
marshaledData, err := input.Marshal()
|
|
s.NoError(err)
|
|
signalPayload := &commonpb.Payloads{
|
|
Payloads: []*commonpb.Payload{
|
|
{
|
|
Metadata: map[string][]byte{
|
|
"encoding": []byte("binary/protobuf"),
|
|
},
|
|
Data: marshaledData,
|
|
},
|
|
},
|
|
}
|
|
|
|
err = s.SendSignal(s.Namespace().String(), workflowExecution, workerdeployment.SyncDrainageSignalName, signalPayload, tv1.ClientIdentity())
|
|
s.Nil(err)
|
|
|
|
// Wait for pollers going away
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err := s.FrontendClient().DescribeTaskQueue(ctx, &workflowservice.DescribeTaskQueueRequest{
|
|
Namespace: s.Namespace().String(),
|
|
TaskQueue: tv1.TaskQueue(),
|
|
TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW,
|
|
})
|
|
require.NoError(t, err)
|
|
require.Empty(t, resp.Pollers)
|
|
}, 5*time.Second, time.Second)
|
|
|
|
// delete succeeds
|
|
s.tryDeleteVersion(ctx, tv1, true)
|
|
|
|
// deployment version does not exist in the deployment list
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() {
|
|
//nolint:staticcheck // SA1019 deprecated Version will clean up later
|
|
a.NotEqual(tv1.DeploymentVersionString(), vs.Version)
|
|
}
|
|
}, time.Second*5, time.Millisecond*200)
|
|
|
|
// Deleting the worker deployment should succeed since there are no associated versions left
|
|
_, err = s.FrontendClient().DeleteWorkerDeployment(ctx, &workflowservice.DeleteWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
Identity: tv1.ClientIdentity(),
|
|
})
|
|
s.Nil(err)
|
|
|
|
// Describe Worker Deployment should give not found
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
_, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
})
|
|
a.Error(err)
|
|
var nfe *serviceerror.NotFound
|
|
a.True(errors.As(err, &nfe))
|
|
}, time.Second*5, time.Millisecond*200)
|
|
|
|
// ListDeployments should not show the closed/deleted Worker Deployment
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
listResp, err := s.FrontendClient().ListWorkerDeployments(ctx, &workflowservice.ListWorkerDeploymentsRequest{
|
|
Namespace: s.Namespace().String(),
|
|
})
|
|
a.Nil(err)
|
|
for _, dInfo := range listResp.GetWorkerDeployments() {
|
|
a.NotEqual(tv1.DeploymentSeries(), dInfo.GetName())
|
|
}
|
|
}, time.Second*5, time.Millisecond*200)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestDeleteWorkerDeployment_Idempotent() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
tv1 := testvars.New(s).WithBuildIDNumber(1)
|
|
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
_, err := s.FrontendClient().DeleteWorkerDeployment(ctx, &workflowservice.DeleteWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
Identity: tv1.ClientIdentity(),
|
|
})
|
|
a.NoError(err)
|
|
}, time.Second*5, time.Millisecond*200)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) TestDeleteWorkerDeployment_InvalidDelete() {
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
tv1 := testvars.New(s).WithBuildIDNumber(1)
|
|
|
|
// Start deployment workflow 1 and wait for the deployment version and deployment workflow to exist
|
|
go s.pollFromDeployment(ctx, tv1)
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeploymentVersion(ctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv1.DeploymentVersionString(),
|
|
})
|
|
a.NoError(err)
|
|
a.Equal(tv1.DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion())
|
|
}, time.Second*5, time.Millisecond*200)
|
|
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
resp, err := s.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
})
|
|
a.NoError(err)
|
|
a.NotEmpty(resp.GetWorkerDeploymentInfo().GetVersionSummaries())
|
|
//nolint:staticcheck // SA1019 deprecated Version will clean up later
|
|
a.Equal(tv1.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetVersionSummaries()[0].Version)
|
|
}, time.Second*5, time.Millisecond*200)
|
|
|
|
// Delete the worker deployment should fail since there are versions associated with it
|
|
_, err := s.FrontendClient().DeleteWorkerDeployment(ctx, &workflowservice.DeleteWorkerDeploymentRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv1.DeploymentSeries(),
|
|
Identity: tv1.ClientIdentity(),
|
|
})
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) tryDeleteVersion(
|
|
ctx context.Context,
|
|
tv *testvars.TestVars,
|
|
expectSuccess bool,
|
|
) {
|
|
_, err := s.FrontendClient().DeleteWorkerDeploymentVersion(ctx, &workflowservice.DeleteWorkerDeploymentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
Version: tv.DeploymentVersionString(),
|
|
})
|
|
if expectSuccess {
|
|
s.Nil(err)
|
|
} else {
|
|
s.Error(err)
|
|
}
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) verifyTimestampEquality(expected, actual *timestamppb.Timestamp, maxDuration time.Duration) {
|
|
s.True((expected == nil) == (actual == nil))
|
|
if expected != nil {
|
|
s.True(expected.AsTime().Sub(actual.AsTime()) < maxDuration)
|
|
}
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) verifyVersionSummary(expected, actual *deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary, maxDuration time.Duration) {
|
|
s.Equal(expected.GetVersion(), actual.GetVersion()) //nolint:staticcheck // SA1019: old worker versioning
|
|
s.Equal(expected.GetDrainageInfo().GetStatus(), actual.GetDrainageInfo().GetStatus())
|
|
s.Equal(expected.GetStatus(), actual.GetStatus())
|
|
|
|
s.verifyTimestampEquality(expected.GetCreateTime(), actual.GetCreateTime(), maxDuration)
|
|
s.verifyTimestampEquality(expected.GetRoutingUpdateTime(), actual.GetRoutingUpdateTime(), maxDuration)
|
|
s.verifyTimestampEquality(expected.GetCurrentSinceTime(), actual.GetCurrentSinceTime(), maxDuration)
|
|
s.verifyTimestampEquality(expected.GetRampingSinceTime(), actual.GetRampingSinceTime(), maxDuration)
|
|
s.verifyTimestampEquality(expected.GetFirstActivationTime(), actual.GetFirstActivationTime(), maxDuration)
|
|
s.verifyTimestampEquality(expected.GetLastDeactivationTime(), actual.GetLastDeactivationTime(), maxDuration)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) verifyRoutingConfig(expected, actual *deploymentpb.RoutingConfig, maxDuration time.Duration) {
|
|
s.Equal(expected.GetRampingVersion(), actual.GetRampingVersion()) //nolint:staticcheck // SA1019: old worker versioning
|
|
s.Equal(expected.GetRampingVersionPercentage(), actual.GetRampingVersionPercentage())
|
|
s.Equal(expected.GetCurrentVersion(), actual.GetCurrentVersion()) //nolint:staticcheck // SA1019: old worker versioning
|
|
|
|
s.verifyTimestampEquality(expected.GetRampingVersionChangedTime(), actual.GetRampingVersionChangedTime(), maxDuration)
|
|
s.verifyTimestampEquality(expected.GetCurrentVersionChangedTime(), actual.GetCurrentVersionChangedTime(), maxDuration)
|
|
s.verifyTimestampEquality(expected.GetRampingVersionPercentageChangedTime(), actual.GetRampingVersionPercentageChangedTime(), maxDuration)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) verifyWorkerDeploymentInfo(expected, actual *deploymentpb.WorkerDeploymentInfo, maxDuration time.Duration) {
|
|
s.Equal(expected.GetName(), actual.GetName())
|
|
s.verifyTimestampEquality(expected.GetCreateTime(), actual.GetCreateTime(), maxDuration)
|
|
s.Equal(expected.GetLastModifierIdentity(), actual.GetLastModifierIdentity())
|
|
s.verifyRoutingConfig(expected.GetRoutingConfig(), actual.GetRoutingConfig(), maxDuration)
|
|
|
|
// Verify version summaries
|
|
for _, expectedSummary := range expected.GetVersionSummaries() {
|
|
found := false
|
|
for _, actualSummary := range actual.GetVersionSummaries() {
|
|
if actualSummary.Version == expectedSummary.Version { //nolint:staticcheck // SA1019: old worker versioning
|
|
s.verifyVersionSummary(expectedSummary, actualSummary, maxDuration)
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
s.True(found)
|
|
}
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) verifyDescribeWorkerDeployment(
|
|
actualResp *workflowservice.DescribeWorkerDeploymentResponse,
|
|
expectedResp *workflowservice.DescribeWorkerDeploymentResponse,
|
|
) {
|
|
// relaxed timestamp constraint since the tests make sync calls, which could theoretically take seconds.
|
|
maxDurationBetweenTimeStamps := 5 * time.Second
|
|
|
|
s.True((actualResp == nil) == (expectedResp == nil))
|
|
s.True((actualResp.GetWorkerDeploymentInfo() == nil) == (expectedResp.GetWorkerDeploymentInfo() == nil))
|
|
s.True((actualResp.GetWorkerDeploymentInfo().GetRoutingConfig() == nil) == (expectedResp.GetWorkerDeploymentInfo().GetRoutingConfig() == nil))
|
|
|
|
s.verifyWorkerDeploymentInfo(expectedResp.GetWorkerDeploymentInfo(), actualResp.GetWorkerDeploymentInfo(), maxDurationBetweenTimeStamps)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) setAndVerifyRampingVersion(
|
|
ctx context.Context,
|
|
tv *testvars.TestVars,
|
|
unset bool,
|
|
percentage int,
|
|
ignoreMissingTaskQueues bool,
|
|
expectedError string,
|
|
expectedResp *workflowservice.SetWorkerDeploymentRampingVersionResponse,
|
|
) {
|
|
s.setAndVerifyRampingVersionUnversionedOption(ctx, tv, false, unset, percentage, ignoreMissingTaskQueues, expectedError, expectedResp)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) setAndVerifyRampingVersionUnversionedOption(
|
|
ctx context.Context,
|
|
tv *testvars.TestVars,
|
|
unversioned bool,
|
|
unset bool,
|
|
percentage int,
|
|
ignoreMissingTaskQueues bool,
|
|
expectedError string,
|
|
expectedResp *workflowservice.SetWorkerDeploymentRampingVersionResponse,
|
|
) {
|
|
version := tv.DeploymentVersionString()
|
|
if unversioned {
|
|
version = worker_versioning.UnversionedVersionId
|
|
}
|
|
if unset {
|
|
version = ""
|
|
percentage = 0
|
|
}
|
|
if !unversioned && !unset {
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
} else {
|
|
s.ensureCreateDeployment(tv)
|
|
}
|
|
resp, err := s.FrontendClient().SetWorkerDeploymentRampingVersion(ctx, &workflowservice.SetWorkerDeploymentRampingVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: version,
|
|
Percentage: float32(percentage),
|
|
Identity: tv.ClientIdentity(),
|
|
IgnoreMissingTaskQueues: ignoreMissingTaskQueues,
|
|
})
|
|
if expectedError != "" {
|
|
s.Error(err)
|
|
s.Contains(err.Error(), expectedError)
|
|
return
|
|
}
|
|
s.NoError(err)
|
|
s.Equal(expectedResp.GetPreviousVersion(), resp.GetPreviousVersion())
|
|
s.Equal(expectedResp.GetPreviousPercentage(), resp.GetPreviousPercentage())
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) setCurrentVersion(ctx context.Context, tv *testvars.TestVars, previousCurrent string, ignoreMissingTaskQueues bool, expectedError string) {
|
|
s.setCurrentVersionUnversionedOption(ctx, tv, false, previousCurrent, ignoreMissingTaskQueues, expectedError)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) setCurrentVersionUnversionedOption(ctx context.Context, tv *testvars.TestVars, unversioned bool, previousCurrent string, ignoreMissingTaskQueues bool, expectedError string) {
|
|
version := tv.DeploymentVersionString()
|
|
if unversioned {
|
|
version = worker_versioning.UnversionedVersionId
|
|
s.ensureCreateDeployment(tv)
|
|
} else {
|
|
s.ensureCreateVersionInDeployment(tv)
|
|
}
|
|
|
|
resp, err := s.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, &workflowservice.SetWorkerDeploymentCurrentVersionRequest{
|
|
Namespace: s.Namespace().String(),
|
|
DeploymentName: tv.DeploymentVersion().GetDeploymentName(),
|
|
Version: version,
|
|
IgnoreMissingTaskQueues: ignoreMissingTaskQueues,
|
|
Identity: tv.ClientIdentity(),
|
|
})
|
|
if expectedError != "" {
|
|
s.Error(err)
|
|
s.Contains(err.Error(), expectedError)
|
|
return
|
|
}
|
|
s.NoError(err)
|
|
s.NotNil(resp.PreviousVersion)
|
|
s.Equal(previousCurrent, resp.PreviousVersion)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) createVersionsInDeployments(ctx context.Context, tv *testvars.TestVars, n int) []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary {
|
|
var expectedDeploymentSummaries []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary
|
|
|
|
for i := 0; i < n; i++ {
|
|
deployment := tv.WithDeploymentSeriesNumber(i)
|
|
version := deployment.WithBuildIDNumber(i)
|
|
|
|
s.startVersionWorkflow(ctx, version)
|
|
s.setCurrentVersion(ctx, version, worker_versioning.UnversionedVersionId, true, "")
|
|
|
|
currentVersionSummary := &deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary{
|
|
Version: version.DeploymentVersionString(),
|
|
CreateTime: timestamppb.Now(),
|
|
CurrentSinceTime: timestamppb.Now(),
|
|
FirstActivationTime: timestamppb.Now(),
|
|
RoutingUpdateTime: timestamppb.Now(),
|
|
DrainageInfo: nil,
|
|
RampingSinceTime: nil,
|
|
LastDeactivationTime: nil,
|
|
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
|
|
}
|
|
|
|
expectedDeployment := s.buildWorkerDeploymentSummary(
|
|
deployment.DeploymentSeries(),
|
|
timestamppb.Now(),
|
|
&deploymentpb.RoutingConfig{
|
|
CurrentVersion: version.DeploymentVersionString(),
|
|
CurrentVersionChangedTime: timestamppb.Now(),
|
|
},
|
|
currentVersionSummary, // latest version added is the current version
|
|
currentVersionSummary,
|
|
nil,
|
|
)
|
|
expectedDeploymentSummaries = append(expectedDeploymentSummaries, expectedDeployment)
|
|
}
|
|
|
|
return expectedDeploymentSummaries
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) verifyWorkerDeploymentSummary(
|
|
expectedSummary *workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary,
|
|
actualSummary *workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary,
|
|
) bool {
|
|
maxDurationBetweenTimeStamps := 5 * time.Second
|
|
if expectedSummary.Name != actualSummary.Name {
|
|
s.Logger.Info("Name mismatch")
|
|
return false
|
|
}
|
|
if expectedSummary.CreateTime.AsTime().Sub(actualSummary.CreateTime.AsTime()) > maxDurationBetweenTimeStamps {
|
|
s.Logger.Info("Create time mismatch")
|
|
return false
|
|
}
|
|
|
|
// Current version checks
|
|
if expectedSummary.RoutingConfig.GetCurrentVersion() != actualSummary.RoutingConfig.GetCurrentVersion() { //nolint:staticcheck // SA1019: old worker versioning
|
|
s.Logger.Info("Current version mismatch")
|
|
return false
|
|
}
|
|
if expectedSummary.RoutingConfig.GetCurrentVersionChangedTime().AsTime().Sub(actualSummary.RoutingConfig.GetCurrentVersionChangedTime().AsTime()) > maxDurationBetweenTimeStamps {
|
|
s.Logger.Info("Current version update time mismatch")
|
|
return false
|
|
}
|
|
|
|
// Ramping version checks
|
|
if expectedSummary.RoutingConfig.GetRampingVersion() != actualSummary.RoutingConfig.GetRampingVersion() { //nolint:staticcheck // SA1019: old worker versioning
|
|
s.Logger.Info("Ramping version mismatch")
|
|
return false
|
|
}
|
|
if expectedSummary.RoutingConfig.GetRampingVersionPercentage() != actualSummary.RoutingConfig.GetRampingVersionPercentage() {
|
|
s.Logger.Info("Ramping version percentage mismatch")
|
|
return false
|
|
}
|
|
if expectedSummary.RoutingConfig.GetRampingVersionChangedTime().AsTime().Sub(actualSummary.RoutingConfig.GetRampingVersionChangedTime().AsTime()) > maxDurationBetweenTimeStamps {
|
|
s.Logger.Info("Ramping version update time mismatch")
|
|
return false
|
|
}
|
|
|
|
// Latest version summary checks
|
|
s.verifyVersionSummary(expectedSummary.LatestVersionSummary, actualSummary.LatestVersionSummary, maxDurationBetweenTimeStamps)
|
|
|
|
// Current version summary checks
|
|
s.verifyVersionSummary(expectedSummary.CurrentVersionSummary, actualSummary.CurrentVersionSummary, maxDurationBetweenTimeStamps)
|
|
|
|
// Ramping version summary checks
|
|
s.verifyVersionSummary(expectedSummary.RampingVersionSummary, actualSummary.RampingVersionSummary, maxDurationBetweenTimeStamps)
|
|
|
|
return true
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) listWorkerDeployments(ctx context.Context, request *workflowservice.ListWorkerDeploymentsRequest) ([]*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary, error) {
|
|
var resp *workflowservice.ListWorkerDeploymentsResponse
|
|
var err error
|
|
var deploymentSummaries []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary
|
|
for resp == nil || len(resp.NextPageToken) > 0 {
|
|
resp, err = s.FrontendClient().ListWorkerDeployments(ctx, request)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
deploymentSummaries = append(deploymentSummaries, resp.GetWorkerDeployments()...)
|
|
request.NextPageToken = resp.NextPageToken
|
|
}
|
|
return deploymentSummaries, nil
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) startAndValidateWorkerDeployments(
|
|
ctx context.Context,
|
|
request *workflowservice.ListWorkerDeploymentsRequest,
|
|
expectedDeploymentSummaries []*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary,
|
|
) {
|
|
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
a := require.New(t)
|
|
|
|
actualDeploymentSummaries, err := s.listWorkerDeployments(ctx, request)
|
|
a.NoError(err)
|
|
if len(actualDeploymentSummaries) < len(expectedDeploymentSummaries) {
|
|
return
|
|
}
|
|
|
|
for _, expectedDeploymentSummary := range expectedDeploymentSummaries {
|
|
deploymentSummaryValidated := false
|
|
for _, actualDeploymentSummary := range actualDeploymentSummaries {
|
|
deploymentSummaryValidated = deploymentSummaryValidated ||
|
|
s.verifyWorkerDeploymentSummary(expectedDeploymentSummary, actualDeploymentSummary)
|
|
}
|
|
a.True(deploymentSummaryValidated)
|
|
}
|
|
}, time.Second*10, time.Millisecond*1000)
|
|
}
|
|
|
|
func (s *WorkerDeploymentSuite) buildWorkerDeploymentSummary(
|
|
deploymentName string, createTime *timestamppb.Timestamp,
|
|
routingConfig *deploymentpb.RoutingConfig,
|
|
latestVersionSummary *deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary,
|
|
currentVersionSummary *deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary,
|
|
rampingVersionSummary *deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary,
|
|
) *workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary {
|
|
return &workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary{
|
|
Name: deploymentName,
|
|
CreateTime: createTime,
|
|
RoutingConfig: routingConfig,
|
|
LatestVersionSummary: latestVersionSummary,
|
|
RampingVersionSummary: rampingVersionSummary,
|
|
CurrentVersionSummary: currentVersionSummary,
|
|
}
|
|
}
|
|
|
|
// Name is used by testvars. We use a shortened test name in variables so that physical task queue IDs
|
|
// do not grow larger than DB column limit (currently as low as 272 chars).
|
|
func (s *WorkerDeploymentSuite) Name() string {
|
|
fullName := s.T().Name()
|
|
if len(fullName) <= 30 {
|
|
return fullName
|
|
}
|
|
short := fmt.Sprintf("%s-%08x",
|
|
fullName[len(fullName)-21:],
|
|
farm.Fingerprint32([]byte(fullName)),
|
|
)
|
|
return strings.Replace(short, ".", "|", -1)
|
|
}
|