Skip to content

Commit 82882df

Browse files
MrAliashanyuancheung
MrAlias
and
hanyuancheung
authored
Have multi-instrument callback return an error (#3576)
* Have multi-inst callback return an error * Update PR number in changelog entry Co-authored-by: Chester Cheung <[email protected]>
1 parent 75a19d1 commit 82882df

File tree

10 files changed

+68
-38
lines changed

10 files changed

+68
-38
lines changed

‎CHANGELOG.md

+1
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
7777
- `InstrumentKindAsyncGauge` is renamed to `InstrumentKindObservableGauge`
7878
- Update the `RegisterCallback` method of the `Meter` in the `go.opentelemetry.io/otel/sdk/metric` package to accept the added `Callback` type instead of an inline function type definition.
7979
The underlying type of a `Callback` is the same `func(context.Context)` that the method used to accept. (#3564)
80+
- The callback function registered with a `Meter` from the `go.opentelemetry.io/otel/metric` package is required to return an error now. (#3576)
8081

8182
### Deprecated
8283

‎example/prometheus/main.go

+2-1
Original file line numberDiff line numberDiff line change
@@ -68,9 +68,10 @@ func main() {
6868
if err != nil {
6969
log.Fatal(err)
7070
}
71-
_, err = meter.RegisterCallback([]instrument.Asynchronous{gauge}, func(ctx context.Context) {
71+
_, err = meter.RegisterCallback([]instrument.Asynchronous{gauge}, func(ctx context.Context) error {
7272
n := -10. + rand.Float64()*(90.) // [-10, 100)
7373
gauge.Observe(ctx, n, attrs...)
74+
return nil
7475
})
7576
if err != nil {
7677
log.Fatal(err)

‎metric/example_test.go

+4-2
Original file line numberDiff line numberDiff line change
@@ -62,13 +62,14 @@ func ExampleMeter_asynchronous_single() {
6262
}
6363

6464
_, err = meter.RegisterCallback([]instrument.Asynchronous{memoryUsage},
65-
func(ctx context.Context) {
65+
func(ctx context.Context) error {
6666
// instrument.WithCallbackFunc(func(ctx context.Context) {
6767
//Do Work to get the real memoryUsage
6868
// mem := GatherMemory(ctx)
6969
mem := 75000
7070

7171
memoryUsage.Observe(ctx, int64(mem))
72+
return nil
7273
})
7374
if err != nil {
7475
fmt.Println("Failed to register callback")
@@ -90,7 +91,7 @@ func ExampleMeter_asynchronous_multiple() {
9091
heapAlloc,
9192
gcCount,
9293
},
93-
func(ctx context.Context) {
94+
func(ctx context.Context) error {
9495
memStats := &runtime.MemStats{}
9596
// This call does work
9697
runtime.ReadMemStats(memStats)
@@ -100,6 +101,7 @@ func ExampleMeter_asynchronous_multiple() {
100101

101102
// This function synchronously records the pauses
102103
computeGCPauses(ctx, gcPause, memStats.PauseNs[:])
104+
return nil
103105
},
104106
)
105107

‎metric/internal/global/meter.go

+1-2
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@ package global // import "go.opentelemetry.io/otel/metric/internal/global"
1616

1717
import (
1818
"container/list"
19-
"context"
2019
"sync"
2120
"sync/atomic"
2221

@@ -323,7 +322,7 @@ func unwrapInstruments(instruments []instrument.Asynchronous) []instrument.Async
323322

324323
type registration struct {
325324
instruments []instrument.Asynchronous
326-
function func(context.Context)
325+
function metric.Callback
327326

328327
unreg func() error
329328
unregMu sync.Mutex

‎metric/internal/global/meter_test.go

+8-5
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ func TestMeterRace(t *testing.T) {
6868
_, _ = mtr.Int64Counter(name)
6969
_, _ = mtr.Int64UpDownCounter(name)
7070
_, _ = mtr.Int64Histogram(name)
71-
_, _ = mtr.RegisterCallback(nil, func(ctx context.Context) {})
71+
_, _ = mtr.RegisterCallback(nil, func(ctx context.Context) error { return nil })
7272
if !once {
7373
wg.Done()
7474
once = true
@@ -88,7 +88,7 @@ func TestMeterRace(t *testing.T) {
8888

8989
func TestUnregisterRace(t *testing.T) {
9090
mtr := &meter{}
91-
reg, err := mtr.RegisterCallback(nil, func(ctx context.Context) {})
91+
reg, err := mtr.RegisterCallback(nil, func(ctx context.Context) error { return nil })
9292
require.NoError(t, err)
9393

9494
wg := &sync.WaitGroup{}
@@ -130,8 +130,9 @@ func testSetupAllInstrumentTypes(t *testing.T, m metric.Meter) (syncfloat64.Coun
130130
_, err = m.Int64ObservableGauge("test_Async_Gauge")
131131
assert.NoError(t, err)
132132

133-
_, err = m.RegisterCallback([]instrument.Asynchronous{afcounter}, func(ctx context.Context) {
133+
_, err = m.RegisterCallback([]instrument.Asynchronous{afcounter}, func(ctx context.Context) error {
134134
afcounter.Observe(ctx, 3)
135+
return nil
135136
})
136137
require.NoError(t, err)
137138

@@ -324,8 +325,9 @@ func TestRegistrationDelegation(t *testing.T) {
324325
require.NoError(t, err)
325326

326327
var called0 bool
327-
reg0, err := m.RegisterCallback([]instrument.Asynchronous{actr}, func(context.Context) {
328+
reg0, err := m.RegisterCallback([]instrument.Asynchronous{actr}, func(context.Context) error {
328329
called0 = true
330+
return nil
329331
})
330332
require.NoError(t, err)
331333
require.Equal(t, 1, mImpl.registry.Len(), "callback not registered")
@@ -334,8 +336,9 @@ func TestRegistrationDelegation(t *testing.T) {
334336
assert.Equal(t, 0, mImpl.registry.Len(), "callback not unregistered")
335337

336338
var called1 bool
337-
reg1, err := m.RegisterCallback([]instrument.Asynchronous{actr}, func(context.Context) {
339+
reg1, err := m.RegisterCallback([]instrument.Asynchronous{actr}, func(context.Context) error {
338340
called1 = true
341+
return nil
339342
})
340343
require.NoError(t, err)
341344
require.Equal(t, 1, mImpl.registry.Len(), "second callback not registered")

‎metric/internal/global/meter_types_test.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ type testMeter struct {
5252
siUDCount int
5353
siHist int
5454

55-
callbacks []func(context.Context)
55+
callbacks []metric.Callback
5656
}
5757

5858
func (m *testMeter) Int64Counter(name string, options ...instrument.Int64Option) (syncint64.Counter, error) {
@@ -145,6 +145,6 @@ func (m *testMeter) collect() {
145145
// Unregister.
146146
continue
147147
}
148-
f(ctx)
148+
_ = f(ctx)
149149
}
150150
}

‎metric/meter.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ type Meter interface {
120120
// the same attributes as another Callback will report.
121121
//
122122
// The function needs to be concurrent safe.
123-
type Callback func(context.Context)
123+
type Callback func(context.Context) error
124124

125125
// Registration is an token representing the unique registration of a callback
126126
// for a set of instruments with a Meter.

‎sdk/metric/meter_test.go

+43-21
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,8 @@ func TestMeterInstrumentConcurrency(t *testing.T) {
9595
wg.Wait()
9696
}
9797

98+
var emptyCallback metric.Callback = func(ctx context.Context) error { return nil }
99+
98100
// A Meter Should be able register Callbacks Concurrently.
99101
func TestMeterCallbackCreationConcurrency(t *testing.T) {
100102
wg := &sync.WaitGroup{}
@@ -103,19 +105,19 @@ func TestMeterCallbackCreationConcurrency(t *testing.T) {
103105
m := NewMeterProvider().Meter("callback-concurrency")
104106

105107
go func() {
106-
_, _ = m.RegisterCallback([]instrument.Asynchronous{}, func(ctx context.Context) {})
108+
_, _ = m.RegisterCallback([]instrument.Asynchronous{}, emptyCallback)
107109
wg.Done()
108110
}()
109111
go func() {
110-
_, _ = m.RegisterCallback([]instrument.Asynchronous{}, func(ctx context.Context) {})
112+
_, _ = m.RegisterCallback([]instrument.Asynchronous{}, emptyCallback)
111113
wg.Done()
112114
}()
113115
wg.Wait()
114116
}
115117

116118
func TestNoopCallbackUnregisterConcurrency(t *testing.T) {
117119
m := NewMeterProvider().Meter("noop-unregister-concurrency")
118-
reg, err := m.RegisterCallback(nil, func(ctx context.Context) {})
120+
reg, err := m.RegisterCallback(nil, emptyCallback)
119121
require.NoError(t, err)
120122

121123
wg := &sync.WaitGroup{}
@@ -143,11 +145,11 @@ func TestCallbackUnregisterConcurrency(t *testing.T) {
143145
require.NoError(t, err)
144146

145147
i := []instrument.Asynchronous{actr}
146-
regCtr, err := meter.RegisterCallback(i, func(ctx context.Context) {})
148+
regCtr, err := meter.RegisterCallback(i, emptyCallback)
147149
require.NoError(t, err)
148150

149151
i = []instrument.Asynchronous{ag}
150-
regG, err := meter.RegisterCallback(i, func(ctx context.Context) {})
152+
regG, err := meter.RegisterCallback(i, emptyCallback)
151153
require.NoError(t, err)
152154

153155
wg := &sync.WaitGroup{}
@@ -183,8 +185,9 @@ func TestMeterCreatesInstruments(t *testing.T) {
183185
}
184186
ctr, err := m.Int64ObservableCounter("aint", instrument.WithInt64Callback(cback))
185187
assert.NoError(t, err)
186-
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
188+
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
187189
ctr.Observe(ctx, 3)
190+
return nil
188191
})
189192
assert.NoError(t, err)
190193

@@ -212,8 +215,9 @@ func TestMeterCreatesInstruments(t *testing.T) {
212215
}
213216
ctr, err := m.Int64ObservableUpDownCounter("aint", instrument.WithInt64Callback(cback))
214217
assert.NoError(t, err)
215-
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
218+
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
216219
ctr.Observe(ctx, 11)
220+
return nil
217221
})
218222
assert.NoError(t, err)
219223

@@ -241,8 +245,9 @@ func TestMeterCreatesInstruments(t *testing.T) {
241245
}
242246
gauge, err := m.Int64ObservableGauge("agauge", instrument.WithInt64Callback(cback))
243247
assert.NoError(t, err)
244-
_, err = m.RegisterCallback([]instrument.Asynchronous{gauge}, func(ctx context.Context) {
248+
_, err = m.RegisterCallback([]instrument.Asynchronous{gauge}, func(ctx context.Context) error {
245249
gauge.Observe(ctx, 11)
250+
return nil
246251
})
247252
assert.NoError(t, err)
248253

@@ -268,8 +273,9 @@ func TestMeterCreatesInstruments(t *testing.T) {
268273
}
269274
ctr, err := m.Float64ObservableCounter("afloat", instrument.WithFloat64Callback(cback))
270275
assert.NoError(t, err)
271-
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
276+
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
272277
ctr.Observe(ctx, 3)
278+
return nil
273279
})
274280
assert.NoError(t, err)
275281

@@ -297,8 +303,9 @@ func TestMeterCreatesInstruments(t *testing.T) {
297303
}
298304
ctr, err := m.Float64ObservableUpDownCounter("afloat", instrument.WithFloat64Callback(cback))
299305
assert.NoError(t, err)
300-
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
306+
_, err = m.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
301307
ctr.Observe(ctx, 11)
308+
return nil
302309
})
303310
assert.NoError(t, err)
304311

@@ -326,8 +333,9 @@ func TestMeterCreatesInstruments(t *testing.T) {
326333
}
327334
gauge, err := m.Float64ObservableGauge("agauge", instrument.WithFloat64Callback(cback))
328335
assert.NoError(t, err)
329-
_, err = m.RegisterCallback([]instrument.Asynchronous{gauge}, func(ctx context.Context) {
336+
_, err = m.RegisterCallback([]instrument.Asynchronous{gauge}, func(ctx context.Context) error {
330337
gauge.Observe(ctx, 11)
338+
return nil
331339
})
332340
assert.NoError(t, err)
333341

@@ -501,16 +509,18 @@ func TestMetersProvideScope(t *testing.T) {
501509
m1 := mp.Meter("scope1")
502510
ctr1, err := m1.Float64ObservableCounter("ctr1")
503511
assert.NoError(t, err)
504-
_, err = m1.RegisterCallback([]instrument.Asynchronous{ctr1}, func(ctx context.Context) {
512+
_, err = m1.RegisterCallback([]instrument.Asynchronous{ctr1}, func(ctx context.Context) error {
505513
ctr1.Observe(ctx, 5)
514+
return nil
506515
})
507516
assert.NoError(t, err)
508517

509518
m2 := mp.Meter("scope2")
510519
ctr2, err := m2.Int64ObservableCounter("ctr2")
511520
assert.NoError(t, err)
512-
_, err = m1.RegisterCallback([]instrument.Asynchronous{ctr2}, func(ctx context.Context) {
521+
_, err = m1.RegisterCallback([]instrument.Asynchronous{ctr2}, func(ctx context.Context) error {
513522
ctr2.Observe(ctx, 7)
523+
return nil
514524
})
515525
assert.NoError(t, err)
516526

@@ -594,7 +604,10 @@ func TestUnregisterUnregisters(t *testing.T) {
594604
floag64Counter,
595605
floag64UpDownCounter,
596606
floag64Gauge,
597-
}, func(context.Context) { called = true })
607+
}, func(context.Context) error {
608+
called = true
609+
return nil
610+
})
598611
require.NoError(t, err)
599612

600613
ctx := context.Background()
@@ -644,7 +657,10 @@ func TestRegisterCallbackDropAggregations(t *testing.T) {
644657
floag64Counter,
645658
floag64UpDownCounter,
646659
floag64Gauge,
647-
}, func(context.Context) { called = true })
660+
}, func(context.Context) error {
661+
called = true
662+
return nil
663+
})
648664
require.NoError(t, err)
649665

650666
data, err := r.Collect(context.Background())
@@ -669,9 +685,10 @@ func TestAttributeFilter(t *testing.T) {
669685
if err != nil {
670686
return err
671687
}
672-
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
688+
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
673689
ctr.Observe(ctx, 1.0, attribute.String("foo", "bar"), attribute.Int("version", 1))
674690
ctr.Observe(ctx, 2.0, attribute.String("foo", "bar"), attribute.Int("version", 2))
691+
return nil
675692
})
676693
return err
677694
},
@@ -696,9 +713,10 @@ func TestAttributeFilter(t *testing.T) {
696713
if err != nil {
697714
return err
698715
}
699-
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
716+
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
700717
ctr.Observe(ctx, 1.0, attribute.String("foo", "bar"), attribute.Int("version", 1))
701718
ctr.Observe(ctx, 2.0, attribute.String("foo", "bar"), attribute.Int("version", 2))
719+
return nil
702720
})
703721
return err
704722
},
@@ -723,9 +741,10 @@ func TestAttributeFilter(t *testing.T) {
723741
if err != nil {
724742
return err
725743
}
726-
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
744+
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
727745
ctr.Observe(ctx, 1.0, attribute.String("foo", "bar"), attribute.Int("version", 1))
728746
ctr.Observe(ctx, 2.0, attribute.String("foo", "bar"), attribute.Int("version", 2))
747+
return nil
729748
})
730749
return err
731750
},
@@ -748,9 +767,10 @@ func TestAttributeFilter(t *testing.T) {
748767
if err != nil {
749768
return err
750769
}
751-
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
770+
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
752771
ctr.Observe(ctx, 10, attribute.String("foo", "bar"), attribute.Int("version", 1))
753772
ctr.Observe(ctx, 20, attribute.String("foo", "bar"), attribute.Int("version", 2))
773+
return nil
754774
})
755775
return err
756776
},
@@ -775,9 +795,10 @@ func TestAttributeFilter(t *testing.T) {
775795
if err != nil {
776796
return err
777797
}
778-
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
798+
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
779799
ctr.Observe(ctx, 10, attribute.String("foo", "bar"), attribute.Int("version", 1))
780800
ctr.Observe(ctx, 20, attribute.String("foo", "bar"), attribute.Int("version", 2))
801+
return nil
781802
})
782803
return err
783804
},
@@ -802,9 +823,10 @@ func TestAttributeFilter(t *testing.T) {
802823
if err != nil {
803824
return err
804825
}
805-
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) {
826+
_, err = mtr.RegisterCallback([]instrument.Asynchronous{ctr}, func(ctx context.Context) error {
806827
ctr.Observe(ctx, 10, attribute.String("foo", "bar"), attribute.Int("version", 1))
807828
ctr.Observe(ctx, 20, attribute.String("foo", "bar"), attribute.Int("version", 2))
829+
return nil
808830
})
809831
return err
810832
},

‎sdk/metric/pipeline.go

+3-1
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,9 @@ func (p *pipeline) produce(ctx context.Context) (metricdata.ResourceMetrics, err
147147
for e := p.multiCallbacks.Front(); e != nil; e = e.Next() {
148148
// TODO make the callbacks parallel. ( #3034 )
149149
f := e.Value.(metric.Callback)
150-
f(ctx)
150+
if err := f(ctx); err != nil {
151+
errs.append(err)
152+
}
151153
if err := ctx.Err(); err != nil {
152154
// This means the context expired before we finished running callbacks.
153155
return metricdata.ResourceMetrics{}, err

0 commit comments

Comments
 (0)