chore(replicationTaskFetcher): Add metrics to task_fetcher (#7462)

<!-- Describe what has changed in this PR -->
**What changed?**
Added metrics to task_fetcher

<!-- Tell your future self why have you made these changes -->
**Why?**
Have visibility over the latency of the taskFetcher relative to the
replication latency

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**


<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

Signed-off-by: fimanishi <[email protected]>

Author: fimanishi

common/metrics/defs.go

@@ -2742,6 +2742,8 @@ const (
 	ReplicationTaskCleanupFailure
 	ReplicationTaskLatency
 	ExponentialReplicationTaskLatency
+	ExponentialReplicationTaskFetchLatency
+	ReplicationTasksFetchedSize
 	MutableStateChecksumMismatch
 	MutableStateChecksumInvalidated
 	FailoverMarkerCount
@@ -3535,6 +3537,8 @@ var MetricDefs = map[ServiceIdx]map[MetricIdx]metricDefinition{
 		ReplicationTaskCleanupFailure:                                {metricName: "replication_task_cleanup_failed", metricType: Counter},
 		ReplicationTaskLatency:                                       {metricName: "replication_task_latency", metricType: Timer},
 		ExponentialReplicationTaskLatency:                            {metricName: "replication_task_latency_ns", metricType: Histogram, exponentialBuckets: Mid1ms24h},
+		ExponentialReplicationTaskFetchLatency:                       {metricName: "replication_task_fetch_latency_ns", metricType: Histogram, exponentialBuckets: Mid1ms24h},
+		ReplicationTasksFetchedSize:                                  {metricName: "replication_tasks_fetched_size", metricType: Gauge},
 		MutableStateChecksumMismatch:                                 {metricName: "mutable_state_checksum_mismatch", metricType: Counter},
 		MutableStateChecksumInvalidated:                              {metricName: "mutable_state_checksum_invalidated", metricType: Counter},
 		FailoverMarkerCount:                                          {metricName: "failover_marker_count", metricType: Counter},

service/history/handler/handler.go

@@ -121,6 +121,7 @@ func (h *handlerImpl) Start() {
 		h.config,
 		h.GetClusterMetadata(),
 		h.GetClientBean(),
+		h.GetMetricsClient(),
 	)
 	if err != nil {
 		h.GetLogger().Fatal("Creating replication task fetchers failed", tag.Error(err))

service/history/replication/task_fetcher.go

@@ -34,6 +34,7 @@ import (
 	"github.com/uber/cadence/common/cluster"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/common/metrics"
 	"github.com/uber/cadence/common/quotas"
 	"github.com/uber/cadence/common/types"
 	"github.com/uber/cadence/service/history/config"
@@ -71,6 +72,7 @@ type (
 		sourceCluster  string
 		config         *config.Config
 		logger         log.Logger
+		metricsScope   metrics.Scope
 		remotePeer     admin.Client
 		rateLimiter    quotas.Limiter
 		timeSource     clock.TimeSource
@@ -99,6 +101,7 @@ func NewTaskFetchers(
 	config *config.Config,
 	clusterMetadata cluster.Metadata,
 	clientBean client.Bean,
+	metricsClient metrics.Client,
 ) (TaskFetchers, error) {
 	currentCluster := clusterMetadata.GetCurrentClusterName()
 	var fetchers []TaskFetcher
@@ -113,6 +116,7 @@ func NewTaskFetchers(
 			currentCluster,
 			config,
 			remoteFrontendClient,
+			metricsClient,
 		)
 		fetchers = append(fetchers, fetcher)
 	}
@@ -160,12 +164,14 @@ func newReplicationTaskFetcher(
 	currentCluster string,
 	config *config.Config,
 	sourceFrontend admin.Client,
+	metricsClient metrics.Client,
 ) TaskFetcher {
 	ctx, cancel := context.WithCancel(context.Background())
 	fetcher := &taskFetcherImpl{
 		status:         common.DaemonStatusInitialized,
 		config:         config,
 		logger:         logger.WithTags(tag.ClusterName(sourceCluster)),
+		metricsScope:   metricsClient.Scope(metrics.ReplicationTaskFetcherScope, metrics.TargetClusterTag(sourceCluster)),
 		remotePeer:     sourceFrontend,
 		currentCluster: currentCluster,
 		sourceCluster:  sourceCluster,
@@ -210,7 +216,13 @@ func (f *taskFetcherImpl) Stop() {
 
 // fetchTasks collects getReplicationTasks request from shards and send out aggregated request to source frontend.
 func (f *taskFetcherImpl) fetchTasks() {
+	startTime := f.timeSource.Now()
 	defer f.wg.Done()
+	defer func() {
+		totalLatency := f.timeSource.Now().Sub(startTime)
+		f.metricsScope.ExponentialHistogram(metrics.ExponentialReplicationTaskFetchLatency, totalLatency)
+	}()
+
 	timer := f.timeSource.NewTimer(backoff.JitDuration(
 		f.config.ReplicationTaskFetcherAggregationInterval(),
 		f.config.ReplicationTaskFetcherTimerJitterCoefficient(),
@@ -273,6 +285,12 @@ func (f *taskFetcherImpl) fetchAndDistributeTasks(requestByShard map[int32]*requ
 		return err
 	}
 
+	totalTasks := 0
+	for _, messages := range messagesByShard {
+		totalTasks += len(messages.ReplicationTasks)
+	}
+	f.metricsScope.UpdateGauge(metrics.ReplicationTasksFetchedSize, float64(totalTasks))
+
 	f.logger.Debug("Successfully fetched replication tasks.", tag.Counter(len(messagesByShard)))
 	for shardID, tasks := range messagesByShard {
 		request := requestByShard[shardID]

service/history/replication/task_fetcher_test.go

@@ -84,6 +84,7 @@ func (s *taskFetcherSuite) SetupTest() {
 		"active",
 		s.config,
 		s.frontendClient,
+		metrics.NewNoopMetricsClient(),
 	).(*taskFetcherImpl)
 }
 
@@ -269,7 +270,7 @@ func TestTaskFetchers(t *testing.T) {
 	cfg := config.NewForTest()
 
 	mockBean.EXPECT().GetRemoteAdminClient(cluster.TestAlternativeClusterName).Return(mockAdminClient, nil)
-	fetchers, err := NewTaskFetchers(logger, cfg, cluster.TestActiveClusterMetadata, mockBean)
+	fetchers, err := NewTaskFetchers(logger, cfg, cluster.TestActiveClusterMetadata, mockBean, metrics.NewNoopMetricsClient())
 	assert.NoError(t, err)
 	assert.NotNil(t, fetchers)
 	assert.Len(t, fetchers.GetFetchers(), len(cluster.TestActiveClusterMetadata.GetRemoteClusterInfo()))