address comments and lints

Author: ShahabT

service/worker/workerdeployment/version_activities.go

@@ -14,10 +14,10 @@ import (
 	taskqueuepb "go.temporal.io/api/taskqueue/v1"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/activity"
+	"go.temporal.io/sdk/log"
 	"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/log/tag"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/worker_versioning"
@@ -53,16 +53,7 @@ func (a *VersionActivities) SyncDeploymentVersionUserData(
 	input *deploymentspb.SyncDeploymentVersionUserDataRequest,
 ) (*deploymentspb.SyncDeploymentVersionUserDataResponse, error) {
 	logger := activity.GetLogger(ctx)
-	scheduledTime := activity.GetInfo(ctx).ScheduledTime
-
-	if scheduledTime.Add(SlowPropagationDelay).Before(time.Now()) {
-		a.Logger.Warn("Slow propagation detected, attempting to sync user data",
-			tag.WorkflowNamespace(a.namespace.Name().String()),
-			tag.Deployment(input.DeploymentName),
-			tag.BuildId(input.GetVersion().GetBuildId()),
-		)
-		a.MetricsHandler.Counter(metrics.SlowVersioningDataPropagationCounter.Name()).Record(1)
-	}
+	defer a.checkSlowPropagation(ctx, logger)
 
 	errs := make(chan error)
 
@@ -127,17 +118,17 @@ func (a *VersionActivities) SyncDeploymentVersionUserData(
 	return &deploymentspb.SyncDeploymentVersionUserDataResponse{TaskQueueMaxVersions: maxVersionByName}, nil
 }
 
-func (a *VersionActivities) CheckWorkerDeploymentUserDataPropagation(ctx context.Context, input *deploymentspb.CheckWorkerDeploymentUserDataPropagationRequest) error {
-	scheduledTime := activity.GetInfo(ctx).ScheduledTime
-
-	if scheduledTime.Add(SlowPropagationDelay).Before(time.Now()) {
-		a.Logger.Warn("Slow propagation detected, awaiting task queue partition propagation",
-			tag.WorkflowNamespace(a.namespace.Name().String()),
-		)
+func (a *VersionActivities) checkSlowPropagation(ctx context.Context, logger log.Logger) {
+	firstAttemptScheduledTime := activity.GetInfo(ctx).ScheduledTime
+	if firstAttemptScheduledTime.Add(SlowPropagationDelay).Before(time.Now()) {
+		logger.Warn("Slow propagation detected", "duration", time.Since(firstAttemptScheduledTime))
 		a.MetricsHandler.Counter(metrics.SlowVersioningDataPropagationCounter.Name()).Record(1)
 	}
+}
 
+func (a *VersionActivities) CheckWorkerDeploymentUserDataPropagation(ctx context.Context, input *deploymentspb.CheckWorkerDeploymentUserDataPropagationRequest) error {
 	logger := activity.GetLogger(ctx)
+	defer a.checkSlowPropagation(ctx, logger)
 
 	errs := make(chan error)
 

service/worker/workerdeployment/version_workflow.go

@@ -377,7 +377,7 @@ func (d *VersionWorkflowRunner) deleteVersionFromTaskQueuesAsync(ctx workflow.Co
 	// Not counting the possible wait for previous propagations in this propagation latency.
 	startTime := workflow.Now(ctx)
 	defer func() {
-		d.metrics.Timer(metrics.VersioningDataPropagationLatency.Name()).Record(time.Since(startTime))
+		d.metrics.Timer(metrics.VersioningDataPropagationLatency.Name()).Record(workflow.Now(ctx).Sub(startTime))
 	}()
 	d.deleteVersionFromTaskQueues(ctx, workflow.WithActivityOptions(ctx, propagationActivityOptions))
 	d.asyncPropagationsInProgress--
@@ -965,7 +965,7 @@ func (d *VersionWorkflowRunner) syncTaskQueuesAsync(
 ) error {
 	startTime := workflow.Now(ctx)
 	defer func() {
-		d.metrics.Timer(metrics.VersioningDataPropagationLatency.Name()).Record(time.Since(startTime))
+		d.metrics.Timer(metrics.VersioningDataPropagationLatency.Name()).Record(workflow.Now(ctx).Sub(startTime))
 	}()
 
 	state := d.GetVersionState()
@@ -1056,8 +1056,9 @@ func (d *VersionWorkflowRunner) executeAndTrackAsyncPropagation(
 
 	if routingConfig != nil {
 		if workflow.GetVersion(ctx, "no-propagation-sync-summary", workflow.DefaultVersion, 0) == workflow.DefaultVersion {
-			// TODO: clean this unnecessary sync up.
-			// No summary changes happen in async propagation that the deployment workflow
+			// TODO: clean this unnecessary sync.
+			// No summary changes need to happen after async propagation because the deployment
+			// workflow has already got the latest summary from update response.
 			d.syncSummary(ctx)
 		}
 		// Signal deployment workflow that routing config propagation completed