Add pipeline uninstall mechanism (#229)

In order to address the issues raised in #225, the following changes are
proposed:

1) Fix the current bug where futures are being dropped instead of
properly awaited by updating:

```diff
- let spawn = |future| tokio::task::spawn_blocking(|| future);
+ let spawn = |future| tokio::task::spawn_blocking(|| futures::executor::block_on(future));
```

2) Because `spawn_blocking` will now block the program from exiting
until the future completes, add a drop guard to pipeline install methods
which uninstalls the global tracer provider (and in the otlp case in the
future also the meter provider) so the exporters properly shut down,
causing the future to resolve.

    Alternatives considered here could be:

    a) let users unset the global(s) themselves (but they would likely
forget, causing the same blocking behavior)
    b) return a method that could be called instead of a drop guard
(similar problems)
    c) Current impl uses custom uninstall objects per exporter to make
the struct `opentelemetry_jaeger::Uninstall` rather than a more generic
`global::TracerProviderGuard` or `global::UninstallPipeline`,
alternative name suggestions could be helpful if they improve clarity.
    c) Return a pipeline object that contains a `Tracer`, and could be
the drop guard itself, maybe cleaner API but less explicit, possibly a
better choice.

3) Because `Tracer`s currently retain strong references to their
`TracerProvider`, and shutdown is called on the inner's drop, it is
still never called. To fix this, update the `Tracer` to instead contain
a `Weak<TracerProviderInner>` reference, which can be upgraded when
needed. This does force other logic like span creation to consider the
case where the `TracerProvider` has been dropped on shutdown, but making
this case explicit seems beneficial in most cases.

4) Additional clarity/consistency tweaks:

    * Rename `global::set_provider` to `global::set_tracer_provider`
    * Rename `global::trace_provider` to `global::tracer_provider`
    * Rename `ProviderInner` to `TracerProviderInner`

Author: jtescher

README.md

@@ -20,7 +20,7 @@ and metrics from your application. You can analyze them using [Prometheus], [Jae
 use opentelemetry::{api::TracerGenerics, global, sdk};
 
 fn main() {
-    global::set_provider(sdk::Provider::default());
+    let _guard = global::set_tracer_provider(sdk::Provider::default());
 
     global::tracer("component-a").in_span("foo", |_context| {
         global::tracer("component-b").in_span("bar", |_context| {

examples/actix-http/src/main.rs

@@ -6,7 +6,7 @@ use opentelemetry::api::{Key, TraceContextExt, Tracer};
 use opentelemetry::{global, sdk};
 use std::error::Error;
 
-fn init_tracer() -> Result<sdk::Tracer, Box<dyn Error>> {
+fn init_tracer() -> Result<(sdk::Tracer, opentelemetry_jaeger::Uninstall), Box<dyn Error>> {
     opentelemetry_jaeger::new_pipeline()
         .with_collector_endpoint("http://127.0.0.1:14268/api/traces")
         .with_service_name("trace-http-demo")
@@ -25,7 +25,7 @@ async fn index() -> &'static str {
 async fn main() -> std::io::Result<()> {
     std::env::set_var("RUST_LOG", "debug");
     env_logger::init();
-    let tracer = init_tracer().expect("Failed to initialise tracer.");
+    let (tracer, _uninstall) = init_tracer().expect("Failed to initialise tracer.");
 
     HttpServer::new(move || {
         let tracer = tracer.clone();

examples/actix-udp/src/main.rs

@@ -6,7 +6,7 @@ use opentelemetry::api::{Key, TraceContextExt, Tracer};
 use opentelemetry::{global, sdk};
 use std::error::Error;
 
-fn init_tracer() -> Result<sdk::Tracer, Box<dyn Error>> {
+fn init_tracer() -> Result<(sdk::Tracer, opentelemetry_jaeger::Uninstall), Box<dyn Error>> {
     opentelemetry_jaeger::new_pipeline()
         .with_agent_endpoint("localhost:6831")
         .with_service_name("trace-udp-demo")
@@ -25,7 +25,7 @@ async fn index() -> &'static str {
 async fn main() -> std::io::Result<()> {
     std::env::set_var("RUST_LOG", "debug");
     env_logger::init();
-    init_tracer().expect("Failed to initialise tracer.");
+    let _uninstall = init_tracer().expect("Failed to initialise tracer.");
 
     HttpServer::new(|| {
         App::new()

examples/async/src/main.rs

@@ -50,15 +50,15 @@ async fn run(addr: &SocketAddr) -> io::Result<usize> {
     write(&mut stream).with_context(cx).await
 }
 
-fn init_tracer() -> Result<sdk::Tracer, Box<dyn Error>> {
+fn init_tracer() -> Result<(sdk::Tracer, opentelemetry_jaeger::Uninstall), Box<dyn Error>> {
     opentelemetry_jaeger::new_pipeline()
         .with_service_name("trace-demo")
         .install()
 }
 
 #[tokio::main]
 pub async fn main() -> Result<(), Box<dyn Error>> {
-    let tracer = init_tracer()?;
+    let (tracer, _uninstall) = init_tracer()?;
     let addr = "127.0.0.1:6142".parse()?;
     let addr2 = "127.0.0.1:6143".parse()?;
     let span = tracer.start("root");

examples/aws-xray/src/client.rs

@@ -3,7 +3,9 @@ use opentelemetry::api::{Context, TraceContextExt, Tracer};
 use opentelemetry::{api, exporter::trace::stdout, global, sdk};
 use opentelemetry_contrib::{XrayIdGenerator, XrayTraceContextPropagator};
 
-fn init_tracer() {
+fn init_tracer() -> (sdk::Tracer, stdout::Uninstall) {
+    global::set_text_map_propagator(XrayTraceContextPropagator::new());
+
     // Install stdout exporter pipeline to be able to retrieve the collected spans.
     // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
     // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
@@ -13,14 +15,12 @@ fn init_tracer() {
             id_generator: Box::new(XrayIdGenerator::default()),
             ..Default::default()
         })
-        .install();
-
-    global::set_text_map_propagator(XrayTraceContextPropagator::new());
+        .install()
 }
 
 #[tokio::main]
 async fn main() -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_tracer();
+    let (_tracer, _guard) = init_tracer();
 
     let client = Client::new();
     let span = global::tracer("example/client").start("say hello");

examples/aws-xray/src/server.rs

@@ -27,7 +27,9 @@ async fn handle(req: Request<Body>) -> Result<Response<Body>, Infallible> {
     ))
 }
 
-fn init_tracer() {
+fn init_tracer() -> (sdk::Tracer, stdout::Uninstall) {
+    global::set_text_map_propagator(XrayTraceContextPropagator::new());
+
     // Install stdout exporter pipeline to be able to retrieve the collected spans.
     // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
     // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
@@ -37,14 +39,12 @@ fn init_tracer() {
             id_generator: Box::new(XrayIdGenerator::default()),
             ..Default::default()
         })
-        .install();
-
-    global::set_text_map_propagator(XrayTraceContextPropagator::new());
+        .install()
 }
 
 #[tokio::main]
 async fn main() {
-    init_tracer();
+    let _guard = init_tracer();
     let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
 
     let make_svc = make_service_fn(|_conn| async { Ok::<_, Infallible>(service_fn(handle)) });

examples/basic-otlp/src/main.rs

@@ -4,9 +4,10 @@ use opentelemetry::api::{BaggageExt, Context, Key, KeyValue, TraceContextExt, Tr
 use opentelemetry::exporter;
 use opentelemetry::sdk::metrics::PushController;
 use opentelemetry::{global, sdk};
+use std::error::Error;
 use std::time::Duration;
 
-fn init_tracer() -> Result<sdk::Tracer, Box<dyn std::error::Error>> {
+fn init_tracer() -> Result<(sdk::Tracer, opentelemetry_otlp::Uninstall), Box<dyn Error>> {
     opentelemetry_otlp::new_pipeline().install()
 }
 
@@ -41,8 +42,8 @@ lazy_static::lazy_static! {
 }
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let _ = init_tracer()?;
+async fn main() -> Result<(), Box<dyn Error>> {
+    let _guard = init_tracer()?;
     let _started = init_meter()?;
 
     let tracer = global::tracer("ex.com/basic");

examples/basic/src/main.rs

@@ -7,7 +7,7 @@ use opentelemetry::{global, sdk};
 use std::error::Error;
 use std::time::Duration;
 
-fn init_tracer() -> Result<sdk::Tracer, Box<dyn Error>> {
+fn init_tracer() -> Result<(sdk::Tracer, opentelemetry_jaeger::Uninstall), Box<dyn Error>> {
     opentelemetry_jaeger::new_pipeline()
         .with_service_name("trace-demo")
         .with_tags(vec![
@@ -49,7 +49,7 @@ lazy_static::lazy_static! {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
-    let _ = init_tracer()?;
+    let _uninstall = init_tracer()?;
     let _started = init_meter()?;
 
     let tracer = global::tracer("ex.com/basic");

examples/datadog/src/main.rs

@@ -1,8 +1,8 @@
 use opentelemetry::api::{Key, Span, TraceContextExt, Tracer};
 use opentelemetry::global;
+use opentelemetry_contrib::datadog::ApiVersion;
 use std::thread;
 use std::time::Duration;
-use opentelemetry_contrib::datadog::ApiVersion;
 
 fn bar() {
     let tracer = global::tracer("component-bar");
@@ -14,7 +14,7 @@ fn bar() {
 }
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let tracer = opentelemetry_contrib::datadog::new_pipeline()
+    let (tracer, _uninstall) = opentelemetry_contrib::datadog::new_pipeline()
         .with_service_name("trace-demo")
         .with_version(ApiVersion::Version05)
         .install()?;

examples/grpc/src/client.rs

@@ -10,15 +10,15 @@ pub mod hello_world {
     tonic::include_proto!("helloworld");
 }
 
-fn tracing_init() -> Result<sdk::Tracer, Box<dyn Error>> {
+fn tracing_init() -> Result<(sdk::Tracer, opentelemetry_jaeger::Uninstall), Box<dyn Error>> {
     opentelemetry_jaeger::new_pipeline()
         .with_service_name("grpc-client")
         .install()
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let tracer = tracing_init()?;
+    let (tracer, _uninstall) = tracing_init()?;
     let mut client = GreeterClient::connect("http://[::1]:50051").await?;
     let propagator = TraceContextPropagator::new();
     let span = tracer.start("client-request");

examples/grpc/src/server.rs

@@ -34,15 +34,15 @@ impl Greeter for MyGreeter {
     }
 }
 
-fn tracing_init() -> Result<sdk::Tracer, Box<dyn Error>> {
+fn tracing_init() -> Result<(sdk::Tracer, opentelemetry_jaeger::Uninstall), Box<dyn Error>> {
     opentelemetry_jaeger::new_pipeline()
         .with_service_name("grpc-server")
         .install()
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
-    let _ = tracing_init()?;
+    let _uninstall = tracing_init()?;
     let addr = "[::1]:50051".parse()?;
     let greeter = MyGreeter::default();
 

examples/http/src/client.rs

@@ -2,7 +2,7 @@ use hyper::{body::Body, Client};
 use opentelemetry::api::{Context, TextMapFormat, TraceContextExt, Tracer};
 use opentelemetry::{api, exporter::trace::stdout, global, sdk};
 
-fn init_tracer() {
+fn init_tracer() -> (sdk::Tracer, stdout::Uninstall) {
     // Install stdout exporter pipeline to be able to retrieve the collected spans.
     // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
     // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
@@ -11,12 +11,12 @@ fn init_tracer() {
             default_sampler: Box::new(sdk::Sampler::AlwaysOn),
             ..Default::default()
         })
-        .install();
+        .install()
 }
 
 #[tokio::main]
 async fn main() -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_tracer();
+    let _guard = init_tracer();
 
     let client = Client::new();
     let propagator = api::TraceContextPropagator::new();

examples/http/src/server.rs

@@ -16,21 +16,22 @@ async fn handle(req: Request<Body>) -> Result<Response<Body>, Infallible> {
     Ok(Response::new("Hello, World!".into()))
 }
 
-fn init_tracer() {
+fn init_tracer() -> (sdk::Tracer, stdout::Uninstall) {
     // Install stdout exporter pipeline to be able to retrieve the collected spans.
+
     // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
     // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
     stdout::new_pipeline()
         .with_trace_config(sdk::Config {
             default_sampler: Box::new(sdk::Sampler::AlwaysOn),
             ..Default::default()
         })
-        .install();
+        .install()
 }
 
 #[tokio::main]
 async fn main() {
-    init_tracer();
+    let _guard = init_tracer();
     let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
 
     let make_svc = make_service_fn(|_conn| async { Ok::<_, Infallible>(service_fn(handle)) });

examples/stdout.rs

@@ -5,7 +5,7 @@ fn main() {
     // Install stdout exporter pipeline to be able to retrieve collected spans.
     // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
     // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
-    let tracer = stdout::new_pipeline()
+    let (tracer, _uninstall) = stdout::new_pipeline()
         .with_trace_config(sdk::Config {
             default_sampler: Box::new(sdk::Sampler::AlwaysOn),
             ..Default::default()

examples/zipkin/src/main.rs

@@ -11,7 +11,7 @@ fn bar() {
 }
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let tracer = opentelemetry_zipkin::new_pipeline()
+    let (tracer, _uninstall) = opentelemetry_zipkin::new_pipeline()
         .with_service_name("trace-demo")
         .install()?;
 

opentelemetry-contrib/src/datadog/mod.rs

@@ -59,7 +59,7 @@
 //! use opentelemetry::sdk::{trace, IdGenerator, Resource, Sampler};
 //!
 //! fn main() -> Result<(), Box<dyn std::error::Error>> {
-//!     let tracer = opentelemetry_contrib::datadog::new_pipeline()
+//!     let (tracer, _uninstall) = opentelemetry_contrib::datadog::new_pipeline()
 //!         .with_service_name("my_app")
 //!         .with_version(opentelemetry_contrib::datadog::ApiVersion::Version05)
 //!         .with_agent_endpoint("http://localhost:8126")
@@ -147,7 +147,7 @@ impl Default for DatadogPipelineBuilder {
 
 impl DatadogPipelineBuilder {
     /// Create `ExporterConfig` struct from current `ExporterConfigBuilder`
-    pub fn install(mut self) -> Result<sdk::Tracer, Box<dyn Error>> {
+    pub fn install(mut self) -> Result<(sdk::Tracer, Uninstall), Box<dyn Error>> {
         let exporter = DatadogExporter::new(
             self.service_name.clone(),
             self.agent_endpoint.parse()?,
@@ -160,9 +160,9 @@ impl DatadogPipelineBuilder {
         }
         let provider = provider_builder.build();
         let tracer = provider.get_tracer("opentelemetry-datadog", Some(env!("CARGO_PKG_VERSION")));
-        global::set_provider(provider);
+        let provider_guard = global::set_tracer_provider(provider);
 
-        Ok(tracer)
+        Ok((tracer, Uninstall(provider_guard)))
     }
 
     /// Assign the service name under which to group traces
@@ -211,3 +211,7 @@ impl trace::SpanExporter for DatadogExporter {
         }
     }
 }
+
+/// Uninstalls the Datadog pipeline on drop
+#[derive(Debug)]
+pub struct Uninstall(global::TracerProviderGuard);

opentelemetry-contrib/src/datadog/model/mod.rs

@@ -65,10 +65,10 @@ impl ApiVersion {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use opentelemetry::api::Key;
     use opentelemetry::sdk::InstrumentationLibrary;
     use opentelemetry::{api, sdk};
     use std::time::{Duration, SystemTime};
-    use opentelemetry::api::Key;
 
     fn get_spans() -> Vec<Arc<trace::SpanData>> {
         let parent_span_id = 1;

opentelemetry-contrib/src/datadog/model/v03.rs

@@ -1,8 +1,8 @@
 use crate::datadog::model::Error;
+use opentelemetry::api::{Key, Value};
 use opentelemetry::exporter::trace;
 use std::sync::Arc;
 use std::time::SystemTime;
-use opentelemetry::api::{Key, Value};
 
 pub(crate) fn encode(
     service_name: &str,

opentelemetry-jaeger/README.md

@@ -22,7 +22,7 @@ exporting telemetry:
 use opentelemetry::api::Tracer;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let tracer = opentelemetry_jaeger::new_pipeline().install()?;
+    let (tracer, _uninstall) = opentelemetry_jaeger::new_pipeline().install()?;
 
     tracer.in_span("doing_work", |cx| {
         // Traced app logic here...
@@ -64,7 +64,7 @@ use opentelemetry::api::Tracer;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     // export OTEL_SERVICE_NAME=my-service-name
-    let tracer = opentelemetry_jaeger::new_pipeline().from_env().install()?;
+    let (tracer, _uninstall) = opentelemetry_jaeger::new_pipeline().from_env().install()?;
 
     tracer.in_span("doing_work", |cx| {
         // Traced app logic here...
@@ -94,7 +94,7 @@ Then you can use the [`with_collector_endpoint`] method to specify the endpoint:
 use opentelemetry::api::Tracer;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let tracer = opentelemetry_jaeger::new_pipeline()
+    let (tracer, _uninstall) = opentelemetry_jaeger::new_pipeline()
         .with_collector_endpoint("http://localhost:14268/api/traces")
         // optionally set username and password as well.
         .with_collector_username("username")
@@ -121,7 +121,7 @@ use opentelemetry::api::{KeyValue, Tracer};
 use opentelemetry::sdk::{trace, IdGenerator, Resource, Sampler};
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let tracer = opentelemetry_jaeger::new_pipeline()
+    let (tracer, _uninstall) = opentelemetry_jaeger::new_pipeline()
         .from_env()
         .with_agent_endpoint("localhost:6831")
         .with_service_name("my_app")