Emit estimated matching task lag metric (#2605)

Author: yycptt

common/metrics/defs.go

@@ -2131,6 +2131,7 @@ const (
 	TaskQueueStoppedCounter
 	TaskWriteThrottlePerTaskQueueCounter
 	TaskWriteLatencyPerTaskQueue
+	TaskLagPerTaskQueueGauge
 
 	NumMatchingMetrics
 )
@@ -2588,6 +2589,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		TaskQueueStoppedCounter:                   NewCounterDef("task_queue_stopped"),
 		TaskWriteThrottlePerTaskQueueCounter:      NewRollupCounterDef("task_write_throttle_count_per_tl", "task_write_throttle_count"),
 		TaskWriteLatencyPerTaskQueue:              NewRollupTimerDef("task_write_latency_per_tl", "task_write_latency"),
+		TaskLagPerTaskQueueGauge:                  NewGaugeDef("task_lag_per_tl"),
 	},
 	Worker: {
 		ReplicatorMessages:                            NewCounterDef("replicator_messages"),

common/metrics/tags.go

@@ -29,6 +29,7 @@ import (
 	"strconv"
 	"strings"
 
+	"go.temporal.io/api/enums/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 )
 
@@ -135,6 +136,10 @@ func TaskQueueTag(value string) Tag {
 	return &tagImpl{key: taskQueue, value: sanitizer.Value(value)}
 }
 
+func TaskQueueTypeTag(tqType enums.TaskQueueType) Tag {
+	return &tagImpl{key: TaskTypeTagName, value: tqType.String()}
+}
+
 // WorkflowTypeTag returns a new workflow type tag.
 func WorkflowTypeTag(value string) Tag {
 	if len(value) == 0 {

common/metrics/temporal_queues.go

@@ -42,5 +42,8 @@ func GetPerTaskQueueScope(
 		metricTaskQueueName = unknownValue
 	}
 
-	return baseScope.Tagged(NamespaceTag(namespaceName), TaskQueueTag(metricTaskQueueName))
+	return baseScope.Tagged(
+		NamespaceTag(namespaceName),
+		TaskQueueTag(metricTaskQueueName),
+	)
 }

service/history/historyEngine.go

@@ -1376,8 +1376,12 @@ func (e *historyEngineImpl) RecordActivityTaskStarted(
 			namespaceName := namespaceEntry.Name()
 			taskQueueName := ai.GetTaskQueue()
 
-			metrics.GetPerTaskQueueScope(metricsScope, namespaceName.String(), taskQueueName, enumspb.TASK_QUEUE_KIND_NORMAL).
-				Tagged(metrics.TaskTypeTag("activity")).
+			metrics.GetPerTaskQueueScope(
+				metricsScope,
+				namespaceName.String(),
+				taskQueueName,
+				enumspb.TASK_QUEUE_KIND_NORMAL,
+			).Tagged(metrics.TaskQueueTypeTag(enumspb.TASK_QUEUE_TYPE_ACTIVITY)).
 				RecordTimer(metrics.TaskScheduleToStartLatency, scheduleToStartLatency)
 
 			response.StartedTime = ai.StartedTime

service/history/workflowTaskHandlerCallbacks.go

@@ -234,8 +234,12 @@ func (handler *workflowTaskHandlerCallbacksImpl) handleWorkflowTaskStarted(
 			workflowScheduleToStartLatency := workflowTask.StartedTime.Sub(*workflowTask.ScheduledTime)
 			namespaceName := namespaceEntry.Name()
 			taskQueue := workflowTask.TaskQueue
-			metrics.GetPerTaskQueueScope(metricsScope, namespaceName.String(), taskQueue.GetName(), taskQueue.GetKind()).
-				Tagged(metrics.TaskTypeTag("workflow")).
+			metrics.GetPerTaskQueueScope(
+				metricsScope,
+				namespaceName.String(),
+				taskQueue.GetName(),
+				taskQueue.GetKind(),
+			).Tagged(metrics.TaskQueueTypeTag(enumspb.TASK_QUEUE_TYPE_WORKFLOW)).
 				RecordTimer(metrics.TaskScheduleToStartLatency, workflowScheduleToStartLatency)
 
 			resp, err = handler.createRecordWorkflowTaskStartedResponse(mutableState, workflowTask, req.PollRequest.GetIdentity())

service/matching/matchingEngine_test.go

@@ -844,7 +844,7 @@ func (s *matchingEngineSuite) TestSyncMatchActivities() {
 	}
 
 	time.Sleep(20 * time.Millisecond) // So any buffer tasks from 0 rps get picked up
-	syncCtr := scope.Snapshot().Counters()["test.sync_throttle_count_per_tl+namespace="+matchingTestNamespace+",operation=TaskQueueMgr,service_name=matching,taskqueue=makeToast"]
+	syncCtr := scope.Snapshot().Counters()["test.sync_throttle_count_per_tl+namespace="+matchingTestNamespace+",operation=TaskQueueMgr,service_name=matching,task_type=Activity,taskqueue=makeToast"]
 	s.Equal(1, int(syncCtr.Value()))                         // Check times zero rps is set = throttle counter
 	s.EqualValues(1, s.taskManager.getCreateTaskCount(tlID)) // Check times zero rps is set = Tasks stored in persistence
 	s.EqualValues(0, s.taskManager.getTaskCount(tlID))

service/matching/taskQueueManager.go

@@ -174,7 +174,7 @@ func newTaskQueueManager(
 		nsName.String(),
 		taskQueue.name,
 		taskQueueKind,
-	)
+	).Tagged(metrics.TaskQueueTypeTag(taskQueue.taskType))
 	tlMgr := &taskQueueManagerImpl{
 		status:              common.DaemonStatusInitialized,
 		namespaceRegistry:   e.namespaceRegistry,

service/matching/taskReader.go

@@ -268,7 +268,9 @@ func (tr *taskReader) addSingleTaskToBuffer(
 }
 
 func (tr *taskReader) persistAckLevel() error {
-	return tr.tlMgr.db.UpdateState(tr.tlMgr.taskAckManager.getAckLevel())
+	ackLevel := tr.tlMgr.taskAckManager.getAckLevel()
+	tr.emitTaskLagMetric(ackLevel)
+	return tr.tlMgr.db.UpdateState(ackLevel)
 }
 
 func (tr *taskReader) isTaskAddedRecently(lastAddTime time.Time) bool {
@@ -282,3 +284,10 @@ func (tr *taskReader) logger() log.Logger {
 func (tr *taskReader) scope() metrics.Scope {
 	return tr.tlMgr.metricScope
 }
+
+func (tr *taskReader) emitTaskLagMetric(ackLevel int64) {
+	// note: this metric is only an estimation for the lag.
+	// taskID in DB may not be continuous, especially when task list ownership changes.
+	maxReadLevel := tr.tlMgr.taskWriter.GetMaxReadLevel()
+	tr.scope().UpdateGauge(metrics.TaskLagPerTaskQueueGauge, float64(maxReadLevel-ackLevel))
+}