Adding reporter to shard context for access to UserScope (#2961)

service/history/shard/context.go

@@ -62,6 +62,7 @@ type (
 		GetLogger() log.Logger
 		GetThrottledLogger() log.Logger
 		GetMetricsClient() metrics.Client
+		GetMetricsReporter() metrics.Reporter
 		GetTimeSource() clock.TimeSource
 
 		GetEngine() (Engine, error)

service/history/shard/context_impl.go

@@ -64,9 +64,7 @@ import (
 	"go.temporal.io/server/service/history/vclock"
 )
 
-var (
-	persistenceOperationRetryPolicy = common.CreatePersistenceRetryPolicy()
-)
+var persistenceOperationRetryPolicy = common.CreatePersistenceRetryPolicy()
 
 const (
 	// See transitionLocked for overview of state transitions.
@@ -90,6 +88,7 @@ type (
 		shardID             int32
 		executionManager    persistence.ExecutionManager
 		metricsClient       metrics.Client
+		metricsReporter     metrics.Reporter
 		eventsCache         events.Cache
 		closeCallback       func(*ContextImpl)
 		config              *configs.Config
@@ -439,7 +438,6 @@ func (s *ContextImpl) UpdateReplicatorDLQAckLevel(
 	sourceCluster string,
 	ackLevel int64,
 ) error {
-
 	s.wLock()
 	defer s.wUnlock()
 
@@ -1116,7 +1114,8 @@ func (s *ContextImpl) renewRangeLocked(isStealing bool) error {
 	defer cancel()
 	err := s.persistenceShardManager.UpdateShard(ctx, &persistence.UpdateShardRequest{
 		ShardInfo:       updatedShardInfo.ShardInfo,
-		PreviousRangeID: s.shardInfo.GetRangeId()})
+		PreviousRangeID: s.shardInfo.GetRangeId(),
+	})
 	if err != nil {
 		// Failure in updating shard to grab new RangeID
 		s.contextTaggedLogger.Error("Persistent store operation failure",
@@ -1818,6 +1817,7 @@ func newContext(
 	clientBean client.Bean,
 	historyClient historyservice.HistoryServiceClient,
 	metricsClient metrics.Client,
+	metricsReporter metrics.Reporter,
 	payloadSerializer serialization.Serializer,
 	timeSource clock.TimeSource,
 	namespaceRegistry namespace.Registry,
@@ -1827,7 +1827,6 @@ func newContext(
 	archivalMetadata archiver.ArchivalMetadata,
 	hostInfoProvider membership.HostInfoProvider,
 ) (*ContextImpl, error) {
-
 	hostIdentity := hostInfoProvider.HostInfo().Identity()
 
 	lifecycleCtx, lifecycleCancel := context.WithCancel(context.Background())
@@ -1837,6 +1836,7 @@ func newContext(
 		shardID:                 shardID,
 		executionManager:        persistenceExecutionManager,
 		metricsClient:           metricsClient,
+		metricsReporter:         metricsReporter,
 		closeCallback:           closeCallback,
 		config:                  config,
 		contextTaggedLogger:     log.With(logger, tag.ShardID(shardID), tag.Address(hostIdentity)),
@@ -1918,6 +1918,7 @@ func copyShardInfo(shardInfo *persistence.ShardInfoWithFailover) *persistence.Sh
 func (s *ContextImpl) GetRemoteAdminClient(cluster string) (adminservice.AdminServiceClient, error) {
 	return s.clientBean.GetRemoteAdminClient(cluster)
 }
+
 func (s *ContextImpl) GetPayloadSerializer() serialization.Serializer {
 	return s.payloadSerializer
 }
@@ -1930,6 +1931,10 @@ func (s *ContextImpl) GetMetricsClient() metrics.Client {
 	return s.metricsClient
 }
 
+func (s *ContextImpl) GetMetricsReporter() metrics.Reporter {
+	return s.metricsReporter
+}
+
 func (s *ContextImpl) GetTimeSource() clock.TimeSource {
 	return s.timeSource
 }
@@ -1941,9 +1946,11 @@ func (s *ContextImpl) GetNamespaceRegistry() namespace.Registry {
 func (s *ContextImpl) GetSearchAttributesProvider() searchattribute.Provider {
 	return s.saProvider
 }
+
 func (s *ContextImpl) GetSearchAttributesMapper() searchattribute.Mapper {
 	return s.saMapper
 }
+
 func (s *ContextImpl) GetClusterMetadata() cluster.Metadata {
 	return s.clusterMetadata
 }

service/history/shard/controller_impl.go

@@ -75,6 +75,7 @@ type (
 		historyClient               historyservice.HistoryServiceClient
 		historyServiceResolver      membership.ServiceResolver
 		metricsClient               metrics.Client
+		metricsReporter             metrics.Reporter
 		payloadSerializer           serialization.Serializer
 		timeSource                  clock.TimeSource
 		namespaceRegistry           namespace.Registry
@@ -242,6 +243,7 @@ func (c *ControllerImpl) getOrCreateShardContext(shardID int32) (*ContextImpl, e
 		c.clientBean,
 		c.historyClient,
 		c.metricsClient,
+		c.metricsReporter,
 		c.payloadSerializer,
 		c.timeSource,
 		c.namespaceRegistry,
@@ -290,14 +292,12 @@ func (c *ControllerImpl) removeShard(shardID int32, expected *ContextImpl) (*Con
 //   b. Periodic ticker
 //   c. ShardOwnershipLostError and subsequent ShardClosedEvents from engine
 func (c *ControllerImpl) shardManagementPump() {
-
 	defer c.shutdownWG.Done()
 
 	acquireTicker := time.NewTicker(c.config.AcquireShardInterval())
 	defer acquireTicker.Stop()
 
 	for {
-
 		select {
 		case <-c.shutdownCh:
 			c.doShutdown()

service/history/shard/fx.go

@@ -58,6 +58,7 @@ func ShardControllerProvider(
 	historyClient historyservice.HistoryServiceClient,
 	historyServiceResolver membership.ServiceResolver,
 	metricsClient metrics.Client,
+	metricsReporter metrics.Reporter,
 	payloadSerializer serialization.Serializer,
 	timeSource clock.TimeSource,
 	namespaceRegistry namespace.Registry,
@@ -84,6 +85,7 @@ func ShardControllerProvider(
 		historyClient:               historyClient,
 		historyServiceResolver:      historyServiceResolver,
 		metricsClient:               metricsClient,
+		metricsReporter:             metricsReporter,
 		payloadSerializer:           payloadSerializer,
 		timeSource:                  timeSource,
 		namespaceRegistry:           namespaceRegistry,