Improve logging and metric for history queue v2 (#7221)

Shaddoll · web-flow · commit a07ce6758258 · 2025-08-29T16:14:11.000-07:00
diff --git a/common/metrics/defs.go b/common/metrics/defs.go
@@ -2420,6 +2420,7 @@ const (
 	TaskRequestsNewScheduler
 	PendingTaskGauge
 	ReschedulerTaskCountGauge
+	NewHistoryTaskCounter
 
 	TaskRequestsPerDomain
 	TaskLatencyPerDomain
@@ -3196,6 +3197,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		TaskRequestsNewScheduler:                      {metricName: "task_requests_new_scheduler", metricType: Counter},
 		PendingTaskGauge:                              {metricName: "pending_task_gauge", metricType: Gauge},
 		ReschedulerTaskCountGauge:                     {metricName: "rescheduler_task_count", metricType: Gauge},
+		NewHistoryTaskCounter:                         {metricName: "new_history_task_counter", metricType: Counter},
 
 		// per domain task metrics
 
diff --git a/service/history/queuev2/pending_task_tracker.go b/service/history/queuev2/pending_task_tracker.go
@@ -35,7 +35,7 @@ type (
 		// AddTask adds a task to the pending task tracker.
 		AddTask(task.Task)
 		// PruneAckedTasks prunes the acked tasks from the pending task tracker.
-		PruneAckedTasks()
+		PruneAckedTasks() int
 		// GetMinimumTaskKey returns the minimum task key in the pending task tracker, if there are no pending tasks, it returns MaximumHistoryTaskKey.
 		GetMinimumTaskKey() (persistence.HistoryTaskKey, bool)
 		// GetTasks returns all the tasks in the pending task tracker, the result should be read-only.
@@ -85,12 +85,14 @@ func (t *pendingTaskTrackerImpl) GetTasks() map[persistence.HistoryTaskKey]task.
 	return t.taskMap
 }
 
-func (t *pendingTaskTrackerImpl) PruneAckedTasks() {
+func (t *pendingTaskTrackerImpl) PruneAckedTasks() int {
+	prunedCount := 0
 	minTaskKey := persistence.MaximumHistoryTaskKey
 	for key, task := range t.taskMap {
 		if task.State() == ctask.TaskStateAcked {
 			delete(t.taskMap, key)
 			t.taskCountPerDomain[task.GetDomainID()]--
+			prunedCount++
 			continue
 		}
 
@@ -99,6 +101,7 @@ func (t *pendingTaskTrackerImpl) PruneAckedTasks() {
 		}
 	}
 	t.minTaskKey = minTaskKey
+	return prunedCount
 }
 
 func (t *pendingTaskTrackerImpl) GetPendingTaskCount() int {
diff --git a/service/history/queuev2/pending_task_tracker_mock.go b/service/history/queuev2/pending_task_tracker_mock.go
diff --git a/service/history/queuev2/pending_task_tracker_test.go b/service/history/queuev2/pending_task_tracker_test.go
@@ -18,6 +18,7 @@ func TestPendingTaskTracker(t *testing.T) {
 		name             string
 		setupTasks       func(ctrl *gomock.Controller) []*task.MockTask
 		pruneAcked       bool
+		pruneAckedCount  int
 		clear            bool
 		wantMinKey       persistence.HistoryTaskKey
 		wantHasMinKey    bool
@@ -92,6 +93,7 @@ func TestPendingTaskTracker(t *testing.T) {
 				return []*task.MockTask{task1, task2, task3}
 			},
 			pruneAcked:       true,
+			pruneAckedCount:  2,
 			wantMinKey:       persistence.NewHistoryTaskKey(testTime, 2),
 			wantHasMinKey:    true,
 			wantTaskCount:    1,
@@ -113,6 +115,7 @@ func TestPendingTaskTracker(t *testing.T) {
 				return []*task.MockTask{task1, task2}
 			},
 			pruneAcked:       true,
+			pruneAckedCount:  2,
 			wantMinKey:       persistence.MaximumHistoryTaskKey,
 			wantHasMinKey:    false,
 			wantTaskCount:    0,
@@ -162,7 +165,8 @@ func TestPendingTaskTracker(t *testing.T) {
 
 			// Prune acked tasks if needed
 			if tt.pruneAcked {
-				tracker.PruneAckedTasks()
+				prunedCount := tracker.PruneAckedTasks()
+				assert.Equal(t, tt.pruneAckedCount, prunedCount)
 			}
 
 			// Clear all tasks if needed
diff --git a/service/history/queuev2/queue_base.go b/service/history/queuev2/queue_base.go
@@ -111,6 +111,7 @@ func newQueueBase(
 	if err != nil {
 		logger.Fatal("Failed to get queue state, probably task category is not supported", tag.Error(err), tag.Dynamic("category", category))
 	}
+	logger.Info("loading queue state", tag.Dynamic("queue-state", persistenceQueueState))
 	queueState := FromPersistenceQueueState(persistenceQueueState)
 	exclusiveAckLevel, _ := getExclusiveAckLevelAndMaxQueueIDFromQueueState(queueState)
 
@@ -266,7 +267,7 @@ func (q *queueBase) processNewTasks() bool {
 	}
 	q.newVirtualSliceState = remainingVirtualSliceState
 
-	newVirtualSlice := NewVirtualSlice(newVirtualSliceState, q.taskInitializer, q.queueReader, NewPendingTaskTracker())
+	newVirtualSlice := NewVirtualSlice(newVirtualSliceState, q.taskInitializer, q.queueReader, NewPendingTaskTracker(), q.logger)
 
 	q.logger.Debug("processing new tasks", tag.Dynamic("inclusiveMinTaskKey", newVirtualSliceState.Range.InclusiveMinTaskKey), tag.Dynamic("exclusiveMaxTaskKey", newVirtualSliceState.Range.ExclusiveMaxTaskKey))
 	q.virtualQueueManager.AddNewVirtualSliceToRootQueue(newVirtualSlice)
@@ -326,7 +327,7 @@ func (q *queueBase) updateQueueState(ctx context.Context) {
 
 	// even though the ack level is not updated, we still need to update the queue state
 	persistenceQueueState := ToPersistenceQueueState(queueState)
-	q.logger.Debug("store queue state", tag.Dynamic("queue-state", persistenceQueueState))
+	q.logger.Debug("store queue state", tag.Dynamic("queue-state", persistenceQueueState), tag.PendingTaskCount(pendingTaskCount))
 	err := q.shard.UpdateQueueState(q.category, persistenceQueueState)
 	if err != nil {
 		q.logger.Error("Failed to update queue state", tag.Error(err))
diff --git a/service/history/queuev2/queue_immediate.go b/service/history/queuev2/queue_immediate.go
@@ -134,11 +134,13 @@ func (q *immediateQueue) LockTaskProcessing() {
 }
 
 func (q *immediateQueue) NotifyNewTask(clusterName string, info *hcommon.NotifyTaskInfo) {
-	if len(info.Tasks) == 0 {
+	numTasks := len(info.Tasks)
+	if numTasks == 0 {
 		return
 	}
 
 	q.notify()
+	q.base.metricsScope.AddCounter(metrics.NewHistoryTaskCounter, int64(numTasks))
 }
 
 func (q *immediateQueue) notify() {
diff --git a/service/history/queuev2/queue_scheduled.go b/service/history/queuev2/queue_scheduled.go
@@ -138,19 +138,21 @@ func (q *scheduledQueue) UnlockTaskProcessing() {
 }
 
 func (q *scheduledQueue) NotifyNewTask(clusterName string, info *hcommon.NotifyTaskInfo) {
-	if len(info.Tasks) == 0 {
+	numTasks := len(info.Tasks)
+	if numTasks == 0 {
 		return
 	}
 
 	nextTime := info.Tasks[0].GetVisibilityTimestamp()
-	for i := 1; i < len(info.Tasks); i++ {
+	for i := 1; i < numTasks; i++ {
 		ts := info.Tasks[i].GetVisibilityTimestamp()
 		if ts.Before(nextTime) {
 			nextTime = ts
 		}
 	}
 
 	q.notify(nextTime)
+	q.base.metricsScope.AddCounter(metrics.NewHistoryTaskCounter, int64(numTasks))
 }
 
 func (q *scheduledQueue) notify(t time.Time) {
diff --git a/service/history/queuev2/virtual_queue.go b/service/history/queuev2/virtual_queue.go
@@ -344,6 +344,8 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {
 
 	pendingTaskCount := q.monitor.GetTotalPendingTaskCount()
 	maxTaskCount := q.queueOptions.MaxPendingTasksCount()
+	// TODO: review the metrics and remove this comment or change the metric from gauge to histogram
+	q.metricsScope.UpdateGauge(metrics.PendingTaskGauge, float64(pendingTaskCount))
 	if pendingTaskCount >= maxTaskCount {
 		q.logger.Warn("Too many pending tasks, pause loading tasks for a while", tag.PendingTaskCount(pendingTaskCount), tag.MaxTaskCount(maxTaskCount))
 		q.pauseController.Pause(q.queueOptions.PollBackoffInterval())
diff --git a/service/history/queuev2/virtual_queue_manager.go b/service/history/queuev2/virtual_queue_manager.go
@@ -96,7 +96,7 @@ func NewVirtualQueueManager(
 	for queueID, states := range virtualQueueStates {
 		virtualSlices := make([]VirtualSlice, len(states))
 		for i, state := range states {
-			virtualSlices[i] = NewVirtualSlice(state, taskInitializer, queueReader, NewPendingTaskTracker())
+			virtualSlices[i] = NewVirtualSlice(state, taskInitializer, queueReader, NewPendingTaskTracker(), logger)
 		}
 		var opts *VirtualQueueOptions
 		if queueID == rootQueueID {
@@ -220,10 +220,13 @@ func (m *virtualQueueManagerImpl) AddNewVirtualSliceToRootQueue(s VirtualSlice)
 
 func (m *virtualQueueManagerImpl) appendOrMergeSlice(vq VirtualQueue, s VirtualSlice) {
 	now := m.timeSource.Now()
+	newVirtualSliceState := s.GetState()
 	if now.After(m.nextForceNewSliceTime) {
+		m.logger.Debug("append new slice to virtual queue", tag.Dynamic("currentTime", now), tag.Dynamic("nextForceNewSliceTime", m.nextForceNewSliceTime), tag.Dynamic("inclusiveMinTaskKey", newVirtualSliceState.Range.InclusiveMinTaskKey), tag.Dynamic("exclusiveMaxTaskKey", newVirtualSliceState.Range.ExclusiveMaxTaskKey))
 		vq.AppendSlices(s)
 		m.nextForceNewSliceTime = now.Add(m.queueManagerOptions.VirtualSliceForceAppendInterval())
 		return
 	}
+	m.logger.Debug("merge slice to virtual queue", tag.Dynamic("currentTime", now), tag.Dynamic("nextForceNewSliceTime", m.nextForceNewSliceTime), tag.Dynamic("inclusiveMinTaskKey", newVirtualSliceState.Range.InclusiveMinTaskKey), tag.Dynamic("exclusiveMaxTaskKey", newVirtualSliceState.Range.ExclusiveMaxTaskKey))
 	vq.MergeSlices(s)
 }
diff --git a/service/history/queuev2/virtual_queue_manager_test.go b/service/history/queuev2/virtual_queue_manager_test.go
@@ -492,6 +492,12 @@ func TestVirtualQueueManager_AddNewVirtualSlice(t *testing.T) {
 			newSlice: nil, // Will be replaced with mock
 			setupMockQueues: func(mocks map[int64]*MockVirtualQueue, slice *MockVirtualSlice) {
 				mocks[rootQueueID].EXPECT().MergeSlices(slice)
+				slice.EXPECT().GetState().Return(VirtualSliceState{
+					Range: Range{
+						InclusiveMinTaskKey: persistence.NewImmediateTaskKey(1),
+						ExclusiveMaxTaskKey: persistence.NewImmediateTaskKey(10),
+					},
+				})
 			},
 			verifyQueues: func(t *testing.T, queues map[int64]VirtualQueue) {
 				assert.Contains(t, queues, int64(rootQueueID))
@@ -506,6 +512,12 @@ func TestVirtualQueueManager_AddNewVirtualSlice(t *testing.T) {
 			newSlice: nil, // Will be replaced with mock
 			setupMockQueues: func(mocks map[int64]*MockVirtualQueue, slice *MockVirtualSlice) {
 				mocks[rootQueueID].EXPECT().AppendSlices(slice)
+				slice.EXPECT().GetState().Return(VirtualSliceState{
+					Range: Range{
+						InclusiveMinTaskKey: persistence.NewImmediateTaskKey(1),
+						ExclusiveMaxTaskKey: persistence.NewImmediateTaskKey(10),
+					},
+				})
 			},
 			verifyQueues: func(t *testing.T, queues map[int64]VirtualQueue) {
 				assert.Contains(t, queues, int64(rootQueueID))
diff --git a/service/history/queuev2/virtual_slice.go b/service/history/queuev2/virtual_slice.go
@@ -26,6 +26,8 @@ package queuev2
 import (
 	"context"
 
+	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/service/history/task"
 )
@@ -54,6 +56,7 @@ type (
 		taskInitializer    task.Initializer
 		queueReader        QueueReader
 		pendingTaskTracker PendingTaskTracker
+		logger             log.Logger
 
 		// progress tracks the read progress of the slice, sorted by the inclusive min task key of the range, ranges are not overlapping
 		// For a virtual slice, the progress is a task key pointing to the next task to read and the next page token
@@ -69,12 +72,14 @@ func NewVirtualSlice(
 	taskInitializer task.Initializer,
 	queueReader QueueReader,
 	pendingTaskTracker PendingTaskTracker,
+	logger log.Logger,
 ) VirtualSlice {
 	return &virtualSliceImpl{
 		state:              state,
 		taskInitializer:    taskInitializer,
 		queueReader:        queueReader,
 		pendingTaskTracker: pendingTaskTracker,
+		logger:             logger,
 		progress: []*GetTaskProgress{
 			{
 				Range:         state.Range,
@@ -148,7 +153,12 @@ func (s *virtualSliceImpl) PendingTaskStats() PendingTaskStats {
 }
 
 func (s *virtualSliceImpl) UpdateAndGetState() VirtualSliceState {
-	s.pendingTaskTracker.PruneAckedTasks()
+	prunedCount := s.pendingTaskTracker.PruneAckedTasks()
+	nextTaskKey := s.state.Range.ExclusiveMaxTaskKey
+	if len(s.progress) > 0 {
+		nextTaskKey = s.progress[0].NextTaskKey
+	}
+	s.logger.Debug("pruned acked tasks", tag.Counter(prunedCount), tag.Dynamic("inclusiveMinTaskKey", s.state.Range.InclusiveMinTaskKey), tag.Dynamic("exclusiveMaxTaskKey", s.state.Range.ExclusiveMaxTaskKey), tag.Dynamic("nextTaskKey", nextTaskKey))
 	minPendingTaskKey, ok := s.pendingTaskTracker.GetMinimumTaskKey()
 	if !ok {
 		if len(s.progress) > 0 { // no pending tasks, and there are more tasks to read
@@ -232,6 +242,7 @@ func (s *virtualSliceImpl) TrySplitByTaskKey(taskKey persistence.HistoryTaskKey)
 		queueReader:        s.queueReader,
 		pendingTaskTracker: leftTracker,
 		progress:           leftProgress,
+		logger:             s.logger,
 	}
 
 	rightSlice := &virtualSliceImpl{
@@ -240,6 +251,7 @@ func (s *virtualSliceImpl) TrySplitByTaskKey(taskKey persistence.HistoryTaskKey)
 		queueReader:        s.queueReader,
 		pendingTaskTracker: rightTracker,
 		progress:           rightProgress,
+		logger:             s.logger,
 	}
 
 	return leftSlice, rightSlice, true
@@ -272,13 +284,15 @@ func (s *virtualSliceImpl) TrySplitByPredicate(predicate Predicate) (VirtualSlic
 		queueReader:        s.queueReader,
 		pendingTaskTracker: passTracker,
 		progress:           passProgress,
+		logger:             s.logger,
 	}
 	failSlice := &virtualSliceImpl{
 		state:              failState,
 		taskInitializer:    s.taskInitializer,
 		queueReader:        s.queueReader,
 		pendingTaskTracker: failTracker,
 		progress:           failProgress,
+		logger:             s.logger,
 	}
 	return passSlice, failSlice, true
 }
@@ -323,6 +337,7 @@ func mergeVirtualSlicesByRange(left, right *virtualSliceImpl) VirtualSlice {
 		queueReader:        left.queueReader,
 		pendingTaskTracker: pendingTaskTracker,
 		progress:           mergedProgress,
+		logger:             left.logger,
 	}
 }
 
@@ -514,6 +529,7 @@ func mergeVirtualSlicesByPredicate(this, that *virtualSliceImpl) VirtualSlice {
 		queueReader:        this.queueReader,
 		pendingTaskTracker: pendingTaskTracker,
 		progress:           mergedProgress,
+		logger:             this.logger,
 	}
 }
 
diff --git a/service/history/queuev2/virtual_slice_test.go b/service/history/queuev2/virtual_slice_test.go
@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 	gomock "go.uber.org/mock/gomock"
 
+	"github.com/uber/cadence/common/log/testlogger"
 	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/service/history/task"
 )
@@ -1317,6 +1318,7 @@ func TestUpdateAndGetState(t *testing.T) {
 
 			tt.slice.queueReader = mockQueueReader
 			tt.slice.pendingTaskTracker = mockPendingTaskTracker
+			tt.slice.logger = testlogger.New(t)
 
 			// Setup mock expectations using the setupMock function
 			tt.setupMock(mockPendingTaskTracker)

Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ type (`
`35`	`35`	`// AddTask adds a task to the pending task tracker.`
`36`	`36`	`AddTask(task.Task)`
`37`	`37`	`// PruneAckedTasks prunes the acked tasks from the pending task tracker.`
`38`		`- PruneAckedTasks()`
	`38`	`+ PruneAckedTasks() int`
`39`	`39`	`// GetMinimumTaskKey returns the minimum task key in the pending task tracker, if there are no pending tasks, it returns MaximumHistoryTaskKey.`
`40`	`40`	`GetMinimumTaskKey() (persistence.HistoryTaskKey, bool)`
`41`	`41`	`// GetTasks returns all the tasks in the pending task tracker, the result should be read-only.`
`@@ -85,12 +85,14 @@ func (t *pendingTaskTrackerImpl) GetTasks() map[persistence.HistoryTaskKey]task.`
`85`	`85`	`return t.taskMap`
`86`	`86`	`}`
`87`	`87`
`88`		`-func (t *pendingTaskTrackerImpl) PruneAckedTasks() {`
	`88`	`+func (t *pendingTaskTrackerImpl) PruneAckedTasks() int {`
	`89`	`+ prunedCount := 0`
`89`	`90`	`minTaskKey := persistence.MaximumHistoryTaskKey`
`90`	`91`	`for key, task := range t.taskMap {`
`91`	`92`	`if task.State() == ctask.TaskStateAcked {`
`92`	`93`	`delete(t.taskMap, key)`
`93`	`94`	`t.taskCountPerDomain[task.GetDomainID()]--`
	`95`	`+ prunedCount++`
`94`	`96`	`continue`
`95`	`97`	`}`
`96`	`98`
`@@ -99,6 +101,7 @@ func (t *pendingTaskTrackerImpl) PruneAckedTasks() {`
`99`	`101`	`}`
`100`	`102`	`}`
`101`	`103`	`t.minTaskKey = minTaskKey`
	`104`	`+ return prunedCount`
`102`	`105`	`}`
`103`	`106`
`104`	`107`	`func (t *pendingTaskTrackerImpl) GetPendingTaskCount() int {`
Original file line number	Diff line number	Diff line change
`@@ -134,11 +134,13 @@ func (q *immediateQueue) LockTaskProcessing() {`
`134`	`134`	`}`
`135`	`135`
`136`	`136`	`func (q immediateQueue) NotifyNewTask(clusterName string, info hcommon.NotifyTaskInfo) {`
`137`		`- if len(info.Tasks) == 0 {`
	`137`	`+ numTasks := len(info.Tasks)`
	`138`	`+ if numTasks == 0 {`
`138`	`139`	`return`
`139`	`140`	`}`
`140`	`141`
`141`	`142`	`q.notify()`
	`143`	`+ q.base.metricsScope.AddCounter(metrics.NewHistoryTaskCounter, int64(numTasks))`
`142`	`144`	`}`
`143`	`145`
`144`	`146`	`func (q *immediateQueue) notify() {`
Original file line number	Diff line number	Diff line change
`@@ -138,19 +138,21 @@ func (q *scheduledQueue) UnlockTaskProcessing() {`
`138`	`138`	`}`
`139`	`139`
`140`	`140`	`func (q scheduledQueue) NotifyNewTask(clusterName string, info hcommon.NotifyTaskInfo) {`
`141`		`- if len(info.Tasks) == 0 {`
	`141`	`+ numTasks := len(info.Tasks)`
	`142`	`+ if numTasks == 0 {`
`142`	`143`	`return`
`143`	`144`	`}`
`144`	`145`
`145`	`146`	`nextTime := info.Tasks[0].GetVisibilityTimestamp()`
`146`		`- for i := 1; i < len(info.Tasks); i++ {`
	`147`	`+ for i := 1; i < numTasks; i++ {`
`147`	`148`	`ts := info.Tasks[i].GetVisibilityTimestamp()`
`148`	`149`	`if ts.Before(nextTime) {`
`149`	`150`	`nextTime = ts`
`150`	`151`	`}`
`151`	`152`	`}`
`152`	`153`
`153`	`154`	`q.notify(nextTime)`
	`155`	`+ q.base.metricsScope.AddCounter(metrics.NewHistoryTaskCounter, int64(numTasks))`
`154`	`156`	`}`
`155`	`157`
`156`	`158`	`func (q *scheduledQueue) notify(t time.Time) {`