Skip to content

Commit c3bdce4

Browse files
jtescherjtescher
authored
metrics: implement batch observer (#429)
This patch adds an implementation of the metrics batch observer. The API is not identical to the example in [the spec](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/metrics/api.md#batch-observer) as it is inconvenient to register metrics after they are moved into the observer callback in rust. Instead the `Meter::batch_observer` method accepts a closure in which instruments can be registered before being moved into the callback. ```rust meter.batch_observer(|batch| { let inst = batch.u64_sum_observer("example").init(); move |result| { result.observe(&[KeyValue::new("a", "1")], inst.observation(42)]); } }); ```
1 parent 6728799 commit c3bdce4

File tree

9 files changed

+237
-53
lines changed

9 files changed

+237
-53
lines changed

‎opentelemetry-prometheus/tests/integration_test.rs‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use opentelemetry::sdk::Resource;
22
use opentelemetry::{
3-
metrics::{MeterProvider, ObserverResult},
3+
metrics::{BatchObserverResult, MeterProvider, ObserverResult},
44
KeyValue,
55
};
66
use opentelemetry_prometheus::PrometheusExporter;
@@ -33,6 +33,34 @@ fn free_unused_instruments() {
3333
compare_export(&exporter, expected);
3434
}
3535

36+
#[test]
37+
fn batch() {
38+
let exporter = opentelemetry_prometheus::exporter()
39+
.with_resource(Resource::new(vec![KeyValue::new("R", "V")]))
40+
.init();
41+
let meter = exporter.provider().unwrap().meter("test", None);
42+
let mut expected = Vec::new();
43+
44+
meter.batch_observer(|batch| {
45+
let uint_observer = batch.u64_value_observer("uint_observer").init();
46+
let float_observer = batch.f64_value_observer("float_observer").init();
47+
48+
move |result: BatchObserverResult| {
49+
result.observe(
50+
&[KeyValue::new("A", "B")],
51+
&[
52+
uint_observer.observation(2),
53+
float_observer.observation(3.1),
54+
],
55+
);
56+
}
57+
});
58+
59+
expected.push(r#"uint_observer{A="B",R="V"} 2"#);
60+
expected.push(r#"float_observer{A="B",R="V"} 3.1"#);
61+
compare_export(&exporter, expected);
62+
}
63+
3664
#[test]
3765
fn test_add() {
3866
let exporter = opentelemetry_prometheus::exporter()

‎opentelemetry/src/metrics/async_instrument.rs‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
11
//! Async metrics
2-
use crate::metrics::{sdk_api, Number};
3-
use crate::KeyValue;
2+
use crate::{
3+
global,
4+
metrics::{sdk_api, MetricsError, Number},
5+
KeyValue,
6+
};
47
use std::fmt;
58
use std::marker;
69
use std::sync::Arc;
@@ -33,17 +36,17 @@ impl Observation {
3336
}
3437

3538
/// A type of callback that `f64` observers run.
36-
type F64ObserverCallback = Box<dyn Fn(ObserverResult<f64>) + Send + Sync + 'static>;
39+
type F64ObserverCallback = Box<dyn Fn(ObserverResult<f64>) + Send + Sync>;
3740

3841
/// A type of callback that `u64` observers run.
39-
type U64ObserverCallback = Box<dyn Fn(ObserverResult<u64>) + Send + Sync + 'static>;
42+
type U64ObserverCallback = Box<dyn Fn(ObserverResult<u64>) + Send + Sync>;
4043

4144
/// A type of callback that `u64` observers run.
42-
type I64ObserverCallback = Box<dyn Fn(ObserverResult<i64>) + Send + Sync + 'static>;
45+
type I64ObserverCallback = Box<dyn Fn(ObserverResult<i64>) + Send + Sync>;
4346

4447
/// A callback argument for use with any Observer instrument that will be
4548
/// reported as a batch of observations.
46-
pub type BatchObserverCallback = Box<dyn Fn(BatchObserverResult) + Send + Sync>;
49+
type BatchObserverCallback = Box<dyn Fn(BatchObserverResult) + Send + Sync>;
4750

4851
/// Data passed to an observer callback to capture observations for one
4952
/// asynchronous metric instrument.
@@ -137,16 +140,17 @@ impl AsyncRunner {
137140
/// implementation can be used for batch runners.)
138141
pub fn run(
139142
&self,
140-
instrument: Arc<dyn sdk_api::AsyncInstrumentCore>,
143+
instrument: &Option<Arc<dyn sdk_api::AsyncInstrumentCore>>,
141144
f: fn(&[KeyValue], &[Observation]),
142145
) {
143-
match self {
144-
AsyncRunner::F64(run) => run(ObserverResult::new(instrument, f)),
145-
AsyncRunner::I64(run) => run(ObserverResult::new(instrument, f)),
146-
AsyncRunner::U64(run) => run(ObserverResult::new(instrument, f)),
147-
// TODO: this should not require an instrument to call. consider
148-
// moving to separate struct
149-
AsyncRunner::Batch(run) => run(BatchObserverResult::new(f)),
146+
match (instrument, self) {
147+
(Some(i), AsyncRunner::F64(run)) => run(ObserverResult::new(i.clone(), f)),
148+
(Some(i), AsyncRunner::I64(run)) => run(ObserverResult::new(i.clone(), f)),
149+
(Some(i), AsyncRunner::U64(run)) => run(ObserverResult::new(i.clone(), f)),
150+
(None, AsyncRunner::Batch(run)) => run(BatchObserverResult::new(f)),
151+
_ => global::handle_error(MetricsError::Other(
152+
"Invalid async runner / instrument pair".into(),
153+
)),
150154
}
151155
}
152156
}

‎opentelemetry/src/metrics/meter.rs‎

Lines changed: 75 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use crate::sdk::InstrumentationLibrary;
22
use crate::{
33
metrics::{
4-
sdk_api, AsyncRunner, BatchObserver, BatchObserverCallback, CounterBuilder, Descriptor,
4+
sdk_api, AsyncRunner, BatchObserver, BatchObserverResult, CounterBuilder, Descriptor,
55
Measurement, NumberKind, ObserverResult, Result, SumObserverBuilder, UpDownCounterBuilder,
66
UpDownSumObserverBuilder, ValueObserverBuilder, ValueRecorderBuilder,
77
},
@@ -93,7 +93,7 @@ impl Meter {
9393
SumObserverBuilder::new(
9494
self,
9595
name.into(),
96-
AsyncRunner::F64(Box::new(callback)),
96+
Some(AsyncRunner::F64(Box::new(callback))),
9797
NumberKind::F64,
9898
)
9999
}
@@ -111,7 +111,7 @@ impl Meter {
111111
UpDownSumObserverBuilder::new(
112112
self,
113113
name.into(),
114-
AsyncRunner::F64(Box::new(callback)),
114+
Some(AsyncRunner::F64(Box::new(callback))),
115115
NumberKind::F64,
116116
)
117117
}
@@ -125,7 +125,7 @@ impl Meter {
125125
ValueObserverBuilder::new(
126126
self,
127127
name.into(),
128-
AsyncRunner::F64(Box::new(callback)),
128+
Some(AsyncRunner::F64(Box::new(callback))),
129129
NumberKind::F64,
130130
)
131131
}
@@ -163,7 +163,7 @@ impl Meter {
163163
SumObserverBuilder::new(
164164
self,
165165
name.into(),
166-
AsyncRunner::I64(Box::new(callback)),
166+
Some(AsyncRunner::I64(Box::new(callback))),
167167
NumberKind::I64,
168168
)
169169
}
@@ -181,7 +181,7 @@ impl Meter {
181181
UpDownSumObserverBuilder::new(
182182
self,
183183
name.into(),
184-
AsyncRunner::I64(Box::new(callback)),
184+
Some(AsyncRunner::I64(Box::new(callback))),
185185
NumberKind::I64,
186186
)
187187
}
@@ -195,7 +195,7 @@ impl Meter {
195195
ValueObserverBuilder::new(
196196
self,
197197
name.into(),
198-
AsyncRunner::I64(Box::new(callback)),
198+
Some(AsyncRunner::I64(Box::new(callback))),
199199
NumberKind::I64,
200200
)
201201
}
@@ -233,7 +233,7 @@ impl Meter {
233233
SumObserverBuilder::new(
234234
self,
235235
name.into(),
236-
AsyncRunner::U64(Box::new(callback)),
236+
Some(AsyncRunner::U64(Box::new(callback))),
237237
NumberKind::U64,
238238
)
239239
}
@@ -251,7 +251,7 @@ impl Meter {
251251
UpDownSumObserverBuilder::new(
252252
self,
253253
name.into(),
254-
AsyncRunner::U64(Box::new(callback)),
254+
Some(AsyncRunner::U64(Box::new(callback))),
255255
NumberKind::U64,
256256
)
257257
}
@@ -265,15 +265,75 @@ impl Meter {
265265
ValueObserverBuilder::new(
266266
self,
267267
name.into(),
268-
AsyncRunner::U64(Box::new(callback)),
268+
Some(AsyncRunner::U64(Box::new(callback))),
269269
NumberKind::U64,
270270
)
271271
}
272272

273-
/// Creates a new `BatchObserver` that supports making batches of observations for
274-
/// multiple instruments.
275-
pub fn batch_observer(&self, callback: BatchObserverCallback) -> BatchObserver<'_> {
276-
BatchObserver::new(self, AsyncRunner::Batch(callback))
273+
/// Creates a new `BatchObserver` that supports making batches of observations
274+
/// for multiple instruments or returns an error if instrument initialization
275+
/// fails.
276+
///
277+
/// # Examples
278+
///
279+
/// ```
280+
/// use opentelemetry::{global, metrics::BatchObserverResult, KeyValue};
281+
///
282+
/// # fn init_observer() -> opentelemetry::metrics::Result<()> {
283+
/// let meter = global::meter("test");
284+
///
285+
/// meter.build_batch_observer(|batch| {
286+
/// let instrument = batch.u64_value_observer("test_instrument").try_init()?;
287+
///
288+
/// Ok(move |result: BatchObserverResult| {
289+
/// result.observe(&[KeyValue::new("my-key", "my-value")], &[instrument.observation(1)]);
290+
/// })
291+
/// })?;
292+
/// # Ok(())
293+
/// # }
294+
/// ```
295+
pub fn build_batch_observer<B, F>(&self, builder: B) -> Result<()>
296+
where
297+
B: Fn(BatchObserver<'_>) -> Result<F>,
298+
F: Fn(BatchObserverResult) + Send + Sync + 'static,
299+
{
300+
let observer = builder(BatchObserver::new(self))?;
301+
self.core
302+
.new_batch_observer(AsyncRunner::Batch(Box::new(observer)))
303+
}
304+
305+
/// Creates a new `BatchObserver` that supports making batches of observations
306+
/// for multiple instruments.
307+
///
308+
/// # Panics
309+
///
310+
/// Panics if instrument initialization or observer registration returns an
311+
/// error.
312+
///
313+
/// # Examples
314+
///
315+
/// ```
316+
/// use opentelemetry::{global, metrics::BatchObserverResult, KeyValue};
317+
///
318+
/// let meter = global::meter("test");
319+
///
320+
/// meter.batch_observer(|batch| {
321+
/// let instrument = batch.u64_value_observer("test_instrument").init();
322+
///
323+
/// move |result: BatchObserverResult| {
324+
/// result.observe(&[KeyValue::new("my-key", "my-value")], &[instrument.observation(1)]);
325+
/// }
326+
/// });
327+
/// ```
328+
pub fn batch_observer<B, F>(&self, builder: B)
329+
where
330+
B: Fn(BatchObserver<'_>) -> F,
331+
F: Fn(BatchObserverResult) + Send + Sync + 'static,
332+
{
333+
let observer = builder(BatchObserver::new(self));
334+
self.core
335+
.new_batch_observer(AsyncRunner::Batch(Box::new(observer)))
336+
.unwrap()
277337
}
278338

279339
/// Atomically record a batch of measurements.
@@ -306,7 +366,7 @@ impl Meter {
306366
pub(crate) fn new_async_instrument(
307367
&self,
308368
descriptor: Descriptor,
309-
runner: AsyncRunner,
369+
runner: Option<AsyncRunner>,
310370
) -> Result<Arc<dyn sdk_api::AsyncInstrumentCore>> {
311371
self.core.new_async_instrument(descriptor, runner)
312372
}

‎opentelemetry/src/metrics/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ mod up_down_counter;
2020
mod value_recorder;
2121

2222
use crate::sdk::export::ExportError;
23-
pub use async_instrument::{AsyncRunner, BatchObserverCallback, Observation, ObserverResult};
23+
pub use async_instrument::{AsyncRunner, BatchObserverResult, Observation, ObserverResult};
2424
pub use config::InstrumentConfig;
2525
pub use counter::{BoundCounter, Counter, CounterBuilder};
2626
pub use descriptor::Descriptor;

‎opentelemetry/src/metrics/noop.rs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ impl MeterCore for NoopMeterCore {
6161
fn new_async_instrument(
6262
&self,
6363
_descriptor: Descriptor,
64-
_runner: AsyncRunner,
64+
_runner: Option<AsyncRunner>,
6565
) -> Result<Arc<dyn AsyncInstrumentCore>> {
6666
Ok(Arc::new(NoopAsyncInstrument::new()))
6767
}
@@ -74,6 +74,10 @@ impl MeterCore for NoopMeterCore {
7474
) {
7575
// Ignored
7676
}
77+
78+
fn new_batch_observer(&self, _runner: AsyncRunner) -> Result<()> {
79+
Ok(())
80+
}
7781
}
7882

7983
/// A no-op sync instrument

0 commit comments

Comments
 (0)