Add service type tag to metrics (#2545)

temporalio · Feb 25, 2022 · 9964786 · 9964786
1 parent f92cc93
commit 9964786
Show file tree

Hide file tree

Showing 23 changed files with 155 additions and 43 deletions.
diff --git a/common/archiver/s3store/historyArchiver_test.go b/common/archiver/s3store/historyArchiver_test.go
@@ -103,9 +103,11 @@ func (s *historyArchiverSuite) TearDownSuite() {
 func (s *historyArchiverSuite) SetupTest() {
 	scope := tally.NewTestScope("test", nil)
 	s.Assertions = require.New(s.T())
+	metricsClient, err := metrics.NewClient(&metrics.ClientConfig{}, scope, metrics.History)
+	s.Require().NoError(err, "metrics.NewClient")
 	s.container = &archiver.HistoryBootstrapContainer{
 		Logger:        log.NewNoopLogger(),
-		MetricsClient: metrics.NewClient(&metrics.ClientConfig{}, scope, metrics.HistoryArchiverScope),
+		MetricsClient: metricsClient,
 	}
 
 	s.controller = gomock.NewController(s.T())

diff --git a/common/archiver/s3store/visibilityArchiver_test.go b/common/archiver/s3store/visibilityArchiver_test.go
@@ -137,10 +137,11 @@ func (s *visibilityArchiverSuite) SetupSuite() {
 
 	s.testArchivalURI, err = archiver.NewURI(testBucketURI)
 	s.Require().NoError(err)
-
+	metricsClient, err := metrics.NewClient(&metrics.ClientConfig{}, scope, metrics.History)
+	s.Require().NoError(err, "metrics.NewClient")
 	s.container = &archiver.VisibilityBootstrapContainer{
 		Logger:        log.NewNoopLogger(),
-		MetricsClient: metrics.NewClient(&metrics.ClientConfig{}, scope, metrics.VisibilityArchiverScope),
+		MetricsClient: metricsClient,
 	}
 }
 

diff --git a/common/metrics/common.go b/common/metrics/common.go
@@ -67,8 +67,32 @@ func GetMetricsServiceIdx(serviceName string, logger log.Logger) ServiceIdx {
 		return Matching
 	case primitives.WorkerService:
 		return Worker
+	case primitives.ServerService:
+		return Server
+	case primitives.UnitTestService:
+		return UnitTestService
 	default:
 		logger.Fatal("Unknown service name for metrics!", tag.Service(serviceName))
 		panic(fmt.Sprintf("Unknown service name for metrics: %s", serviceName))
 	}
 }
+
+// GetMetricsServiceIdx returns service id corresponding to serviceName
+func MetricsServiceIdxToServiceName(serviceIdx ServiceIdx) (string, error) {
+	switch serviceIdx {
+	case Server:
+		return primitives.ServerService, nil
+	case Frontend:
+		return primitives.FrontendService, nil
+	case History:
+		return primitives.HistoryService, nil
+	case Matching:
+		return primitives.MatchingService, nil
+	case Worker:
+		return primitives.WorkerService, nil
+	case UnitTestService:
+		return primitives.UnitTestService, nil
+	default:
+		return "", fmt.Errorf(fmt.Sprintf("Unknown service idx for metrics: %d", serviceIdx))
+	}
+}
diff --git a/common/metrics/metric_client_tests.go b/common/metrics/metric_client_tests.go
@@ -29,6 +29,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
+
+	"go.temporal.io/server/common/primitives"
 )
 
 type (
@@ -73,7 +75,7 @@ func (s *MetricTestSuiteBase) TestClientReportCounter() {
 		TestScope1,
 		TestCounterMetric1, 66)
 	testDef := MetricDefs[UnitTestService][TestCounterMetric1]
-	assert.NoError(s.T(), s.metricTestUtility.ContainsCounter(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation}, 66))
+	assert.NoError(s.T(), s.metricTestUtility.ContainsCounter(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation, serviceName: primitives.UnitTestService}, 66))
 	assert.Equal(s.T(), 1, s.metricTestUtility.CollectionSize())
 }
 
@@ -82,7 +84,7 @@ func (s *MetricTestSuiteBase) TestClientReportGauge() {
 		TestScope1,
 		TestGaugeMetric1, 66)
 	testDef := MetricDefs[UnitTestService][TestGaugeMetric1]
-	assert.NoError(s.T(), s.metricTestUtility.ContainsGauge(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation}, 66))
+	assert.NoError(s.T(), s.metricTestUtility.ContainsGauge(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation, serviceName: primitives.UnitTestService}, 66))
 	assert.Equal(s.T(), 1, s.metricTestUtility.CollectionSize())
 }
 
@@ -92,7 +94,7 @@ func (s *MetricTestSuiteBase) TestClientReportTimer() {
 		TestScope1,
 		TestTimerMetric1, targetDuration)
 	testDef := MetricDefs[UnitTestService][TestTimerMetric1]
-	assert.NoError(s.T(), s.metricTestUtility.ContainsTimer(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation}, targetDuration))
+	assert.NoError(s.T(), s.metricTestUtility.ContainsTimer(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation, serviceName: primitives.UnitTestService}, targetDuration))
 	assert.Equal(s.T(), 1, s.metricTestUtility.CollectionSize())
 }
 
@@ -101,28 +103,28 @@ func (s *MetricTestSuiteBase) TestClientReportHistogram() {
 		TestScope1,
 		TestDimensionlessHistogramMetric1, 66)
 	testDef := MetricDefs[UnitTestService][TestDimensionlessHistogramMetric1]
-	assert.NoError(s.T(), s.metricTestUtility.ContainsHistogram(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation}, 66))
+	assert.NoError(s.T(), s.metricTestUtility.ContainsHistogram(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation, serviceName: primitives.UnitTestService}, 66))
 	assert.Equal(s.T(), 1, s.metricTestUtility.CollectionSize())
 }
 
 func (s *MetricTestSuiteBase) TestScopeReportCounter() {
 	s.testClient.Scope(TestScope1).AddCounter(TestCounterMetric1, 66)
 	testDef := MetricDefs[UnitTestService][TestCounterMetric1]
-	assert.NoError(s.T(), s.metricTestUtility.ContainsCounter(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation}, 66))
+	assert.NoError(s.T(), s.metricTestUtility.ContainsCounter(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation, serviceName: primitives.UnitTestService}, 66))
 	assert.Equal(s.T(), 1, s.metricTestUtility.CollectionSize())
 }
 
 func (s *MetricTestSuiteBase) TestScopeReportGauge() {
 	s.testClient.Scope(TestScope1).UpdateGauge(TestGaugeMetric1, 66)
 	testDef := MetricDefs[UnitTestService][TestGaugeMetric1]
-	assert.NoError(s.T(), s.metricTestUtility.ContainsGauge(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation}, 66))
+	assert.NoError(s.T(), s.metricTestUtility.ContainsGauge(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation, serviceName: primitives.UnitTestService}, 66))
 	assert.Equal(s.T(), 1, s.metricTestUtility.CollectionSize())
 }
 
 func (s *MetricTestSuiteBase) TestScopeReportTimer() {
 	targetDuration := time.Second * 100
 	s.testClient.Scope(TestScope1).RecordTimer(TestTimerMetric1, targetDuration)
 	testDef := MetricDefs[UnitTestService][TestTimerMetric1]
-	assert.NoError(s.T(), s.metricTestUtility.ContainsTimer(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation}, targetDuration))
+	assert.NoError(s.T(), s.metricTestUtility.ContainsTimer(testDef.metricName, map[string]string{namespace: namespaceAllValue, OperationTagName: ScopeDefs[UnitTestService][TestScope1].operation, serviceName: primitives.UnitTestService}, targetDuration))
 	assert.Equal(s.T(), 1, s.metricTestUtility.CollectionSize())
 }
diff --git a/common/metrics/opentelemetry_client.go b/common/metrics/opentelemetry_client.go
@@ -25,6 +25,7 @@
 package metrics
 
 import (
+	"fmt"
 	"time"
 
 	"go.temporal.io/server/common/log"
@@ -55,6 +56,17 @@ func NewOpentelemeteryClient(clientConfig *ClientConfig, serviceIdx ServiceIdx,
 
 	globalRootScope := newOpentelemetryScope(serviceIdx, reporter.GetMeterMust(), nil, clientConfig.Tags, getMetricDefs(serviceIdx), false, gaugeCache, false)
 
+	serviceTypeTagValue, err := MetricsServiceIdxToServiceName(serviceIdx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize metrics client: %w", err)
+	}
+
+	rootTags := make(map[string]string, len(clientConfig.Tags)+1)
+	for k, v := range clientConfig.Tags {
+		rootTags[k] = v
+	}
+	rootTags[serviceName] = serviceTypeTagValue
+
 	totalScopes := len(ScopeDefs[Common]) + len(ScopeDefs[serviceIdx])
 	metricsClient := &opentelemetryClient{
 		rootScope:    globalRootScope,

diff --git a/common/metrics/opentelemetry_reporter.go b/common/metrics/opentelemetry_reporter.go
@@ -93,7 +93,6 @@ func (r *opentelemetryReporterImpl) GetMeterMust() metric.MeterMust {
 }
 
 func (r *opentelemetryReporterImpl) NewClient(logger log.Logger, serviceIdx ServiceIdx) (Client, error) {
-
 	return NewOpentelemeteryClient(r.clientConfig, serviceIdx, r, logger, r.gaugeCache)
 }
 

diff --git a/common/metrics/tags.go b/common/metrics/tags.go
@@ -46,6 +46,7 @@ const (
 	workflowType  = "workflowType"
 	activityType  = "activityType"
 	commandType   = "commandType"
+	serviceName   = "service_name"
 
 	namespaceAllValue = "all"
 	unknownValue      = "_unknown_"
@@ -227,3 +228,7 @@ func HttpStatusTag(value int) Tag {
 func ResourceExhaustedCauseTag(cause enumspb.ResourceExhaustedCause) Tag {
 	return &tagImpl{key: resourceExhaustedTag, value: cause.String()}
 }
+
+func ServiceTypeTag(value string) Tag {
+	return &tagImpl{key: serviceName, value: value}
+}
diff --git a/common/metrics/tally_client.go b/common/metrics/tally_client.go
@@ -25,11 +25,14 @@
 package metrics
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/uber-go/tally/v4"
 )
 
+var _ Client = (*TallyClient)(nil)
+
 // TallyClient is used for reporting metrics by various Temporal services
 type TallyClient struct {
 	// This is the scope provided by user to the client. It contains no client-specific tags.
@@ -46,7 +49,7 @@ type TallyClient struct {
 // Client implementation
 // reporter holds the common tags for the service
 // serviceIdx indicates the service type in (InputhostIndex, ... StorageIndex)
-func NewClient(clientConfig *ClientConfig, scope tally.Scope, serviceIdx ServiceIdx) Client {
+func NewClient(clientConfig *ClientConfig, scope tally.Scope, serviceIdx ServiceIdx) (Client, error) {
 	tagsFilterConfig := NewTagFilteringScopeConfig(clientConfig.ExcludeTags)
 
 	perUnitBuckets := make(map[MetricUnit]tally.Buckets)
@@ -58,6 +61,13 @@ func NewClient(clientConfig *ClientConfig, scope tally.Scope, serviceIdx Service
 		return NewTagFilteringScope(tagsFilterConfig, impl)
 	}
 
+	serviceTypeTagValue, err := MetricsServiceIdxToServiceName(serviceIdx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize metrics client: %w", err)
+	}
+
+	rootScope := scope.Tagged(map[string]string{serviceName: serviceTypeTagValue})
+
 	totalScopes := len(ScopeDefs[Common]) + len(ScopeDefs[serviceIdx])
 	metricsClient := &TallyClient{
 		globalRootScope: scope,
@@ -75,7 +85,7 @@ func NewClient(clientConfig *ClientConfig, scope tally.Scope, serviceIdx Service
 			namespace:        namespaceAllValue,
 		}
 		mergeMapToRight(def.tags, scopeTags)
-		metricsClient.childScopes[idx] = scope.Tagged(scopeTags)
+		metricsClient.childScopes[idx] = rootScope.Tagged(scopeTags)
 	}
 
 	for idx, def := range ScopeDefs[serviceIdx] {
@@ -84,10 +94,10 @@ func NewClient(clientConfig *ClientConfig, scope tally.Scope, serviceIdx Service
 			namespace:        namespaceAllValue,
 		}
 		mergeMapToRight(def.tags, scopeTags)
-		metricsClient.childScopes[idx] = scope.Tagged(scopeTags)
+		metricsClient.childScopes[idx] = rootScope.Tagged(scopeTags)
 	}
 
-	return metricsClient
+	return metricsClient, nil
 }
 
 // IncCounter increments one for a counter and emits

diff --git a/common/metrics/tally_metric_test_utility.go b/common/metrics/tally_metric_test_utility.go
@@ -46,7 +46,11 @@ func NewTallyMetricTestUtility() *TallyMetricTestUtility {
 }
 
 func (t *TallyMetricTestUtility) GetClient(config *ClientConfig, idx ServiceIdx) Client {
-	return NewClient(config, t.scope, idx)
+	result, err := NewClient(config, t.scope, idx)
+	if err != nil {
+		panic(err)
+	}
+	return result
 }
 
 func (t *TallyMetricTestUtility) ContainsCounter(name MetricName, labels map[string]string, value int64) error {

diff --git a/common/metrics/tally_reporter.go b/common/metrics/tally_reporter.go
@@ -51,7 +51,7 @@ func NewTallyReporter(
 }
 
 func (tr *TallyReporter) NewClient(logger log.Logger, serviceIdx ServiceIdx) (Client, error) {
-	return NewClient(tr.clientConfig, tr.scope, serviceIdx), nil
+	return NewClient(tr.clientConfig, tr.scope, serviceIdx)
 }
 
 func (tr *TallyReporter) GetScope() tally.Scope {

diff --git a/common/persistence/persistence-tests/persistenceTestBase.go b/common/persistence/persistence-tests/persistenceTestBase.go
@@ -211,7 +211,9 @@ func (s *TestBase) Setup(clusterMetadataConfig *cluster.Config) {
 
 	cfg := s.DefaultTestCluster.Config()
 	scope := tally.NewTestScope(common.HistoryServiceName, make(map[string]string))
-	metricsClient := metrics.NewClient(&metrics.ClientConfig{}, scope, metrics.GetMetricsServiceIdx(common.HistoryServiceName, s.Logger))
+	var metricsClient metrics.Client
+	metricsClient, err = metrics.NewClient(&metrics.ClientConfig{}, scope, metrics.GetMetricsServiceIdx(common.HistoryServiceName, s.Logger))
+	s.fatalOnError("metrics.NewClient", err)
 	factory := client.NewFactoryImpl(&cfg, resolver.NewNoopResolver(), nil, serialization.NewSerializer(), s.AbstractDataStoreFactory, clusterName, metricsClient, s.Logger)
 
 	s.TaskMgr, err = factory.NewTaskManager()

diff --git a/common/primitives/role.go b/common/primitives/role.go
@@ -31,4 +31,6 @@ const (
 	HistoryService  = "history"
 	MatchingService = "matching"
 	WorkerService   = "worker"
+	ServerService   = "server"
+	UnitTestService = "unittest"
 )
diff --git a/common/resource/resourceTest.go b/common/resource/resourceTest.go
@@ -165,6 +165,11 @@ func NewTest(
 
 	scope := tally.NewTestScope("test", nil)
 
+	metricClient, err := metrics.NewClient(&metrics.ClientConfig{}, scope, serviceMetricsIndex)
+	if err != nil {
+		panic(err)
+	}
+
 	return &Test{
 		MetricsScope:             scope,
 		ClusterMetadata:          cluster.NewMockMetadata(controller),
@@ -177,7 +182,7 @@ func NewTest(
 		NamespaceCache:    namespace.NewMockRegistry(controller),
 		TimeSource:        clock.NewRealTimeSource(),
 		PayloadSerializer: serialization.NewSerializer(),
-		MetricsClient:     metrics.NewClient(&metrics.ClientConfig{}, scope, serviceMetricsIndex),
+		MetricsClient:     metricClient,
 		ArchivalMetadata:  archiver.NewMockArchivalMetadata(controller),
 		ArchiverProvider:  provider.NewMockArchiverProvider(controller),
 

diff --git a/common/tasks/benchmark_test.go b/common/tasks/benchmark_test.go
@@ -53,11 +53,14 @@ func BenchmarkInterleavedWeightedRoundRobinScheduler(b *testing.B) {
 		3: 1,
 	}
 	logger := log.NewTestLogger()
-	metricsClient := metrics.NewClient(
+	metricsClient, err := metrics.NewClient(
 		&metrics.ClientConfig{},
 		tally.NewTestScope("test", nil),
 		metrics.Common,
 	)
+	if err != nil {
+		panic(err)
+	}
 
 	scheduler := NewInterleavedWeightedRoundRobinScheduler(
 		InterleavedWeightedRoundRobinSchedulerOptions{

diff --git a/common/tasks/interleaved_weighted_round_robin_test.go b/common/tasks/interleaved_weighted_round_robin_test.go
@@ -76,11 +76,14 @@ func (s *interleavedWeightedRoundRobinSchedulerSuite) SetupTest() {
 		3: 1,
 	}
 	logger := log.NewTestLogger()
-	metricsClient := metrics.NewClient(
+	metricsClient, err := metrics.NewClient(
 		&metrics.ClientConfig{},
 		tally.NewTestScope("test", nil),
-		metrics.Common,
+		metrics.History,
 	)
+	if err != nil {
+		s.NoError(err, "metrics.NewClient")
+	}
 
 	s.scheduler = NewInterleavedWeightedRoundRobinScheduler(
 		InterleavedWeightedRoundRobinSchedulerOptions{

diff --git a/common/tasks/parallel_processor_test.go b/common/tasks/parallel_processor_test.go
@@ -62,12 +62,17 @@ func (s *parallelProcessorSuite) SetupTest() {
 
 	s.controller = gomock.NewController(s.T())
 
+	metricsClient, err := metrics.NewClient(&metrics.ClientConfig{}, tally.NoopScope, metrics.History)
+	if err != nil {
+		s.NoError(err, "metrics.NewClient")
+	}
+
 	s.processor = NewParallelProcessor(
 		&ParallelProcessorOptions{
 			QueueSize:   1,
 			WorkerCount: 1,
 		},
-		metrics.NewClient(&metrics.ClientConfig{}, tally.NoopScope, metrics.Common),
+		metricsClient,
 		log.NewNoopLogger(),
 	)
 	s.retryPolicy = backoff.NewExponentialRetryPolicy(time.Millisecond)