Use newPipeline instead of direct construction (#4285)

* Use newPipeline instead of direct construction

The newPipeline function exists to create a new pipeline, but it is not
currently used. This switches to using that as the default creation
method and removes the test for construction of a pipeline directly.

* Update sdk/metric/pipeline.go

sdk/metric/pipeline.go

@@ -57,10 +57,10 @@ func newPipeline(res *resource.Resource, reader Reader, views []View) *pipeline
 		res = resource.Empty()
 	}
 	return &pipeline{
-		resource:     res,
-		reader:       reader,
-		views:        views,
-		aggregations: make(map[instrumentation.Scope][]instrumentSync),
+		resource: res,
+		reader:   reader,
+		views:    views,
+		// aggregations is lazy allocated when needed.
 	}
 }
 
@@ -486,11 +486,7 @@ type pipelines []*pipeline
 func newPipelines(res *resource.Resource, readers []Reader, views []View) pipelines {
 	pipes := make([]*pipeline, 0, len(readers))
 	for _, r := range readers {
-		p := &pipeline{
-			resource: res,
-			reader:   r,
-			views:    views,
-		}
+		p := newPipeline(res, r, views)
 		r.register(p)
 		pipes = append(pipes, p)
 	}

sdk/metric/pipeline_test.go

@@ -39,31 +39,6 @@ func (testSumAggregator) Aggregation() metricdata.Aggregation {
 		DataPoints:  []metricdata.DataPoint[int64]{}}
 }
 
-func TestEmptyPipeline(t *testing.T) {
-	pipe := &pipeline{}
-
-	output := metricdata.ResourceMetrics{}
-	err := pipe.produce(context.Background(), &output)
-	require.NoError(t, err)
-	assert.Nil(t, output.Resource)
-	assert.Len(t, output.ScopeMetrics, 0)
-
-	iSync := instrumentSync{"name", "desc", "1", testSumAggregator{}}
-	assert.NotPanics(t, func() {
-		pipe.addSync(instrumentation.Scope{}, iSync)
-	})
-
-	require.NotPanics(t, func() {
-		pipe.addMultiCallback(func(context.Context) error { return nil })
-	})
-
-	err = pipe.produce(context.Background(), &output)
-	require.NoError(t, err)
-	assert.Nil(t, output.Resource)
-	require.Len(t, output.ScopeMetrics, 1)
-	require.Len(t, output.ScopeMetrics[0].Metrics, 1)
-}
-
 func TestNewPipeline(t *testing.T) {
 	pipe := newPipeline(nil, nil, nil)