[ISSUE #8090]♻️Report telemetry shutdown health and dropped log lines (#8091)

Author: mxsm

rocketmq-broker/src/bin/broker_bootstrap_server.rs

@@ -111,6 +111,7 @@ async fn run(service_context: ServiceContext) -> Result<()> {
     if let Err(error) = verify_rocketmq_home() {
         telemetry_guard
             .shutdown()
+            .into_result()
             .context("failed to shutdown broker telemetry bootstrap after ROCKETMQ_HOME validation failed")?;
         return Err(error);
     }
@@ -120,6 +121,7 @@ async fn run(service_context: ServiceContext) -> Result<()> {
         print_config(&broker_config, &message_store_config, args.print_important_config);
         telemetry_guard
             .shutdown()
+            .into_result()
             .context("failed to shutdown broker telemetry bootstrap after printing broker configuration")?;
         return Ok(());
     }

rocketmq-broker/src/broker_runtime.rs

@@ -520,6 +520,17 @@ impl BrokerShutdownComponentReport {
         }
     }
 
+    pub(crate) fn completed_with_detail(name: &'static str, elapsed: Duration, detail: impl Into<String>) -> Self {
+        Self {
+            name,
+            present: true,
+            healthy: true,
+            timed_out: false,
+            elapsed,
+            detail: Some(detail.into()),
+        }
+    }
+
     pub(crate) fn unhealthy(name: &'static str, elapsed: Duration, detail: impl Into<String>) -> Self {
         Self {
             name,
@@ -555,6 +566,18 @@ impl BrokerShutdownComponentReport {
             Self::unhealthy(name, elapsed, report.to_json())
         }
     }
+
+    pub(crate) fn from_telemetry_shutdown_report(
+        report: &rocketmq_observability::TelemetryShutdownReport,
+        elapsed: Duration,
+    ) -> Self {
+        let detail = report.to_json();
+        if report.is_healthy() {
+            Self::completed_with_detail("observability", elapsed, detail)
+        } else {
+            Self::unhealthy("observability", elapsed, detail)
+        }
+    }
 }
 
 impl BrokerBasicServiceShutdownReport {
@@ -971,14 +994,15 @@ impl BrokerRuntime {
         {
             let started = Instant::now();
             if let Some(guard) = self.inner.observability_guard.take() {
-                if let Err(error) = guard.shutdown() {
-                    warn!("Failed to shutdown observability runtime: {error}");
-                    shutdown_report.observability =
-                        BrokerShutdownComponentReport::unhealthy("observability", started.elapsed(), error.to_string());
-                } else {
-                    shutdown_report.observability =
-                        BrokerShutdownComponentReport::completed("observability", started.elapsed());
+                let telemetry_report = guard.shutdown();
+                if !telemetry_report.is_healthy() {
+                    warn!(
+                        report = %telemetry_report.to_json(),
+                        "Failed to shutdown observability runtime cleanly"
+                    );
                 }
+                shutdown_report.observability =
+                    BrokerShutdownComponentReport::from_telemetry_shutdown_report(&telemetry_report, started.elapsed());
             } else {
                 shutdown_report.observability = BrokerShutdownComponentReport::skipped("observability");
             }
@@ -5148,6 +5172,57 @@ mod tests {
         assert_eq!(report.timed_out_component_names(), vec!["message_store"]);
     }
 
+    #[test]
+    fn broker_shutdown_report_records_healthy_telemetry_detail() {
+        let telemetry_report = rocketmq_observability::TelemetryShutdownReport {
+            subscriber_installed: true,
+            file_log_enabled: true,
+            dropped_log_lines: 0,
+            logs_shutdown_ok: true,
+            traces_shutdown_ok: true,
+            metrics_shutdown_ok: true,
+            logs_shutdown_error: None,
+            traces_shutdown_error: None,
+            metrics_shutdown_error: None,
+        };
+
+        let component =
+            BrokerShutdownComponentReport::from_telemetry_shutdown_report(&telemetry_report, Duration::from_millis(4));
+
+        assert!(component.present);
+        assert!(component.healthy);
+        assert_eq!(component.name, "observability");
+        let detail = component.detail.expect("telemetry detail should be recorded");
+        assert!(detail.contains("\"subscriber_installed\": true"));
+        assert!(detail.contains("\"file_log_enabled\": true"));
+    }
+
+    #[test]
+    fn broker_shutdown_report_marks_telemetry_provider_failure_unhealthy() {
+        let telemetry_report = rocketmq_observability::TelemetryShutdownReport {
+            subscriber_installed: true,
+            file_log_enabled: false,
+            dropped_log_lines: 0,
+            logs_shutdown_ok: false,
+            traces_shutdown_ok: true,
+            metrics_shutdown_ok: true,
+            logs_shutdown_error: Some("logger provider failed".to_string()),
+            traces_shutdown_error: None,
+            metrics_shutdown_error: None,
+        };
+
+        let component =
+            BrokerShutdownComponentReport::from_telemetry_shutdown_report(&telemetry_report, Duration::from_millis(5));
+
+        assert!(component.present);
+        assert!(!component.healthy);
+        assert_eq!(component.name, "observability");
+        assert!(component
+            .detail
+            .expect("telemetry failure detail should be recorded")
+            .contains("logger provider failed"));
+    }
+
     #[tokio::test]
     async fn initial_rpc_hooks_rejects_malformed_inner_client_credentials() {
         let broker_config = Arc::new(BrokerConfig {

rocketmq-controller/src/bin/controller_bootstrap.rs

@@ -128,12 +128,13 @@ async fn run(service_context: ServiceContext) -> Result<()> {
     let controller_result = run_controller(config, service_context).await;
     let shutdown_result = telemetry_guard
         .shutdown()
+        .into_result()
         .context("failed to shutdown controller telemetry bootstrap");
 
     match (controller_result, shutdown_result) {
         (Err(error), _) => Err(error),
         (Ok(()), Err(error)) => Err(error),
-        (Ok(()), Ok(())) => Ok(()),
+        (Ok(()), Ok(_report)) => Ok(()),
     }
 }
 

rocketmq-namesrv/src/bin/namesrv_bootstrap_server.rs

@@ -138,12 +138,13 @@ async fn run(service_context: ServiceContext) -> Result<()> {
         .map_err(anyhow::Error::from);
     let shutdown_result = telemetry_guard
         .shutdown()
+        .into_result()
         .context("failed to shutdown namesrv telemetry bootstrap");
 
     match (boot_result, shutdown_result) {
         (Err(error), _) => Err(error),
         (Ok(()), Err(error)) => Err(error),
-        (Ok(()), Ok(())) => Ok(()),
+        (Ok(()), Ok(_report)) => Ok(()),
     }
 }
 

rocketmq-observability/src/init.rs

@@ -20,6 +20,48 @@ use crate::config::SubscriberInstallPolicy;
 use crate::config::SubscriberInstallStatus;
 use crate::error::ObservabilityError;
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct TelemetryProviderShutdownReport {
+    pub(crate) logs_shutdown_ok: bool,
+    pub(crate) traces_shutdown_ok: bool,
+    pub(crate) metrics_shutdown_ok: bool,
+    pub(crate) logs_shutdown_error: Option<String>,
+    pub(crate) traces_shutdown_error: Option<String>,
+    pub(crate) metrics_shutdown_error: Option<String>,
+}
+
+impl Default for TelemetryProviderShutdownReport {
+    fn default() -> Self {
+        Self {
+            logs_shutdown_ok: true,
+            traces_shutdown_ok: true,
+            metrics_shutdown_ok: true,
+            logs_shutdown_error: None,
+            traces_shutdown_error: None,
+            metrics_shutdown_error: None,
+        }
+    }
+}
+
+impl TelemetryProviderShutdownReport {
+    pub(crate) fn is_healthy(&self) -> bool {
+        self.logs_shutdown_ok && self.traces_shutdown_ok && self.metrics_shutdown_ok
+    }
+
+    pub(crate) fn into_result(self) -> Result<(), ObservabilityError> {
+        if let Some(error) = self.logs_shutdown_error {
+            return Err(ObservabilityError::logs_shutdown(error));
+        }
+        if let Some(error) = self.traces_shutdown_error {
+            return Err(ObservabilityError::traces_shutdown(error));
+        }
+        if let Some(error) = self.metrics_shutdown_error {
+            return Err(ObservabilityError::metrics_shutdown(error));
+        }
+        Ok(())
+    }
+}
+
 #[derive(Debug, Default)]
 pub struct TelemetryGuard {
     subscriber_install_status: SubscriberInstallStatus,
@@ -39,6 +81,10 @@ impl TelemetryGuard {
     }
 
     pub fn shutdown(self) -> Result<(), ObservabilityError> {
+        self.shutdown_with_report().into_result()
+    }
+
+    pub(crate) fn shutdown_with_report(self) -> TelemetryProviderShutdownReport {
         self.shutdown_inner()
     }
 
@@ -51,33 +97,44 @@ impl TelemetryGuard {
     }
 
     #[cfg(any(feature = "otel-metrics", feature = "otel-traces", feature = "otel-logs"))]
-    fn shutdown_inner(mut self) -> Result<(), ObservabilityError> {
-        #[cfg(feature = "prometheus")]
-        if let Some(prometheus_http) = self.prometheus_http.take() {
-            prometheus_http.shutdown();
-        }
+    fn shutdown_inner(mut self) -> TelemetryProviderShutdownReport {
+        let mut report = TelemetryProviderShutdownReport::default();
 
-        #[cfg(feature = "otel-metrics")]
-        if let Some(provider) = self.meter_provider.take() {
-            provider.shutdown().map_err(ObservabilityError::metrics_shutdown)?;
+        #[cfg(feature = "otel-logs")]
+        if let Some(provider) = self.logger_provider.take() {
+            if let Err(error) = provider.shutdown() {
+                report.logs_shutdown_ok = false;
+                report.logs_shutdown_error = Some(error.to_string());
+            }
         }
 
         #[cfg(feature = "otel-traces")]
         if let Some(provider) = self.tracer_provider.take() {
-            provider.shutdown().map_err(ObservabilityError::traces_shutdown)?;
+            if let Err(error) = provider.shutdown() {
+                report.traces_shutdown_ok = false;
+                report.traces_shutdown_error = Some(error.to_string());
+            }
         }
 
-        #[cfg(feature = "otel-logs")]
-        if let Some(provider) = self.logger_provider.take() {
-            provider.shutdown().map_err(ObservabilityError::logs_shutdown)?;
+        #[cfg(feature = "otel-metrics")]
+        if let Some(provider) = self.meter_provider.take() {
+            if let Err(error) = provider.shutdown() {
+                report.metrics_shutdown_ok = false;
+                report.metrics_shutdown_error = Some(error.to_string());
+            }
         }
 
-        Ok(())
+        #[cfg(feature = "prometheus")]
+        if let Some(prometheus_http) = self.prometheus_http.take() {
+            prometheus_http.shutdown();
+        }
+
+        report
     }
 
     #[cfg(not(any(feature = "otel-metrics", feature = "otel-traces", feature = "otel-logs")))]
-    fn shutdown_inner(self) -> Result<(), ObservabilityError> {
-        Ok(())
+    fn shutdown_inner(self) -> TelemetryProviderShutdownReport {
+        TelemetryProviderShutdownReport::default()
     }
 
     #[cfg(feature = "otel-metrics")]

rocketmq-observability/src/lib.rs

@@ -47,6 +47,7 @@ pub use logging::install_global;
 pub use logging::FileLogLayer;
 pub use logging::LoggingGuard;
 pub use logging::TelemetryRuntimeGuard;
+pub use logging::TelemetryShutdownReport;
 
 #[cfg(feature = "prometheus")]
 #[doc(hidden)]