sketching/
pipeline.rs

1use opentelemetry::{global, trace::TracerProvider as _, KeyValue};
2use opentelemetry_otlp::{tonic_types::metadata::MetadataMap, WithExportConfig, WithTonicConfig};
3use opentelemetry_sdk::{
4    trace::{Sampler, SdkTracerProvider},
5    Resource,
6};
7use opentelemetry_semantic_conventions::{
8    attribute::{DEPLOYMENT_ENVIRONMENT_NAME, SERVICE_VERSION},
9    SCHEMA_URL,
10};
11use std::str::FromStr;
12use std::time::Duration;
13use tracing::Subscriber;
14use tracing_core::Level;
15use tracing_opentelemetry::OpenTelemetryLayer;
16use tracing_subscriber::{filter::Directive, prelude::*, EnvFilter, Registry};
17
18const MAX_EVENTS_PER_SPAN: u32 = 64 * 1024;
19const MAX_ATTRIBUTES_PER_SPAN: u32 = 128;
20
21/// This does all the startup things for the logging pipeline
22pub fn start_logging_pipeline(
23    otlp_endpoint: &Option<String>,
24    log_filter: crate::LogLevel,
25) -> Result<(Option<SdkTracerProvider>, Box<dyn Subscriber + Send + Sync>), String> {
26    // Always force the event span to be generated at the correct level, regardless
27    // of what the user set.
28    let kanidmd_core_directives = [
29        Directive::from_str("kanidmd_core::https::trace=info")
30            .map_err(|err| format!("Invalid directive during log setup: {}", err))?,
31        Directive::from_str("kanidmd_core::https::middleware=info")
32            .map_err(|err| format!("Invalid directive during log setup: {}", err))?,
33    ];
34
35    let mut logging_filter = EnvFilter::builder()
36        .with_default_directive(log_filter.into())
37        .parse("")
38        .map_err(|err| format!("Failed to create OTEL logging filter: {}", err))?;
39    for directive in kanidmd_core_directives.iter() {
40        logging_filter = logging_filter.add_directive(directive.clone());
41    }
42    logging_filter = logging_filter
43        // tell the tonic/grpc/h2 layers to trace at warn, so we can see connectivity issues
44        .add_directive(
45            Directive::from_str("tonic=warn")
46                .map_err(|err| format!("Failed to set tonic logging to warn: {}", err))?,
47        )
48        .add_directive(
49            Directive::from_str("hyper=warn")
50                .map_err(|err| format!("Failed to set hyper logging to warn: {}", err))?,
51        )
52        .add_directive(
53            Directive::from_str("h2=warn")
54                .map_err(|err| format!("Failed to set h2 logging to warn: {}", err))?,
55        )
56        .add_directive(
57            Directive::from_str("h2::proto::streams::prioritize=warn").map_err(|err| {
58                format!(
59                    "Failed to set h2::proto::streams::prioritize logging to warn: {}",
60                    err
61                )
62            })?,
63        );
64
65    eprintln!(
66        "Logging filter initialized: {:?}",
67        logging_filter.to_string()
68    );
69
70    // TODO: work out how to do metrics things
71    if let Some(endpoint) = otlp_endpoint {
72        eprintln!(
73            "Attempting to start OTLP logging pipeline endpoint={}",
74            endpoint
75        );
76
77        // setup metadata so we can auth to third-party services
78        let mut tonic_metadata = MetadataMap::new();
79
80        if let Some(headers) = std::env::var_os("OTEL_EXPORTER_OTLP_HEADERS") {
81            let headers = headers.to_string_lossy();
82            for header in headers.split(',') {
83                if let Some((key, value)) = header.split_once('=') {
84                    if !key.is_empty() && key.eq_ignore_ascii_case("authorization") {
85                        if let Ok(header_value) = tonic::metadata::MetadataValue::from_str(value) {
86                            tonic_metadata.insert("authorization", header_value);
87                        } else {
88                            eprintln!(
89                            "Warning: could not parse OTEL_EXPORTER_OTLP_HEADERS environment variable, skipping this: {}", header
90                        );
91                        };
92                    }
93                }
94            }
95        }
96        let otlp_exporter = opentelemetry_otlp::SpanExporter::builder()
97            .with_tonic()
98            .with_endpoint(endpoint)
99            .with_metadata(tonic_metadata)
100            .with_timeout(Duration::from_secs(5))
101            .build()
102            .map_err(|err| err.to_string())?;
103
104        // this env var gets set at build time, if we can pull it, add it to the metadata
105        let git_rev = match option_env!("KANIDM_PKG_COMMIT_REV") {
106            Some(rev) => format!("-{rev}"),
107            None => "".to_string(),
108        };
109
110        let version = format!("{}{}", env!("CARGO_PKG_VERSION"), git_rev);
111        let hostname = gethostname::gethostname();
112        let hostname = hostname.to_string_lossy();
113        let hostname = hostname.to_lowercase();
114
115        let mut resource = Resource::builder().with_schema_url(
116            [
117                // TODO: it'd be really nice to be able to set the instance ID here, from the server UUID so we know *which* instance on this host is logging
118                KeyValue::new(SERVICE_VERSION, version),
119                KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, hostname),
120            ],
121            SCHEMA_URL,
122        );
123
124        // only set the service name if it's not already set in the environment because the SDK defaults to "unknown_service"
125        if std::env::var("OTEL_SERVICE_NAME").is_err() {
126            resource = resource.with_service_name("kanidmd");
127        }
128        let resource = resource.build();
129
130        let provider = opentelemetry_sdk::trace::TracerProviderBuilder::default()
131            .with_batch_exporter(otlp_exporter)
132            // we want *everything!*
133            .with_sampler(Sampler::AlwaysOn)
134            .with_max_events_per_span(MAX_EVENTS_PER_SPAN)
135            .with_max_attributes_per_span(MAX_ATTRIBUTES_PER_SPAN)
136            .with_resource(resource)
137            .build();
138
139        let provider_handle = provider.clone();
140
141        global::set_tracer_provider(provider.clone());
142        provider.tracer("tracing-otel-subscriber");
143
144        let registry = tracing_subscriber::registry()
145            .with(
146                tracing_subscriber::filter::LevelFilter::from_level(Level::INFO)
147                    .with_filter(logging_filter.clone()),
148            )
149            .with(
150                OpenTelemetryLayer::new(provider.tracer("tracing-otel-subscriber"))
151                    .with_filter(logging_filter),
152            );
153
154        Ok((Some(provider_handle), Box::new(registry)))
155    } else {
156        let forest_layer = tracing_forest::ForestLayer::default().with_filter(logging_filter);
157        Ok((None, Box::new(Registry::default().with(forest_layer))))
158    }
159}
160
161/// This helps with cleanly shutting down the tracing/logging providers when done,
162/// so we don't lose traces.
163pub struct TracingPipelineGuard(pub Option<SdkTracerProvider>);
164
165impl Drop for TracingPipelineGuard {
166    fn drop(&mut self) {
167        if let Some(provider) = self.0.take() {
168            if let Err(err) = provider.shutdown() {
169                eprintln!("Error shutting down logging pipeline: {}", err);
170            } else {
171                eprintln!("Logging pipeline completed shutdown");
172            }
173        }
174    }
175}
sketching/pipeline.rs

sketching/
pipeline.rs
