Improve extensibility of optentelemetry metrics (#2521)

Author: Ardagan

common/metrics/config.go

@@ -267,7 +267,7 @@ func InitReporterFromPrometheusConfig(logger log.Logger, config *PrometheusConfi
 	case FrameworkTally:
 		return NewTallyReporterFromPrometheusConfig(logger, config, clientConfig), nil
 	case FrameworkOpentelemetry:
-		return NewOpentelemeteryReporter(logger, config, clientConfig)
+		return NewOpentelemeteryReporterWithMust(logger, config, clientConfig)
 	default:
 		err := fmt.Errorf("unsupported framework type specified in config: %q", config.Framework)
 		logger.Error(err.Error())

common/metrics/opentelemetry_client.go

@@ -46,7 +46,7 @@ type (
 
 // NewOpentelemeteryClientByReporter creates and returns a new instance of Client implementation
 // serviceIdx indicates the service type in (InputhostIndex, ... StorageIndex)
-func newOpentelemeteryClient(clientConfig *ClientConfig, serviceIdx ServiceIdx, reporter OpentelemetryReporter, logger log.Logger, gaugeCache OtelGaugeCache) (Client, error) {
+func NewOpentelemeteryClient(clientConfig *ClientConfig, serviceIdx ServiceIdx, reporter OpentelemetryReporter, logger log.Logger, gaugeCache OtelGaugeCache) (Client, error) {
 	tagsFilterConfig := NewTagFilteringScopeConfig(clientConfig.ExcludeTags)
 
 	scopeWrapper := func(impl internalScope) internalScope {

common/metrics/opentelemetry_must_provider.go

@@ -0,0 +1,137 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package metrics
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"go.opentelemetry.io/otel/exporters/prometheus"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
+	controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
+	processor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
+	"go.opentelemetry.io/otel/sdk/resource"
+
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
+)
+
+var _ OpentelemetryMustProvider = (*opentelemetryMustProviderImpl)(nil)
+
+type (
+	OpentelemetryMustProvider interface {
+		Stop(logger log.Logger)
+		GetMeterMust() metric.MeterMust
+	}
+
+	opentelemetryMustProviderImpl struct {
+		exporter  *prometheus.Exporter
+		meter     metric.Meter
+		meterMust metric.MeterMust
+		config    *PrometheusConfig
+		server    *http.Server
+	}
+)
+
+func NewOpentelemetryMustProvider(
+	logger log.Logger,
+	prometheusConfig *PrometheusConfig,
+	clientConfig *ClientConfig,
+) (*opentelemetryMustProviderImpl, error) {
+	histogramBoundaries := prometheusConfig.DefaultHistogramBoundaries
+	if len(histogramBoundaries) == 0 {
+		histogramBoundaries = defaultHistogramBoundaries
+	}
+
+	c := controller.New(
+		processor.NewFactory(
+			NewOtelAggregatorSelector(
+				histogramBoundaries,
+				clientConfig.PerUnitHistogramBoundaries,
+			),
+			aggregation.CumulativeTemporalitySelector(),
+			processor.WithMemory(true),
+		),
+		controller.WithResource(resource.Empty()),
+	)
+	exporter, err := prometheus.New(prometheus.Config{DefaultHistogramBoundaries: histogramBoundaries}, c)
+
+	if err != nil {
+		logger.Error("Failed to initialize prometheus exporter.", tag.Error(err))
+		return nil, err
+	}
+
+	metricServer := initPrometheusListener(prometheusConfig, logger, exporter)
+
+	meter := c.Meter("temporal")
+	meterMust := metric.Must(meter)
+	reporter := &opentelemetryMustProviderImpl{
+		exporter:  exporter,
+		meter:     meter,
+		meterMust: meterMust,
+		config:    prometheusConfig,
+		server:    metricServer,
+	}
+
+	return reporter, nil
+}
+
+func initPrometheusListener(config *PrometheusConfig, logger log.Logger, exporter *prometheus.Exporter) *http.Server {
+	handlerPath := config.HandlerPath
+	if handlerPath == "" {
+		handlerPath = "/metrics"
+	}
+
+	handler := http.NewServeMux()
+	handler.HandleFunc(handlerPath, exporter.ServeHTTP)
+
+	if config.ListenAddress == "" {
+		logger.Fatal("Listen address must be specified.", tag.Address(config.ListenAddress))
+	}
+	server := &http.Server{Addr: config.ListenAddress, Handler: handler}
+
+	go func() {
+		err := server.ListenAndServe()
+		if err != http.ErrServerClosed {
+			logger.Fatal("Failed to initialize prometheus listener.", tag.Address(config.ListenAddress))
+		}
+	}()
+
+	return server
+}
+
+func (r *opentelemetryMustProviderImpl) GetMeterMust() metric.MeterMust {
+	return r.meterMust
+}
+
+func (r *opentelemetryMustProviderImpl) Stop(logger log.Logger) {
+	ctx, closeCtx := context.WithTimeout(context.Background(), time.Second)
+	defer closeCtx()
+	if err := r.server.Shutdown(ctx); !(err == nil || err == http.ErrServerClosed) {
+		logger.Error("Prometheus metrics server shutdown failure.", tag.Address(r.config.ListenAddress), tag.Error(err))
+	}
+}

common/metrics/opentelemetry_reporter.go

@@ -25,23 +25,15 @@
 package metrics
 
 import (
-	"context"
-	"net/http"
-	"time"
-
 	"go.opentelemetry.io/otel/exporters/prometheus"
 	"go.opentelemetry.io/otel/metric"
-	"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
-	controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
-	processor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
-	"go.opentelemetry.io/otel/sdk/resource"
 
 	"go.temporal.io/server/common/log"
-	"go.temporal.io/server/common/log/tag"
 )
 
 var _ Reporter = (OpentelemetryReporter)(nil)
 var _ OpentelemetryReporter = (*opentelemetryReporterImpl)(nil)
+var _ OpentelemetryMustProvider = (*opentelemetryMustProviderImpl)(nil)
 
 type (
 	OpentelemetryReporter interface {
@@ -56,105 +48,57 @@ type (
 		exporter     *prometheus.Exporter
 		meter        metric.Meter
 		meterMust    metric.MeterMust
-		config       *PrometheusConfig
-		server       *http.Server
 		clientConfig *ClientConfig
 		gaugeCache   OtelGaugeCache
 		userScope    UserScope
+		mustProvider OpentelemetryMustProvider
 	}
 
 	OpentelemetryListener struct {
 	}
 )
 
-func NewOpentelemeteryReporter(
-	logger log.Logger,
+func NewOpentelemeteryReporterWithMust(
+	logger log.Logger, // keeping this to maintain API in case of adding more logging later
 	prometheusConfig *PrometheusConfig,
 	clientConfig *ClientConfig,
 ) (*opentelemetryReporterImpl, error) {
-	histogramBoundaries := prometheusConfig.DefaultHistogramBoundaries
-	if len(histogramBoundaries) == 0 {
-		histogramBoundaries = defaultHistogramBoundaries
-	}
-
-	c := controller.New(
-		processor.NewFactory(
-			NewOtelAggregatorSelector(
-				histogramBoundaries,
-				clientConfig.PerUnitHistogramBoundaries,
-			),
-			aggregation.CumulativeTemporalitySelector(),
-			processor.WithMemory(true),
-		),
-		controller.WithResource(resource.Empty()),
-	)
-	exporter, err := prometheus.New(
-		prometheus.Config{DefaultHistogramBoundaries: histogramBoundaries}, c)
-
+	mustProvider, err := NewOpentelemetryMustProvider(logger, prometheusConfig, clientConfig)
 	if err != nil {
-		logger.Error("Failed to initialize prometheus exporter.", tag.Error(err))
 		return nil, err
 	}
+	return NewOpentelemeteryReporter(logger, clientConfig, mustProvider)
+}
 
-	metricServer := initPrometheusListener(prometheusConfig, logger, exporter)
-
-	meter := c.Meter("temporal")
-	meterMust := metric.Must(meter)
+func NewOpentelemeteryReporter(
+	logger log.Logger, // keeping this to maintain API in case of adding more logging later
+	clientConfig *ClientConfig,
+	mustProvider OpentelemetryMustProvider,
+) (*opentelemetryReporterImpl, error) {
+	meterMust := mustProvider.GetMeterMust()
 	gaugeCache := NewOtelGaugeCache(meterMust)
-	userScope := newOpentelemetryUserScope(meterMust, clientConfig.Tags, gaugeCache)
+	userScope := NewOpentelemetryUserScope(meterMust, clientConfig.Tags, gaugeCache)
 	reporter := &opentelemetryReporterImpl{
-		exporter:     exporter,
-		meter:        meter,
-		meterMust:    meterMust,
-		config:       prometheusConfig,
-		server:       metricServer,
 		clientConfig: clientConfig,
 		gaugeCache:   gaugeCache,
 		userScope:    userScope,
+		mustProvider: mustProvider,
 	}
 
 	return reporter, nil
 }
 
-func initPrometheusListener(config *PrometheusConfig, logger log.Logger, exporter *prometheus.Exporter) *http.Server {
-	handlerPath := config.HandlerPath
-	if handlerPath == "" {
-		handlerPath = "/metrics"
-	}
-
-	handler := http.NewServeMux()
-	handler.HandleFunc(handlerPath, exporter.ServeHTTP)
-
-	if config.ListenAddress == "" {
-		logger.Fatal("Listen address must be specified.", tag.Address(config.ListenAddress))
-	}
-	server := &http.Server{Addr: config.ListenAddress, Handler: handler}
-
-	go func() {
-		err := server.ListenAndServe()
-		if err != http.ErrServerClosed {
-			logger.Fatal("Failed to initialize prometheus listener.", tag.Address(config.ListenAddress))
-		}
-	}()
-
-	return server
-}
-
 func (r *opentelemetryReporterImpl) GetMeterMust() metric.MeterMust {
-	return r.meterMust
+	return r.mustProvider.GetMeterMust()
 }
 
 func (r *opentelemetryReporterImpl) NewClient(logger log.Logger, serviceIdx ServiceIdx) (Client, error) {
 
-	return newOpentelemeteryClient(r.clientConfig, serviceIdx, r, logger, r.gaugeCache)
+	return NewOpentelemeteryClient(r.clientConfig, serviceIdx, r, logger, r.gaugeCache)
 }
 
 func (r *opentelemetryReporterImpl) Stop(logger log.Logger) {
-	ctx, closeCtx := context.WithTimeout(context.Background(), time.Second)
-	defer closeCtx()
-	if err := r.server.Shutdown(ctx); !(err == nil || err == http.ErrServerClosed) {
-		logger.Error("Prometheus metrics server shutdown failure.", tag.Address(r.config.ListenAddress), tag.Error(err))
-	}
+	r.mustProvider.Stop(logger)
 }
 
 func (r *opentelemetryReporterImpl) UserScope() UserScope {

common/metrics/opentelemetry_user_scope.go

@@ -40,7 +40,7 @@ type opentelemetryUserScope struct {
 	gaugeCache OtelGaugeCache
 }
 
-func newOpentelemetryUserScope(
+func NewOpentelemetryUserScope(
 	meterMust metric.MeterMust,
 	tags map[string]string,
 	gaugeCache OtelGaugeCache,
@@ -95,5 +95,5 @@ func (o opentelemetryUserScope) Tagged(tags map[string]string) UserScope {
 	for key, value := range tags {
 		tagMap[key] = value
 	}
-	return newOpentelemetryUserScope(o.meterMust, tagMap, o.gaugeCache)
+	return NewOpentelemetryUserScope(o.meterMust, tagMap, o.gaugeCache)
 }

common/metrics/otel_metric_test_utility.go

@@ -62,7 +62,7 @@ func NewOtelMetricTestUtility() *OtelMetricTestUtility {
 }
 
 func (t *OtelMetricTestUtility) GetClient(config *ClientConfig, idx ServiceIdx) Client {
-	result, err := newOpentelemeteryClient(config, idx, t.reporter, log.NewNoopLogger(), t.gaugeCache)
+	result, err := NewOpentelemeteryClient(config, idx, t.reporter, log.NewNoopLogger(), t.gaugeCache)
 	if err != nil {
 		panic(err)
 	}