add expired tasks state to the metric (temporalio#8059)

## What changed?
add "task expire state" tag to the "ExpiredTasksPerTaskQueueCounter"
metric.
Possible states - "read", "process", "memory"

## Why?
For fairness, we remove TTL. We expect that this will lead to reading
too many expired tasks.
To be sure we want a metrics tag that will allows us to separate tasks
that are expired when we read them from other tasks.

## How did you test it?
- [X] built

Author: ychebotarev

common/metrics/tags.go

@@ -41,6 +41,7 @@ const (
 	// See server.api.enums.v1.ReplicationTaskType
 	replicationTaskType     = "replicationTaskType"
 	replicationTaskPriority = "replicationTaskPriority"
+	taskExpireStage         = "task_expire_stage"
 	versioningBehavior      = "versioning_behavior"
 	isFirstAttempt          = "first-attempt"
 	workflowStatus          = "workflow_status"
@@ -477,3 +478,7 @@ func ToUnversionedTag(version string) Tag {
 	}
 	return &tagImpl{key: toUnversioned, value: falseValue}
 }
+
+var TaskExpireStageReadTag Tag = &tagImpl{key: taskExpireStage, value: "read"}
+var TaskExpireStageMemoryTag Tag = &tagImpl{key: taskExpireStage, value: "memory"}
+var TaskInvalidTag Tag = &tagImpl{key: taskExpireStage, value: "invalid"}

service/matching/physical_task_queue_manager.go

@@ -338,8 +338,10 @@ func (c *physicalTaskQueueManagerImpl) PollTask(
 		// there. In that case, go back for another task.
 		// If we didn't do this, the task would be rejected when we call RecordXTaskStarted on
 		// history, but this is more efficient.
+
 		if task.event != nil && IsTaskExpired(task.event.AllocatedTaskInfo) {
-			c.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1)
+			// task is expired while polling
+			c.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, metrics.TaskExpireStageMemoryTag)
 			task.finish(nil, false)
 			continue
 		}
@@ -382,7 +384,9 @@ func (c *physicalTaskQueueManagerImpl) ProcessSpooledTask(
 ) error {
 	if !c.taskValidator.maybeValidate(task.event.AllocatedTaskInfo, c.queue.TaskType()) {
 		task.finish(nil, false)
-		c.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1)
+
+		var invalidTaskTag = getInvalidTaskTag(task)
+		c.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, invalidTaskTag)
 		// Don't try to set read level here because it may have been advanced already.
 		return nil
 	}

service/matching/pri_matcher.go

@@ -187,7 +187,10 @@ func (tm *priTaskMatcher) forwardTask(task *internalTask) error {
 		maybeValid := tm.validator.maybeValidate(task.event.AllocatedTaskInfo, tm.fwdr.partition.TaskType())
 		if !maybeValid {
 			task.finish(nil, false)
-			tm.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1)
+			var invalidTaskTag = getInvalidTaskTag(task)
+
+			// consider this task expired while processing.
+			tm.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, invalidTaskTag)
 			return nil
 		}
 
@@ -243,7 +246,9 @@ func (tm *priTaskMatcher) validateTasksOnRoot(lim quotas.RateLimiter, retrier ba
 		if !maybeValid {
 			// We found an invalid one, complete it and go back for another immediately.
 			task.finish(nil, false)
-			tm.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1)
+			var invalidStageTag = getInvalidTaskTag(task)
+			tm.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, invalidStageTag)
+
 			retrier.Reset()
 		} else {
 			// Task was valid, put it back and slow down checking.
@@ -611,3 +616,10 @@ func (tm *priTaskMatcher) emitForwardedSourceStats(
 		metrics.LocalToLocalMatchPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
 	}
 }
+
+func getInvalidTaskTag(task *internalTask) metrics.Tag {
+	if IsTaskExpired(task.event.AllocatedTaskInfo) {
+		return metrics.TaskExpireStageMemoryTag
+	}
+	return metrics.TaskInvalidTag
+}

service/matching/pri_task_reader.go

@@ -247,7 +247,8 @@ func (tr *priTaskReader) processTaskBatch(tasks []*persistencespb.AllocatedTaskI
 		tr.readLevel = max(tr.readLevel, t.TaskId)
 
 		if IsTaskExpired(t) {
-			metrics.ExpiredTasksPerTaskQueueCounter.With(tr.backlogMgr.metricsHandler).Record(1)
+			// task expired when we read it
+			metrics.ExpiredTasksPerTaskQueueCounter.With(tr.backlogMgr.metricsHandler).Record(1, metrics.TaskExpireStageReadTag)
 			return true
 		}
 

service/matching/task_reader.go

@@ -254,7 +254,8 @@ func (tr *taskReader) addTasksToBuffer(
 ) error {
 	for _, t := range tasks {
 		if IsTaskExpired(t) {
-			metrics.ExpiredTasksPerTaskQueueCounter.With(tr.taggedMetricsHandler()).Record(1)
+			// task is expired when "add tasks to buffer" is called, so when we read it
+			metrics.ExpiredTasksPerTaskQueueCounter.With(tr.taggedMetricsHandler()).Record(1, metrics.TaskExpireStageReadTag)
 			// Also increment readLevel for expired tasks otherwise it could result in
 			// looping over the same tasks if all tasks read in the batch are expired
 			tr.backlogMgr.taskAckManager.setReadLevel(t.GetTaskId())