export logs over logs stream (#73)

Author: davidhewitt

Cargo.toml

@@ -19,9 +19,9 @@ tonic = { version = "0.13", optional = true }
 
 rand = "0.9.0"
 
-opentelemetry = { version = "0.30", default-features = false, features = ["trace"] }
-opentelemetry_sdk = { version = "0.30", default-features = false, features = ["trace", "experimental_metrics_custom_reader"] }
-opentelemetry-otlp = { version = "0.30", default-features = false, features = ["trace", "metrics"] }
+opentelemetry = { version = "0.30", default-features = false, features = ["trace", "logs"] }
+opentelemetry_sdk = { version = "0.30", default-features = false, features = ["trace", "experimental_metrics_custom_reader", "logs"] }
+opentelemetry-otlp = { version = "0.30", default-features = false, features = ["trace", "metrics", "logs"] }
 futures-util = "0.3"
 
 tracing = "0.1"

src/bridges/tracing.rs

@@ -1783,7 +1783,7 @@ mod tests {
         let output = std::str::from_utf8(&output).unwrap();
         let output = remap_timestamps_in_console_output(output);
 
-        assert_snapshot!(output, @r#"
+        assert_snapshot!(output, @r"
         1970-01-01T00:00:00.000000Z  INFO logfire::bridges::tracing::tests root event
         1970-01-01T00:00:00.000001Z  INFO logfire::bridges::tracing::tests root event with value field_value=1
         1970-01-01T00:00:00.000002Z  INFO logfire::bridges::tracing::tests root span
@@ -1792,6 +1792,6 @@ mod tests {
         1970-01-01T00:00:00.000005Z DEBUG logfire::bridges::tracing::tests debug span with explicit parent
         1970-01-01T00:00:00.000006Z  INFO logfire::bridges::tracing::tests hello world log
         1970-01-01T00:00:00.000007Z  INFO logfire::bridges::tracing::tests hello world log with value field_value=1
-        "#);
+        ");
     }
 }

src/config.rs

@@ -9,6 +9,7 @@ use std::{
 };
 
 use opentelemetry_sdk::{
+    logs::LogProcessor,
     metrics::reader::MetricReader,
     trace::{IdGenerator, SpanProcessor},
 };
@@ -193,15 +194,15 @@ pub struct AdvancedOptions {
     pub(crate) id_generator: Option<BoxedIdGenerator>,
     /// Resource to override default resource detection.
     pub(crate) resource: Option<opentelemetry_sdk::Resource>,
+
+    // Configuration for OpenTelemetry logging. This is experimental and may be removed.
+    pub(crate) log_record_processors: Vec<BoxedLogProcessor>,
     //
     //
     // TODO: arguments below supported by Python
 
     // /// Generator for nanosecond start and end timestamps of spans.
     // pub ns_timestamp_generator: Option,
-
-    // /// Configuration for OpenTelemetry logging. This is experimental and may be removed.
-    // pub log_record_processors: Vec<Box<dyn LogRecordProcessor>>,
 }
 
 impl AdvancedOptions {
@@ -228,6 +229,17 @@ impl AdvancedOptions {
         self.resource = Some(resource);
         self
     }
+
+    /// Add a log processor to the list of log processors.
+    #[must_use]
+    pub fn with_log_processor<T: LogProcessor + Send + Sync + 'static>(
+        mut self,
+        processor: T,
+    ) -> Self {
+        self.log_record_processors
+            .push(BoxedLogProcessor::new(Box::new(processor)));
+        self
+    }
 }
 
 struct RegionData {
@@ -388,6 +400,37 @@ impl MetricReader for BoxedMetricReader {
     }
 }
 
+/// Boxed log processor for dynamic dispatch
+#[derive(Debug)]
+pub(crate) struct BoxedLogProcessor {
+    inner: Box<dyn LogProcessor + Send + Sync>,
+}
+
+impl BoxedLogProcessor {
+    /// Create a new boxed log processor.
+    pub fn new(processor: Box<dyn LogProcessor + Send + Sync>) -> Self {
+        Self { inner: processor }
+    }
+}
+
+impl LogProcessor for BoxedLogProcessor {
+    fn emit(
+        &self,
+        log_record: &mut opentelemetry_sdk::logs::SdkLogRecord,
+        instrumentation_scope: &opentelemetry::InstrumentationScope,
+    ) {
+        self.inner.emit(log_record, instrumentation_scope);
+    }
+
+    fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+        self.inner.force_flush()
+    }
+
+    fn shutdown(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+        self.inner.shutdown()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::config::SendToLogfire;

src/exporters.rs

@@ -1,8 +1,11 @@
 //! Helper functions to configure mo
 use std::collections::HashMap;
 
-use opentelemetry_otlp::{MetricExporter, Protocol};
-use opentelemetry_sdk::{metrics::exporter::PushMetricExporter, trace::SpanExporter};
+use opentelemetry_otlp::{LogExporter, MetricExporter, Protocol};
+use opentelemetry_sdk::{
+    logs::LogExporter as LogExporterTrait, metrics::exporter::PushMetricExporter,
+    trace::SpanExporter,
+};
 
 use crate::{
     ConfigureError, get_optional_env,
@@ -170,6 +173,76 @@ pub fn metric_exporter(
     }
 }
 
+/// Build a [`LogExporter`] for passing to log processors.
+///
+/// This uses `OTEL_EXPORTER_OTLP_PROTOCOL` and `OTEL_EXPORTER_OTLP_LOGS_PROTOCOL` environment
+/// variables to determine the protocol to use (or otherwise defaults to [`Protocol::HttpBinary`]).
+///
+/// # Errors
+///
+/// Returns an error if the protocol specified by the env var is not supported or if the required feature is not enabled for
+/// the given protocol.
+///
+/// Returns an error if the endpoint is not a valid URI.
+///
+/// Returns an error if any headers are not valid HTTP headers.
+pub fn log_exporter(
+    endpoint: &str,
+    headers: Option<HashMap<String, String>>,
+) -> Result<impl LogExporterTrait + use<>, ConfigureError> {
+    let (source, protocol) = protocol_from_env("OTEL_EXPORTER_OTLP_LOGS_PROTOCOL")?;
+
+    let builder = LogExporter::builder();
+
+    // FIXME: it would be nice to let `opentelemetry-rust` handle this; ideally we could detect if
+    // OTEL_EXPORTER_OTLP_PROTOCOL or OTEL_EXPORTER_OTLP_LOGS_PROTOCOL is set and let the SDK
+    // make a builder. (If unset, we could supply our preferred exporter.)
+    //
+    // But at the moment otel-rust ignores these env vars; see
+    // https://github.com/open-telemetry/opentelemetry-rust/issues/1983
+    match protocol {
+        Protocol::Grpc => {
+            feature_required!("export-grpc", source, {
+                use opentelemetry_otlp::WithTonicConfig;
+                Ok(builder
+                    .with_tonic()
+                    .with_channel(
+                        tonic::transport::Channel::builder(
+                            endpoint.try_into().map_err(|e: http::uri::InvalidUri| {
+                                ConfigureError::Other(e.into())
+                            })?,
+                        )
+                        .connect_lazy(),
+                    )
+                    .with_metadata(build_metadata_from_headers(headers.as_ref())?)
+                    .build()?)
+            })
+        }
+        Protocol::HttpBinary => {
+            feature_required!("export-http-protobuf", source, {
+                use opentelemetry_otlp::{WithExportConfig, WithHttpConfig};
+                Ok(builder
+                    .with_http()
+                    .with_protocol(Protocol::HttpBinary)
+                    .with_headers(headers.unwrap_or_default())
+                    .with_endpoint(format!("{endpoint}/v1/logs"))
+                    .build()?)
+            })
+        }
+        Protocol::HttpJson => {
+            feature_required!("export-http-json", source, {
+                use opentelemetry_otlp::{WithExportConfig, WithHttpConfig};
+                Ok(builder
+                    .with_http()
+                    .with_protocol(Protocol::HttpBinary)
+                    .with_headers(headers.unwrap_or_default())
+                    .with_endpoint(format!("{endpoint}/v1/logs"))
+                    .build()?)
+            })
+        }
+    }
+}
+
 #[cfg(feature = "export-grpc")]
 fn build_metadata_from_headers(
     headers: Option<&HashMap<String, String>>,