Set MeterProvider resource for all pipelines

Resolves open-telemetry#3208
MrAlias · Sep 21, 2022 · 2469880 · 2469880
1 parent 038248b
commit 2469880
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 8 deletions.
diff --git a/sdk/metric/meter_test.go b/sdk/metric/meter_test.go
@@ -31,6 +31,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/instrumentation"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+	"go.opentelemetry.io/otel/sdk/resource"
 )
 
 func TestMeterRegistry(t *testing.T) {
@@ -469,6 +470,7 @@ func TestMetersProvideScope(t *testing.T) {
 	assert.NoError(t, err)
 
 	want := metricdata.ResourceMetrics{
+		Resource: resource.Default(),
 		ScopeMetrics: []metricdata.ScopeMetrics{
 			{
 				Scope: instrumentation.Scope{

diff --git a/sdk/metric/pipeline.go b/sdk/metric/pipeline.go
@@ -165,10 +165,10 @@ type pipelineRegistry struct {
 	pipelines map[Reader]*pipeline
 }
 
-func newPipelineRegistries(views map[Reader][]view.View) *pipelineRegistry {
+func newPipelineRegistries(res *resource.Resource, views map[Reader][]view.View) *pipelineRegistry {
 	pipelines := map[Reader]*pipeline{}
 	for rdr := range views {
-		pipe := &pipeline{}
+		pipe := &pipeline{resource: res}
 		rdr.register(pipe)
 		pipelines[rdr] = pipe
 	}

diff --git a/sdk/metric/pipeline_registry_test.go b/sdk/metric/pipeline_registry_test.go
@@ -23,10 +23,12 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric/unit"
 	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/internal"
 	"go.opentelemetry.io/otel/sdk/metric/view"
+	"go.opentelemetry.io/otel/sdk/resource"
 )
 
 type invalidAggregation struct {
@@ -324,9 +326,9 @@ func TestPipelineRegistryCreateAggregators(t *testing.T) {
 
 	for _, tt := range testCases {
 		t.Run(tt.name, func(t *testing.T) {
-			reg := newPipelineRegistries(tt.views)
+			reg := newPipelineRegistries(resource.Empty(), tt.views)
 			testPipelineRegistryCreateIntAggregators(t, reg, tt.wantCount)
-			reg = newPipelineRegistries(tt.views)
+			reg = newPipelineRegistries(resource.Empty(), tt.views)
 			testPipelineRegistryCreateFloatAggregators(t, reg, tt.wantCount)
 		})
 	}
@@ -350,6 +352,19 @@ func testPipelineRegistryCreateFloatAggregators(t *testing.T, reg *pipelineRegis
 	require.Len(t, aggs, wantCount)
 }
 
+func TestPipelineRegistryResource(t *testing.T) {
+	v, err := view.New(view.MatchInstrumentName("bar"), view.WithRename("foo"))
+	require.NoError(t, err)
+	views := map[Reader][]view.View{
+		NewManualReader(): {{}, v},
+	}
+	res := resource.NewSchemaless(attribute.String("key", "val"))
+	reg := newPipelineRegistries(res, views)
+	for _, p := range reg.pipelines {
+		assert.True(t, res.Equal(p.resource), "resource not set")
+	}
+}
+
 func TestPipelineRegistryCreateAggregatorsIncompatibleInstrument(t *testing.T) {
 	testRdrHistogram := NewManualReader(WithAggregationSelector(func(ik view.InstrumentKind) aggregation.Aggregation { return aggregation.ExplicitBucketHistogram{} }))
 
@@ -358,14 +373,14 @@ func TestPipelineRegistryCreateAggregatorsIncompatibleInstrument(t *testing.T) {
 			{},
 		},
 	}
-	reg := newPipelineRegistries(views)
+	reg := newPipelineRegistries(resource.Empty(), views)
 	inst := view.Instrument{Name: "foo", Kind: view.AsyncGauge}
 
 	intAggs, err := createAggregators[int64](reg, inst, unit.Dimensionless)
 	assert.Error(t, err)
 	assert.Len(t, intAggs, 0)
 
-	reg = newPipelineRegistries(views)
+	reg = newPipelineRegistries(resource.Empty(), views)
 
 	floatAggs, err := createAggregators[float64](reg, inst, unit.Dimensionless)
 	assert.Error(t, err)
@@ -387,7 +402,7 @@ func TestPipelineRegistryCreateAggregatorsDuplicateErrors(t *testing.T) {
 	fooInst := view.Instrument{Name: "foo", Kind: view.SyncCounter}
 	barInst := view.Instrument{Name: "bar", Kind: view.SyncCounter}
 
-	reg := newPipelineRegistries(views)
+	reg := newPipelineRegistries(resource.Empty(), views)
 
 	intAggs, err := createAggregators[int64](reg, fooInst, unit.Dimensionless)
 	assert.NoError(t, err)

diff --git a/sdk/metric/provider.go b/sdk/metric/provider.go
@@ -51,7 +51,7 @@ func NewMeterProvider(options ...Option) *MeterProvider {
 
 	flush, sdown := conf.readerSignals()
 
-	registry := newPipelineRegistries(conf.readers)
+	registry := newPipelineRegistries(conf.res, conf.readers)
 
 	return &MeterProvider{
 		res: conf.res,