Add shard id to queue processor related metrics

cadence-workflow · Dec 28, 2023 · 4ccb876 · 4ccb876
1 parent d663ce7
commit 4ccb876
Show file tree

Hide file tree

Showing 6 changed files with 20 additions and 12 deletions.
diff --git a/service/history/queue/processor_base.go b/service/history/queue/processor_base.go
@@ -23,6 +23,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
@@ -94,7 +95,7 @@ func newProcessorBase(
 	logger log.Logger,
 	metricsClient metrics.Client,
 ) *processorBase {
-	metricsScope := metricsClient.Scope(options.MetricScope)
+	metricsScope := metricsClient.Scope(options.MetricScope).Tagged(metrics.ShardIDTag(strconv.Itoa(shard.GetShardID())))
 	return &processorBase{
 		shard:         shard,
 		taskProcessor: taskProcessor,

diff --git a/service/history/queue/timer_queue_processor.go b/service/history/queue/timer_queue_processor.go
@@ -23,6 +23,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -425,12 +426,15 @@ func (t *timerQueueProcessor) completeTimer() error {
 	}
 
 	newAckLevelTimestamp := newAckLevel.(timerTaskKey).visibilityTimestamp
-	t.logger.Debugf("Start completing timer task from: %v, to %v", t.ackLevel, newAckLevelTimestamp)
 	if !t.ackLevel.Before(newAckLevelTimestamp) {
+		t.logger.Debugf("Skipping timer task completion because new ack level %v is not before ack level %v", newAckLevelTimestamp, t.ackLevel)
 		return nil
 	}
 
-	t.metricsClient.IncCounter(metrics.TimerQueueProcessorScope, metrics.TaskBatchCompleteCounter)
+	t.logger.Debugf("Start completing timer task from: %v, to %v", t.ackLevel, newAckLevelTimestamp)
+	t.metricsClient.Scope(metrics.TimerQueueProcessorScope).
+		Tagged(metrics.ShardIDTag(strconv.Itoa(t.shard.GetShardID()))).
+		IncCounter(metrics.TaskBatchCompleteCounter)
 
 	for {
 		pageSize := t.config.TimerTaskDeleteBatchSize()

diff --git a/service/history/queue/timer_queue_processor_base.go b/service/history/queue/timer_queue_processor_base.go
@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -542,6 +543,7 @@ func (t *timerQueueProcessorBase) notifyNewTimers(timerTasks []persistence.Task)
 
 	isActive := t.options.MetricScope == metrics.TimerActiveQueueProcessorScope
 	minNewTime := timerTasks[0].GetVisibilityTimestamp()
+	shardIDTag := metrics.ShardIDTag(strconv.Itoa(t.shard.GetShardID()))
 	for _, timerTask := range timerTasks {
 		ts := timerTask.GetVisibilityTimestamp()
 		if ts.Before(minNewTime) {
@@ -552,7 +554,7 @@ func (t *timerQueueProcessorBase) notifyNewTimers(timerTasks []persistence.Task)
 			timerTask.GetType(),
 			isActive,
 		)
-		t.metricsClient.IncCounter(taskScopeIdx, metrics.NewTimerCounter)
+		t.metricsClient.Scope(taskScopeIdx).Tagged(shardIDTag).IncCounter(metrics.NewTimerNotifyCounter)
 	}
 
 	t.notifyNewTimer(minNewTime)

diff --git a/service/history/queue/transfer_queue_processor.go b/service/history/queue/transfer_queue_processor.go
@@ -25,6 +25,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -423,7 +424,9 @@ func (t *transferQueueProcessor) completeTransfer() error {
 		return nil
 	}
 
-	t.metricsClient.IncCounter(metrics.TransferQueueProcessorScope, metrics.TaskBatchCompleteCounter)
+	t.metricsClient.Scope(metrics.TransferQueueProcessorScope).
+		Tagged(metrics.ShardIDTag(strconv.Itoa(t.shard.GetShardID()))).
+		IncCounter(metrics.TaskBatchCompleteCounter)
 
 	for {
 		pageSize := t.config.TransferTaskDeleteBatchSize()

diff --git a/service/history/queue/transfer_queue_processor_base.go b/service/history/queue/transfer_queue_processor_base.go
@@ -46,7 +46,6 @@ const (
 
 var (
 	loadQueueTaskThrottleRetryDelay = 5 * time.Second
-
 	persistenceOperationRetryPolicy = common.CreatePersistenceRetryPolicy()
 )
 
@@ -145,7 +144,7 @@ func newTransferQueueProcessorBase(
 			transferQueueProcessorBase,
 			options.ValidationInterval,
 			logger,
-			metricsClient.Scope(options.MetricScope),
+			processorBase.metricsScope,
 		)
 	}
 

diff --git a/service/history/queue/transfer_queue_validator.go b/service/history/queue/transfer_queue_validator.go
@@ -71,11 +71,10 @@ func newTransferQueueValidator(
 ) *transferQueueValidator {
 	timeSource := processor.shard.GetTimeSource()
 	return &transferQueueValidator{
-		processor:    processor,
-		timeSource:   timeSource,
-		logger:       logger,
-		metricsScope: metricsScope,
-
+		processor:          processor,
+		timeSource:         timeSource,
+		logger:             logger,
+		metricsScope:       metricsScope,
 		pendingTaskInfos:   make(map[int64]pendingTaskInfo),
 		maxReadLevels:      make(map[int]task.Key),
 		minReadTaskID:      0,