Produce workflow_duration metric on completion

common/metrics/metric_defs.go

@@ -926,6 +926,7 @@ var (
 	WorkflowTimeoutCount                  = NewCounterDef("workflow_timeout")
 	WorkflowTerminateCount                = NewCounterDef("workflow_terminate")
 	WorkflowContinuedAsNewCount           = NewCounterDef("workflow_continued_as_new")
+	WorkflowDuration                      = NewTimerDef("workflow_duration")
 	ReplicationStreamPanic                = NewCounterDef("replication_stream_panic")
 	ReplicationStreamError                = NewCounterDef("replication_stream_error")
 	ReplicationServiceError               = NewCounterDef("replication_service_error")

service/history/ndc/workflow_resetter.go

@@ -369,6 +369,7 @@ func (r *workflowResetterImpl) persistToDB(
 			workflow.MutableStateFailoverVersion(resetWorkflow.GetMutableState()),
 			resetWorkflowSnapshot,
 			resetWorkflowEventsSeq,
+			currentWorkflow.GetMutableState().IsWorkflow(),
 		); err != nil {
 			return err
 		}
@@ -409,6 +410,7 @@ func (r *workflowResetterImpl) persistToDB(
 		&currentRunVerson,
 		currentWorkflowMutation,
 		currentWorkflowEventsSeq,
+		currentWorkflow.GetMutableState().IsWorkflow(),
 	); err != nil {
 		return err
 	}

service/history/ndc/workflow_resetter_test.go

@@ -137,6 +137,7 @@ func (s *workflowResetterSuite) TestPersistToDB_CurrentTerminated() {
 	}).AnyTimes()
 
 	currentMutableState.EXPECT().GetCurrentVersion().Return(int64(0)).AnyTimes()
+	currentMutableState.EXPECT().IsWorkflow().Return(true).AnyTimes()
 	currentNewEventsSize := int64(3444)
 	currentMutation := &persistence.WorkflowMutation{
 		ExecutionInfo: &persistencespb.WorkflowExecutionInfo{
@@ -201,6 +202,7 @@ func (s *workflowResetterSuite) TestPersistToDB_CurrentTerminated() {
 		util.Ptr(int64(0)),
 		resetSnapshot,
 		resetEventsSeq,
+		true, // isWorkflow
 	).Return(currentNewEventsSize, resetNewEventsSize, nil)
 
 	err := s.workflowResetter.persistToDB(context.Background(), currentWorkflow, currentWorkflow, currentMutation, currentEventsSeq, resetWorkflow)
@@ -226,6 +228,7 @@ func (s *workflowResetterSuite) TestPersistToDB_CurrentNotTerminated() {
 	currentMutation := &persistence.WorkflowMutation{}
 	currentEventsSeq := []*persistence.WorkflowEvents{{}}
 	currentMutableState.EXPECT().GetCurrentVersion().Return(int64(0)).AnyTimes()
+	currentMutableState.EXPECT().IsWorkflow().Return(true).AnyTimes()
 	currentMutableState.EXPECT().CloseTransactionAsMutation(historyi.TransactionPolicyActive).Return(currentMutation, currentEventsSeq, nil)
 
 	resetWorkflow := NewMockWorkflow(s.controller)
@@ -263,6 +266,7 @@ func (s *workflowResetterSuite) TestPersistToDB_CurrentNotTerminated() {
 		util.Ptr(int64(0)),
 		resetSnapshot,
 		resetEventsSeq,
+		true, // isWorkflow
 	).Return(int64(0), int64(0), nil)
 
 	err := s.workflowResetter.persistToDB(context.Background(), currentWorkflow, currentWorkflow, nil, nil, resetWorkflow)

service/history/workflow/context.go

@@ -246,6 +246,7 @@ func (c *ContextImpl) CreateWorkflowExecution(
 		shardContext,
 		newMutableState.GetCurrentVersion(),
 		createRequest,
+		newMutableState.IsWorkflow(),
 	)
 	if err != nil {
 		return err
@@ -363,6 +364,7 @@ func (c *ContextImpl) ConflictResolveWorkflowExecution(
 		MutableStateFailoverVersion(currentMutableState),
 		currentWorkflow,
 		currentWorkflowEventsSeq,
+		resetMutableState.IsWorkflow(),
 	); err != nil {
 		return err
 	}
@@ -576,6 +578,7 @@ func (c *ContextImpl) UpdateWorkflowExecutionWithNew(
 		MutableStateFailoverVersion(newMutableState),
 		newWorkflow,
 		newWorkflowEventsSeq,
+		c.MutableState.IsWorkflow(),
 	); err != nil {
 		return err
 	}

service/history/workflow/metrics.go

@@ -77,19 +77,22 @@ func emitMutableStateStatus(
 func emitWorkflowCompletionStats(
 	metricsHandler metrics.Handler,
 	namespace namespace.Name,
-	namespaceState string,
-	taskQueue string,
-	workflowTypeName string,
-	status enumspb.WorkflowExecutionStatus,
+	completion completionMetric,
 	config *configs.Config,
 ) {
-	handler := GetPerTaskQueueFamilyScope(metricsHandler, namespace, taskQueue, config,
+	// Only emit metrics for Workflows, not other Chasm archetypes
+	if !completion.isWorkflow {
+		return
+	}
+
+	handler := GetPerTaskQueueFamilyScope(metricsHandler, namespace, completion.taskQueue, config,
 		metrics.OperationTag(metrics.WorkflowCompletionStatsScope),
-		metrics.NamespaceStateTag(namespaceState),
-		metrics.WorkflowTypeTag(workflowTypeName),
+		metrics.NamespaceStateTag(completion.namespaceState),
+		metrics.WorkflowTypeTag(completion.workflowTypeName),
 	)
 
-	switch status {
+	closed := true
+	switch completion.status {
 	case enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED:
 		metrics.WorkflowSuccessCount.With(handler).Record(1)
 	case enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED:
@@ -102,6 +105,15 @@ func emitWorkflowCompletionStats(
 		metrics.WorkflowTerminateCount.With(handler).Record(1)
 	case enumspb.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW:
 		metrics.WorkflowContinuedAsNewCount.With(handler).Record(1)
+	case enumspb.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING:
+		closed = false
+	}
+	if closed && completion.startTime != nil && completion.closeTime != nil {
+		startTime := completion.startTime.AsTime()
+		closeTime := completion.closeTime.AsTime()
+		if closeTime.After(startTime) {
+			metrics.WorkflowDuration.With(handler).Record(closeTime.Sub(startTime))
+		}
 	}
 }
 

service/history/workflow/metrics_test.go

@@ -0,0 +1,65 @@
+package workflow
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/metrics/metricstest"
+	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/service/history/configs"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func TestEmitWorkflowCompletionStats_WorkflowDuration(t *testing.T) {
+	logger := log.NewTestLogger()
+	testHandler, _ := metricstest.NewHandler(logger, metrics.ClientConfig{})
+	testNamespace := namespace.Name("test-namespace")
+	config := &configs.Config{
+		BreakdownMetricsByTaskQueue: dynamicconfig.GetBoolPropertyFnFilteredByTaskQueue(true),
+	}
+
+	completionMetric := completionMetric{
+		initialized:      true,
+		isWorkflow:       true,
+		taskQueue:        "test-task-queue",
+		namespaceState:   "active",
+		workflowTypeName: "test-workflow",
+		status:           enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
+		startTime:        timestamppb.New(time.Unix(100, 0)),
+		closeTime:        timestamppb.New(time.Unix(130, 0)),
+	}
+
+	emitWorkflowCompletionStats(testHandler, testNamespace, completionMetric, config)
+
+	snapshot, err := testHandler.Snapshot()
+	require.NoError(t, err)
+	buckets, err := snapshot.Histogram("workflow_duration_milliseconds",
+
+		metrics.StringTag("namespace", "test-namespace"),
+		metrics.StringTag("namespace_state", "active"),
+		metrics.StringTag("workflowType", "test-workflow"),
+		metrics.StringTag("operation", "CompletionStats"),
+		metrics.StringTag("taskqueue", "test-task-queue"),
+		metrics.StringTag("otel_scope_name", "temporal"),
+		metrics.StringTag("otel_scope_version", ""),
+	)
+	require.NoError(t, err)
+	require.NotEmpty(t, buckets)
+}
+
+func TestEmitWorkflowCompletionStats_SkipNonWorkflow(t *testing.T) {
+	logger := log.NewTestLogger()
+	testHandler, _ := metricstest.NewHandler(logger, metrics.ClientConfig{})
+	testNamespace := namespace.Name("test-namespace")
+	completionMetric := completionMetric{isWorkflow: false}
+	emitWorkflowCompletionStats(testHandler, testNamespace, completionMetric, nil)
+	snapshot, err := testHandler.Snapshot()
+	require.NoError(t, err)
+	_, err = snapshot.Histogram("workflow_duration_milliseconds")
+	require.Error(t, err)
+}

service/history/workflow/transaction.go

@@ -15,6 +15,7 @@ type (
 			newWorkflowFailoverVersion int64,
 			newWorkflowSnapshot *persistence.WorkflowSnapshot,
 			newWorkflowEventsSeq []*persistence.WorkflowEvents,
+			isWorkflow bool,
 		) (int64, error)
 
 		ConflictResolveWorkflowExecution(
@@ -29,6 +30,7 @@ type (
 			currentWorkflowFailoverVersion *int64,
 			currentWorkflowMutation *persistence.WorkflowMutation,
 			currentWorkflowEventsSeq []*persistence.WorkflowEvents,
+			isWorkflow bool,
 		) (int64, int64, int64, error)
 
 		UpdateWorkflowExecution(
@@ -40,6 +42,7 @@ type (
 			newWorkflowFailoverVersion *int64,
 			newWorkflowSnapshot *persistence.WorkflowSnapshot,
 			newWorkflowEventsSeq []*persistence.WorkflowEvents,
+			isWorkflow bool,
 		) (int64, int64, error)
 
 		SetWorkflowExecution(