feat: add opentelemetry layers

Author: ymgyt

Cargo.lock

@@ -1,11 +1,12 @@
+use synd_o11y::{opentelemetry::OpenTelemetryGuard, tracing_subscriber::otel_metrics};
 use tracing::{error, info};
 
 use synd_api::{args, config, dependency::Dependency, serve::listen_and_serve};
 
-fn init_tracing() {
+fn init_tracing() -> Option<OpenTelemetryGuard> {
     use synd_o11y::{
         opentelemetry::init_propagation,
-        tracing_subscriber::{audit, otel_log},
+        tracing_subscriber::{audit, otel_log, otel_trace},
     };
     use tracing_subscriber::{
         filter::EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt as _, Layer as _,
@@ -19,14 +20,25 @@ fn init_tracing() {
     let show_src = true;
     let show_target = !show_src;
 
-    let otel_log_layer = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
-        .ok()
-        .filter(|endpoint| !endpoint.is_empty())
-        .map(|endpoint| {
-            let resource =
-                synd_o11y::opentelemetry::resource(config::NAME, config::VERSION, "local");
-            otel_log::layer(endpoint, resource)
-        });
+    let (opentelemetry_layers, guard) = {
+        match std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok() {
+            None => (None, None),
+            Some(endpoint) if endpoint.is_empty() => (None, None),
+            Some(endpoint) => {
+                let resource =
+                    synd_o11y::opentelemetry::resource(config::NAME, config::VERSION, "local");
+
+                let trace_layer = otel_trace::layer(resource.clone());
+                let log_layer = otel_log::layer(endpoint, resource.clone());
+                let metrics_layer = otel_metrics::layer(resource);
+
+                (
+                    Some(trace_layer.and_then(log_layer).and_then(metrics_layer)),
+                    Some(synd_o11y::opentelemetry::OpenTelemetryGuard),
+                )
+            }
+        }
+    };
 
     Registry::default()
         .with(
@@ -36,7 +48,7 @@ fn init_tracing() {
                 .with_file(show_src)
                 .with_line_number(show_src)
                 .with_target(show_target)
-                .and_then(otel_log_layer)
+                .and_then(opentelemetry_layers)
                 .with_filter(
                     EnvFilter::try_from_default_env()
                         .or_else(|_| EnvFilter::try_new("info"))
@@ -49,13 +61,15 @@ fn init_tracing() {
 
     // Set text map propagator globally
     init_propagation();
+
+    guard
 }
 
 #[tokio::main]
 async fn main() {
     let args = args::parse();
 
-    init_tracing();
+    let _guard = init_tracing();
 
     let dep = Dependency::new(args.kvsd).await.unwrap();
 

crates/synd_api/src/main.rs

@@ -11,7 +11,7 @@ impl<B> tower_http::trace::MakeSpan<B> for MakeSpan {
 
         let request_id = cx
             .baggage()
-            .get("request.id")
+            .get(synd_o11y::REQUEST_ID_KEY)
             .map_or("?".into(), |v| v.as_str());
 
         let span = tracing::span!(

crates/synd_api/src/serve/layer/trace.rs

@@ -20,6 +20,7 @@ opentelemetry-http                 = { workspace = true }
 opentelemetry-otlp                 = { version = "0.14.0", default-features = false, features = ["trace", "metrics", "logs", "grpc-tonic"] }
 opentelemetry-semantic-conventions = { workspace = true }
 opentelemetry_sdk                  = { workspace = true, features = ["logs", "rt-tokio"] }
+rand                               = { workspace = true }
 reqwest                            = { workspace = true }
 tracing                            = { workspace = true }
 tracing-opentelemetry              = { workspace = true }

crates/synd_o11y/Cargo.toml

@@ -1,2 +1,19 @@
+use ::opentelemetry::KeyValue;
+
 pub mod opentelemetry;
 pub mod tracing_subscriber;
+
+/// Request id key for opentelemetry baggage
+pub const REQUEST_ID_KEY: &str = "request.id";
+
+/// Generate random request id
+pub fn request_id() -> String {
+    // https://stackoverflow.com/questions/54275459/how-do-i-create-a-random-string-by-sampling-from-alphanumeric-characters
+    use rand::distributions::{Alphanumeric, DistString};
+    Alphanumeric.sample_string(&mut rand::thread_rng(), 10)
+}
+
+/// Generate random request id key value
+pub fn request_id_key_value() -> KeyValue {
+    KeyValue::new(REQUEST_ID_KEY, request_id())
+}

crates/synd_o11y/src/lib.rs

@@ -0,0 +1,10 @@
+/// `OpenTelemetry` terminination process handler
+pub struct OpenTelemetryGuard;
+
+impl Drop for OpenTelemetryGuard {
+    fn drop(&mut self) {
+        opentelemetry::global::shutdown_tracer_provider();
+        opentelemetry::global::shutdown_meter_provider();
+        opentelemetry::global::shutdown_logger_provider();
+    }
+}

crates/synd_o11y/src/opentelemetry/guard.rs

@@ -5,3 +5,6 @@ mod propagation;
 pub use propagation::{extension, http, init_propagation};
 
 pub use opentelemetry::KeyValue;
+
+mod guard;
+pub use guard::OpenTelemetryGuard;

crates/synd_o11y/src/opentelemetry/mod.rs

@@ -1,2 +1,4 @@
 pub mod audit;
 pub mod otel_log;
+pub mod otel_metrics;
+pub mod otel_trace;

crates/synd_o11y/src/tracing_subscriber/mod.rs

@@ -0,0 +1,27 @@
+use std::time::Duration;
+
+use opentelemetry_sdk::{metrics::MeterProvider, runtime, Resource};
+use tracing::Subscriber;
+use tracing_opentelemetry::MetricsLayer;
+use tracing_subscriber::{registry::LookupSpan, Layer};
+
+pub fn layer<S>(resource: Resource) -> impl Layer<S>
+where
+    S: Subscriber + for<'span> LookupSpan<'span>,
+{
+    MetricsLayer::new(init_meter_provider(resource))
+}
+
+fn init_meter_provider(resource: Resource) -> MeterProvider {
+    let provider = opentelemetry_otlp::new_pipeline()
+        .metrics(runtime::Tokio)
+        .with_resource(resource)
+        .with_exporter(opentelemetry_otlp::new_exporter().tonic())
+        .with_period(Duration::from_secs(60 * 3))
+        .build()
+        .unwrap();
+
+    opentelemetry::global::set_meter_provider(provider.clone());
+
+    provider
+}

crates/synd_o11y/src/tracing_subscriber/otel_metrics/mod.rs

@@ -0,0 +1,38 @@
+use opentelemetry_sdk::{
+    runtime,
+    trace::{BatchConfig, Sampler, Tracer},
+    Resource,
+};
+use tracing::Subscriber;
+use tracing_opentelemetry::OpenTelemetryLayer;
+use tracing_subscriber::{registry::LookupSpan, Layer};
+
+pub fn layer<S>(resource: Resource) -> impl Layer<S>
+where
+    S: Subscriber + for<'span> LookupSpan<'span>,
+{
+    OpenTelemetryLayer::new(init_tracer(resource))
+}
+
+fn init_tracer(resource: Resource) -> Tracer {
+    // https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/
+    let sampler_ratio = std::env::var("OTEL_TRACES_SAMPLER_ARG")
+        .ok()
+        .and_then(|v| v.parse::<f64>().ok())
+        .unwrap_or(0.);
+
+    // TODO: construct tonic transport channel
+    opentelemetry_otlp::new_pipeline()
+        .tracing()
+        .with_trace_config(
+            opentelemetry_sdk::trace::Config::default()
+                .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
+                    sampler_ratio,
+                ))))
+                .with_resource(resource),
+        )
+        .with_batch_config(BatchConfig::default())
+        .with_exporter(opentelemetry_otlp::new_exporter().tonic())
+        .install_batch(runtime::Tokio)
+        .unwrap()
+}

crates/synd_o11y/src/tracing_subscriber/otel_trace/mod.rs

@@ -30,7 +30,6 @@ futures-util         = "0.3.30"
 graphql_client       = { workspace = true }
 html2text            = { version = "0.12" }
 open                 = "5.0.1"
-rand                 = { workspace = true }
 ratatui              = { git = "https://github.com/ratatui-org/ratatui.git", branch = "main" }
 reqwest              = { workspace = true }
 serde                = { workspace = true, features = ["derive"] }

crates/synd_term/Cargo.toml

@@ -4,7 +4,7 @@ use anyhow::anyhow;
 use graphql_client::{GraphQLQuery, Response};
 use reqwest::header::{self, HeaderValue};
 use serde::{de::DeserializeOwned, Serialize};
-use synd_o11y::opentelemetry::{extension::*, KeyValue};
+use synd_o11y::opentelemetry::extension::*;
 use tracing::{error, Span};
 use url::Url;
 
@@ -124,13 +124,10 @@ impl Client {
             .json(&body)
             .build()?;
 
-        // TODO: use trace_id
-        let request_id = Self::request_id();
-
         synd_o11y::opentelemetry::http::inject_with_baggage(
             &Span::current().context(),
             request.headers_mut(),
-            std::iter::once(KeyValue::new("request.id", request_id)),
+            std::iter::once(synd_o11y::request_id_key_value()),
         );
 
         let response: Response<ResponseData> = self
@@ -152,10 +149,4 @@ impl Client {
             _ => Err(anyhow::anyhow!("unexpected response",)),
         }
     }
-
-    fn request_id() -> String {
-        // https://stackoverflow.com/questions/54275459/how-do-i-create-a-random-string-by-sampling-from-alphanumeric-characters
-        use rand::distributions::{Alphanumeric, DistString};
-        Alphanumeric.sample_string(&mut rand::thread_rng(), 10)
-    }
 }