Update metrics API and SDK

This change aligns metrics with the spec, changes include:

* Rename `MeterProvider::meter` to `MeterProvider::versioned_meter` for
  consistency with `TracerProvider` trait.
* Move metrics sdk api types to `opentelemetry-sdk`
* Consolidate instrument builders into `InstrumentBuilder`
* Remove value observers and add gauges.
* Move from batch observer to registered callbacks.
* Rename `ExportKindFor` to `TemporalitySelector`
* Consolidate `PushController` and `PullController` into
  `BasicController`
* Remove `MinMaxSumCountAggregator` and `ArrayAggregator`
* Update examples and exporters for new api/sdk

Cargo.toml

@@ -23,7 +23,6 @@ members = [
     "examples/aws-xray",
     "examples/basic",
     "examples/basic-otlp",
-    "examples/basic-otlp-with-selector",
     "examples/basic-otlp-http",
     "examples/datadog",
     "examples/dynatrace",

examples/basic-otlp/src/main.rs

@@ -1,15 +1,16 @@
-use futures_util::{Stream, StreamExt as _};
 use opentelemetry::global::shutdown_tracer_provider;
-use opentelemetry::sdk::metrics::{selectors, PushController};
+use opentelemetry::runtime;
+use opentelemetry::sdk::export::metrics::aggregation::cumulative_temporality_selector;
+use opentelemetry::sdk::metrics::controllers::BasicController;
+use opentelemetry::sdk::metrics::selectors;
 use opentelemetry::sdk::Resource;
 use opentelemetry::trace::TraceError;
+use opentelemetry::{global, sdk::trace as sdktrace};
 use opentelemetry::{
-    baggage::BaggageExt,
-    metrics::{self, ObserverResult},
+    metrics,
     trace::{TraceContextExt, Tracer},
     Context, Key, KeyValue,
 };
-use opentelemetry::{global, sdk::trace as sdktrace};
 use opentelemetry_otlp::{ExportConfig, WithExportConfig};
 use std::error::Error;
 use std::time::Duration;
@@ -31,29 +32,25 @@ fn init_tracer() -> Result<sdktrace::Tracer, TraceError> {
         .install_batch(opentelemetry::runtime::Tokio)
 }
 
-// Skip first immediate tick from tokio, not needed for async_std.
-fn delayed_interval(duration: Duration) -> impl Stream<Item = tokio::time::Instant> {
-    opentelemetry::sdk::util::tokio_interval_stream(duration).skip(1)
-}
-
-fn init_meter() -> metrics::Result<PushController> {
+fn init_metrics() -> metrics::Result<BasicController> {
     let export_config = ExportConfig {
         endpoint: "http://localhost:4317".to_string(),
         ..ExportConfig::default()
     };
     opentelemetry_otlp::new_pipeline()
-        .metrics(tokio::spawn, delayed_interval)
+        .metrics(
+            selectors::simple::inexpensive(),
+            cumulative_temporality_selector(),
+            runtime::Tokio,
+        )
         .with_exporter(
             opentelemetry_otlp::new_exporter()
                 .tonic()
                 .with_export_config(export_config),
         )
-        .with_aggregator_selector(selectors::simple::Selector::Exact)
         .build()
 }
 
-const FOO_KEY: Key = Key::from_static_str("ex.com/foo");
-const BAR_KEY: Key = Key::from_static_str("ex.com/bar");
 const LEMONS_KEY: Key = Key::from_static_str("lemons");
 const ANOTHER_KEY: Key = Key::from_static_str("ex.com/another");
 
@@ -72,39 +69,20 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // matches the containing block, reporting traces and metrics during the whole
     // execution.
     let _ = init_tracer()?;
-    let _started = init_meter()?;
+    let metrics_controller = init_metrics()?;
+    let cx = Context::new();
 
     let tracer = global::tracer("ex.com/basic");
     let meter = global::meter("ex.com/basic");
 
-    let one_metric_callback =
-        |res: ObserverResult<f64>| res.observe(1.0, COMMON_ATTRIBUTES.as_ref());
-    let _ = meter
-        .f64_value_observer("ex.com.one", one_metric_callback)
-        .with_description("A ValueObserver set to 1.0")
+    let gauge = meter
+        .f64_observable_gauge("ex.com.one")
+        .with_description("A gauge set to 1.0")
         .init();
+    meter.register_callback(move |cx| gauge.observe(cx, 1.0, COMMON_ATTRIBUTES.as_ref()))?;
 
-    let histogram_two = meter.f64_histogram("ex.com.two").init();
-
-    // Needed for code coverage reasons.
-    #[allow(deprecated)]
-    let a_recorder = meter.f64_value_recorder("ex.recorder.a").init();
-    a_recorder.record(5.5, COMMON_ATTRIBUTES.as_ref());
-    #[allow(deprecated)]
-    let b_recorder = meter.u64_value_recorder("ex.recorder.b").init();
-    b_recorder.record(5, COMMON_ATTRIBUTES.as_ref());
-    #[allow(deprecated)]
-    let c_recorder = meter.i64_value_recorder("ex.recorder.c").init();
-    c_recorder.record(5, COMMON_ATTRIBUTES.as_ref());
-
-    let another_histogram = meter.f64_histogram("ex.com.two").init();
-    another_histogram.record(5.5, COMMON_ATTRIBUTES.as_ref());
-
-    let _baggage =
-        Context::current_with_baggage(vec![FOO_KEY.string("foo1"), BAR_KEY.string("bar1")])
-            .attach();
-
-    let histogram = histogram_two.bind(COMMON_ATTRIBUTES.as_ref());
+    let histogram = meter.f64_histogram("ex.com.two").init();
+    histogram.record(&cx, 5.5, COMMON_ATTRIBUTES.as_ref());
 
     tracer.in_span("operation", |cx| {
         let span = cx.span();
@@ -114,27 +92,21 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         );
         span.set_attribute(ANOTHER_KEY.string("yes"));
 
-        meter.record_batch_with_context(
-            // Note: call-site variables added as context Entries:
-            &Context::current_with_baggage(vec![ANOTHER_KEY.string("xyz")]),
-            COMMON_ATTRIBUTES.as_ref(),
-            vec![histogram_two.measurement(2.0)],
-        );
-
         tracer.in_span("Sub operation...", |cx| {
             let span = cx.span();
             span.set_attribute(LEMONS_KEY.string("five"));
 
             span.add_event("Sub span event", vec![]);
 
-            histogram.record(1.3);
+            histogram.record(&cx, 1.3, &[]);
         });
     });
 
     // wait for 1 minutes so that we could see metrics being pushed via OTLP every 10 seconds.
     tokio::time::sleep(Duration::from_secs(60)).await;
 
     shutdown_tracer_provider();
+    metrics_controller.stop(&cx)?;
 
     Ok(())
 }