Versioning Metrics pt2: DeploymentTransition + VersioningOverride (#7825)

## What changed?
- metrics for DeploymentTransition + VersioningOverride

## Why?
- versioning-0.32

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
- a little new to this side of the code involving versioning so a bit
scared about doing things wrong, hopefully I am not

Author: Shivs11

common/metrics/metric_defs.go

@@ -1180,6 +1180,10 @@ var (
 		WithDescription("The number of schedule actions that failed to start"),
 	)
 
+	// Worker Versioning
+	WorkerDeploymentVersioningOverrideCounter = NewCounterDef("worker_deployment_versioning_override_count")
+	StartDeploymentTransitionCounter          = NewCounterDef("start_deployment_transition_count")
+
 	// Force replication
 	EncounterZombieWorkflowCount        = NewCounterDef("encounter_zombie_workflow_count")
 	EncounterNotFoundWorkflowCount      = NewCounterDef("encounter_not_found_workflow_count")

common/metrics/tags.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/common/locks"
 	"go.temporal.io/server/common/primitives"
@@ -40,13 +41,30 @@ const (
 	// See server.api.enums.v1.ReplicationTaskType
 	replicationTaskType     = "replicationTaskType"
 	replicationTaskPriority = "replicationTaskPriority"
-
-	namespaceAllValue = "all"
-	unknownValue      = "_unknown_"
-	totalMetricSuffix = "_total"
-	tagExcludedValue  = "_tag_excluded_"
-
-	errorPrefix = "*"
+	versioningBehavior      = "versioning_behavior"
+	isFirstAttempt          = "first-attempt"
+	workflowStatus          = "workflow_status"
+	behaviorBefore          = "behavior_before"
+	behaviorAfter           = "behavior_after"
+	runInitiator            = "run_initiator"
+	fromUnversioned         = "from_unversioned"
+	toUnversioned           = "to_unversioned"
+	queryType               = "query_type"
+	namespaceAllValue       = "all"
+	unknownValue            = "_unknown_"
+	totalMetricSuffix       = "_total"
+	tagExcludedValue        = "_tag_excluded_"
+	falseValue              = "false"
+	trueValue               = "true"
+	errorPrefix             = "*"
+
+	newRun      = "new"
+	existingRun = "existing"
+	childRun    = "child"
+	canRun      = "can"
+	retryRun    = "retry"
+	cronRun     = "cron"
+	unknownRun  = "unknown"
 )
 
 // Tag is an interface to define metrics tags
@@ -383,3 +401,50 @@ func DestinationTag(value string) Tag {
 		value: value,
 	}
 }
+
+func VersioningBehaviorBeforeOverrideTag(behavior enumspb.VersioningBehavior) Tag {
+	return &tagImpl{key: behaviorBefore, value: behavior.String()}
+}
+
+func VersioningBehaviorAfterOverrideTag(behavior enumspb.VersioningBehavior) Tag {
+	return &tagImpl{key: behaviorAfter, value: behavior.String()}
+}
+
+// RunInitiatorTag creates a tag indicating how a workflow run was initiated.
+// It handles both new workflow runs and continuations from previous runs.
+// When attributes is nil (e.g. during AddWorkflowExecutionOptionsUpdatedEvent),
+// it returns a tag indicating an existing run.
+func RunInitiatorTag(prevRunID string, attributes *historypb.WorkflowExecutionStartedEventAttributes) Tag {
+	if attributes == nil {
+		return &tagImpl{key: runInitiator, value: existingRun}
+	} else if attributes.GetParentWorkflowExecution() != nil {
+		return &tagImpl{key: runInitiator, value: childRun}
+	}
+
+	switch attributes.GetInitiator() {
+	case enumspb.CONTINUE_AS_NEW_INITIATOR_UNSPECIFIED:
+		return &tagImpl{key: runInitiator, value: newRun}
+	case enumspb.CONTINUE_AS_NEW_INITIATOR_WORKFLOW:
+		return &tagImpl{key: runInitiator, value: canRun}
+	case enumspb.CONTINUE_AS_NEW_INITIATOR_RETRY:
+		return &tagImpl{key: runInitiator, value: retryRun}
+	case enumspb.CONTINUE_AS_NEW_INITIATOR_CRON_SCHEDULE:
+		return &tagImpl{key: runInitiator, value: cronRun}
+	default:
+		return &tagImpl{key: runInitiator, value: unknownRun}
+	}
+}
+
+func FromUnversionedTag(version string) Tag {
+	if version == "_unversioned_" {
+		return &tagImpl{key: fromUnversioned, value: trueValue}
+	}
+	return &tagImpl{key: fromUnversioned, value: falseValue}
+}
+
+func ToUnversionedTag(version string) Tag {
+	if version == "_unversioned_" {
+		return &tagImpl{key: toUnversioned, value: trueValue}
+	}
+	return &tagImpl{key: toUnversioned, value: falseValue}
+}

common/worker_versioning/worker_versioning.go

@@ -375,6 +375,17 @@ func OverrideIsPinned(override *workflowpb.VersioningOverride) bool {
 		override.GetPinned().GetBehavior() == workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED
 }
 
+func ExtractVersioningBehaviorFromOverride(override *workflowpb.VersioningOverride) enumspb.VersioningBehavior {
+	if override.GetAutoUpgrade() {
+		return enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE
+	} else if override.GetPinned() != nil {
+		return enumspb.VERSIONING_BEHAVIOR_PINNED
+	}
+
+	//nolint:staticcheck // SA1019: worker versioning v0.31
+	return override.GetBehavior()
+}
+
 func ValidateVersioningOverride(override *workflowpb.VersioningOverride) error {
 	if override == nil {
 		return nil

service/history/workflow/mutable_state_impl.go

@@ -2463,6 +2463,18 @@ func (ms *MutableStateImpl) AddWorkflowExecutionStartedEventWithOptions(
 	); err != nil {
 		return nil, err
 	}
+
+	// Versioning Override set on StartWorkflowExecutionRequest
+	if startRequest.GetStartRequest().GetVersioningOverride() != nil {
+		metrics.WorkerDeploymentVersioningOverrideCounter.With(
+			ms.metricsHandler.WithTags(
+				metrics.NamespaceTag(ms.namespaceEntry.Name().String()),
+				metrics.VersioningBehaviorBeforeOverrideTag(enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED),
+				metrics.VersioningBehaviorAfterOverrideTag(worker_versioning.ExtractVersioningBehaviorFromOverride(startRequest.GetStartRequest().GetVersioningOverride())),
+				metrics.RunInitiatorTag(prevRunID, event.GetWorkflowExecutionStartedEventAttributes()),
+			),
+		).Record(1)
+	}
 	return event, nil
 }
 
@@ -4804,9 +4816,25 @@ func (ms *MutableStateImpl) AddWorkflowExecutionOptionsUpdatedEvent(
 		attachCompletionCallbacks,
 		links,
 	)
+	prevEffectiveVersioningBehavior := ms.GetEffectiveVersioningBehavior()
+	prevEffectiveDeployment := ms.GetEffectiveDeployment()
+
 	if err := ms.ApplyWorkflowExecutionOptionsUpdatedEvent(event); err != nil {
 		return nil, err
 	}
+
+	if !proto.Equal(ms.GetEffectiveDeployment(), prevEffectiveDeployment) ||
+		ms.GetEffectiveVersioningBehavior() != prevEffectiveVersioningBehavior {
+		metrics.WorkerDeploymentVersioningOverrideCounter.With(
+			ms.metricsHandler.WithTags(
+				metrics.NamespaceTag(ms.namespaceEntry.Name().String()),
+				metrics.VersioningBehaviorBeforeOverrideTag(prevEffectiveVersioningBehavior),
+				metrics.VersioningBehaviorAfterOverrideTag(ms.GetEffectiveVersioningBehavior()),
+				metrics.RunInitiatorTag("", event.GetWorkflowExecutionStartedEventAttributes()),
+			),
+		).Record(1)
+	}
+
 	return event, nil
 }
 
@@ -4833,6 +4861,7 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionOptionsUpdatedEvent(event *his
 			return err
 		}
 	}
+
 	return nil
 }
 
@@ -8087,7 +8116,8 @@ func (ms *MutableStateImpl) StartDeploymentTransition(deployment *deploymentpb.D
 		ms.GetExecutionInfo().VersioningInfo = versioningInfo
 	}
 
-	if ms.GetEffectiveDeployment().Equal(deployment) {
+	preTransitionEffectiveDeployment := ms.GetEffectiveDeployment()
+	if preTransitionEffectiveDeployment.Equal(deployment) {
 		return serviceerror.NewInternal("start transition should receive a version different from effective version")
 	}
 
@@ -8125,6 +8155,16 @@ func (ms *MutableStateImpl) StartDeploymentTransition(deployment *deploymentpb.D
 			ms.logInfo("start transition did not reschedule pending speculative task")
 		}
 	}
+
+	// DeploymentTransition has taken place, so we increment the DeploymentTransition metric
+	metrics.StartDeploymentTransitionCounter.With(
+		ms.metricsHandler.WithTags(
+			metrics.NamespaceTag(ms.namespaceEntry.Name().String()),
+			metrics.FromUnversionedTag(worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(preTransitionEffectiveDeployment))),
+			metrics.ToUnversionedTag(worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(deployment))),
+		),
+	).Record(1)
+
 	return nil
 }