Fix bugs for history queue v2 (#7178)

Author: Shaddoll

common/log/tag/tags.go

@@ -1207,6 +1207,11 @@ func PendingTaskCount(count int) Tag {
 	return newInt("pending-task-count", count)
 }
 
+// MaxTaskCount returns a tag for max task count
+func MaxTaskCount(count int) Tag {
+	return newInt("max-task-count", count)
+}
+
 // VirtualQueueID returns a tag for virtual queue id
 func VirtualQueueID(id int64) Tag {
 	return newInt64("virtual-queue-id", id)

common/metrics/defs.go

@@ -2697,6 +2697,9 @@ const (
 	WorkflowIDCacheRequestsExternalMaxRequestsPerSecondsTimer
 	WorkflowIDCacheRequestsInternalMaxRequestsPerSecondsTimer
 	WorkflowIDCacheRequestsInternalRatelimitedCounter
+	VirtualQueueCountGauge
+	VirtualQueuePausedGauge
+	VirtualQueueRunningGauge
 	NumHistoryMetrics
 )
 
@@ -3465,6 +3468,9 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		WorkflowIDCacheRequestsExternalMaxRequestsPerSecondsTimer:    {metricName: "workflow_id_external_requests_max_requests_per_seconds", metricType: Timer},
 		WorkflowIDCacheRequestsInternalMaxRequestsPerSecondsTimer:    {metricName: "workflow_id_internal_requests_max_requests_per_seconds", metricType: Timer},
 		WorkflowIDCacheRequestsInternalRatelimitedCounter:            {metricName: "workflow_id_internal_requests_ratelimited", metricType: Counter},
+		VirtualQueueCountGauge:                                       {metricName: "virtual_queue_count", metricType: Gauge},
+		VirtualQueuePausedGauge:                                      {metricName: "virtual_queue_paused", metricType: Gauge},
+		VirtualQueueRunningGauge:                                     {metricName: "virtual_queue_running", metricType: Gauge},
 	},
 	Matching: {
 		PollSuccessPerTaskListCounter:                           {metricName: "poll_success_per_tl", metricRollupName: "poll_success"},

service/history/queuev2/mitigator.go

@@ -11,6 +11,7 @@ import (
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/types"
 )
 
 const (
@@ -53,10 +54,11 @@ func NewMitigator(
 	options *MitigatorOptions,
 ) Mitigator {
 	m := &mitigatorImpl{
-		monitor:      monitor,
-		logger:       logger,
-		metricsScope: metricsScope,
-		options:      options,
+		virtualQueueManager: virtualQueueManager,
+		monitor:             monitor,
+		logger:              logger,
+		metricsScope:        metricsScope,
+		options:             options,
 	}
 	m.handlers = map[AlertType]func(Alert){
 		AlertTypeQueuePendingTaskCount: m.handleQueuePendingTaskCount,
@@ -83,17 +85,49 @@ func (m *mitigatorImpl) handleQueuePendingTaskCount(alert Alert) {
 		virtualQueue.UpdateAndGetState()
 	}
 	if m.monitor.GetTotalPendingTaskCount() <= alert.AlertAttributesQueuePendingTaskCount.CriticalPendingTaskCount {
+		m.logger.Debug("mitigating queue alert, skip mitigation because the alert is no longer valid")
 		return
 	}
 	// Second, getting the stats of pending tasks. We need:
 	stats := m.collectPendingTaskStats()
 
 	// Third, find virtual slices to split given the target pending task count and the stats of pending tasks
 	targetPendingTaskCount := int(float64(alert.AlertAttributesQueuePendingTaskCount.CriticalPendingTaskCount) * targetLoadFactor)
+	if m.logger.DebugOn() {
+		sliceStatesPerDomain := make(map[string][]*types.VirtualSliceState)
+		for domain, slices := range stats.slicesPerDomain {
+			for _, s := range slices {
+				sliceStatesPerDomain[domain] = append(sliceStatesPerDomain[domain], ToPersistenceVirtualSliceState(s.GetState()))
+			}
+		}
+		for s, domainStats := range stats.pendingTaskCountPerDomainPerSlice {
+			m.logger.Debug("mitigating queue alert, get task stats per slice", tag.Dynamic("slice", ToPersistenceVirtualSliceState(s.GetState())), tag.Dynamic("domain-stats", domainStats))
+		}
+		m.logger.Debug("mitigating queue alert, get task stats",
+			tag.AlertType(int(alert.AlertType)),
+			tag.Dynamic("pending-task-count-per-domain", stats.pendingTaskCountPerDomain),
+			tag.Dynamic("slices-per-domain", sliceStatesPerDomain),
+			tag.Dynamic("pending-task-count", stats.totalPendingTaskCount),
+			tag.Dynamic("target-task-count", targetPendingTaskCount),
+		)
+	}
 	domainsToClearPerSlice := m.findDomainsToClear(stats, targetPendingTaskCount)
+	if m.logger.DebugOn() {
+		for s, domains := range domainsToClearPerSlice {
+			m.logger.Debug("mitigating queue alert, get domains to clear", tag.Dynamic("slice", ToPersistenceVirtualSliceState(s.GetState())), tag.WorkflowDomainIDs(domains))
+		}
+	}
 
 	// Finally, split and clear the slices
 	m.processQueueSplitsAndClear(virtualQueues, domainsToClearPerSlice)
+	if m.logger.DebugOn() {
+		virtualQueues := m.virtualQueueManager.VirtualQueues()
+		state := make(map[int64]*types.VirtualQueueState)
+		for queueID, vq := range virtualQueues {
+			state[queueID] = ToPersistenceVirtualQueueState(vq.GetState())
+		}
+		m.logger.Debug("mitigating queue alert, get queue state after mitigation", tag.Dynamic("queue-state", state))
+	}
 }
 
 // The stats of pending tasks are used to calculate the domains to clear. We need:

service/history/queuev2/queue_base.go

@@ -111,7 +111,7 @@ func newQueueBase(
 		logger.Fatal("Failed to get queue state, probably task category is not supported", tag.Error(err), tag.Dynamic("category", category))
 	}
 	queueState := FromPersistenceQueueState(persistenceQueueState)
-	exclusiveAckLevel := getExclusiveAckLevelFromQueueState(queueState)
+	exclusiveAckLevel, _ := getExclusiveAckLevelAndMaxQueueIDFromQueueState(queueState)
 
 	redispatcher := task.NewRedispatcher(
 		taskProcessor,
@@ -277,7 +277,8 @@ func (q *queueBase) updateQueueState(ctx context.Context) {
 		VirtualQueueStates:    q.virtualQueueManager.UpdateAndGetState(),
 		ExclusiveMaxReadLevel: q.newVirtualSliceState.Range.InclusiveMinTaskKey,
 	}
-	newExclusiveAckLevel := getExclusiveAckLevelFromQueueState(queueState)
+	newExclusiveAckLevel, maxQueueID := getExclusiveAckLevelAndMaxQueueIDFromQueueState(queueState)
+	q.metricsScope.UpdateGauge(metrics.VirtualQueueCountGauge, float64(maxQueueID+1))
 
 	// for backward compatibility, we record the timer metrics in shard info scope
 	pendingTaskCount := q.monitor.GetTotalPendingTaskCount()
@@ -318,7 +319,9 @@ func (q *queueBase) updateQueueState(ctx context.Context) {
 	}
 
 	// even though the ack level is not updated, we still need to update the queue state
-	err := q.shard.UpdateQueueState(q.category, ToPersistenceQueueState(queueState))
+	persistenceQueueState := ToPersistenceQueueState(queueState)
+	q.logger.Debug("store queue state", tag.Dynamic("queue-state", persistenceQueueState))
+	err := q.shard.UpdateQueueState(q.category, persistenceQueueState)
 	if err != nil {
 		q.logger.Error("Failed to update queue state", tag.Error(err))
 		q.metricsScope.IncCounter(metrics.AckLevelUpdateFailedCounter)
@@ -339,12 +342,14 @@ func (q *queueBase) handleAlert(ctx context.Context, alert *Alert) {
 	q.updateQueueStateFn(ctx)
 }
 
-func getExclusiveAckLevelFromQueueState(state *QueueState) persistence.HistoryTaskKey {
+func getExclusiveAckLevelAndMaxQueueIDFromQueueState(state *QueueState) (persistence.HistoryTaskKey, int64) {
+	maxQueueID := int64(0)
 	newExclusiveAckLevel := state.ExclusiveMaxReadLevel
-	for _, virtualQueueState := range state.VirtualQueueStates {
+	for queueID, virtualQueueState := range state.VirtualQueueStates {
 		if len(virtualQueueState) != 0 {
 			newExclusiveAckLevel = persistence.MinHistoryTaskKey(newExclusiveAckLevel, virtualQueueState[0].Range.InclusiveMinTaskKey)
 		}
+		maxQueueID = max(maxQueueID, queueID)
 	}
-	return newExclusiveAckLevel
+	return newExclusiveAckLevel, maxQueueID
 }

service/history/queuev2/virtual_queue.go

@@ -339,15 +339,21 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {
 	}
 
 	pendingTaskCount := q.monitor.GetTotalPendingTaskCount()
-	if pendingTaskCount > q.options.MaxPendingTasksCount() {
-		q.logger.Warn("Too many pending tasks, pause loading tasks for a while", tag.PendingTaskCount(pendingTaskCount))
+	maxTaskCount := q.options.MaxPendingTasksCount()
+	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.options.PollBackoffInterval())
 	}
 
 	if q.pauseController.IsPaused() {
+		// emit a metric indicating that the virtual queue is paused
+		q.metricsScope.UpdateGauge(metrics.VirtualQueuePausedGauge, 1.0)
+		q.logger.Debug("virtual queue is paused")
 		return
 	}
 
+	// emit a metric indicating that the virtual queue is alive
+	q.metricsScope.UpdateGauge(metrics.VirtualQueueRunningGauge, 1.0)
 	sliceToRead := q.sliceToRead.Value.(VirtualSlice)
 	tasks, err := sliceToRead.GetTasks(q.ctx, q.options.PageSize())
 	if err != nil {

service/history/queuev2/virtual_queue_manager.go

@@ -79,7 +79,7 @@ type (
 		sync.RWMutex
 		status               int32
 		virtualQueues        map[int64]VirtualQueue
-		createVirtualQueueFn func(VirtualSlice, int64) VirtualQueue
+		createVirtualQueueFn func(int64, ...VirtualSlice) VirtualQueue
 
 		nextForceNewSliceTime time.Time
 	}
@@ -127,14 +127,14 @@ func NewVirtualQueueManager(
 		nonRootQueueOptions: nonRootQueueOptions,
 		status:              common.DaemonStatusInitialized,
 		virtualQueues:       virtualQueues,
-		createVirtualQueueFn: func(s VirtualSlice, queueID int64) VirtualQueue {
+		createVirtualQueueFn: func(queueID int64, s ...VirtualSlice) VirtualQueue {
 			var options *VirtualQueueOptions
 			if queueID == rootQueueID {
 				options = rootQueueOptions
 			} else {
 				options = nonRootQueueOptions
 			}
-			return NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, []VirtualSlice{s}, options)
+			return NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, s, options)
 		},
 	}
 }
@@ -184,7 +184,8 @@ func (m *virtualQueueManagerImpl) GetOrCreateVirtualQueue(queueID int64) Virtual
 	if vq, ok := m.virtualQueues[queueID]; ok {
 		return vq
 	}
-	m.virtualQueues[queueID] = m.createVirtualQueueFn(nil, queueID)
+	m.virtualQueues[queueID] = m.createVirtualQueueFn(queueID)
+	m.virtualQueues[queueID].Start()
 	return m.virtualQueues[queueID]
 }
 
@@ -221,7 +222,7 @@ func (m *virtualQueueManagerImpl) AddNewVirtualSliceToRootQueue(s VirtualSlice)
 		return
 	}
 
-	m.virtualQueues[rootQueueID] = m.createVirtualQueueFn(s, rootQueueID)
+	m.virtualQueues[rootQueueID] = m.createVirtualQueueFn(rootQueueID, s)
 	m.virtualQueues[rootQueueID].Start()
 }
 

service/history/queuev2/virtual_queue_manager_test.go

@@ -571,7 +571,7 @@ func TestVirtualQueueManager_AddNewVirtualSlice(t *testing.T) {
 				},
 				status:        common.DaemonStatusInitialized,
 				virtualQueues: virtualQueues,
-				createVirtualQueueFn: func(s VirtualSlice, queueID int64) VirtualQueue {
+				createVirtualQueueFn: func(queueID int64, s ...VirtualSlice) VirtualQueue {
 					vq := NewMockVirtualQueue(ctrl)
 					vq.EXPECT().Start()
 					return vq