rocketmq_observability/
lib.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod attributes;
16mod config;
17mod environment;
18mod error;
19mod exporter;
20mod exporter_types;
21mod handle;
22mod init;
23mod log_filter;
24mod logging;
25pub mod logs;
26pub mod metrics;
27mod noop;
28mod propagation;
29mod resolver;
30mod resource;
31mod runtime_diagnostics;
32mod sampling;
33pub mod semantic;
34pub mod statistics;
35pub mod stats;
36mod status;
37pub mod trace;
38
39pub use attributes::base_attributes;
40pub use attributes::Attribute;
41pub use attributes::MetricAttributes;
42pub use config::ConsoleLogConfig;
43pub use config::FileLogConfig;
44pub use config::LogFormat;
45pub use config::LogRotation;
46pub use config::LoggingConfig;
47pub use config::LogsExporter;
48pub use config::LogsOverrides;
49pub use config::MetricsExporter;
50pub use config::MetricsOverrides;
51pub use config::NonBlockingLogConfig;
52pub use config::ObservabilityConfig;
53pub use config::ObservabilityOverrides;
54pub use config::OtlpOverrides;
55pub use config::OtlpProtocol;
56pub use config::PrometheusOverrides;
57pub use config::ReloadConfig;
58pub use config::SubscriberInstallPolicy;
59pub use config::SubscriberInstallStatus;
60pub use config::TelemetryBootstrapConfig;
61pub use config::TraceExporter;
62pub use config::TracesConfig;
63pub use config::TracesOverrides;
64pub use environment::apply_standard_otlp_environment;
65pub use environment::apply_standard_otlp_environment_values;
66pub use environment::StandardOtlpEnvironmentStatus;
67pub use environment::OTEL_EXPORTER_OTLP_ENDPOINT;
68pub use environment::OTEL_EXPORTER_OTLP_PROTOCOL;
69pub use error::ObservabilityError;
70pub use error::ObservabilityFailureDetail;
71pub use error::ObservabilityOperation;
72pub use error::ObservabilityResult;
73pub use exporter::outage::*;
74pub use exporter_types::LogExporterType;
75pub use exporter_types::MetricsExporterType;
76pub use exporter_types::TraceExporterType;
77pub use handle::MetricsRuntimePolicy;
78#[cfg(feature = "otel-metrics")]
79pub use handle::ReleaseIdentityRegistrationError;
80pub use handle::TelemetryHandle;
81pub use handle::TelemetryRecorder;
82pub use handle::TelemetryState;
83pub use handle::TracePolicy;
84pub use handle::BROKER_METER_SCOPE;
85pub use handle::CLIENT_METER_SCOPE;
86pub use handle::CONTROLLER_METER_SCOPE;
87pub use handle::DASHBOARD_METER_SCOPE;
88pub use handle::MCP_METER_SCOPE;
89pub use handle::NAMESRV_METER_SCOPE;
90pub use handle::PROXY_METER_SCOPE;
91pub use handle::RUNTIME_METER_SCOPE;
92pub use handle::SRE_CONNECTOR_METER_SCOPE;
93pub use handle::SRE_CONTROL_PLANE_METER_SCOPE;
94pub use handle::STORE_METER_SCOPE;
95pub use handle::TIERED_STORE_METER_SCOPE;
96pub use handle::TRANSPORT_METER_SCOPE;
97pub use init::init_observability;
98pub use init::init_observability_with_service_context;
99pub use log_filter::read_rust_log;
100pub use log_filter::LogFilterHandle;
101pub use log_filter::LogFilterInputs;
102pub use log_filter::LogFilterReloadRequest;
103pub use log_filter::LogFilterResolver;
104pub use log_filter::LogFilterSource;
105pub use log_filter::LoggingOverrideConfig;
106pub use log_filter::LoggingOverrides;
107pub use log_filter::ResolvedLogFilter;
108pub use log_filter::DEFAULT_LOG_FILTER;
109pub use logging::install_global;
110pub use logging::install_global_with_filter;
111pub use logging::install_global_with_filter_and_service_context;
112pub use logging::install_global_with_service_context;
113pub use logging::reserve_telemetry_flush_lease;
114pub use logging::FileLogLayer;
115pub use logging::LoggingGuard;
116pub use logging::TelemetryRuntimeGuard;
117pub use logging::TelemetryShutdownReport;
118pub use metrics::dashboard::DashboardStorageBackend;
119pub use metrics::dashboard::DashboardStorageFailureLabel;
120pub use metrics::dashboard::DashboardStorageMetricsRecorder;
121pub use metrics::dashboard::DashboardStorageOperation;
122pub use metrics::dashboard::DashboardStorageOperationResult;
123pub use metrics::dashboard::DashboardStorageResult;
124pub use metrics::labels::MetricLabelPolicy;
125pub use metrics::labels::METRIC_LABEL_SENTINEL;
126#[cfg(feature = "otel-traces")]
127pub use propagation::add_current_span_event_with_status;
128#[cfg(feature = "otel-traces")]
129#[doc(hidden)]
130pub use propagation::bench_extract_trace_context;
131#[cfg(feature = "otel-traces")]
132pub use propagation::inject_current_context_with_handle;
133#[cfg(feature = "otel-traces")]
134pub use propagation::record_span_parent_assignment_error;
135#[cfg(feature = "otel-traces")]
136pub use propagation::set_span_parent_from_properties_with_handle;
137#[cfg(feature = "otel-traces")]
138pub use propagation::SpanParentError;
139pub use propagation::TRACEPARENT;
140pub use propagation::TRACESTATE;
141pub use resolver::normalize_and_validate;
142pub use resolver::resolve_telemetry_from_env;
143pub use resolver::resolve_telemetry_values;
144pub use resolver::TelemetryEnvironmentSpec;
145pub use resolver::TelemetryEnvironmentValues;
146pub use resolver::TelemetryResolution;
147pub use runtime_diagnostics::start_runtime_diagnostics_endpoint;
148pub use runtime_diagnostics::start_runtime_diagnostics_endpoint_from_env;
149pub use runtime_diagnostics::start_runtime_diagnostics_endpoint_from_env_with_telemetry;
150pub use runtime_diagnostics::RuntimeDiagnosticsEndpointConfig;
151pub use runtime_diagnostics::RuntimeDiagnosticsEndpointHandle;
152pub use runtime_diagnostics::RuntimeDiagnosticsSources;
153pub use runtime_diagnostics::RUNTIME_DIAGNOSTICS_ALLOW_INSECURE_HTTP_ENV;
154pub use runtime_diagnostics::RUNTIME_DIAGNOSTICS_BIND_ADDR_ENV;
155pub use runtime_diagnostics::RUNTIME_DIAGNOSTICS_ENDPOINT_SCHEMA;
156pub use runtime_diagnostics::RUNTIME_DIAGNOSTICS_PATH;
157pub use runtime_diagnostics::RUNTIME_DIAGNOSTICS_SAMPLE_INTERVAL_SECONDS_ENV;
158pub use runtime_diagnostics::RUNTIME_DIAGNOSTICS_SCOPE;
159pub use runtime_diagnostics::RUNTIME_DIAGNOSTICS_TOKEN_FILE_ENV;
160pub use runtime_diagnostics::RUNTIME_METADATA_TARGETS_PATH;
161pub use runtime_diagnostics::{
162    RuntimeDiagnosticsDataProvider, RuntimeDiagnosticsMode, RuntimeDiagnosticsService, RuntimeDiagnosticsStarted,
163    RUNTIME_DIAGNOSTICS_MODE_ENV,
164};
165pub use sampling::SamplingGate;
166pub use status::ObservabilityCompiledFeaturesV1;
167pub use status::ObservabilitySignalStatusV1;
168pub use status::ObservabilityStatusHandle;
169pub use status::ObservabilityStatusValueV1;
170pub use status::ObservabilityStatusViewV1;
171pub use status::ObservabilitySubscriberStatusV1;
172
173#[cfg(feature = "prometheus")]
174#[doc(hidden)]
175pub mod bench_support {
176    use std::time::Duration;
177    use std::time::Instant;
178
179    use opentelemetry::metrics::MeterProvider;
180    use rocketmq_runtime::ChildServiceContext;
181    use rocketmq_runtime::ShutdownReport;
182    use serde::Serialize;
183    use tokio::io::AsyncReadExt;
184    use tokio::io::AsyncWriteExt;
185    use tokio::net::TcpStream;
186
187    use crate::config::MetricsExporter;
188    use crate::config::ObservabilityConfig;
189
190    #[derive(Clone, Debug, Serialize)]
191    pub struct PrometheusLifecycleProbe {
192        pub task_count_before_scrape: usize,
193        pub task_count_before_shutdown: usize,
194        pub response_status_ok: bool,
195        pub response_contains_metric: bool,
196        pub response_bytes: usize,
197        pub shutdown_elapsed_us: u128,
198        pub shutdown_report: ShutdownReport,
199        pub healthy: bool,
200    }
201
202    pub async fn run_prometheus_lifecycle_probe(service_context: ChildServiceContext) -> PrometheusLifecycleProbe {
203        let mut config = ObservabilityConfig {
204            enabled: true,
205            ..ObservabilityConfig::default()
206        };
207        config.metrics.enabled = true;
208        config.metrics.exporter = MetricsExporter::Prometheus;
209        config.prometheus.host = "127.0.0.1".to_string();
210        config.prometheus.port = 0;
211        config.prometheus.path = "/metrics".to_string();
212
213        let prometheus = crate::exporter::prometheus::init_prometheus_metrics(&config)
214            .expect("prometheus metrics should initialize");
215        let meter = prometheus.provider().meter("observability-lifecycle-bench");
216        let counter = meter.u64_counter("rocketmq_observability_lifecycle_total").build();
217        counter.add(1, &[]);
218
219        let handle = crate::exporter::prometheus::spawn_prometheus_http_endpoint(
220            &config,
221            prometheus.registry().clone(),
222            &service_context,
223        )
224        .expect("prometheus HTTP endpoint should start");
225        let task_count_before_scrape = handle.task_count();
226
227        let response = scrape_metrics(handle.local_addr(), config.prometheus.path.as_str()).await;
228        let response_status_ok = response.starts_with("HTTP/1.1 200 OK");
229        let response_contains_metric = response.contains("rocketmq_observability_lifecycle_total");
230        let response_bytes = response.len();
231        let task_count_before_shutdown = handle.task_count();
232
233        let shutdown_started_at = Instant::now();
234        let shutdown_report = handle.shutdown_gracefully(Duration::from_secs(5)).await;
235        let shutdown_elapsed_us = shutdown_started_at.elapsed().as_micros();
236        let parent_report = service_context.task_group().shutdown(Duration::from_secs(1)).await;
237        let finished_accept_tasks = shutdown_report.completed + shutdown_report.cancelled;
238        let healthy = response_status_ok
239            && response_contains_metric
240            && shutdown_report.is_healthy()
241            && parent_report.is_healthy()
242            && finished_accept_tasks >= 1
243            && shutdown_report.leaked == 0
244            && shutdown_report.timed_out == 0;
245
246        PrometheusLifecycleProbe {
247            task_count_before_scrape,
248            task_count_before_shutdown,
249            response_status_ok,
250            response_contains_metric,
251            response_bytes,
252            shutdown_elapsed_us,
253            shutdown_report,
254            healthy,
255        }
256    }
257
258    async fn scrape_metrics(addr: std::net::SocketAddr, path: &str) -> String {
259        let mut stream = TcpStream::connect(addr)
260            .await
261            .expect("prometheus scrape should connect");
262        let request = format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
263        stream
264            .write_all(request.as_bytes())
265            .await
266            .expect("prometheus scrape request should write");
267        let mut response = Vec::new();
268        stream
269            .read_to_end(&mut response)
270            .await
271            .expect("prometheus scrape response should read");
272        String::from_utf8(response).expect("prometheus response should be utf8")
273    }
274}
275
276#[cfg(all(test, feature = "prometheus"))]
277mod bench_support_tests {
278    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
279    async fn prometheus_lifecycle_probe_reports_clean_shutdown() {
280        let runtime = rocketmq_runtime::RuntimeContext::from_current("prometheus-lifecycle-test");
281        let probe =
282            super::bench_support::run_prometheus_lifecycle_probe(runtime.service_context("prometheus-probe")).await;
283
284        assert!(probe.healthy, "{probe:?}");
285        assert!(probe.response_status_ok, "{probe:?}");
286        assert!(
287            probe.shutdown_report.is_healthy(),
288            "{}",
289            probe.shutdown_report.to_json()
290        );
291    }
292}
rocketmq_observability/lib.rs

rocketmq_observability/
lib.rs
