Remove tasks_expired metric, superseded by tasks_dropped (#10642)

## What changed?

Removes the per-task-queue **`tasks_expired`** counter
(`ExpiredTasksPerTaskQueueCounter`) and everything that existed only for
it: its tags (`TaskExpireStage*`, `TaskInvalidTag`, the
`taskExpireStage` key), the `getInvalidTaskTag` helper, all six emission
sites in matching, and the `TestGetInvalidTaskTag` test. Surrounding
expiry/validation logic and the `tasks_dropped` emissions are untouched.

## Why?

`tasks_expired` is fully superseded by `tasks_dropped` (#10468), which
reports the same events with finer `reason` tags (`expired_read`,
`expired_memory`, `invalid`). This is the agreed post-release cleanup.

## How did you test it?

- [x] built (`make bins` / `go build ./...`)
- [x] run locally and tested manually
- [x] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

Signed-off-by: Sandeep Balaji <sandeep.balaji@temporal.io>

Author: cbsandeep10

common/metrics/metric_defs.go

@@ -1240,7 +1240,6 @@ var (
 	)
 	SyncThrottlePerTaskQueueCounter                   = NewCounterDef("sync_throttle_count")
 	BufferThrottlePerTaskQueueCounter                 = NewCounterDef("buffer_throttle_count")
-	ExpiredTasksPerTaskQueueCounter                   = NewCounterDef("tasks_expired") // TODO: remove tasks_expired since it is superseded by tasks_dropped (expired_read / expired_memory reasons).
 	ForwardedPerTaskQueueCounter                      = NewCounterDef("forwarded_per_tl")
 	PriorityBacklogForwardedPerTaskQueueCounter       = NewCounterDef("priority_backlog_forwarded")
 	ForwardTaskErrorsPerTaskQueue                     = NewCounterDef("forward_task_errors")

common/metrics/tags.go

@@ -47,7 +47,6 @@ const (
 	// See server.api.enums.v1.ReplicationTaskType
 	replicationTaskType                            = "replicationTaskType"
 	replicationTaskPriority                        = "replicationTaskPriority"
-	taskExpireStage                                = "task_expire_stage"
 	taskAddResult                                  = "task_add_result"
 	versioningBehavior                             = "versioning_behavior"
 	continueAsNewVersioningBehavior                = "continue_as_new_versioning_behavior"
@@ -587,10 +586,6 @@ func ToUnversionedTag(version string) Tag {
 	return Tag{Key: toUnversioned, Value: falseValue}
 }
 
-var TaskExpireStageReadTag = Tag{Key: taskExpireStage, Value: "read"}
-var TaskExpireStageMemoryTag = Tag{Key: taskExpireStage, Value: "memory"}
-var TaskInvalidTag = Tag{Key: taskExpireStage, Value: "invalid"}
-
 // ClientNameTag returns a new client_name tag for the SDK client name.
 func ClientNameTag(value string) Tag {
 	if len(value) == 0 {

service/matching/fair_task_reader.go

@@ -513,7 +513,6 @@ func (tr *fairTaskReader) mergeTasksLocked(tasks []*persistencespb.AllocatedTask
 			// Expired tasks are added as pre-acked (nil) so they participate in
 			// readLevel calculation above and advance ackLevel + get GC'd below.
 			tr.outstandingTasks.Put(level, nil)
-			metrics.ExpiredTasksPerTaskQueueCounter.With(tr.backlogMgr.metricsHandler).Record(1, metrics.TaskExpireStageReadTag)
 			recordDroppedTask(tr.backlogMgr.metricsHandler, dropReasonExpiredRead)
 			continue
 		}

service/matching/metrics_util.go

@@ -36,15 +36,6 @@ func (r dropReason) tag() metrics.Tag {
 	}
 }
 
-// getInvalidTaskTag returns the tasks_expired stage tag for a task being dropped
-// due to in-memory expiry or validation failure.
-func getInvalidTaskTag(task *internalTask) metrics.Tag {
-	if IsTaskExpired(task.event.AllocatedTaskInfo) {
-		return metrics.TaskExpireStageMemoryTag
-	}
-	return metrics.TaskInvalidTag
-}
-
 // getDroppedTaskExpiryReason returns the drop reason for a task being dropped due to
 // in-memory expiry or validation failure.
 func getDroppedTaskExpiryReason(task *internalTask) dropReason {

service/matching/metrics_util_test.go

@@ -22,18 +22,6 @@ func backlogTaskWithExpiry(t *testing.T, expiry *timestamppb.Timestamp) *interna
 	}, func(*internalTask, taskResponse) {})
 }
 
-func TestGetInvalidTaskTag(t *testing.T) {
-	t.Run("expired -> memory stage", func(t *testing.T) {
-		task := backlogTaskWithExpiry(t, timestamppb.New(time.Now().Add(-time.Minute)))
-		require.Equal(t, metrics.TaskExpireStageMemoryTag, getInvalidTaskTag(task))
-	})
-
-	t.Run("not expired -> invalid", func(t *testing.T) {
-		task := backlogTaskWithExpiry(t, nil)
-		require.Equal(t, metrics.TaskInvalidTag, getInvalidTaskTag(task))
-	})
-}
-
 func TestGetDroppedTaskExpiryReason(t *testing.T) {
 	t.Run("expired -> expired_memory", func(t *testing.T) {
 		task := backlogTaskWithExpiry(t, timestamppb.New(time.Now().Add(-time.Minute)))

service/matching/physical_task_queue_manager.go

@@ -525,7 +525,6 @@ func (c *physicalTaskQueueManagerImpl) PollTask(
 		// history, but this is more efficient.
 		if task.event != nil && IsTaskExpired(task.event.AllocatedTaskInfo) {
 			// task is expired while polling
-			c.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, metrics.TaskExpireStageMemoryTag)
 			task.finish(taskFinishResult{dropReason: dropReasonExpiredMemory})
 			continue
 		}
@@ -580,8 +579,6 @@ func (c *physicalTaskQueueManagerImpl) ProcessSpooledTask(
 	task *internalTask,
 ) error {
 	if !c.taskValidator.maybeValidate(task.event.AllocatedTaskInfo, c.queue.TaskType()) {
-		var invalidTaskTag = getInvalidTaskTag(task)
-		c.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, invalidTaskTag)
 		task.finish(taskFinishResult{dropReason: getDroppedTaskExpiryReason(task)})
 		// Don't try to set read level here because it may have been advanced already.
 

service/matching/pri_matcher.go

@@ -210,10 +210,7 @@ func (tm *priTaskMatcher) forwardTask(task *internalTask) (bool, error) {
 		// to the head of the backlog, which is what taskValidator expects.
 		maybeValid := tm.validator.maybeValidate(task.event.AllocatedTaskInfo, tm.fwdr.partition.TaskType())
 		if !maybeValid {
-			var invalidTaskTag = getInvalidTaskTag(task)
-
 			// consider this task expired while processing.
-			tm.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, invalidTaskTag)
 			task.finish(taskFinishResult{dropReason: getDroppedTaskExpiryReason(task)})
 
 			// Stay alive as long as we're invalidating tasks
@@ -269,8 +266,6 @@ func (tm *priTaskMatcher) validateTasksOnRoot(retrier backoff.Retrier) {
 		maybeValid := tm.validator == nil || tm.validator.maybeValidate(task.event.AllocatedTaskInfo, tm.partition.TaskType())
 		if !maybeValid {
 			// We found an invalid one, complete it and go back for another immediately.
-			var invalidStageTag = getInvalidTaskTag(task)
-			tm.metricsHandler.Counter(metrics.ExpiredTasksPerTaskQueueCounter.Name()).Record(1, invalidStageTag)
 			task.finish(taskFinishResult{dropReason: getDroppedTaskExpiryReason(task)})
 
 			// Stay alive as long as we're invalidating tasks

service/matching/pri_task_reader.go

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

service/matching/task_reader.go

@@ -253,7 +253,6 @@ func (tr *taskReader) addTasksToBuffer(
 	for _, t := range tasks {
 		if IsTaskExpired(t) {
 			// task is expired when "add tasks to buffer" is called, so when we read it
-			metrics.ExpiredTasksPerTaskQueueCounter.With(tr.taggedMetricsHandler()).Record(1, metrics.TaskExpireStageReadTag)
 			recordDroppedTask(tr.taggedMetricsHandler(), dropReasonExpiredRead)
 			// 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