Workflow execution and scheduling latency

pkg/diagnostics/workflow_monitoring.go

@@ -52,9 +52,13 @@
 	activityExecutionCount *stats.Int64Measure
 	// activityExecutionLatency records time taken to run an activity to completion.
 	activityExecutionLatency *stats.Float64Measure
-	appID                    string
-	enabled                  bool
-	namespace                string
+	// workflowExecutionLatency records time taken to run a workflow to completion.
+	workflowExecutionLatency *stats.Float64Measure
+	// workflowSchedulingLatency records time taken between workflow execution request and actual workflow execution
+	workflowSchedulingLatency *stats.Float64Measure
+	appID                     string
+	enabled                   bool
+	namespace                 string
 }
 
 func newWorkflowMetrics() *workflowMetrics {
@@ -79,6 +83,14 @@
 			"runtime/workflow/activity/execution/latency",
 			"The total time taken to run an activity to completion.",
 			stats.UnitMilliseconds),
+		workflowExecutionLatency: stats.Float64(
+			"runtime/workflow/execution/latency",
+			"The total time taken to run workflow to completion.",
+			stats.UnitMilliseconds),
+		workflowSchedulingLatency: stats.Float64(
+			"runtime/workflow/scheduling/latency",
+			"The time taken between workflow execution request and actual workflow execution.",
+			stats.UnitMilliseconds),
 	}
 }
 
@@ -97,7 +109,9 @@
 		diagUtils.NewMeasureView(w.workflowOperationLatency, []tag.Key{appIDKey, namespaceKey, operationKey, statusKey}, defaultLatencyDistribution),
 		diagUtils.NewMeasureView(w.workflowExecutionCount, []tag.Key{appIDKey, namespaceKey, workflowNameKey, statusKey}, view.Count()),
 		diagUtils.NewMeasureView(w.activityExecutionCount, []tag.Key{appIDKey, namespaceKey, activityNameKey, statusKey}, view.Count()),
-		diagUtils.NewMeasureView(w.activityExecutionLatency, []tag.Key{appIDKey, namespaceKey, activityNameKey, statusKey}, defaultLatencyDistribution))
+		diagUtils.NewMeasureView(w.activityExecutionLatency, []tag.Key{appIDKey, namespaceKey, activityNameKey, statusKey}, defaultLatencyDistribution),
+		diagUtils.NewMeasureView(w.workflowExecutionLatency, []tag.Key{appIDKey, namespaceKey, workflowNameKey, statusKey}, defaultLatencyDistribution),
+		diagUtils.NewMeasureView(w.workflowSchedulingLatency, []tag.Key{appIDKey, namespaceKey, workflowNameKey, statusKey}, defaultLatencyDistribution))
 }
 
 // WorkflowOperationEvent records total number of Successful/Failed workflow Operations requests. It also records latency for those requests.
@@ -123,6 +137,26 @@
 	stats.RecordWithTags(ctx, diagUtils.WithTags(w.workflowExecutionCount.Name(), appIDKey, w.appID, namespaceKey, w.namespace, workflowNameKey, workflowName, statusKey, status), w.workflowExecutionCount.M(1))
 }
 
+func (w *workflowMetrics) WorkflowExecutionLatency(ctx context.Context, workflowName, status string, elapsed float64) {
+	if !w.IsEnabled() {
+		return
+	}
+
+	if elapsed > 0 {
+		stats.RecordWithTags(ctx, diagUtils.WithTags(w.workflowExecutionLatency.Name(), appIDKey, w.appID, namespaceKey, w.namespace, workflowNameKey, workflowName, statusKey, status), w.workflowExecutionLatency.M(elapsed))
+	}
+}
+
+func (w *workflowMetrics) WorkflowSchedulingLatency(ctx context.Context, workflowName, status string, elapsed float64) {
+	if !w.IsEnabled() {
+		return
+	}
+
+	if elapsed > 0 {
+		stats.RecordWithTags(ctx, diagUtils.WithTags(w.workflowSchedulingLatency.Name(), appIDKey, w.appID, namespaceKey, w.namespace, workflowNameKey, workflowName, statusKey, status), w.workflowSchedulingLatency.M(elapsed))
+	}
+}
+
 // ActivityExecutionEvent records total number of Successful/Failed/Recoverable workflow executions. It also records latency for these executions.
 func (w *workflowMetrics) ActivityExecutionEvent(ctx context.Context, activityName, status string, elapsed float64) {
 	if !w.IsEnabled() {

pkg/diagnostics/workflow_monitoring_test.go

@@ -222,7 +222,10 @@ func TestExecution(t *testing.T) {
 
 	t.Run("record workflow executions", func(t *testing.T) {
 		countMetricName := "runtime/workflow/execution/count"
+		executionLatencyMetricName := "runtime/workflow/execution/latency"
+		schedulingLatencyMetricName := "runtime/workflow/scheduling/latency"
 		workflowName := "test-workflow"
+
 		t.Run("Failed with retryable error", func(t *testing.T) {
 			w := initWorkflowMetrics()
 
@@ -255,5 +258,29 @@ func TestExecution(t *testing.T) {
 
 			allTagsPresent(t, v, viewData[0].Tags)
 		})
+
+		t.Run("workflow execution latency", func(t *testing.T) {
+			w := initWorkflowMetrics()
+
+			w.WorkflowExecutionLatency(context.Background(), workflowName, StatusSuccess, 20)
+
+			viewData, _ := view.RetrieveData(executionLatencyMetricName)
+			v := view.Find(executionLatencyMetricName)
+
+			allTagsPresent(t, v, viewData[0].Tags)
+			assert.InEpsilon(t, float64(20), viewData[0].Data.(*view.DistributionData).Min, 0)
+		})
+
+		t.Run("workflow scheduling latency", func(t *testing.T) {
+			w := initWorkflowMetrics()
+
+			w.WorkflowSchedulingLatency(context.Background(), workflowName, StatusSuccess, 10)
+
+			viewData, _ := view.RetrieveData(schedulingLatencyMetricName)
+			v := view.Find(schedulingLatencyMetricName)
+
+			allTagsPresent(t, v, viewData[0].Tags)
+			assert.InEpsilon(t, float64(10), viewData[0].Data.(*view.DistributionData).Min, 0)
+		})
 	})
 }

pkg/runtime/wfengine/backends/actors/workflow_actor.go

@@ -417,13 +417,24 @@
 
 	// The logic/for loop below purges/removes any leftover state from a completed or failed activity
 	transactionalRequests := make(map[string][]actors.TransactionalOperation)
+	isWorkflowExecutionStartedEvent := false
+	scheduledStartTimestamp := time.Time{}
+
 	for _, e := range state.Inbox {
 		var taskID int32
 		if ts := e.GetTaskCompleted(); ts != nil {
 			taskID = ts.GetTaskScheduledId()
 		} else if tf := e.GetTaskFailed(); tf != nil {
 			taskID = tf.GetTaskScheduledId()
 		} else {
+			// If the event is an execution started event, then we need to record the scheduled start timestamp
+			if es := e.GetExecutionStarted(); es != nil {
+				isWorkflowExecutionStartedEvent = true
+				timestamp := es.GetScheduledStartTimestamp()
+				if timestamp != nil {
+					scheduledStartTimestamp = timestamp.AsTime()
+				}
+			}
 			continue
 		}
 		op := actors.TransactionalOperation{
@@ -482,11 +493,14 @@
 		}
 		return newRecoverableError(fmt.Errorf("failed to schedule a workflow execution: %w", err))
 	}
-	// Record metrics for workflow execution
+
+	wf.recordWorkflowSchedulingLatency(ctx, isWorkflowExecutionStartedEvent, state, scheduledStartTimestamp, workflowName, executionStatus)
+	wfExecutionElapsedTime := float64(0)
+
 	defer func() {
 		if executionStatus != "" {
-			// execution latency for workflow is not supported yet.
 			diag.DefaultWorkflowMonitoring.WorkflowExecutionEvent(ctx, workflowName, executionStatus)
+			diag.DefaultWorkflowMonitoring.WorkflowExecutionLatency(ctx, workflowName, executionStatus, wfExecutionElapsedTime)
 		}
 	}()
 
@@ -635,11 +649,24 @@
 				// Setting executionStatus to failed if workflow has failed/terminated/cancelled
 				executionStatus = diag.StatusFailed
 			}
+			wfExecutionElapsedTime = diag.ElapsedSince(state.workflowStartTime)
 		}
 	}
 	return nil
 }
 
+func (wf *workflowActor) recordWorkflowSchedulingLatency(ctx context.Context, isWorkflowExecutionStartedEvent bool, state *workflowState, scheduledStartTimestamp time.Time, workflowName string, executionStatus string) {
+	if isWorkflowExecutionStartedEvent {
+		state.workflowStartTime = time.Now()
+		// If scheduledStartTimestamp is zero, then scheduling latency is zero
+		if !scheduledStartTimestamp.IsZero() {
+			// get the time diff between state.workflowStartTime and scheduledStartTimestamp
+			wfSchedulingLatency := float64(state.workflowStartTime.Sub(scheduledStartTimestamp).Milliseconds())
+			diag.DefaultWorkflowMonitoring.WorkflowSchedulingLatency(ctx, workflowName, executionStatus, wfSchedulingLatency)
+		}
+	}
+}
+
 func (wf *workflowActor) loadInternalState(ctx context.Context, actorID string) (*workflowState, error) {
 	if !wf.cachingDisabled {
 		// see if the state for this actor is already cached in memory

pkg/runtime/wfengine/backends/actors/workflowstate.go

@@ -28,10 +28,11 @@
 )
 
 const (
-	inboxKeyPrefix   = "inbox"
-	historyKeyPrefix = "history"
-	customStatusKey  = "customStatus"
-	metadataKey      = "metadata"
+	inboxKeyPrefix       = "inbox"
+	historyKeyPrefix     = "history"
+	customStatusKey      = "customStatus"
+	workflowStartTimeKey = "workflowStartTime"
+	metadataKey          = "metadata"
 )
 
 type workflowState struct {
@@ -45,6 +46,7 @@
 	inboxRemovedCount   int
 	historyAddedCount   int
 	historyRemovedCount int
+	workflowStartTime   time.Time
 	config              actorsBackendConfig
 }
 
@@ -127,6 +129,14 @@
 		return nil, err
 	}
 
+	// if s.workflowStartTime is not nil then append req.operations with a new operation to update the workflowStartTime
+	if !s.workflowStartTime.IsZero() {
+		req.Operations = append(req.Operations, actors.TransactionalOperation{
+			Operation: actors.Upsert,
+			Request:   actors.TransactionalUpsert{Key: workflowStartTimeKey, Value: s.workflowStartTime},
+		})
+	}
+
 	// We update the custom status only when the workflow itself has been updated, and not when
 	// we're saving changes only to the workflow inbox.
 	// CONSIDER: Only save custom status if it has changed. However, need a way to track this.
@@ -252,8 +262,8 @@
 	bulkReq := &actors.GetBulkStateRequest{
 		ActorType: config.workflowActorType,
 		ActorID:   actorID,
-		// Initializing with size for all the inbox, history, and custom status
-		Keys: make([]string, metadata.InboxLength+metadata.HistoryLength+1),
+		// Initializing with size for all the inbox, history, custom status and workflow execution start time
+		Keys: make([]string, metadata.InboxLength+metadata.HistoryLength+2),
 	}
 
 	var n int
@@ -268,6 +278,8 @@
 		n++
 	}
 
+	bulkReq.Keys[n] = workflowStartTimeKey
+
 	// Perform the request
 	bulkRes, err := actorRuntime.GetBulkState(ctx, bulkReq)
 	if err != nil {
@@ -305,6 +317,13 @@
 		}
 	}
 
+	if len(bulkRes[workflowStartTimeKey]) > 0 {
+		err = json.Unmarshal(bulkRes[workflowStartTimeKey], &state.workflowStartTime)
+		if err != nil {
+			return nil, fmt.Errorf("failed to unmarshal JSON from workflow start time key entry: %w", err)
+		}
+	}
+
 	wfLogger.Infof("%s: loaded %d state records in %v", actorID, loadedRecords, time.Since(loadStartTime))
 	return state, nil
 }
@@ -336,6 +355,10 @@
 			Operation: actors.Delete,
 			Request:   actors.TransactionalDelete{Key: metadataKey},
 		},
+		actors.TransactionalOperation{
+			Operation: actors.Delete,
+			Request:   actors.TransactionalDelete{Key: workflowStartTimeKey},
+		},
 	)
 
 	return req, nil