Add task count metric (#2869)

common/metrics/defs.go

@@ -2125,6 +2125,7 @@ const (
 	SignalInfoCount
 	RequestCancelInfoCount
 	BufferedEventsCount
+	TaskCount
 	WorkflowRetryBackoffTimerCount
 	WorkflowCronBackoffTimerCount
 	WorkflowCleanupDeleteCount
@@ -2607,6 +2608,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		SignalInfoCount:                                   NewDimensionlessHistogramDef("signal_info_count"),
 		RequestCancelInfoCount:                            NewDimensionlessHistogramDef("request_cancel_info_count"),
 		BufferedEventsCount:                               NewDimensionlessHistogramDef("buffered_events_count"),
+		TaskCount:                                         NewDimensionlessHistogramDef("task_count"),
 		WorkflowRetryBackoffTimerCount:                    NewCounterDef("workflow_retry_backoff_timer"),
 		WorkflowCronBackoffTimerCount:                     NewCounterDef("workflow_cron_backoff_timer"),
 		WorkflowCleanupDeleteCount:                        NewCounterDef("workflow_cleanup_delete"),

common/persistence/dataInterfaces.go

@@ -688,6 +688,7 @@ type (
 		SignalInfoCount        int
 		SignalRequestIDCount   int
 		BufferedEventsCount    int
+		TaskCountByCategory    map[string]int
 	}
 
 	HistoryStatistics struct {

common/persistence/size.go

@@ -24,6 +24,8 @@
 
 package persistence
 
+import "go.temporal.io/server/service/history/tasks"
+
 func statusOfInternalWorkflow(
 	state *InternalWorkflowMutableState,
 	historyStatistics *HistoryStatistics,
@@ -144,7 +146,8 @@ func statusOfInternalWorkflowMutation(
 		bufferedEventsSize = mutation.NewBufferedEvents.Size()
 	}
 
-	// TODO what about tasks?
+	taskCountByCategory := taskCountsByCategory(&mutation.Tasks)
+
 	// TODO what about checksum?
 
 	totalSize := executionInfoSize
@@ -184,7 +187,17 @@ func statusOfInternalWorkflowMutation(
 
 		BufferedEventsSize:  bufferedEventsSize,
 		BufferedEventsCount: bufferedEventsCount,
+
+		TaskCountByCategory: taskCountByCategory,
+	}
+}
+
+func taskCountsByCategory(t *map[tasks.Category][]InternalHistoryTask) map[string]int {
+	counts := make(map[string]int)
+	for category, tasks := range *t {
+		counts[category.Name()] = len(tasks)
 	}
+	return counts
 }
 
 func statusOfInternalWorkflowSnapshot(
@@ -229,6 +242,8 @@ func statusOfInternalWorkflowSnapshot(
 	totalSize += signalRequestIDSize
 	totalSize += bufferedEventsSize
 
+	taskCountByCategory := taskCountsByCategory(&snapshot.Tasks)
+
 	return &MutableStateStatistics{
 		TotalSize:         totalSize,
 		HistoryStatistics: historyStatistics,
@@ -256,5 +271,7 @@ func statusOfInternalWorkflowSnapshot(
 
 		BufferedEventsSize:  bufferedEventsSize,
 		BufferedEventsCount: bufferedEventsCount,
+
+		TaskCountByCategory: taskCountByCategory,
 	}
 }

service/history/workflow/metrics.go

@@ -78,6 +78,10 @@ func emitMutableStateStatus(
 		scope.RecordDistribution(metrics.HistorySize, stats.HistoryStatistics.SizeDiff)
 		scope.RecordDistribution(metrics.HistoryCount, stats.HistoryStatistics.CountDiff)
 	}
+
+	for category, taskCount := range stats.TaskCountByCategory {
+		scope.Tagged(metrics.TaskCategoryTag(category)).RecordDistribution(metrics.TaskCount, taskCount)
+	}
 }
 
 func emitWorkflowCompletionStats(