Versioning Metrics: pt1 (#7822)

## What changed?
- adding helpful metrics for queries so that we can understand customer
impact

Metrics added in this PR are:

(Directly related to entity workflows)
- WorkerDeploymentCreated
- WorkerDeploymentVersionCreated
- WorkerDeploymentVersionCreatedManagedByController
- WorkerDeploymentVersionVisibilityQueryCount

(Not directly related to entity workflows)
- WorkflowResetCount
- WorkflowQuerySuccessCount
- WorkflowQueryTimeoutCount
- WorkflowQueryFailureCount
- WorkflowTasksCompleted


## Why?
- vis

## 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
- None

---------

Co-authored-by: ShahabT <[email protected]>

Author: Shivs11

common/metrics/metric_defs.go

@@ -321,6 +321,7 @@ const (
 	HistoryRecordActivityTaskHeartbeatScope = "RecordActivityTaskHeartbeat"
 	// HistoryRespondWorkflowTaskCompletedScope tracks RespondWorkflowTaskCompleted API calls received by service
 	HistoryRespondWorkflowTaskCompletedScope = "RespondWorkflowTaskCompleted"
+	HistoryRespondWorkflowTaskFailedScope    = "RespondWorkflowTaskFailed"
 	// HistoryRespondActivityTaskCompletedScope tracks RespondActivityTaskCompleted API calls received by service
 	HistoryRespondActivityTaskCompletedScope = "RespondActivityTaskCompleted"
 	// HistoryRespondActivityTaskFailedScope tracks RespondActivityTaskFailed API calls received by service
@@ -357,6 +358,7 @@ const (
 	HistoryReapplyEventsScope = "ReapplyEvents"
 	// HistoryQueryWorkflowScope tracks QueryWorkflow API calls received by service
 	HistoryQueryWorkflowScope = "QueryWorkflow"
+	HistoryResetWorkflowScope = "HistoryResetWorkflow"
 	// HistoryProcessDeleteHistoryEventScope tracks ProcessDeleteHistoryEvent processing calls
 	HistoryProcessDeleteHistoryEventScope = "ProcessDeleteHistoryEvent"
 	// HistoryDeleteWorkflowExecutionScope tracks DeleteWorkflowExecutions API calls
@@ -1181,8 +1183,18 @@ var (
 	)
 
 	// Worker Versioning
-	WorkerDeploymentVersioningOverrideCounter = NewCounterDef("worker_deployment_versioning_override_count")
-	StartDeploymentTransitionCounter          = NewCounterDef("start_deployment_transition_count")
+	WorkerDeploymentCreated                           = NewCounterDef("worker_deployment_created")
+	WorkerDeploymentVersionCreated                    = NewCounterDef("worker_deployment_version_created")
+	WorkerDeploymentVersionCreatedManagedByController = NewCounterDef("worker_deployment_version_created_managed_by_controller")
+	WorkerDeploymentVersionVisibilityQueryCount       = NewCounterDef("worker_deployment_version_visibility_query_count")
+	WorkerDeploymentVersioningOverrideCounter         = NewCounterDef("worker_deployment_versioning_override_count")
+	StartDeploymentTransitionCounter                  = NewCounterDef("start_deployment_transition_count")
+
+	WorkflowResetCount        = NewCounterDef("workflow_reset_count")
+	WorkflowQuerySuccessCount = NewCounterDef("workflow_query_success_count")
+	WorkflowQueryFailureCount = NewCounterDef("workflow_query_failure_count")
+	WorkflowQueryTimeoutCount = NewCounterDef("workflow_query_timeout_count")
+	WorkflowTasksCompleted    = NewCounterDef("workflow_tasks_completed")
 
 	// Force replication
 	EncounterZombieWorkflowCount        = NewCounterDef("encounter_zombie_workflow_count")

common/metrics/tags.go

@@ -58,6 +58,11 @@ const (
 	trueValue               = "true"
 	errorPrefix             = "*"
 
+	queryTypeStackTrace       = "__stack_trace"
+	queryTypeOpenSessions     = "__open_sessions"
+	queryTypeWorkflowMetadata = "__temporal_workflow_metadata"
+	queryTypeUserDefined      = "__user_defined"
+
 	newRun      = "new"
 	existingRun = "existing"
 	childRun    = "child"
@@ -241,6 +246,14 @@ func FailureTag(value string) Tag {
 	return &tagImpl{key: FailureTagName, value: value}
 }
 
+func FirstAttemptTag(attempt int32) Tag {
+	value := falseValue
+	if attempt == 1 {
+		value = trueValue
+	}
+	return &tagImpl{key: isFirstAttempt, value: value}
+}
+
 func FailureSourceTag(value string) Tag {
 	if len(value) == 0 {
 		value = unknownValue
@@ -402,6 +415,22 @@ func DestinationTag(value string) Tag {
 	}
 }
 
+func VersioningBehaviorTag(behavior enumspb.VersioningBehavior) Tag {
+	return &tagImpl{versioningBehavior, behavior.String()}
+}
+
+func WorkflowStatusTag(status string) Tag {
+	return &tagImpl{key: workflowStatus, value: status}
+}
+
+func QueryTypeTag(queryType string) Tag {
+	if queryType == queryTypeStackTrace || queryType == queryTypeOpenSessions || queryType == queryTypeWorkflowMetadata {
+		return &tagImpl{key: queryType, value: queryType}
+	}
+	// group all user defined queries into a single tag value
+	return &tagImpl{key: queryType, value: queryTypeUserDefined}
+}
+
 func VersioningBehaviorBeforeOverrideTag(behavior enumspb.VersioningBehavior) Tag {
 	return &tagImpl{key: behaviorBefore, value: behavior.String()}
 }

service/history/api/queryworkflow/api.go

@@ -154,6 +154,7 @@ func Invoke(
 			return queryDirectlyThroughMatching(
 				ctx,
 				msResp,
+				nsEntry,
 				request.GetNamespaceId(),
 				req,
 				shardContext,
@@ -178,6 +179,12 @@ func Invoke(
 	}
 	queryID, completionCh := queryReg.BufferQuery(req.GetQuery())
 	defer queryReg.RemoveQuery(queryID)
+
+	msResp, err := api.MutableStateToGetResponse(mutableState)
+	if err != nil {
+		return nil, err
+	}
+
 	workflowLease.GetReleaseFn()(nil) // release the lock - no access to mutable state beyond this point!
 	select {
 	case <-completionCh:
@@ -191,13 +198,28 @@ func Invoke(
 			result := completionState.Result
 			switch result.GetResultType() {
 			case enumspb.QUERY_RESULT_TYPE_ANSWERED:
+				emitWorkflowQueryMetrics(
+					scope,
+					nsEntry,
+					msResp,
+					req.GetQuery().GetQueryType(),
+					nil,
+				)
 				return &historyservice.QueryWorkflowResponse{
 					Response: &workflowservice.QueryWorkflowResponse{
 						QueryResult: result.GetAnswer(),
 					},
 				}, nil
 			case enumspb.QUERY_RESULT_TYPE_FAILED:
-				return nil, serviceerror.NewQueryFailedWithFailure(result.GetErrorMessage(), result.GetFailure())
+				err := serviceerror.NewQueryFailedWithFailure(result.GetErrorMessage(), result.GetFailure())
+				emitWorkflowQueryMetrics(
+					scope,
+					nsEntry,
+					msResp,
+					req.GetQuery().GetQueryType(),
+					err,
+				)
+				return nil, err
 			default:
 				metrics.QueryRegistryInvalidStateCount.With(scope).Record(1)
 				return nil, consts.ErrQueryEnteredInvalidState
@@ -211,6 +233,7 @@ func Invoke(
 			return queryDirectlyThroughMatching(
 				ctx,
 				msResp,
+				nsEntry,
 				request.GetNamespaceId(),
 				req,
 				shardContext,
@@ -221,12 +244,27 @@ func Invoke(
 				priority,
 			)
 		case workflow.QueryCompletionTypeFailed:
-			return nil, completionState.Err
+			err = completionState.Err
+			emitWorkflowQueryMetrics(
+				scope,
+				nsEntry,
+				msResp,
+				req.GetQuery().GetQueryType(),
+				err,
+			)
+			return nil, err
 		default:
 			metrics.QueryRegistryInvalidStateCount.With(scope).Record(1)
 			return nil, consts.ErrQueryEnteredInvalidState
 		}
 	case <-ctx.Done():
+		emitWorkflowQueryMetrics(
+			scope,
+			nsEntry,
+			msResp,
+			req.GetQuery().GetQueryType(),
+			ctx.Err(),
+		)
 		metrics.ConsistentQueryTimeoutCount.With(scope).Record(1)
 		return nil, ctx.Err()
 	}
@@ -259,6 +297,7 @@ func queryWillTimeoutsBeforeFirstWorkflowTaskStart(
 func queryDirectlyThroughMatching(
 	ctx context.Context,
 	msResp *historyservice.GetMutableStateResponse,
+	nsEntry *namespace.Namespace,
 	namespaceID string,
 	queryRequest *workflowservice.QueryWorkflowRequest,
 	shard historyi.ShardContext,
@@ -267,11 +306,18 @@ func queryDirectlyThroughMatching(
 	matchingClient matchingservice.MatchingServiceClient,
 	metricsHandler metrics.Handler,
 	priority *commonpb.Priority,
-) (*historyservice.QueryWorkflowResponse, error) {
+) (resp *historyservice.QueryWorkflowResponse, retError error) {
 
 	startTime := time.Now().UTC()
 	defer func() {
 		metrics.DirectQueryDispatchLatency.With(metricsHandler).Record(time.Since(startTime))
+		emitWorkflowQueryMetrics(
+			metricsHandler,
+			nsEntry,
+			msResp,
+			queryRequest.GetQuery().GetQueryType(),
+			retError,
+		)
 	}()
 
 	directive := worker_versioning.MakeDirectiveForWorkflowTask(
@@ -355,3 +401,27 @@ func queryDirectlyThroughMatching(
 			QueryRejected: matchingResp.GetQueryRejected(),
 		}}, err
 }
+
+func emitWorkflowQueryMetrics(
+	metricsHandler metrics.Handler,
+	nsEntry *namespace.Namespace,
+	msResp *historyservice.GetMutableStateResponse,
+	queryType string,
+	err error,
+) {
+	commonTags := []metrics.Tag{
+		metrics.OperationTag(metrics.HistoryQueryWorkflowScope),
+		metrics.NamespaceTag(nsEntry.Name().String()),
+		metrics.VersioningBehaviorTag(workflow.GetEffectiveVersioningBehavior(msResp.GetVersioningInfo())),
+		metrics.WorkflowStatusTag(msResp.GetWorkflowStatus().String()),
+		metrics.QueryTypeTag(queryType),
+	}
+
+	if err == nil {
+		metrics.WorkflowQuerySuccessCount.With(metricsHandler).Record(1, commonTags...)
+	} else if common.IsContextDeadlineExceededErr(err) {
+		metrics.WorkflowQueryTimeoutCount.With(metricsHandler).Record(1, commonTags...)
+	} else {
+		metrics.WorkflowQueryFailureCount.With(metricsHandler).Record(1, commonTags...)
+	}
+}

service/history/api/resetworkflow/api.go

@@ -12,6 +12,7 @@ import (
 	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/locks"
 	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence/versionhistory"
 	"go.temporal.io/server/common/worker_versioning"
@@ -130,6 +131,15 @@ func Invoke(
 	if err != nil {
 		return nil, err
 	}
+
+	metrics.WorkflowResetCount.With(
+		shardContext.GetMetricsHandler().WithTags(
+			metrics.NamespaceTag(namespaceEntry.Name().String()),
+			metrics.OperationTag(metrics.HistoryResetWorkflowScope),
+			metrics.VersioningBehaviorTag(baseMutableState.GetEffectiveVersioningBehavior()),
+		),
+	).Record(1)
+
 	allowResetWithPendingChildren := shardContext.GetConfig().AllowResetWithPendingChildren(namespaceEntry.Name().String())
 	if err := ndc.NewWorkflowResetter(
 		shardContext,

service/history/api/respondworkflowtaskcompleted/api.go

@@ -431,7 +431,12 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 
 		metrics.FailedWorkflowTasksCounter.With(handler.metricsHandler).Record(
 			1,
-			metrics.OperationTag(metrics.HistoryRespondWorkflowTaskCompletedScope))
+			metrics.OperationTag(metrics.HistoryRespondWorkflowTaskCompletedScope),
+			metrics.NamespaceTag(namespaceEntry.Name().String()),
+			metrics.VersioningBehaviorTag(ms.GetEffectiveVersioningBehavior()),
+			metrics.FailureTag(wtFailedCause.failedCause.String()),
+			metrics.FirstAttemptTag(currentWorkflowTask.Attempt),
+		)
 		handler.logger.Info("Failing the workflow task.",
 			tag.Value(wtFailedCause.Message()),
 			tag.WorkflowID(token.GetWorkflowId()),

service/history/api/respondworkflowtaskfailed/api.go

@@ -7,6 +7,7 @@ import (
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/definition"
+	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/tasktoken"
 	"go.temporal.io/server/service/history/api"
@@ -41,6 +42,11 @@ func Invoke(
 			token.RunId,
 		),
 		func(workflowLease api.WorkflowLease) (*api.UpdateWorkflowAction, error) {
+			namespaceEntry, err := api.GetActiveNamespace(shardContext, namespace.ID(req.GetNamespaceId()))
+			if err != nil {
+				return nil, err
+			}
+
 			mutableState := workflowLease.GetMutableState()
 			if !mutableState.IsWorkflowExecutionRunning() {
 				return nil, consts.ErrWorkflowCompleted
@@ -73,6 +79,15 @@ func Invoke(
 				return nil, err
 			}
 
+			metrics.FailedWorkflowTasksCounter.With(shardContext.GetMetricsHandler()).Record(
+				1,
+				metrics.OperationTag(metrics.HistoryRespondWorkflowTaskFailedScope),
+				metrics.NamespaceTag(namespaceEntry.Name().String()),
+				metrics.VersioningBehaviorTag(mutableState.GetEffectiveVersioningBehavior()),
+				metrics.FailureTag(request.GetCause().String()),
+				metrics.FirstAttemptTag(workflowTask.Attempt),
+			)
+
 			// TODO (alex-update): if it was speculative WT that failed, and there is nothing but pending updates,
 			//  new WT also should be create as speculative (or not?). Currently, it will be recreated as normal WT.
 			return &api.UpdateWorkflowAction{

service/history/timer_queue_active_task_executor.go

@@ -312,6 +312,8 @@ func (t *timerQueueActiveTaskExecutor) processSingleActivityTimeoutTask(
 		namespace.ID(mutableState.GetExecutionInfo().NamespaceId),
 		metrics.TimerActiveTaskActivityTimeoutScope,
 		timerSequenceID.TimerType,
+		mutableState.GetEffectiveVersioningBehavior(),
+		ai.Attempt,
 	)
 	if _, err = mutableState.AddActivityTaskTimedOutEvent(
 		ai.ScheduledEventId,
@@ -380,6 +382,8 @@ func (t *timerQueueActiveTaskExecutor) executeWorkflowTaskTimeoutTask(
 			namespace.ID(mutableState.GetExecutionInfo().NamespaceId),
 			operationMetricsTag,
 			enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
+			mutableState.GetEffectiveVersioningBehavior(),
+			workflowTask.Attempt,
 		)
 		if _, err := mutableState.AddWorkflowTaskTimedOutEvent(
 			workflowTask,
@@ -398,6 +402,8 @@ func (t *timerQueueActiveTaskExecutor) executeWorkflowTaskTimeoutTask(
 			namespace.ID(mutableState.GetExecutionInfo().NamespaceId),
 			operationMetricsTag,
 			enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
+			mutableState.GetEffectiveVersioningBehavior(),
+			workflowTask.Attempt,
 		)
 		_, err := mutableState.AddWorkflowTaskScheduleToStartTimeoutEvent(workflowTask)
 		if err != nil {
@@ -856,6 +862,8 @@ func (t *timerQueueActiveTaskExecutor) emitTimeoutMetricScopeWithNamespaceTag(
 	namespaceID namespace.ID,
 	operation string,
 	timerType enumspb.TimeoutType,
+	effectiveVersioningBehavior enumspb.VersioningBehavior,
+	taskAttempt int32,
 ) {
 	namespaceEntry, err := t.registry.GetNamespaceByID(namespaceID)
 	if err != nil {
@@ -864,6 +872,8 @@ func (t *timerQueueActiveTaskExecutor) emitTimeoutMetricScopeWithNamespaceTag(
 	metricsScope := t.metricsHandler.WithTags(
 		metrics.OperationTag(operation),
 		metrics.NamespaceTag(namespaceEntry.Name().String()),
+		metrics.VersioningBehaviorTag(effectiveVersioningBehavior),
+		metrics.FirstAttemptTag(taskAttempt),
 	)
 	switch timerType {
 	case enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START:

service/history/timer_queue_active_task_executor_test.go

@@ -2233,6 +2233,7 @@ func (s *timerQueueActiveTaskExecutorSuite) TestProcessSingleActivityTimeoutTask
 				ms.EXPECT().GetExecutionInfo().Return(info).AnyTimes()
 				ms.EXPECT().AddActivityTaskTimedOutEvent(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
 			}
+			ms.EXPECT().GetEffectiveVersioningBehavior().Return(enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED).AnyTimes()
 
 			result, err := s.timerQueueActiveTaskExecutor.processSingleActivityTimeoutTask(
 				ms, tc.timerSequenceID, tc.ai)

service/history/workflow/workflow_task_state_machine.go

@@ -745,6 +745,13 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskCompletedEvent(
 	if err != nil {
 		return nil, err
 	}
+
+	metrics.WorkflowTasksCompleted.With(m.metricsHandler).Record(1,
+		metrics.NamespaceTag(m.ms.GetNamespaceEntry().Name().String()),
+		metrics.VersioningBehaviorTag(vb),
+		metrics.FirstAttemptTag(workflowTask.Attempt),
+	)
+
 	return event, nil
 }