Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions service/worker/workerdeployment/version_activities.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"go.temporal.io/sdk/temporal"
deploymentspb "go.temporal.io/server/api/deployment/v1"
"go.temporal.io/server/api/matchingservice/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/worker_versioning"
Expand Down Expand Up @@ -199,6 +200,19 @@ func (a *VersionActivities) GetVersionDrainageStatus(ctx context.Context, versio
}, nil
}

func (a *VersionActivities) DescribeWorkerControllerInstanceStatus(ctx context.Context, version *deploymentspb.WorkerDeploymentVersion) (*deploymentpb.ComputeStatus, error) {
resp, _, err := a.WorkerControllerInstanceClient.DescribeWorkerControllerInstance(ctx, a.namespace, &deploymentpb.WorkerDeploymentVersion{DeploymentName: version.GetDeploymentName(), BuildId: version.GetBuildId()})
if err != nil {
if common.IsNotFoundError(err) {
return nil, nil
}

return nil, err
}

return wciValidationStatusToComputeStatus(resp.ValidationStatus), nil
}

func (a *VersionActivities) UpdateWorkerControllerInstance(ctx context.Context, input *deploymentspb.UpdateWorkerControllerInstanceInput) (*computepb.ComputeConfigSummary, error) {
upserts := scalingGroupUpdatesToWCI(input.GetUpsertScalingGroups())
resp, err := a.WorkerControllerInstanceClient.UpdateWorkerControllerInstance(ctx, a.namespace, input.GetVersion(), nil, input.GetIdentity(), upserts, input.GetRemoveScalingGroups())
Expand Down
30 changes: 30 additions & 0 deletions service/worker/workerdeployment/version_workflow.go
Original file line number Diff line number Diff line change
Expand Up @@ -345,6 +345,12 @@ func (d *VersionWorkflowRunner) run(ctx workflow.Context) error {
}
}

// When creating a compute provider and version together, there is a race condition between the two coming up. Making sure to have pulled
// the latest state from the compute provider if this happens to be the slower one.
if err := d.syncVersionDataToComputeStatus(ctx); err != nil {
Comment thread
02strich marked this conversation as resolved.
return err
}

// Listen to signals in a different goroutine to make business logic clearer
workflow.Go(ctx, d.listenToSignals)
Comment thread
02strich marked this conversation as resolved.

Expand Down Expand Up @@ -1244,6 +1250,30 @@ func (d *VersionWorkflowRunner) syncVersionStatusAfterDrainageStatusChange(ctx w
return d.syncVersionDataToTaskQueues(ctx, versionData)
}

// syncVersionDataToComputeStatus is a helper that syncs the compute status from WCI to the worker deployment version
func (d *VersionWorkflowRunner) syncVersionDataToComputeStatus(ctx workflow.Context) error {
if workflow.GetVersion(ctx, "sync-compute-validation-status", workflow.DefaultVersion, 0) == workflow.DefaultVersion {
return nil
}

state := d.GetVersionState()

if state.ComputeStatus == nil && state.ComputeConfig != nil {
workflow.Go(ctx, func(ctx workflow.Context) {
logger := workflow.GetLogger(ctx)

var result deploymentpb.ComputeStatus
resp := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, defaultActivityOptions), d.a.DescribeWorkerControllerInstanceStatus, state.GetVersion())
if err := resp.Get(ctx, &result); err != nil {
logger.Error("failed to sync compute status", "error", err)
} else if result.ProviderValidation != nil {
state.ComputeStatus = &result
}
})
}
return nil
}

// syncVersionDataToTaskQueues is a helper that syncs the provided version data to all task queues.
// This function does NOT acquire the workflow lock - the caller is responsible for that.
func (d *VersionWorkflowRunner) syncVersionDataToTaskQueues(ctx workflow.Context, versionData *deploymentspb.DeploymentVersionData) error {
Expand Down
5 changes: 5 additions & 0 deletions service/worker/workerdeployment/version_workflow_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,11 @@ func (s *VersionWorkflowSuite) SetupTest() {
}
s.env.RegisterWorkflowWithOptions(versionWorkflow, workflow.RegisterOptions{Name: WorkerDeploymentVersionWorkflowType})

var a *VersionActivities
s.env.OnActivity(a.DescribeWorkerControllerInstanceStatus, mock.Anything, mock.Anything).
Return(&deploymentpb.ComputeStatus{}, nil).
Maybe()

// Initialize an empty ClientImpl to use its helper methods
s.workerDeploymentClient = &ClientImpl{}
}
Expand Down
Loading