fix(telemetry): ensure tracer providers are shut down in all services (…

…#1098)

## Summary
Calls `opentelemetry::global::shutdown_tracer_providers` upon exiting
services

## Background
Because OTLP is pushed-based, users must ensure that all spans are
flushed and sent to the OTLP endpoint.

Note that `opentelemetry::global::shutdown_trace_providers` is a
blocking operation and can potentially stall a service from exiting. We
must observe if this actually ends up being a problem in practice.

## Changes
- Add a drop guard as the return value of
`astria_telemetry::Config::try_init`, which in turns calls
`opentelemetry::global::shutdown_trace_providers` on drop

## Testing
Happens outside of testable code. Must be observed on devnet/in
production.

crates/astria-composer/src/main.rs

@@ -42,13 +42,16 @@ async fn main() -> ExitCode {
             .register_metrics(metrics_init::register);
     }
 
-    if let Err(e) = telemetry_conf
+    let _telemetry_guard = match telemetry_conf
         .try_init()
         .wrap_err("failed to setup telemetry")
     {
-        eprintln!("initializing composer failed:\n{e:?}");
-        return ExitCode::FAILURE;
-    }
+        Err(e) => {
+            eprintln!("initializing composer failed:\n{e:?}");
+            return ExitCode::FAILURE;
+        }
+        Ok(guard) => guard,
+    };
 
     let cfg_ser = serde_json::to_string(&cfg)
         .expect("the json serializer should never fail when serializing to a string");

crates/astria-conductor/src/main.rs

@@ -57,13 +57,16 @@ async fn main() -> ExitCode {
             .register_metrics(|| {}); // conductor currently has no metrics
     }
 
-    if let Err(e) = telemetry_conf
+    let _telemetry_guard = match telemetry_conf
         .try_init()
         .wrap_err("failed to setup telemetry")
     {
-        eprintln!("initializing conductor failed:\n{e:?}");
-        return ExitCode::FAILURE;
-    }
+        Err(e) => {
+            eprintln!("initializing conductor failed:\n{e:?}");
+            return ExitCode::FAILURE;
+        }
+        Ok(guard) => guard,
+    };
 
     info!(
         config = serde_json::to_string(&cfg).expect("serializing to a string cannot fail"),

crates/astria-sequencer-relayer/src/main.rs

@@ -39,13 +39,16 @@ async fn main() -> ExitCode {
             .register_metrics(metrics_init::register);
     }
 
-    if let Err(e) = telemetry_conf
+    let _telemetry_guard = match telemetry_conf
         .try_init()
         .wrap_err("failed to setup telemetry")
     {
-        eprintln!("initializing sequencer-relayer failed:\n{e:?}");
-        return ExitCode::FAILURE;
-    }
+        Err(e) => {
+            eprintln!("initializing sequencer-relayer failed:\n{e:?}");
+            return ExitCode::FAILURE;
+        }
+        Ok(guard) => guard,
+    };
 
     info!(
         config = %telemetry::display::json(&cfg),

crates/astria-sequencer/src/main.rs

@@ -40,13 +40,17 @@ async fn main() -> ExitCode {
             .register_metrics(metrics_init::register);
     }
 
-    if let Err(e) = telemetry_conf
+    let _telemetry_guard = match telemetry_conf
         .try_init()
         .context("failed to setup telemetry")
     {
-        eprintln!("initializing sequencer failed:\n{e:?}");
-        return ExitCode::FAILURE;
-    }
+        Err(e) => {
+            eprintln!("initializing sequencer failed:\n{e:?}");
+            return ExitCode::FAILURE;
+        }
+        Ok(guard) => guard,
+    };
+
     info!(
         config = serde_json::to_string(&cfg).expect("serializing to a string cannot fail"),
         "initializing sequencer"

crates/astria-telemetry/src/lib.rs

@@ -261,7 +261,7 @@ impl Config {
     /// # Errors
     /// Fails if the filter directives could not be parsed, if communication with the OTLP
     /// endpoint failed, or if the global tracing subscriber could not be installed.
-    pub fn try_init(self) -> Result<(), Error> {
+    pub fn try_init(self) -> Result<Guard, Error> {
         let Self {
             filter_directives,
             force_stdout,
@@ -350,6 +350,24 @@ impl Config {
             metrics_builder.install().map_err(Error::exporter_install)?;
         }
 
-        Ok(())
+        Ok(Guard {
+            run_otel_shutdown: !no_otel,
+        })
+    }
+}
+
+/// A drop guard for terminating all `OpenTelemetry` tracer providers on drop.
+///
+/// *Note:* Shutting down the tracer providers can potentially block a thread
+/// indefinitely.
+pub struct Guard {
+    run_otel_shutdown: bool,
+}
+
+impl Drop for Guard {
+    fn drop(&mut self) {
+        if self.run_otel_shutdown {
+            opentelemetry::global::shutdown_tracer_provider();
+        }
     }
 }