comments

Author: Shivs11

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"
@@ -38,24 +39,31 @@ const (
 	// Generic reason tag can be used anywhere a reason is needed.
 	reason = "reason"
 	// See server.api.enums.v1.ReplicationTaskType
-	replicationTaskType                 = "replicationTaskType"
-	replicationTaskPriority             = "replicationTaskPriority"
-	versioningBehavior                  = "versioning_behavior"
-	isFirstAttempt                      = "first-attempt"
-	workflowStatus                      = "workflow_status"
-	workflowBehaviorBeforeOverride      = "workflow_behavior_before_override"
-	workflowBehaviorAfterOverride       = "workflow_behavior_after_override"
-	versioningOverrideOnNewWorkflow     = "versioning_override_on_new_workflow"
-	effectiveDeploymentBeforeTransition = "effective_deployment_before_transition"
-	effectiveDeploymentAfterTransition  = "effective_deployment_after_transition"
-	queryType                           = "query_type"
-	namespaceAllValue                   = "all"
-	unknownValue                        = "_unknown_"
-	totalMetricSuffix                   = "_total"
-	tagExcludedValue                    = "_tag_excluded_"
-	falseValue                          = "false"
-	trueValue                           = "true"
-	errorPrefix                         = "*"
+	replicationTaskType     = "replicationTaskType"
+	replicationTaskPriority = "replicationTaskPriority"
+	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"
 )
 
 // Tag is an interface to define metrics tags
@@ -394,21 +402,42 @@ func DestinationTag(value string) Tag {
 }
 
 func VersioningBehaviorBeforeOverrideTag(behavior enumspb.VersioningBehavior) Tag {
-	return &tagImpl{key: workflowBehaviorBeforeOverride, value: behavior.String()}
+	return &tagImpl{key: behaviorBefore, value: behavior.String()}
 }
 
 func VersioningBehaviorAfterOverrideTag(behavior enumspb.VersioningBehavior) Tag {
-	return &tagImpl{key: workflowBehaviorAfterOverride, value: behavior.String()}
+	return &tagImpl{key: behaviorAfter, value: behavior.String()}
 }
 
-func VersioningOverrideOnNewWorkflowTag(isNewWorkflow bool) Tag {
-	return &tagImpl{key: versioningOverrideOnNewWorkflow, value: strconv.FormatBool(isNewWorkflow)}
+func RunInitiatorTag(prevRunID string, attributes *historypb.WorkflowExecutionStartedEventAttributes) Tag {
+	// UpdateWorkflowExecutionOptions API is the caller
+	if attributes == nil {
+		return &tagImpl{key: runInitiator, value: existingRun}
+	}
+
+	// StartWorkflowExecution API is the caller
+	if prevRunID == "" {
+		return &tagImpl{key: runInitiator, value: newRun}
+	} else if attributes.GetInitiator() == enumspb.CONTINUE_AS_NEW_INITIATOR_WORKFLOW {
+		return &tagImpl{key: runInitiator, value: childRun}
+	} else if attributes.GetInitiator() == enumspb.CONTINUE_AS_NEW_INITIATOR_RETRY {
+		return &tagImpl{key: runInitiator, value: retryRun}
+	} else if attributes.GetInitiator() == enumspb.CONTINUE_AS_NEW_INITIATOR_CRON_SCHEDULE {
+		return &tagImpl{key: runInitiator, value: cronRun}
+	}
+	return &tagImpl{key: runInitiator, value: existingRun}
 }
 
-func EffectiveDeploymentBeforeTransitionTag(version string) Tag {
-	return &tagImpl{key: effectiveDeploymentBeforeTransition, value: version}
+func FromUnversionedTag(version string) Tag {
+	if version == "_unversioned_" {
+		return &tagImpl{key: fromUnversioned, value: trueValue}
+	}
+	return &tagImpl{key: fromUnversioned, value: falseValue}
 }
 
-func EffectiveDeploymentAfterTransitionTag(version string) Tag {
-	return &tagImpl{key: effectiveDeploymentAfterTransition, value: version}
+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

@@ -2470,8 +2470,8 @@ func (ms *MutableStateImpl) AddWorkflowExecutionStartedEventWithOptions(
 			ms.metricsHandler.WithTags(
 				metrics.NamespaceTag(ms.namespaceEntry.Name().String()),
 				metrics.VersioningBehaviorBeforeOverrideTag(enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED),
-				metrics.VersioningBehaviorAfterOverrideTag(startRequest.GetStartRequest().GetVersioningOverride().GetBehavior()), //nolint:staticcheck // SA1019: worker versioning v0.31
-				metrics.VersioningOverrideOnNewWorkflowTag(prevRunID == ""),
+				metrics.VersioningBehaviorAfterOverrideTag(worker_versioning.ExtractVersioningBehaviorFromOverride(startRequest.GetStartRequest().GetVersioningOverride())),
+				metrics.RunInitiatorTag(prevRunID, event.GetWorkflowExecutionStartedEventAttributes()),
 			),
 		).Record(1)
 	}
@@ -4817,18 +4817,20 @@ func (ms *MutableStateImpl) AddWorkflowExecutionOptionsUpdatedEvent(
 		links,
 	)
 	prevEffectiveVersioningBehavior := ms.GetEffectiveVersioningBehavior()
+	prevEffectiveDeployment := ms.GetEffectiveDeployment()
+
 	if err := ms.ApplyWorkflowExecutionOptionsUpdatedEvent(event); err != nil {
 		return nil, err
 	}
 
-	// Versioning Override set via UpdateWorkflowExecutionOptionsRequest
-	if versioningOverride != nil {
+	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.VersioningOverrideOnNewWorkflowTag(false),
+				metrics.RunInitiatorTag("", event.GetWorkflowExecutionStartedEventAttributes()),
 			),
 		).Record(1)
 	}
@@ -8158,8 +8160,8 @@ func (ms *MutableStateImpl) StartDeploymentTransition(deployment *deploymentpb.D
 	metrics.StartDeploymentTransitionCounter.With(
 		ms.metricsHandler.WithTags(
 			metrics.NamespaceTag(ms.namespaceEntry.Name().String()),
-			metrics.EffectiveDeploymentBeforeTransitionTag(worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(preTransitionEffectiveDeployment))),
-			metrics.EffectiveDeploymentAfterTransitionTag(worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(deployment))),
+			metrics.FromUnversionedTag(worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(preTransitionEffectiveDeployment))),
+			metrics.ToUnversionedTag(worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(deployment))),
 		),
 	).Record(1)