Produce workflow_duration metric on completion (#8185)

simvlad · lina-temporal · commit 298d5b687c90 · 2025-08-27T17:08:09.000-07:00
## What changed?
This PR produces the metric workflow_duration, when the workflow
execution completes.

## Why?
Currently there is no metric that captures the duration of the workflow
execution. It's also valuable to have the duration broken down by task
queue, namespace, workflow type, which this PR enables

## How did you test it?
- [X] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [X] added new unit test(s)
- [ ] added new functional test(s)
diff --git a/common/metrics/metric_defs.go b/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")
diff --git a/service/history/ndc/workflow_resetter.go b/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
 	}
diff --git a/service/history/ndc/workflow_resetter_test.go b/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)
diff --git a/service/history/workflow/context.go b/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
 	}
diff --git a/service/history/workflow/metrics.go b/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))
+		}
 	}
 }
 
diff --git a/service/history/workflow/metrics_test.go b/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)
+}
diff --git a/service/history/workflow/transaction.go b/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(
diff --git a/service/history/workflow/transaction_impl.go b/service/history/workflow/transaction_impl.go
diff --git a/service/history/workflow/transaction_mock.go b/service/history/workflow/transaction_mock.go
diff --git a/service/history/workflow/transaction_test.go b/service/history/workflow/transaction_test.go
