Merge branch 'main' into concurrent_collect_bugfix

sdk/metric/manual_reader.go

@@ -123,6 +123,7 @@ func (mr *ManualReader) Shutdown(context.Context) error {
 //
 // Collect will return an error if called after shutdown.
 // Collect will return an error if rm is a nil ResourceMetrics.
+// Collect will return an error if the context's Done channel is closed.
 //
 // This method is safe to call concurrently.
 func (mr *ManualReader) Collect(ctx context.Context, rm *metricdata.ResourceMetrics) error {

sdk/metric/manual_reader_test.go

@@ -15,11 +15,14 @@
 package metric // import "go.opentelemetry.io/otel/sdk/metric"
 
 import (
+ "context"
  "testing"
+ "time"
 
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/suite"
 
+ "go.opentelemetry.io/otel/metric"
  "go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
@@ -71,3 +74,50 @@ func TestManualReaderTemporality(t *testing.T) {
  })
  }
 }
+
+func TestManualReaderCollect(t *testing.T) {
+ expiredCtx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-1))
+ defer cancel()
+
+ tests := []struct {
+ name string
+
+ ctx context.Context
+ resourceMetrics *metricdata.ResourceMetrics
+
+ expectedErr error
+ }{
+ {
+ name: "with a valid context",
+
+ ctx: context.Background(),
+ resourceMetrics: &metricdata.ResourceMetrics{},
+ },
+ {
+ name: "with an expired context",
+
+ ctx: expiredCtx,
+ resourceMetrics: &metricdata.ResourceMetrics{},
+
+ expectedErr: context.DeadlineExceeded,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ rdr := NewManualReader()
+ mp := NewMeterProvider(WithReader(rdr))
+ meter := mp.Meter("test")
+
+ // Ensure the pipeline has a callback setup
+ testM, err := meter.Int64ObservableCounter("test")
+ assert.NoError(t, err)
+ _, err = meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+ return nil
+ }, testM)
+ assert.NoError(t, err)
+
+ assert.Equal(t, tt.expectedErr, rdr.Collect(tt.ctx, tt.resourceMetrics))
+ })
+ }
+}

sdk/metric/periodic_reader.go

@@ -239,8 +239,10 @@ func (r *PeriodicReader) collectAndExport(ctx context.Context) error {
 // data is not exported to the configured exporter, it is left to the caller to
 // handle that if desired.
 //
-// An error is returned if this is called after Shutdown, if rm is nil or if
-// the duration of the collect and export exceeded the timeout.
+// Collect will return an error if called after shutdown.
+// Collect will return an error if rm is a nil ResourceMetrics.
+// Collect will return an error if the context's Done channel is closed.
+//
 // This method is safe to call concurrently.
 func (r *PeriodicReader) Collect(ctx context.Context, rm *metricdata.ResourceMetrics) error {
  if rm == nil {

sdk/metric/periodic_reader_test.go

@@ -23,6 +23,7 @@ import (
  "github.com/stretchr/testify/suite"
 
  "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/metric"
  "go.opentelemetry.io/otel/sdk/metric/aggregation"
  "go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
@@ -377,3 +378,50 @@ func TestPeriodiclReaderTemporality(t *testing.T) {
  })
  }
 }
+
+func TestPeriodicReaderCollect(t *testing.T) {
+ expiredCtx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-1))
+ defer cancel()
+
+ tests := []struct {
+ name string
+
+ ctx context.Context
+ resourceMetrics *metricdata.ResourceMetrics
+
+ expectedErr error
+ }{
+ {
+ name: "with a valid context",
+
+ ctx: context.Background(),
+ resourceMetrics: &metricdata.ResourceMetrics{},
+ },
+ {
+ name: "with an expired context",
+
+ ctx: expiredCtx,
+ resourceMetrics: &metricdata.ResourceMetrics{},
+
+ expectedErr: context.DeadlineExceeded,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ rdr := NewPeriodicReader(new(fnExporter))
+ mp := NewMeterProvider(WithReader(rdr))
+ meter := mp.Meter("test")
+
+ // Ensure the pipeline has a callback setup
+ testM, err := meter.Int64ObservableCounter("test")
+ assert.NoError(t, err)
+ _, err = meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+ return nil
+ }, testM)
+ assert.NoError(t, err)
+
+ assert.Equal(t, tt.expectedErr, rdr.Collect(tt.ctx, tt.resourceMetrics))
+ })
+ }
+}

sdk/metric/reader.go

@@ -74,7 +74,8 @@ type Reader interface {
  // the SDK and stores it in out. An error is returned if this is called
  // after Shutdown or if out is nil.
  //
- // This method needs to be concurrent safe.
+ // This method needs to be concurrent safe, and the cancelation of the
+ // passed context is expected to be honored.
  Collect(ctx context.Context, rm *metricdata.ResourceMetrics) error
 
  // ForceFlush flushes all metric measurements held in an export pipeline.