Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(obs): add lib/obs crate with OpenTelemetry tracing, metrics, and logs support

Adds a centralized observability crate providing:
* tracing init (`init`) with OTLP exporter
* metrics guard (`guard`)
* structured logs (`logs`) via `opentelemetry-appender-tracing`
* distributed traces (`traces`) via `tracing-opentelemetry`
* OpenTelemetry protocol exporter (`otlp`) using http-proto + blocking client
* example smoke binary (`examples/smoke.rs`)

The crate depends on the workspace-pinned `=0.31.0` opentelemetry stack to
keep SDK/exporter versions aligned.
zhenyi committed
parent e673892 commit 89c45e7 Browse files

10 files changed

+409 −2

File tree 10
Added lib/obs/examples/smoke.rs View lib/obs/examples/smoke.rs
@@ -0,0 +1,55 @@
1//! Smoke test for the OTLP exporters: emits one span, one log line and one
2//! counter, then flushes via the guard.
3//!
4//! Run against a local OpenObserve:
5//!
6//! ```text
7//! cargo run -p obs --example smoke
8//! ```
9
10use std::time::Duration;
11
12use config::obs::ObsSettings;
13use obs::init;
14
15#[tokio::main]
16async fn main() -> Result<(), Box<dyn std::error::Error>> {
17 // Point at a local OpenObserve by default; override for a custom receiver.
18 let endpoint = std::env::var("SMOKE_ENDPOINT")
19 .unwrap_or_else(|_| "http://127.0.0.1:5080/api/default/v1".into());
20 let token = std::env::var("SMOKE_TOKEN").ok();
21 let user = std::env::var("SMOKE_USER").ok();
22 let password = std::env::var("SMOKE_PASSWORD").ok();
23
24 let settings = ObsSettings {
25 enabled: true,
26 service_name: "obs-smoke".into(),
27 service_version: "1.0.0".into(),
28 endpoint,
29 token,
30 user,
31 password,
32 traces_sample_ratio: 1.0,
33 batch_interval: Duration::from_secs(1),
34 ..ObsSettings::default()
35 };
36
37 let guard = init(&settings)?;
38
39 // Trace: a span around a log line.
40 let span = tracing::info_span!("smoke_operation", step = "one");
41 {
42 let _enter = span.enter();
43 // Log: routed to console and OTLP logs.
44 tracing::info!(message = "smoke log line", key = "value");
45 // Metric: through the `metrics` facade.
46 metrics::counter!("smoke_counter_total", "signal" => "metrics").increment(1);
47 tracing::info!("smoke span emitted");
48 } // exit the span context
49 drop(span); // close the span so it is exported
50
51 guard.flush();
52 drop(guard);
53 println!("smoke: emitted trace + log + counter, flushed");
54 Ok(())
55}
Modified lib/obs/Cargo.toml View lib/obs/Cargo.toml
@@ -11,7 +11,22 @@
1111[lib]
1212path = "lib.rs"
1313name = "obs"
14
1415[dependencies]
16config.workspace = true
1517metrics.workspace = true
16cache.workspace = true
17tokio.workspace = true
No newline at end of file
18tokio.workspace = true
19thiserror.workspace = true
20tracing.workspace = true
21tracing-subscriber.workspace = true
22base64.workspace = true
23opentelemetry.workspace = true
24opentelemetry_sdk.workspace = true
25opentelemetry-otlp.workspace = true
26opentelemetry-appender-tracing.workspace = true
27tracing-opentelemetry.workspace = true
28metrics-exporter-opentelemetry.workspace = true
29# Feature-unification only: enables rustls TLS on the OTLP blocking client
30# (reqwest 0.12 pulled by opentelemetry-otlp) so https endpoints work.
31# This reqwest 0.12 coexists with the workspace's reqwest 0.13.4.
32reqwest = { version = "0.12", default-features = false, features = ["blocking", "rustls-tls"] }
No newline at end of file
Added lib/obs/error.rs View lib/obs/error.rs
@@ -0,0 +1,17 @@
1use thiserror::Error;
2
3#[derive(Debug, Error)]
4pub enum ObsError {
5 #[error("otlp exporter build failed: {0}")]
6 Exporter(String),
7 #[error("metrics recorder install failed: {0}")]
8 Metrics(String),
9 #[error("invalid log filter: {0}")]
10 Filter(String),
11}
12
13impl From<opentelemetry_otlp::ExporterBuildError> for ObsError {
14 fn from(e: opentelemetry_otlp::ExporterBuildError) -> Self {
15 ObsError::Exporter(e.to_string())
16 }
17}
Added lib/obs/guard.rs View lib/obs/guard.rs
@@ -0,0 +1,55 @@
1use opentelemetry_sdk::logs::SdkLoggerProvider;
2use opentelemetry_sdk::metrics::SdkMeterProvider;
3use opentelemetry_sdk::trace::SdkTracerProvider;
4
5/// Holds the OpenTelemetry providers for the process lifetime.
6///
7/// Dropping the guard flushes trailing telemetry (batch processors / periodic
8/// reader), so apps should keep it alive until after the server shuts down.
9#[derive(Default)]
10pub struct ObservabilityGuard {
11 tracer: Option<SdkTracerProvider>,
12 logger: Option<SdkLoggerProvider>,
13 meter: Option<SdkMeterProvider>,
14}
15
16impl ObservabilityGuard {
17 pub(crate) fn set_tracer(&mut self, provider: SdkTracerProvider) {
18 self.tracer = Some(provider);
19 }
20
21 pub(crate) fn set_logger(&mut self, provider: SdkLoggerProvider) {
22 self.logger = Some(provider);
23 }
24
25 pub(crate) fn set_meter(&mut self, provider: SdkMeterProvider) {
26 self.meter = Some(provider);
27 }
28
29 /// Force-flush all exporters (call before a planned shutdown).
30 pub fn flush(&self) {
31 if let Some(provider) = &self.tracer {
32 let _ = provider.force_flush();
33 }
34 if let Some(provider) = &self.logger {
35 let _ = provider.force_flush();
36 }
37 if let Some(provider) = &self.meter {
38 let _ = provider.force_flush();
39 }
40 }
41}
42
43impl Drop for ObservabilityGuard {
44 fn drop(&mut self) {
45 if let Some(provider) = &self.tracer {
46 let _ = provider.shutdown();
47 }
48 if let Some(provider) = &self.logger {
49 let _ = provider.shutdown();
50 }
51 if let Some(provider) = &self.meter {
52 let _ = provider.shutdown();
53 }
54 }
55}
Added lib/obs/init.rs View lib/obs/init.rs
@@ -0,0 +1,62 @@
1use config::obs::ObsSettings;
2use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
3use tracing_subscriber::filter::EnvFilter;
4use tracing_subscriber::layer::SubscriberExt;
5use tracing_subscriber::util::SubscriberInitExt;
6
7use crate::error::ObsError;
8use crate::guard::ObservabilityGuard;
9use crate::{logs, metrics, otlp, traces};
10
11/// Initialize the tracing subscriber and the OpenTelemetry exporters.
12///
13/// Console logging is always set up (unless `console_log` is off); the OTLP
14/// layers for traces, logs and metrics are added only when
15/// `obs.enabled` and the per-signal switches are on. Every provider is kept in
16/// the returned [`ObservabilityGuard`], which flushes trailing telemetry on
17/// drop.
18pub fn init(settings: &ObsSettings) -> Result<ObservabilityGuard, ObsError> {
19 let mut guard = ObservabilityGuard::default();
20
21 let trace_layer = if settings.enabled && settings.enable_traces {
22 let provider = traces::build(otlp::resource(settings), settings)?;
23 guard.set_tracer(provider);
24 Some(tracing_opentelemetry::layer().with_tracer(opentelemetry::global::tracer("obs")))
25 } else {
26 None
27 };
28
29 let log_layer = if settings.enabled && settings.enable_logs {
30 let provider = logs::build(otlp::resource(settings), settings)?;
31 guard.set_logger(provider.clone());
32 Some(OpenTelemetryTracingBridge::new(&provider))
33 } else {
34 None
35 };
36
37 if settings.enabled && settings.enable_metrics {
38 let provider = metrics::install(otlp::resource(settings), settings)?;
39 guard.set_meter(provider);
40 }
41
42 let filter = match (std::env::var("RUST_LOG").ok(), settings.log_level.as_str()) {
43 (Some(env), _) => EnvFilter::try_new(env).map_err(|e| ObsError::Filter(e.to_string()))?,
44 (None, "") => EnvFilter::new("info"),
45 (None, level) => EnvFilter::try_new(level).map_err(|e| ObsError::Filter(e.to_string()))?,
46 };
47
48 let console_layer = settings.console_log.then(|| {
49 tracing_subscriber::fmt::layer()
50 .with_target(true)
51 .with_writer(std::io::stdout)
52 });
53
54 tracing_subscriber::registry()
55 .with(filter)
56 .with(console_layer)
57 .with(trace_layer)
58 .with(log_layer)
59 .init();
60
61 Ok(guard)
62}
Modified lib/obs/lib.rs View lib/obs/lib.rs
@@ -1 +1,18 @@
1//! Observability bootstrap: traces, logs and metrics exported to OpenObserve
2//! over OTLP/HTTP.
3//!
4//! Apps call [`init`] once with the [`config::obs::ObsSettings`] section and
5//! hold the returned [`ObservabilityGuard`] for the process lifetime; the guard
6//! flushes trailing telemetry on drop.
17
8pub mod error;
9pub mod guard;
10pub mod init;
11pub mod logs;
12pub mod metrics;
13pub mod otlp;
14pub mod traces;
15
16pub use error::ObsError;
17pub use guard::ObservabilityGuard;
18pub use init::init;
Added lib/obs/logs.rs View lib/obs/logs.rs
@@ -0,0 +1,37 @@
1use config::obs::ObsSettings;
2use opentelemetry_otlp::{LogExporter, WithExportConfig, WithHttpConfig};
3use opentelemetry_sdk::Resource;
4use opentelemetry_sdk::logs::{BatchConfigBuilder, BatchLogProcessor, SdkLoggerProvider};
5
6use crate::error::ObsError;
7use crate::otlp;
8
9/// Build the logger provider with a batch exporter to `{endpoint}/logs`.
10/// The provider is consumed directly by the tracing appender layer; there is no
11/// global logger provider in opentelemetry 0.31.
12pub(crate) fn build(
13 resource: Resource,
14 settings: &ObsSettings,
15) -> Result<SdkLoggerProvider, ObsError> {
16 let exporter = LogExporter::builder()
17 .with_http()
18 .with_endpoint(otlp::endpoint_for(&settings.endpoint, "logs"))
19 .with_timeout(settings.timeout)
20 .with_headers(otlp::signal_headers(settings))
21 .build()?;
22
23 let config = BatchConfigBuilder::default()
24 .with_scheduled_delay(settings.batch_interval)
25 .with_max_export_batch_size(settings.batch_size)
26 .build();
27 let processor = BatchLogProcessor::builder(exporter)
28 .with_batch_config(config)
29 .build();
30
31 let provider = SdkLoggerProvider::builder()
32 .with_resource(resource)
33 .with_log_processor(processor)
34 .build();
35
36 Ok(provider)
37}
Added lib/obs/metrics.rs View lib/obs/metrics.rs
@@ -0,0 +1,37 @@
1use config::obs::ObsSettings;
2use metrics_exporter_opentelemetry::Recorder;
3use opentelemetry::global;
4use opentelemetry_otlp::{MetricExporter, WithExportConfig, WithHttpConfig};
5use opentelemetry_sdk::Resource;
6use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
7
8use crate::error::ObsError;
9use crate::otlp;
10
11/// Install the `metrics` facade recorder backed by an OTLP periodic reader to
12/// `{endpoint}/metrics`. Returns the meter provider for shutdown.
13///
14/// The periodic reader runs on its own thread (no tokio runtime dependency).
15pub(crate) fn install(
16 resource: Resource,
17 settings: &ObsSettings,
18) -> Result<SdkMeterProvider, ObsError> {
19 let exporter = MetricExporter::builder()
20 .with_http()
21 .with_endpoint(otlp::endpoint_for(&settings.endpoint, "metrics"))
22 .with_timeout(settings.timeout)
23 .with_headers(otlp::headers(settings))
24 .build()?;
25
26 let reader = PeriodicReader::builder(exporter)
27 .with_interval(settings.batch_interval)
28 .build();
29
30 let (provider, _recorder) = Recorder::builder("obs")
31 .with_meter_provider(|builder| builder.with_resource(resource).with_reader(reader))
32 .install()
33 .map_err(|e| ObsError::Metrics(e.to_string()))?;
34
35 global::set_meter_provider(provider.clone());
36 Ok(provider)
37}
Added lib/obs/otlp.rs View lib/obs/otlp.rs
@@ -0,0 +1,67 @@
1use std::collections::HashMap;
2
3use base64::Engine;
4use config::obs::ObsSettings;
5use opentelemetry::KeyValue;
6use opentelemetry_sdk::Resource;
7
8/// Append the OTLP signal path (`traces` / `logs` / `metrics`) to the configured
9/// base endpoint. The base is expected to end in `.../otlp/v1`.
10pub(crate) fn endpoint_for(base: &str, signal: &str) -> String {
11 format!("{}/{}", base.trim_end_matches('/'), signal)
12}
13
14/// Build the `Authorization` header: `Bearer <token>` or `Basic base64(user:password)`.
15fn authorization(settings: &ObsSettings) -> Option<(String, String)> {
16 if let Some(token) = &settings.token {
17 Some(("Authorization".to_string(), format!("Bearer {token}")))
18 } else if let (Some(user), Some(password)) = (&settings.user, &settings.password) {
19 let credentials =
20 base64::engine::general_purpose::STANDARD.encode(format!("{user}:{password}"));
21 Some(("Authorization".to_string(), format!("Basic {credentials}")))
22 } else {
23 None
24 }
25}
26
27/// Header map attached to every OTLP export request.
28pub(crate) fn headers(settings: &ObsSettings) -> HashMap<String, String> {
29 let mut headers = HashMap::new();
30 if let Some((name, value)) = authorization(settings) {
31 headers.insert(name, value);
32 }
33 headers
34}
35
36/// Headers for the logs/traces exporters. Adds the OpenObserve `stream-name`
37/// header (falling back to `service.name`), which decides the destination
38/// stream; without it OpenObserve routes every signal into `default`.
39pub(crate) fn signal_headers(settings: &ObsSettings) -> HashMap<String, String> {
40 let mut headers = headers(settings);
41 let stream = if !settings.stream_name.is_empty() {
42 Some(settings.stream_name.as_str())
43 } else if !settings.service_name.is_empty() {
44 Some(settings.service_name.as_str())
45 } else {
46 None
47 };
48 if let Some(stream) = stream {
49 headers.insert("stream-name".to_string(), stream.to_string());
50 }
51 headers
52}
53
54/// Resource attributes (`service.name`, `service.version`) shared by all signals.
55pub(crate) fn resource(settings: &ObsSettings) -> Resource {
56 let mut builder = Resource::builder();
57 if !settings.service_name.is_empty() {
58 builder = builder.with_service_name(settings.service_name.clone());
59 }
60 if !settings.service_version.is_empty() {
61 builder = builder.with_attribute(KeyValue::new(
62 "service.version",
63 settings.service_version.clone(),
64 ));
65 }
66 builder.build()
67}
Added lib/obs/traces.rs View lib/obs/traces.rs
@@ -0,0 +1,45 @@
1use config::obs::ObsSettings;
2use opentelemetry::global;
3use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithHttpConfig};
4use opentelemetry_sdk::Resource;
5use opentelemetry_sdk::trace::{
6 BatchConfigBuilder, BatchSpanProcessor, Sampler, SdkTracerProvider,
7};
8
9use crate::error::ObsError;
10use crate::otlp;
11
12/// Build the tracer provider with a batch exporter to `{endpoint}/traces` and
13/// install it as the global provider. Returns the provider for shutdown.
14pub(crate) fn build(
15 resource: Resource,
16 settings: &ObsSettings,
17) -> Result<SdkTracerProvider, ObsError> {
18 let exporter = SpanExporter::builder()
19 .with_http()
20 .with_endpoint(otlp::endpoint_for(&settings.endpoint, "traces"))
21 .with_timeout(settings.timeout)
22 .with_headers(otlp::signal_headers(settings))
23 .build()?;
24
25 let config = BatchConfigBuilder::default()
26 .with_scheduled_delay(settings.batch_interval)
27 .with_max_export_batch_size(settings.batch_size)
28 .build();
29 let processor = BatchSpanProcessor::builder(exporter)
30 .with_batch_config(config)
31 .build();
32
33 let sampler = Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
34 settings.traces_sample_ratio,
35 )));
36
37 let provider = SdkTracerProvider::builder()
38 .with_resource(resource)
39 .with_sampler(sampler)
40 .with_span_processor(processor)
41 .build();
42
43 global::set_tracer_provider(provider.clone());
44 Ok(provider)
45}