First draft of logs enrichment example

Author: stefanobaghino

examples/logs-enrichment/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "logs-enrichment"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0"
+rust-version = "1.75.0"
+publish = false
+autobenches = false
+
+[[bin]]
+name = "logs-enrichment"
+path = "src/main.rs"
+bench = false
+required-features = ["spec_unstable_logs_enabled"]
+
+[features]
+spec_unstable_logs_enabled = ["opentelemetry/spec_unstable_logs_enabled", "opentelemetry_sdk/spec_unstable_logs_enabled"]
+
+[dependencies]
+opentelemetry_sdk = { workspace = true, features = ["logs", "spec_unstable_logs_enabled"] }
+opentelemetry-stdout = { workspace = true, features = ["logs"] }
+opentelemetry-appender-tracing = { workspace = true }
+tracing = { workspace = true, features = ["std"]}
+tracing-subscriber = { workspace = true, features = ["env-filter","registry", "std", "fmt"] }
+opentelemetry = { version = "0.31.0", features = ["spec_unstable_logs_enabled"] }

examples/logs-enrichment/README.md

@@ -0,0 +1,17 @@
+# OpenTelemetry Log Appender for tracing -  Example
+
+This example shows how to use the opentelemetry-appender-tracing crate, which is a
+[logging
+appender](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/glossary.md#log-appender--bridge)
+that bridges logs from the [tracing crate](https://tracing.rs/tracing/#events) to
+OpenTelemetry. The example setups a LoggerProvider with stdout exporter, so logs
+are emitted to stdout.
+
+## Usage
+
+Run the following, and Logs emitted using [tracing](https://docs.rs/tracing/latest/tracing/)
+will be written out to stdout.
+
+```shell
+cargo run
+```

examples/logs-enrichment/src/main.rs

@@ -0,0 +1,128 @@
+use std::thread::sleep;
+use std::time::Duration;
+use opentelemetry_appender_tracing::layer;
+use opentelemetry_sdk::logs::{LogProcessor, SdkLogRecord, SdkLoggerProvider, SimpleLogProcessor};
+use opentelemetry_sdk::Resource;
+use tracing::{info, error};
+use tracing_subscriber::{prelude::*, EnvFilter};
+use opentelemetry::InstrumentationScope;
+use opentelemetry::logs::Severity;
+use opentelemetry_sdk::error::OTelSdkResult;
+
+fn main() {
+    let exporter = opentelemetry_stdout::LogExporter::default();
+    let filtering_processor = FilteringLogProcessor::new(
+        SimpleLogProcessor::new(exporter),
+        Severity::Error,
+    );
+    let provider: SdkLoggerProvider = SdkLoggerProvider::builder()
+        .with_resource(
+            Resource::builder()
+                .with_service_name("log-appender-tracing-example")
+                .build(),
+        )
+        .with_log_processor(SlowEnrichmentLogProcessor::new(filtering_processor))
+        .build();
+
+    // To prevent a telemetry-induced-telemetry loop, OpenTelemetry's own internal
+    // logging is properly suppressed. However, logs emitted by external components
+    // (such as reqwest, tonic, etc.) are not suppressed as they do not propagate
+    // OpenTelemetry context. Until this issue is addressed
+    // (https://github.com/open-telemetry/opentelemetry-rust/issues/2877),
+    // filtering like this is the best way to suppress such logs.
+    //
+    // The filter levels are set as follows:
+    // - Allow `info` level and above by default.
+    // - Completely restrict logs from `hyper`, `tonic`, `h2`, and `reqwest`.
+    //
+    // Note: This filtering will also drop logs from these components even when
+    // they are used outside of the OTLP Exporter.
+    let filter_otel = EnvFilter::new("info")
+        .add_directive("hyper=off".parse().unwrap())
+        .add_directive("tonic=off".parse().unwrap())
+        .add_directive("h2=off".parse().unwrap())
+        .add_directive("reqwest=off".parse().unwrap());
+    let otel_layer = layer::OpenTelemetryTracingBridge::new(&provider).with_filter(filter_otel);
+
+    // Create a new tracing::Fmt layer to print the logs to stdout. It has a
+    // default filter of `info` level and above, and `debug` and above for logs
+    // from OpenTelemetry crates. The filter levels can be customized as needed.
+    let filter_fmt = EnvFilter::new("info").add_directive("opentelemetry=debug".parse().unwrap());
+    let fmt_layer = tracing_subscriber::fmt::layer()
+        .with_thread_names(true)
+        .with_filter(filter_fmt);
+
+    tracing_subscriber::registry()
+        .with(otel_layer)
+        .with(fmt_layer)
+        .init();
+
+    for i in 0..10 {
+        info!(name: "my-event-name", target: "my-system", event_id = 20 + i, user_name = "otel", user_email = "[email protected]", message = "This is an example message");
+    }
+    error!(name: "my-event-name", target: "my-system", event_id = 50, user_name = "otel", user_email = "[email protected]", message = "This is an example message");
+    let _ = provider.shutdown();
+}
+
+#[derive(Debug)]
+pub struct FilteringLogProcessor<P: LogProcessor> {
+    delegate: P,
+    min_severity: Severity,
+}
+
+impl<P: LogProcessor> FilteringLogProcessor<P> {
+    pub fn new(delegate: P, min_severity: Severity) -> FilteringLogProcessor<P> {
+        FilteringLogProcessor {
+            delegate: delegate,
+            min_severity: min_severity,
+        }
+    }
+}
+
+impl<P: LogProcessor> LogProcessor for FilteringLogProcessor<P> {
+    fn emit(&self, data: &mut SdkLogRecord, instrumentation: &InstrumentationScope) {
+        self.delegate.emit(data, instrumentation)
+    }
+
+    fn force_flush(&self) -> OTelSdkResult {
+        self.delegate.force_flush()
+    }
+
+    #[cfg(feature = "spec_unstable_logs_enabled")]
+    fn event_enabled(&self, level: Severity, target: &str, name: Option<&str>) -> bool {
+        println!("filtering: severity is {level:?}");
+        self.delegate.event_enabled(level, target, name) && level >= self.min_severity
+    }
+}
+
+#[derive(Debug)]
+pub struct SlowEnrichmentLogProcessor<P: LogProcessor> {
+    /// The wrapped processor that will receive enriched log records
+    delegate: P,
+}
+
+impl<P: LogProcessor> SlowEnrichmentLogProcessor<P> {
+    pub fn new(delegate: P) -> SlowEnrichmentLogProcessor<P> {
+        SlowEnrichmentLogProcessor {
+            delegate: delegate,
+        }
+    }
+}
+
+impl<P: LogProcessor> LogProcessor for SlowEnrichmentLogProcessor<P> {
+    fn emit(&self, data: &mut SdkLogRecord, instrumentation: &InstrumentationScope) {
+        println!("slow enrichment log processor is waiting 1 s");
+        sleep(Duration::from_secs(1));
+        self.delegate.emit(data, instrumentation);
+    }
+
+    fn force_flush(&self) -> OTelSdkResult {
+        self.delegate.force_flush()
+    }
+
+    #[cfg(feature = "spec_unstable_logs_enabled")]
+    fn event_enabled(&self, level: Severity, target: &str, name: Option<&str>) -> bool {
+        println!("slow enrichment: severity is {level:?}");
+        self.delegate.event_enabled(level, target, name)
+    }
+}