Update OpenCensus metric bridge to use the metric.Producer interface (#…

…3541) * update OpenCensus metric bridge to use the metric.Producer interface * return nil for empty metrics * fix nits Co-authored-by: Tyler Yahn <[email protected]>
open-telemetry · Dec 19, 2022 · a724cf8 · a724cf8
1 parent 0617172
commit a724cf8
Show file tree

Hide file tree

Showing 4 changed files with 173 additions and 16 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -21,8 +21,14 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Changed
 
 - Global error handler uses an atomic value instead of a mutex. (#3543)
+- Add `Producer` interface and `Reader.RegisterProducer(Producer)` to `go.opentelemetry.io/otel/sdk/metric` to enable external metric Producers. (#3524)
+- Add `NewMetricProducer` to `go.opentelemetry.io/otel/bridge/opencensus`, which can be used to pass OpenCensus metrics to an OpenTelemetry Reader. (#3541)
 - Global logger uses an atomic value instead of a mutex. (#3545)
 
+### Deprecated
+
+- The `NewMetricExporter` in `go.opentelemetry.io/otel/bridge/opencensus` is deprecated.  Use `NewMetricProducer` instead. (#3541)
+
 ## [1.11.2/0.34.0] 2022-12-05
 
 ### Added

diff --git a/bridge/opencensus/metric.go b/bridge/opencensus/metric.go
@@ -22,6 +22,7 @@ import (
 
 	ocmetricdata "go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/metric/metricexport"
+	"go.opencensus.io/metric/metricproducer"
 
 	"go.opentelemetry.io/otel"
 	internal "go.opentelemetry.io/otel/bridge/opencensus/internal/ocmetric"
@@ -33,6 +34,36 @@ import (
 
 const scopeName = "go.opentelemetry.io/otel/bridge/opencensus"
 
+type producer struct {
+	manager *metricproducer.Manager
+}
+
+// NewMetricProducer returns a metric.Producer that fetches metrics from
+// OpenCensus.
+func NewMetricProducer() metric.Producer {
+	return &producer{
+		manager: metricproducer.GlobalManager(),
+	}
+}
+
+func (p *producer) Produce(context.Context) ([]metricdata.ScopeMetrics, error) {
+	producers := p.manager.GetAll()
+	data := []*ocmetricdata.Metric{}
+	for _, ocProducer := range producers {
+		data = append(data, ocProducer.Read()...)
+	}
+	otelmetrics, err := internal.ConvertMetrics(data)
+	if len(otelmetrics) == 0 {
+		return nil, err
+	}
+	return []metricdata.ScopeMetrics{{
+		Scope: instrumentation.Scope{
+			Name: scopeName,
+		},
+		Metrics: otelmetrics,
+	}}, err
+}
+
 // exporter implements the OpenCensus metric Exporter interface using an
 // OpenTelemetry base exporter.
 type exporter struct {
@@ -42,6 +73,7 @@ type exporter struct {
 
 // NewMetricExporter returns an OpenCensus exporter that exports to an
 // OpenTelemetry (push) exporter.
+// Deprecated: Use NewMetricProducer instead.
 func NewMetricExporter(base metric.Exporter, res *resource.Resource) metricexport.Exporter {
 	return &exporter{base: base, res: res}
 }

diff --git a/bridge/opencensus/metric_test.go b/bridge/opencensus/metric_test.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	ocmetricdata "go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/metric/metricproducer"
 	ocresource "go.opencensus.io/resource"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -32,6 +33,134 @@ import (
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
+func TestMetricProducer(t *testing.T) {
+	now := time.Now()
+	for _, tc := range []struct {
+		desc      string
+		input     []*ocmetricdata.Metric
+		expected  []metricdata.ScopeMetrics
+		expectErr bool
+	}{
+		{
+			desc:     "empty",
+			expected: nil,
+		},
+		{
+			desc: "success",
+			input: []*ocmetricdata.Metric{
+				{
+					Resource: &ocresource.Resource{
+						Labels: map[string]string{
+							"R1": "V1",
+							"R2": "V2",
+						},
+					},
+					TimeSeries: []*ocmetricdata.TimeSeries{
+						{
+							StartTime: now,
+							Points: []ocmetricdata.Point{
+								{Value: int64(123), Time: now},
+							},
+						},
+					},
+				},
+			},
+			expected: []metricdata.ScopeMetrics{{
+				Scope: instrumentation.Scope{
+					Name: scopeName,
+				},
+				Metrics: []metricdata.Metrics{
+					{
+						Data: metricdata.Gauge[int64]{
+							DataPoints: []metricdata.DataPoint[int64]{
+								{
+									Attributes: attribute.NewSet(),
+									StartTime:  now,
+									Time:       now,
+									Value:      123,
+								},
+							},
+						},
+					},
+				},
+			}},
+		},
+		{
+			desc: "partial success",
+			input: []*ocmetricdata.Metric{
+				{
+					Descriptor: ocmetricdata.Descriptor{
+						Name:        "foo.com/bad-point",
+						Description: "a bad type",
+						Unit:        ocmetricdata.UnitDimensionless,
+						Type:        ocmetricdata.TypeGaugeDistribution,
+					},
+				},
+				{
+					Resource: &ocresource.Resource{
+						Labels: map[string]string{
+							"R1": "V1",
+							"R2": "V2",
+						},
+					},
+					TimeSeries: []*ocmetricdata.TimeSeries{
+						{
+							StartTime: now,
+							Points: []ocmetricdata.Point{
+								{Value: int64(123), Time: now},
+							},
+						},
+					},
+				},
+			},
+			expected: []metricdata.ScopeMetrics{{
+				Scope: instrumentation.Scope{
+					Name: scopeName,
+				},
+				Metrics: []metricdata.Metrics{
+					{
+						Data: metricdata.Gauge[int64]{
+							DataPoints: []metricdata.DataPoint[int64]{
+								{
+									Attributes: attribute.NewSet(),
+									StartTime:  now,
+									Time:       now,
+									Value:      123,
+								},
+							},
+						},
+					},
+				},
+			}},
+			expectErr: true,
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			fakeProducer := &fakeOCProducer{metrics: tc.input}
+			metricproducer.GlobalManager().AddProducer(fakeProducer)
+			defer metricproducer.GlobalManager().DeleteProducer(fakeProducer)
+			output, err := NewMetricProducer().Produce(context.Background())
+			if tc.expectErr {
+				require.Error(t, err)
+			} else {
+				require.Nil(t, err)
+			}
+			require.Equal(t, len(output), len(tc.expected))
+			for i := range output {
+				metricdatatest.AssertEqual(t, tc.expected[i], output[i])
+			}
+		})
+	}
+}
+
+type fakeOCProducer struct {
+	metrics []*ocmetricdata.Metric
+}
+
+func (f *fakeOCProducer) Read() []*ocmetricdata.Metric {
+	return f.metrics
+}
+
 func TestPushMetricsExporter(t *testing.T) {
 	now := time.Now()
 	for _, tc := range []struct {

diff --git a/example/opencensus/main.go b/example/opencensus/main.go
@@ -22,7 +22,6 @@ import (
 
 	ocmetric "go.opencensus.io/metric"
 	"go.opencensus.io/metric/metricdata"
-	"go.opencensus.io/metric/metricexport"
 	"go.opencensus.io/metric/metricproducer"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
@@ -34,7 +33,6 @@ import (
 	"go.opentelemetry.io/otel/exporters/stdout/stdoutmetric"
 	"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
 	"go.opentelemetry.io/otel/sdk/metric"
-	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 )
 
@@ -103,20 +101,12 @@ func tracing(otExporter sdktrace.SpanExporter) {
 // monitoring demonstrates creating an IntervalReader using the OpenTelemetry
 // exporter to send metrics to the exporter by using either an OpenCensus
 // registry or an OpenCensus view.
-func monitoring(otExporter metric.Exporter) error {
-	log.Println("Using the OpenTelemetry stdoutmetric exporter to export OpenCensus metrics.  This allows routing telemetry from both OpenTelemetry and OpenCensus to a single exporter.")
-	ocExporter := opencensus.NewMetricExporter(otExporter, resource.Default())
-	intervalReader, err := metricexport.NewIntervalReader(&metricexport.Reader{}, ocExporter)
-	if err != nil {
-		return fmt.Errorf("failed to create interval reader: %w", err)
-	}
-	intervalReader.ReportingInterval = 10 * time.Second
-	log.Println("Emitting metrics using OpenCensus APIs.  These should be printed out using the OpenTelemetry stdoutmetric exporter.")
-	err = intervalReader.Start()
-	if err != nil {
-		return fmt.Errorf("failed to start interval reader: %w", err)
-	}
-	defer intervalReader.Stop()
+func monitoring(exporter metric.Exporter) error {
+	log.Println("Adding the OpenCensus metric Producer to an OpenTelemetry Reader to export OpenCensus metrics using the OpenTelemetry stdout exporter.")
+	reader := metric.NewPeriodicReader(exporter)
+	// Register the OpenCensus metric Producer to add metrics from OpenCensus to the output.
+	reader.RegisterProducer(opencensus.NewMetricProducer())
+	metric.NewMeterProvider(metric.WithReader(reader))
 
 	log.Println("Registering a gauge metric using an OpenCensus registry.")
 	r := ocmetric.NewRegistry()