Produce workflow_duration metric on completion

Author: simvlad

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/workflow/metrics.go

@@ -77,19 +77,26 @@ 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,
+	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 {
+	if completion.state == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED &&
+		completion.executionTime != nil && completion.closeTime != nil {
+
+		executionTime := completion.executionTime.AsTime()
+		closeTime := completion.closeTime.AsTime()
+		if closeTime.After(executionTime) {
+			metrics.WorkflowDuration.With(handler).Record(closeTime.Sub(executionTime))
+		}
+	}
+
+	switch completion.status {
 	case enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED:
 		metrics.WorkflowSuccessCount.With(handler).Record(1)
 	case enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED:
@@ -102,6 +109,8 @@ 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:
+		// skip, the workflow is not completed
 	}
 }
 

service/history/workflow/metrics_test.go

@@ -0,0 +1,55 @@
+package workflow
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
+	enumsspb "go.temporal.io/server/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, err := metricstest.NewHandler(logger, metrics.ClientConfig{})
+	require.NoError(t, err)
+	testNamespace := namespace.Name("test-namespace")
+	config := &configs.Config{
+		BreakdownMetricsByTaskQueue: dynamicconfig.GetBoolPropertyFnFilteredByTaskQueue(true),
+	}
+
+	completionMetric := completionMetric{
+		initialized:      true,
+		taskQueue:        "test-task-queue",
+		namespaceState:   "active",
+		workflowTypeName: "test-workflow",
+		status:           enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
+		state:            enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
+		executionTime:    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)
+}

service/history/workflow/transaction_impl.go

@@ -6,6 +6,7 @@ import (
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
+	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
@@ -14,6 +15,7 @@ import (
 	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/service/history/events"
 	historyi "go.temporal.io/server/service/history/interfaces"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type (
@@ -23,6 +25,9 @@ type (
 		namespaceState   string
 		workflowTypeName string
 		status           enumspb.WorkflowExecutionStatus
+		state            enumsspb.WorkflowExecutionState
+		executionTime    *timestamppb.Timestamp
+		closeTime        *timestamppb.Timestamp
 	}
 	TransactionImpl struct {
 		shard  historyi.ShardContext
@@ -680,12 +685,16 @@ func snapshotToCompletionMetric(
 	if workflowSnapshot == nil {
 		return completionMetric{initialized: false}
 	}
+
 	return completionMetric{
 		initialized:      true,
 		taskQueue:        workflowSnapshot.ExecutionInfo.TaskQueue,
 		namespaceState:   namespaceState,
 		workflowTypeName: workflowSnapshot.ExecutionInfo.WorkflowTypeName,
 		status:           workflowSnapshot.ExecutionState.Status,
+		state:            workflowSnapshot.ExecutionState.State,
+		executionTime:    workflowSnapshot.ExecutionInfo.ExecutionTime,
+		closeTime:        workflowSnapshot.ExecutionInfo.CloseTime,
 	}
 }
 
@@ -696,12 +705,16 @@ func mutationToCompletionMetric(
 	if workflowMutation == nil {
 		return completionMetric{initialized: false}
 	}
+
 	return completionMetric{
 		initialized:      true,
 		taskQueue:        workflowMutation.ExecutionInfo.TaskQueue,
 		namespaceState:   namespaceState,
 		workflowTypeName: workflowMutation.ExecutionInfo.WorkflowTypeName,
 		status:           workflowMutation.ExecutionState.Status,
+		state:            workflowMutation.ExecutionState.State,
+		executionTime:    workflowMutation.ExecutionInfo.ExecutionTime,
+		closeTime:        workflowMutation.ExecutionInfo.CloseTime,
 	}
 }
 
@@ -717,13 +730,11 @@ func emitCompletionMetrics(
 		if !completionMetric.initialized {
 			continue
 		}
+
 		emitWorkflowCompletionStats(
 			metricsHandler,
 			namespaceName,
-			completionMetric.namespaceState,
-			completionMetric.taskQueue,
-			completionMetric.workflowTypeName,
-			completionMetric.status,
+			completionMetric,
 			shardContext.GetConfig(),
 		)
 	}