Fix task schedule latency metric (#7055)

Author: Shaddoll

service/history/queuev2/queue_base.go

@@ -204,21 +204,22 @@ func (q *queueBase) LockTaskProcessing() {}
 
 func (q *queueBase) UnlockTaskProcessing() {}
 
-func (q *queueBase) processNewTasks() {
+func (q *queueBase) processNewTasks() bool {
 	newExclusiveMaxTaskKey := q.shard.UpdateIfNeededAndGetQueueMaxReadLevel(q.category, q.shard.GetClusterMetadata().GetCurrentClusterName())
 	if q.category.Type() == persistence.HistoryTaskCategoryTypeImmediate {
 		newExclusiveMaxTaskKey = persistence.NewImmediateTaskKey(newExclusiveMaxTaskKey.GetTaskID() + 1)
 	}
 
 	newVirtualSliceState, remainingVirtualSliceState, ok := q.newVirtualSliceState.TrySplitByTaskKey(newExclusiveMaxTaskKey)
 	if !ok {
-		return
+		return false
 	}
 	q.newVirtualSliceState = remainingVirtualSliceState
 
 	newVirtualSlice := NewVirtualSlice(newVirtualSliceState, q.taskInitializer, q.queueReader, NewPendingTaskTracker())
 
 	q.virtualQueueManager.AddNewVirtualSliceToRootQueue(newVirtualSlice)
+	return true
 }
 
 func (q *queueBase) updateQueueState(ctx context.Context) {

service/history/queuev2/queue_immediate.go

@@ -153,8 +153,11 @@ func (q *immediateQueue) processPollTimer() {
 	// thus this periodic poll is not needed, but we keep it for now to provide a fallback mechanism in case
 	// there is a bug in the notification mechanism
 	if q.lastPollTime.Add(q.base.options.PollBackoffInterval()).Before(q.base.timeSource.Now()) {
-		q.base.logger.Info("processing new tasks because poll timer fired")
-		q.base.processNewTasks()
+		newTasks := q.base.processNewTasks()
+		if newTasks {
+			// TODO: consider changing it to warn level
+			q.base.logger.Info("processing new tasks because poll timer fired")
+		}
 		q.lastPollTime = q.base.timeSource.Now()
 	}
 

service/history/task/task.go

@@ -134,11 +134,7 @@ func (t *taskImpl) Execute() error {
 	if t.State() != ctask.TaskStatePending {
 		return nil
 	}
-	if t.GetAttempt() == 0 {
-		// domain level metrics for the duration between task being submitted to task scheduler and being executed
-		t.scope.RecordHistogramDuration(metrics.TaskScheduleLatencyPerDomain, time.Since(t.initialSubmitTime))
-	}
-
+	scheduleLatency := t.timeSource.Now().Sub(t.initialSubmitTime)
 	var err error
 	t.shouldProcessTask, err = t.taskFilter(t.Task)
 	if err != nil {
@@ -158,6 +154,10 @@ func (t *taskImpl) Execute() error {
 	}()
 	executeResponse, err := t.taskExecutor.Execute(t)
 	t.scope = executeResponse.Scope
+	if t.GetAttempt() == 0 {
+		// domain level metrics for the duration between task being submitted to task scheduler and being executed
+		t.scope.RecordHistogramDuration(metrics.TaskScheduleLatencyPerDomain, scheduleLatency)
+	}
 	if t.isPreviousExecutorActive != executeResponse.IsActiveTask {
 		t.resetAttempt()
 	}