remove telemetry configuration hot reloading

apollo-router/src/configuration/mod.rs

@@ -143,6 +143,17 @@ impl Configuration {
 
         plugins
     }
+
+    //. checks that we can reload configuration from the current one to the new one
+    pub fn is_compatible(&self, new: &Configuration) -> Result<(), &'static str> {
+        if self.apollo_plugins.plugins.get("telemetry")
+            == new.apollo_plugins.plugins.get("telemetry")
+        {
+            Ok(())
+        } else {
+            Err("incompatible telemetry configuration. Telemetry cannot be reloaded and its configuration must stay the same for the entire life of the process")
+        }
+    }
 }
 
 impl FromStr for Configuration {

apollo-router/src/executable.rs

@@ -15,6 +15,8 @@ use clap::CommandFactory;
 use clap::Parser;
 use directories::ProjectDirs;
 use once_cell::sync::OnceCell;
+use tracing::dispatcher::{with_default, Dispatch};
+use tracing::instrument::WithSubscriber;
 use tracing_subscriber::EnvFilter;
 use url::ParseError;
 use url::Url;
@@ -26,10 +28,8 @@ use crate::router::ApolloRouter;
 use crate::router::ConfigurationKind;
 use crate::router::SchemaKind;
 use crate::router::ShutdownKind;
-use crate::subscriber::set_global_subscriber;
-use crate::subscriber::RouterSubscriber;
 
-static GLOBAL_ENV_FILTER: OnceCell<String> = OnceCell::new();
+pub(crate) static GLOBAL_ENV_FILTER: OnceCell<String> = OnceCell::new();
 
 /// Options for the router
 #[derive(Parser, Debug)]
@@ -192,24 +192,31 @@ impl Executable {
             return Ok(());
         }
 
-        // This is more complex than I'd like it to be. Really, we just want to pass
-        // a FmtSubscriber to set_global_subscriber(), but we can't because of the
-        // generic nature of FmtSubscriber. See: https://github.com/tokio-rs/tracing/issues/380
-        // for more details.
         let builder = tracing_subscriber::fmt::fmt().with_env_filter(
             EnvFilter::try_new(&opt.log_level).context("could not parse log configuration")?,
         );
 
-        let subscriber: RouterSubscriber = if atty::is(atty::Stream::Stdout) {
-            RouterSubscriber::TextSubscriber(builder.finish())
+        let dispatcher = if atty::is(atty::Stream::Stdout) {
+            Dispatch::new(builder.finish())
         } else {
-            RouterSubscriber::JsonSubscriber(builder.json().finish())
+            Dispatch::new(builder.json().finish())
         };
 
-        set_global_subscriber(subscriber)?;
+        GLOBAL_ENV_FILTER.set(opt.log_level.clone()).unwrap();
 
-        GLOBAL_ENV_FILTER.set(opt.log_level).unwrap();
+        // The dispatcher we created is passed explicitely here to make sure we display the logs
+        // in the initialization pahse and in the state machine code, before a global subscriber
+        // is set using the configuration file
+        Self::inner_start(router_builder_fn, opt, dispatcher.clone())
+            .with_subscriber(dispatcher)
+            .await
+    }
 
+    async fn inner_start(
+        router_builder_fn: Option<fn(ConfigurationKind, SchemaKind) -> ApolloRouter>,
+        opt: Opt,
+        dispatcher: Dispatch,
+    ) -> Result<()> {
         let current_directory = std::env::current_dir()?;
 
         let configuration = opt
@@ -229,11 +236,12 @@ impl Executable {
                 }
             })
             .unwrap_or_else(|| Configuration::builder().build().into());
+
         let apollo_router_msg = format!("Apollo Router v{} // (c) Apollo Graph, Inc. // Licensed as ELv2 (https://go.apollo.dev/elv2)", std::env!("CARGO_PKG_VERSION"));
         let schema = match (opt.supergraph_path, opt.apollo_key) {
             (Some(supergraph_path), _) => {
                 tracing::info!("{apollo_router_msg}");
-                setup_panic_handler();
+                setup_panic_handler(dispatcher.clone());
 
                 let supergraph_path = if supergraph_path.is_relative() {
                     current_directory.join(supergraph_path)
@@ -248,6 +256,7 @@ impl Executable {
             }
             (None, Some(apollo_key)) => {
                 tracing::info!("{apollo_router_msg}");
+
                 let apollo_graph_ref = opt.apollo_graph_ref.ok_or_else(||anyhow!("cannot fetch the supergraph from Apollo Studio without setting the APOLLO_GRAPH_REF environment variable"))?;
                 if opt.apollo_uplink_poll_interval < Duration::from_secs(10) {
                     return Err(anyhow!("Apollo poll interval must be at least 10s"));
@@ -322,21 +331,23 @@ impl Executable {
     }
 }
 
-fn setup_panic_handler() {
+fn setup_panic_handler(dispatcher: Dispatch) {
     // Redirect panics to the logs.
     let backtrace_env = std::env::var("RUST_BACKTRACE");
     let show_backtraces =
         backtrace_env.as_deref() == Ok("1") || backtrace_env.as_deref() == Ok("full");
     if show_backtraces {
-        tracing::warn!("RUST_BACKTRACE={} detected. This use useful for diagnostics but will have a performance impact and may leak sensitive information", backtrace_env.as_ref().unwrap());
+        tracing::info!("RUST_BACKTRACE={} detected. This use useful for diagnostics but will have a performance impact and may leak sensitive information", backtrace_env.as_ref().unwrap());
     }
     std::panic::set_hook(Box::new(move |e| {
-        if show_backtraces {
-            let backtrace = backtrace::Backtrace::new();
-            tracing::error!("{}\n{:?}", e, backtrace)
-        } else {
-            tracing::error!("{}", e)
-        }
+        with_default(&dispatcher, || {
+            if show_backtraces {
+                let backtrace = backtrace::Backtrace::new();
+                tracing::error!("{}\n{:?}", e, backtrace)
+            } else {
+                tracing::error!("{}", e)
+            }
+        });
     }));
 }
 

apollo-router/src/lib.rs

@@ -62,15 +62,13 @@ pub mod layers;
 pub mod plugin;
 pub mod plugins;
 pub mod query_planner;
-mod reload;
 mod request;
 mod response;
 mod router;
 mod router_factory;
 pub mod services;
 mod spec;
 mod state_machine;
-pub mod subscriber;
 mod traits;
 
 pub use configuration::Configuration;

apollo-router/src/plugins/telemetry/mod.rs

@@ -8,6 +8,7 @@ use std::time::Duration;
 use std::time::Instant;
 
 use ::tracing::info_span;
+use ::tracing::subscriber::set_global_default;
 use ::tracing::Span;
 use apollo_spaceport::server::ReportSpaceport;
 use apollo_spaceport::StatsContext;
@@ -18,14 +19,14 @@ use futures::StreamExt;
 use http::HeaderValue;
 use http::StatusCode;
 use metrics::apollo::Sender;
+use once_cell::sync::OnceCell;
 use opentelemetry::global;
 use opentelemetry::propagation::TextMapPropagator;
 use opentelemetry::sdk::propagation::BaggagePropagator;
 use opentelemetry::sdk::propagation::TextMapCompositePropagator;
 use opentelemetry::sdk::propagation::TraceContextPropagator;
 use opentelemetry::sdk::trace::Builder;
 use opentelemetry::trace::SpanKind;
-use opentelemetry::trace::Tracer;
 use opentelemetry::trace::TracerProvider;
 use opentelemetry::KeyValue;
 use router_bridge::planner::UsageReporting;
@@ -35,11 +36,14 @@ use tower::util::BoxService;
 use tower::BoxError;
 use tower::ServiceBuilder;
 use tower::ServiceExt;
+use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt;
+use tracing_subscriber::EnvFilter;
 use url::Url;
 
 use self::config::Conf;
 use self::metrics::AttributesForwardConf;
 use self::metrics::MetricsAttributesConf;
+use crate::executable::GLOBAL_ENV_FILTER;
 use crate::http_ext;
 use crate::layers::ServiceBuilderExt;
 use crate::plugin::Handler;
@@ -59,7 +63,6 @@ use crate::plugins::telemetry::metrics::MetricsExporterHandle;
 use crate::plugins::telemetry::tracing::TracingConfigurator;
 use crate::query_planner::USAGE_REPORTING;
 use crate::register_plugin;
-use crate::subscriber::replace_layer;
 use crate::Context;
 use crate::ExecutionRequest;
 use crate::ExecutionResponse;
@@ -85,9 +88,10 @@ pub(crate) static STUDIO_EXCLUDE: &str = "apollo_telemetry::studio::exclude";
 const SERVICE_NAME_RESOURCE: &str = "service.name";
 const DEFAULT_SERVICE_NAME: &str = "apollo-router";
 
+static TELEMETRY_LOADED: OnceCell<bool> = OnceCell::new();
+
 pub struct Telemetry {
     config: config::Conf,
-    tracer_provider: Option<opentelemetry::sdk::trace::TracerProvider>,
     // Do not remove _metrics_exporters. Metrics will not be exported if it is removed.
     // Typically the handles are a PushController but may be something else. Dropping the handle will
     // shutdown exporter.
@@ -133,19 +137,6 @@ fn setup_metrics_exporter<T: MetricsConfigurator>(
 
 impl Drop for Telemetry {
     fn drop(&mut self) {
-        if let Some(tracer_provider) = self.tracer_provider.take() {
-            // Tracer providers must be flushed. This may happen as part of otel if the provider was set
-            // as the global, but may also happen in the case of an failed config reload.
-            // If the tracer prover is present then it was not handed over so we must flush it.
-            // We must make the call to force_flush() from a separate thread to prevent hangs:
-            // see https://github.com/open-telemetry/opentelemetry-rust/issues/536.
-            ::tracing::debug!("flushing telemetry");
-            let jh = tokio::task::spawn_blocking(move || {
-                opentelemetry::trace::TracerProvider::force_flush(&tracer_provider);
-            });
-            futures::executor::block_on(jh).expect("failed to flush tracer provider");
-        }
-
         if let Some(sender) = self.spaceport_shutdown.take() {
             ::tracing::debug!("notifying spaceport to shut down");
             let _ = sender.send(());
@@ -157,29 +148,7 @@ impl Drop for Telemetry {
 impl Plugin for Telemetry {
     type Config = config::Conf;
 
-    fn activate(&mut self) {
-        // The active service is about to be swapped in.
-        // The rest of this code in this method is expected to succeed.
-        // The issue is that Otel uses globals for a bunch of stuff.
-        // If we move to a completely tower based architecture then we could make this nicer.
-        let tracer_provider = self
-            .tracer_provider
-            .take()
-            .expect("trace_provider will have been set in startup, qed");
-        let tracer = tracer_provider.versioned_tracer(
-            "apollo-router",
-            Some(env!("CARGO_PKG_VERSION")),
-            None,
-        );
-        let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
-        Self::replace_tracer_provider(tracer_provider);
-
-        replace_layer(Box::new(telemetry))
-            .expect("set_global_subscriber() was not called at startup, fatal");
-        opentelemetry::global::set_error_handler(handle_error)
-            .expect("otel error handler lock poisoned, fatal");
-        global::set_text_map_propagator(Self::create_propagator(&self.config));
-    }
+    fn activate(&mut self) {}
 
     async fn new(init: PluginInit<Self::Config>) -> Result<Self, BoxError> {
         // Apollo config is special because we enable tracing if some env variables are present.
@@ -221,17 +190,52 @@ impl Plugin for Telemetry {
         // eventually be one.
         let mut builder = Self::create_metrics_exporters(&config)?;
 
-        //// THIS IS IMPORTANT
-        // Once the trace provider has been created this method MUST NOT FAIL
-        // The trace provider will not be shut down if drop is not called and it will result in a hang.
-        // Don't add anything fallible after the tracer provider has been created.
-        let tracer_provider = Self::create_tracer_provider(&config)?;
+        // the global tracer and subscriber initialization step must be performed only once
+        TELEMETRY_LOADED.get_or_try_init::<_, BoxError>(|| {
+            use anyhow::Context;
+            let tracer_provider = Self::create_tracer_provider(&config)?;
+
+            let tracer = tracer_provider.versioned_tracer(
+                "apollo-router",
+                Some(env!("CARGO_PKG_VERSION")),
+                None,
+            );
+
+            opentelemetry::global::set_tracer_provider(tracer_provider);
+            opentelemetry::global::set_error_handler(handle_error)
+                .expect("otel error handler lock poisoned, fatal");
+            global::set_text_map_propagator(Self::create_propagator(&config));
+
+            let log_level = GLOBAL_ENV_FILTER.get().unwrap();
+
+            let sub_builder = tracing_subscriber::fmt::fmt().with_env_filter(
+                EnvFilter::try_new(log_level).context("could not parse log configuration")?,
+            );
+
+            if atty::is(atty::Stream::Stdout) {
+                let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
+
+                let subscriber = sub_builder.finish().with(telemetry);
+                if let Err(e) = set_global_default(subscriber) {
+                    ::tracing::error!("cannot set global subscriber: {:?}", e);
+                }
+            } else {
+                let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
+
+                let subscriber = sub_builder.json().finish().with(telemetry);
+                if let Err(e) = set_global_default(subscriber) {
+                    ::tracing::error!("cannot set global subscriber: {:?}", e);
+                }
+            };
+
+            Ok(true)
+        })?;
+
         //
         // let metrics_response_queries = Self::create_metrics_queries(&config)?;
 
         let plugin = Ok(Telemetry {
             spaceport_shutdown: shutdown_tx,
-            tracer_provider: Some(tracer_provider),
             custom_endpoints: builder.custom_endpoints(),
             _metrics_exporters: builder.exporters(),
             meter_provider: builder.meter_provider(),
@@ -670,20 +674,6 @@ impl Telemetry {
         )
     }
 
-    fn replace_tracer_provider<T>(tracer_provider: T)
-    where
-        T: TracerProvider + Send + Sync + 'static,
-        <T as TracerProvider>::Tracer: Send + Sync + 'static,
-        <<T as opentelemetry::trace::TracerProvider>::Tracer as Tracer>::Span:
-            Send + Sync + 'static,
-    {
-        let jh = tokio::task::spawn_blocking(|| {
-            opentelemetry::global::force_flush_tracer_provider();
-            opentelemetry::global::set_tracer_provider(tracer_provider);
-        });
-        futures::executor::block_on(jh).expect("failed to replace tracer provider");
-    }
-
     fn router_service_span(config: apollo::Config) -> impl Fn(&RouterRequest) -> Span + Clone {
         let client_name_header = config.client_name_header;
         let client_version_header = config.client_version_header;