Custom OTLP File Exporter + opentelemetry updates

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- `libcnb`:
+  - Implemented custom OTLP File Exporter instead of `opentelemetry-stdout` and updated `opentelemetry` libraries to `0.28`. ([#909](https://github.com/heroku/libcnb.rs/pull/909/))
 
 ## [0.26.1] - 2024-12-10
 

libcnb/Cargo.toml

@@ -16,9 +16,16 @@ workspace = true
 
 [features]
 trace = [
+    "dep:futures-core",
     "dep:opentelemetry",
+    "opentelemetry/trace",
     "dep:opentelemetry_sdk",
-    "dep:opentelemetry-stdout",
+    "opentelemetry_sdk/trace",
+    "dep:opentelemetry-proto",
+    "opentelemetry-proto/trace",
+    "opentelemetry-proto/gen-tonic-messages",
+    "opentelemetry-proto/with-serde",
+    "dep:serde_json",
 ]
 
 [dependencies]
@@ -27,15 +34,15 @@ cyclonedx-bom = { version = "0.8.0", optional = true }
 libcnb-common.workspace = true
 libcnb-data.workspace = true
 libcnb-proc-macros.workspace = true
-opentelemetry = { version = "0.24", optional = true }
-opentelemetry_sdk = { version = "0.24", optional = true }
-opentelemetry-stdout = { version = "0.5", optional = true, features = [
-    "trace",
-] }
+futures-core = { version = "0.3", optional = true }
+opentelemetry = { version = "0.28.0", optional = true, default-features = false }
+opentelemetry_sdk = { version = "0.28.0", optional = true, default-features = false }
+opentelemetry-proto = { version = "0.28.0", optional = true, default-features = false }
 serde = { version = "1.0.215", features = ["derive"] }
+serde_json = { version = "1.0.133", optional = true }
 thiserror = "2.0.6"
 toml.workspace = true
 
 [dev-dependencies]
-serde_json = "1.0.133"
 tempfile = "3.14.0"
+serde_json = "1.0.133"

libcnb/src/tracing.rs

@@ -1,14 +1,24 @@
+use futures_core::future::BoxFuture;
 use libcnb_data::buildpack::Buildpack;
 use opentelemetry::{
-    global,
+    global::{self, BoxedSpan},
     trace::{Span as SpanTrait, Status, Tracer, TracerProvider as TracerProviderTrait},
-    KeyValue,
+    InstrumentationScope, KeyValue,
 };
+use opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema;
+use opentelemetry_proto::transform::trace::tonic::group_spans_by_resource_and_scope;
 use opentelemetry_sdk::{
-    trace::{Config, Span, TracerProvider},
+    error::{OTelSdkError, OTelSdkResult},
+    trace::SdkTracerProvider,
+    trace::SpanExporter,
     Resource,
 };
-use std::{io::BufWriter, path::Path};
+use std::{
+    fmt::Debug,
+    io::{LineWriter, Write},
+    path::Path,
+    sync::{Arc, Mutex},
+};
 
 // This is the directory in which `BuildpackTrace` stores OpenTelemetry File
 // Exports. Services which intend to export the tracing data from libcnb.rs
@@ -22,8 +32,8 @@ const TELEMETRY_EXPORT_ROOT: &str = "/tmp/libcnb-telemetry";
 /// Represents an OpenTelemetry tracer provider and single span tracing
 /// a single CNB build or detect phase.
 pub(crate) struct BuildpackTrace {
-    provider: TracerProvider,
-    span: Span,
+    provider: SdkTracerProvider,
+    span: BoxedSpan,
 }
 
 /// Start an OpenTelemetry trace and span that exports to an
@@ -40,45 +50,44 @@ pub(crate) fn start_trace(buildpack: &Buildpack, phase_name: &'static str) -> Bu
     if let Some(parent_dir) = tracing_file_path.parent() {
         let _ = std::fs::create_dir_all(parent_dir);
     }
-    let exporter = match std::fs::File::options()
-        .create(true)
-        .append(true)
-        .open(&tracing_file_path)
-    {
-        // Write tracing data to a file, which may be read by other
-        // services. Wrap with a BufWriter to prevent serde from sending each
-        // JSON token to IO, and instead send entire JSON objects to IO.
-        Ok(file) => opentelemetry_stdout::SpanExporter::builder()
-            .with_writer(BufWriter::new(file))
-            .build(),
-        // Failed tracing shouldn't fail a build, and any logging here would
-        // likely confuse the user, so send telemetry to /dev/null on errors.
-        Err(_) => opentelemetry_stdout::SpanExporter::builder()
-            .with_writer(std::io::sink())
-            .build(),
-    };
 
-    let provider = TracerProvider::builder()
-        .with_simple_exporter(exporter)
-        .with_config(Config::default().with_resource(Resource::new([
-            // Associate the tracer provider with service attributes. The buildpack
-            // name/version seems to map well to the suggestion here
-            // https://opentelemetry.io/docs/specs/semconv/resource/#service.
+    let resource = Resource::builder()
+        // Define a resource that defines the trace provider.
+        // The buildpack name/version seems to map well to the suggestion here
+        // https://opentelemetry.io/docs/specs/semconv/resource/#service.
+        .with_attributes([
             KeyValue::new("service.name", buildpack.id.to_string()),
             KeyValue::new("service.version", buildpack.version.to_string()),
-        ])))
+        ])
         .build();
 
+    let provider_builder = SdkTracerProvider::builder().with_resource(resource.clone());
+
+    let provider = match std::fs::File::options()
+        .create(true)
+        .append(true)
+        .open(&tracing_file_path)
+        .map(|file| FileExporter::new(file, resource))
+    {
+        // Write tracing data to a file, which may be read by other services
+        Ok(exporter) => provider_builder.with_batch_exporter(exporter),
+        // Failed tracing shouldn't fail a build, and any export logging here
+        // would likely confuse the user; don't export when the file has IO errors
+        Err(_) => provider_builder,
+    }
+    .build();
+
     // Set the global tracer provider so that buildpacks may use it.
     global::set_tracer_provider(provider.clone());
 
     // Get a tracer identified by the instrumentation scope/library. The libcnb
     // crate name/version seems to map well to the suggestion here:
     // https://opentelemetry.io/docs/specs/otel/trace/api/#get-a-tracer.
-    let tracer = provider
-        .tracer_builder(env!("CARGO_PKG_NAME"))
-        .with_version(env!("CARGO_PKG_VERSION"))
-        .build();
+    let tracer = global::tracer_provider().tracer_with_scope(
+        InstrumentationScope::builder(env!("CARGO_PKG_NAME"))
+            .with_version(env!("CARGO_PKG_VERSION"))
+            .build(),
+    );
 
     let mut span = tracer.start(trace_name);
     span.set_attributes([
@@ -109,8 +118,60 @@ impl BuildpackTrace {
 impl Drop for BuildpackTrace {
     fn drop(&mut self) {
         self.span.end();
-        self.provider.force_flush();
-        global::shutdown_tracer_provider();
+        self.provider.force_flush().ok();
+        self.provider.shutdown().ok();
+    }
+}
+
+#[derive(Debug)]
+struct FileExporter<W: Write + Send + Debug> {
+    writer: Arc<Mutex<LineWriter<W>>>,
+    resource: Resource,
+}
+
+impl<W: Write + Send + Debug> FileExporter<W> {
+    fn new(writer: W, resource: Resource) -> Self {
+        Self {
+            writer: Arc::new(Mutex::new(LineWriter::new(writer))),
+            resource,
+        }
+    }
+}
+
+impl<W: Write + Send + Debug> SpanExporter for FileExporter<W> {
+    fn export(
+        &mut self,
+        batch: Vec<opentelemetry_sdk::trace::SpanData>,
+    ) -> BoxFuture<'static, OTelSdkResult> {
+        let resource = ResourceAttributesWithSchema::from(&self.resource);
+        let data = group_spans_by_resource_and_scope(batch, &resource);
+        let mut writer = match self.writer.lock() {
+            Ok(f) => f,
+            Err(e) => {
+                return Box::pin(std::future::ready(Err(OTelSdkError::InternalFailure(
+                    e.to_string(),
+                ))));
+            }
+        };
+        Box::pin(std::future::ready(
+            serde_json::to_writer(writer.get_mut(), &data)
+                .map_err(|e| OTelSdkError::InternalFailure(e.to_string())),
+        ))
+    }
+
+    fn force_flush(&mut self) -> OTelSdkResult {
+        let mut writer = self
+            .writer
+            .lock()
+            .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
+
+        writer
+            .flush()
+            .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))
+    }
+
+    fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) {
+        self.resource = res.clone();
     }
 }