Only registers callbacks if non-drop aggregation is used

CHANGELOG.md

@@ -26,6 +26,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Cumulative metrics from the OpenCensus bridge (`go.opentelemetry.io/otel/bridge/opencensus`) are defined as monotonic sums, instead of non-monotonic. (#3389)
 - Asynchronous counters (`Counter` and `UpDownCounter`) from the metric SDK now produce delta sums when configured with delta temporality. (#3398)
 - Exported `Status` codes in the `go.opentelemetry.io/otel/exporters/zipkin` exporter are now exported as all upper case values. (#3340)
+- Asynchronous callbacks are only called if they are registered with at least one instrument that does not use drop aggragation. (#3408)
 
 ## [1.11.1/0.33.0] 2022-10-19
 

sdk/metric/instrument_provider.go

@@ -15,8 +15,6 @@
 package metric // import "go.opentelemetry.io/otel/sdk/metric"
 
 import (
-	"fmt"
-
 	"go.opentelemetry.io/otel/metric/instrument"
 	"go.opentelemetry.io/otel/metric/instrument/asyncfloat64"
 	"go.opentelemetry.io/otel/metric/instrument/asyncint64"
@@ -43,13 +41,7 @@ func (p asyncInt64Provider) Counter(name string, opts ...instrument.Option) (asy
 		Description: cfg.Description(),
 		Kind:        view.AsyncCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-
-	return &instrumentImpl[int64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[int64]{aggregators: aggs}, err
 }
 
 // UpDownCounter creates an instrument for recording changes of a value.
@@ -62,12 +54,7 @@ func (p asyncInt64Provider) UpDownCounter(name string, opts ...instrument.Option
 		Description: cfg.Description(),
 		Kind:        view.AsyncUpDownCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[int64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[int64]{aggregators: aggs}, err
 }
 
 // Gauge creates an instrument for recording the current value.
@@ -80,12 +67,7 @@ func (p asyncInt64Provider) Gauge(name string, opts ...instrument.Option) (async
 		Description: cfg.Description(),
 		Kind:        view.AsyncGauge,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[int64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[int64]{aggregators: aggs}, err
 }
 
 type asyncFloat64Provider struct {
@@ -105,12 +87,7 @@ func (p asyncFloat64Provider) Counter(name string, opts ...instrument.Option) (a
 		Description: cfg.Description(),
 		Kind:        view.AsyncCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[float64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[float64]{aggregators: aggs}, err
 }
 
 // UpDownCounter creates an instrument for recording changes of a value.
@@ -123,12 +100,7 @@ func (p asyncFloat64Provider) UpDownCounter(name string, opts ...instrument.Opti
 		Description: cfg.Description(),
 		Kind:        view.AsyncUpDownCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[float64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[float64]{aggregators: aggs}, err
 }
 
 // Gauge creates an instrument for recording the current value.
@@ -141,12 +113,7 @@ func (p asyncFloat64Provider) Gauge(name string, opts ...instrument.Option) (asy
 		Description: cfg.Description(),
 		Kind:        view.AsyncGauge,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[float64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[float64]{aggregators: aggs}, err
 }
 
 type syncInt64Provider struct {
@@ -166,12 +133,7 @@ func (p syncInt64Provider) Counter(name string, opts ...instrument.Option) (sync
 		Description: cfg.Description(),
 		Kind:        view.SyncCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[int64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[int64]{aggregators: aggs}, err
 }
 
 // UpDownCounter creates an instrument for recording changes of a value.
@@ -184,12 +146,7 @@ func (p syncInt64Provider) UpDownCounter(name string, opts ...instrument.Option)
 		Description: cfg.Description(),
 		Kind:        view.SyncUpDownCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[int64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[int64]{aggregators: aggs}, err
 }
 
 // Histogram creates an instrument for recording the current value.
@@ -202,12 +159,7 @@ func (p syncInt64Provider) Histogram(name string, opts ...instrument.Option) (sy
 		Description: cfg.Description(),
 		Kind:        view.SyncHistogram,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[int64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[int64]{aggregators: aggs}, err
 }
 
 type syncFloat64Provider struct {
@@ -227,12 +179,7 @@ func (p syncFloat64Provider) Counter(name string, opts ...instrument.Option) (sy
 		Description: cfg.Description(),
 		Kind:        view.SyncCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[float64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[float64]{aggregators: aggs}, err
 }
 
 // UpDownCounter creates an instrument for recording changes of a value.
@@ -245,12 +192,7 @@ func (p syncFloat64Provider) UpDownCounter(name string, opts ...instrument.Optio
 		Description: cfg.Description(),
 		Kind:        view.SyncUpDownCounter,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[float64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[float64]{aggregators: aggs}, err
 }
 
 // Histogram creates an instrument for recording the current value.
@@ -263,10 +205,5 @@ func (p syncFloat64Provider) Histogram(name string, opts ...instrument.Option) (
 		Description: cfg.Description(),
 		Kind:        view.SyncHistogram,
 	}, cfg.Unit())
-	if len(aggs) == 0 && err != nil {
-		err = fmt.Errorf("instrument does not match any view: %w", err)
-	}
-	return &instrumentImpl[float64]{
-		aggregators: aggs,
-	}, err
+	return &instrumentImpl[float64]{aggregators: aggs}, err
 }

sdk/metric/meter.go

@@ -16,6 +16,7 @@ package metric // import "go.opentelemetry.io/otel/sdk/metric"
 
 import (
 	"context"
+	"errors"
 
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/instrument"
@@ -77,7 +78,23 @@ func (m *meter) AsyncFloat64() asyncfloat64.InstrumentProvider {
 // RegisterCallback registers the function f to be called when any of the
 // insts Collect method is called.
 func (m *meter) RegisterCallback(insts []instrument.Asynchronous, f func(context.Context)) error {
-	m.pipes.registerCallback(f)
+	for _, inst := range insts {
+		switch t := inst.(type) {
+		case *instrumentImpl[int64]:
+			if len(t.aggregators) > 0 {
+				m.pipes.registerCallback(f)
+				return nil
+			}
+		case *instrumentImpl[float64]:
+			if len(t.aggregators) > 0 {
+				m.pipes.registerCallback(f)
+				return nil
+			}
+		default:
+			return errors.New("invalid asynchronous instrument")
+		}
+	}
+	// Only instrument using drop aggregation passed.
 	return nil
 }
 

sdk/metric/meter_test.go

@@ -26,8 +26,10 @@ import (
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/instrument"
 	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+	"go.opentelemetry.io/otel/sdk/metric/view"
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
@@ -474,3 +476,60 @@ func TestMetersProvideScope(t *testing.T) {
 	assert.NoError(t, err)
 	metricdatatest.AssertEqual(t, want, got, metricdatatest.IgnoreTimestamp())
 }
+
+func TestRegisterCallbackDropAggregations(t *testing.T) {
+	aggFn := func(view.InstrumentKind) aggregation.Aggregation {
+		return aggregation.Drop{}
+	}
+	r := NewManualReader(WithAggregationSelector(aggFn))
+	mp := NewMeterProvider(WithReader(r))
+	m := mp.Meter("testRegisterCallbackDropAggregations")
+
+	int64Counter, err := m.AsyncInt64().Counter("int64.counter")
+	require.NoError(t, err)
+
+	int64UpDownCounter, err := m.AsyncInt64().UpDownCounter("int64.up_down_counter")
+	require.NoError(t, err)
+
+	int64Gauge, err := m.AsyncInt64().Gauge("int64.gauge")
+	require.NoError(t, err)
+
+	floag64Counter, err := m.AsyncFloat64().Counter("floag64.counter")
+	require.NoError(t, err)
+
+	floag64UpDownCounter, err := m.AsyncFloat64().UpDownCounter("floag64.up_down_counter")
+	require.NoError(t, err)
+
+	floag64Gauge, err := m.AsyncFloat64().Gauge("floag64.gauge")
+	require.NoError(t, err)
+
+	var called bool
+	require.NoError(t, m.RegisterCallback([]instrument.Asynchronous{
+		int64Counter,
+		int64UpDownCounter,
+		int64Gauge,
+		floag64Counter,
+		floag64UpDownCounter,
+		floag64Gauge,
+	}, func(context.Context) { called = true }))
+
+	data, err := r.Collect(context.Background())
+	require.NoError(t, err)
+
+	assert.False(t, called, "callback called for all drop instruments")
+	assert.Len(t, data.ScopeMetrics, 0, "metrics exported for drop instruments")
+}
+
+func TestRegisterCallbackErrorForNonSDKInstrument(t *testing.T) {
+	type alien struct{ instrument.Asynchronous }
+
+	r := NewManualReader()
+	mp := NewMeterProvider(WithReader(r))
+	m := mp.Meter("TestRegisterCallbackErrorForNonSDKInstrument")
+
+	err := m.RegisterCallback([]instrument.Asynchronous{&alien{}}, func(context.Context) {
+		panic("should not be registered")
+	})
+	assert.Error(t, err)
+	assert.NotPanics(t, func() { _, _ = r.Collect(context.Background()) })
+}

sdk/metric/pipeline.go

@@ -430,7 +430,6 @@ func newPipelines(res *resource.Resource, readers []Reader, views []view.View) p
 	return pipes
 }
 
-// TODO (#3053) Only register callbacks if any instrument matches in a view.
 func (p pipelines) registerCallback(fn func(context.Context)) {
 	for _, pipe := range p {
 		pipe.addCallback(fn)