feat(telemetry): Send logs to OTel collector directly using OTel libraries

Signed-off-by: Caleb Schoepp <[email protected]>

Author: calebschoepp

Cargo.lock

@@ -9,8 +9,8 @@ anyhow = { workspace = true }
 http0 = { version = "0.2.9", package = "http" }
 http1 = { version = "1.0.0", package = "http" }
 opentelemetry = { version = "0.22.0", features = [ "metrics", "trace"] }
-opentelemetry_sdk = { version = "0.22.1", features = ["rt-tokio"] }
-opentelemetry-otlp = { version = "0.15.0", default-features=false, features = ["http-proto", "trace", "http", "reqwest-client", "metrics", "grpc-tonic"] }
+opentelemetry_sdk = { version = "0.22.1", features = ["rt-tokio", "logs_level_enabled"] }
+opentelemetry-otlp = { version = "0.15.0", default-features=false, features = ["http-proto", "trace", "http", "reqwest-client", "metrics", "grpc-tonic", "logs"] }
 opentelemetry-semantic-conventions = "0.14.0"
 tracing = { version = "0.1.37", features = ["log"] }
 tracing-appender = "0.2.2"

crates/telemetry/Cargo.toml

@@ -1,13 +1,15 @@
 use std::env::VarError;
 
 use opentelemetry_otlp::{
-    OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, OTEL_EXPORTER_OTLP_PROTOCOL,
+    OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_LOGS_ENDPOINT,
+    OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, OTEL_EXPORTER_OTLP_PROTOCOL,
     OTEL_EXPORTER_OTLP_TRACES_ENDPOINT,
 };
 
 const OTEL_SDK_DISABLED: &str = "OTEL_SDK_DISABLED";
 const OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL";
 const OTEL_EXPORTER_OTLP_METRICS_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL";
+const OTEL_EXPORTER_OTLP_LOGS_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_LOGS_PROTOCOL";
 const SPIN_DISABLE_LOG_TO_TRACING: &str = "SPIN_DISABLE_LOG_TO_TRACING";
 
 /// Returns a boolean indicating if the OTEL tracing layer should be enabled.
@@ -38,6 +40,20 @@ pub(crate) fn otel_metrics_enabled() -> bool {
     ]) && !otel_sdk_disabled()
 }
 
+/// Returns a boolean indicating if the OTEL log layer should be enabled.
+///
+/// It is considered enabled if any of the following environment variables are set and not empty:
+/// - `OTEL_EXPORTER_OTLP_ENDPOINT`
+/// - `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT`
+///
+/// Note that this is overridden if OTEL_SDK_DISABLED is set and not empty.
+pub(crate) fn otel_logs_enabled() -> bool {
+    any_vars_set(&[
+        OTEL_EXPORTER_OTLP_ENDPOINT,
+        OTEL_EXPORTER_OTLP_LOGS_ENDPOINT,
+    ]) && !otel_sdk_disabled()
+}
+
 /// Returns a boolean indicating if the compatibility layer that emits tracing events from
 /// applications logs should be disabled.
 ///
@@ -84,6 +100,14 @@ impl OtlpProtocol {
         )
     }
 
+    /// Returns the protocol to be used for exporting logs as defined by the environment.
+    pub(crate) fn logs_protocol_from_env() -> Self {
+        Self::protocol_from_env(
+            std::env::var(OTEL_EXPORTER_OTLP_LOGS_PROTOCOL),
+            std::env::var(OTEL_EXPORTER_OTLP_PROTOCOL),
+        )
+    }
+
     fn protocol_from_env(
         specific_protocol: Result<String, VarError>,
         general_protocol: Result<String, VarError>,

crates/telemetry/src/env.rs

@@ -1,13 +1,14 @@
 use std::io::IsTerminal;
 
+use env::otel_logs_enabled;
 use env::otel_metrics_enabled;
 use env::otel_tracing_enabled;
 use opentelemetry_sdk::propagation::TraceContextPropagator;
 use tracing_subscriber::{fmt, prelude::*, registry, EnvFilter, Layer};
 
 pub mod detector;
 mod env;
-pub mod log;
+pub mod logs;
 pub mod metrics;
 mod propagation;
 mod traces;
@@ -73,7 +74,7 @@ pub fn init(spin_version: String) -> anyhow::Result<ShutdownGuard> {
     };
 
     let otel_metrics_layer = if otel_metrics_enabled() {
-        Some(metrics::otel_metrics_layer(spin_version)?)
+        Some(metrics::otel_metrics_layer(spin_version.clone())?)
     } else {
         None
     };
@@ -89,6 +90,10 @@ pub fn init(spin_version: String) -> anyhow::Result<ShutdownGuard> {
     // layer is disabled we still want to propagate trace context.
     opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
 
+    if otel_logs_enabled() {
+        logs::init_otel_logging_backend(spin_version)?;
+    }
+
     Ok(ShutdownGuard)
 }
 

crates/telemetry/src/lib.rs

@@ -0,0 +1,112 @@
+use std::{ascii::escape_default, sync::OnceLock, time::Duration};
+
+use anyhow::bail;
+use opentelemetry::{
+    global,
+    logs::{Logger, LoggerProvider},
+};
+use opentelemetry_otlp::LogExporterBuilder;
+use opentelemetry_sdk::{
+    logs::{BatchConfigBuilder, BatchLogProcessor},
+    resource::{EnvResourceDetector, TelemetryResourceDetector},
+    Resource,
+};
+
+use crate::{
+    detector::SpinResourceDetector,
+    env::{self, otel_logs_enabled, OtlpProtocol},
+};
+
+/// Handle an application log. Has the potential to both forward the log to OTel and to emit it as a
+/// tracing event.
+pub fn handle_app_log(buf: &[u8]) {
+    app_log_to_otel(buf);
+    app_log_to_tracing_event(buf);
+}
+
+/// Forward the app log to OTel.
+fn app_log_to_otel(buf: &[u8]) {
+    if !otel_logs_enabled() {
+        return;
+    }
+
+    let logger = global::logger_provider().logger("spin");
+    if let Ok(s) = std::str::from_utf8(buf) {
+        logger.emit(
+            opentelemetry::logs::LogRecord::builder()
+                .with_body(s.to_owned())
+                .build(),
+        );
+    } else {
+        logger.emit(
+            opentelemetry::logs::LogRecord::builder()
+                .with_body(escape_non_utf8_buf(buf))
+                .with_attribute("app_log_non_utf8", true)
+                .build(),
+        );
+    }
+}
+
+/// Takes a Spin application log and emits it as a tracing event. This acts as a compatibility layer
+/// to easily get Spin app logs as events in our OTel traces.
+fn app_log_to_tracing_event(buf: &[u8]) {
+    static CELL: OnceLock<bool> = OnceLock::new();
+    if *CELL.get_or_init(env::spin_disable_log_to_tracing) {
+        return;
+    }
+
+    if let Ok(s) = std::str::from_utf8(buf) {
+        tracing::info!(app_log = s);
+    } else {
+        tracing::info!(app_log_non_utf8 = escape_non_utf8_buf(buf));
+    }
+}
+
+fn escape_non_utf8_buf(buf: &[u8]) -> String {
+    buf.iter()
+        .take(50)
+        .map(|&x| escape_default(x).to_string())
+        .collect::<String>()
+}
+
+/// Initialize the OTel logging backend.
+pub(crate) fn init_otel_logging_backend(spin_version: String) -> anyhow::Result<()> {
+    let resource = Resource::from_detectors(
+        Duration::from_secs(5),
+        vec![
+            // Set service.name from env OTEL_SERVICE_NAME > env OTEL_RESOURCE_ATTRIBUTES > spin
+            // Set service.version from Spin metadata
+            Box::new(SpinResourceDetector::new(spin_version)),
+            // Sets fields from env OTEL_RESOURCE_ATTRIBUTES
+            Box::new(EnvResourceDetector::new()),
+            // Sets telemetry.sdk{name, language, version}
+            Box::new(TelemetryResourceDetector),
+        ],
+    );
+
+    // This will configure the exporter based on the OTEL_EXPORTER_* environment variables. We
+    // currently default to using the HTTP exporter but in the future we could select off of the
+    // combination of OTEL_EXPORTER_OTLP_PROTOCOL and OTEL_EXPORTER_OTLP_LOGS_PROTOCOL to
+    // determine whether we should use http/protobuf or grpc.
+    let exporter_builder: LogExporterBuilder = match OtlpProtocol::logs_protocol_from_env() {
+        OtlpProtocol::Grpc => opentelemetry_otlp::new_exporter().tonic().into(),
+        OtlpProtocol::HttpProtobuf => opentelemetry_otlp::new_exporter().http().into(),
+        OtlpProtocol::HttpJson => bail!("http/json OTLP protocol is not supported"),
+    };
+
+    let provider = opentelemetry_sdk::logs::LoggerProvider::builder()
+        .with_config(opentelemetry_sdk::logs::config().with_resource(resource))
+        .with_log_processor(
+            BatchLogProcessor::builder(
+                exporter_builder.build_log_exporter()?,
+                opentelemetry_sdk::runtime::Tokio,
+            )
+            .with_batch_config(BatchConfigBuilder::default().build())
+            .build(),
+        )
+        .build();
+
+    global::set_logger_provider(provider);
+
+    Ok(())
+}

crates/telemetry/src/log.rs

@@ -292,7 +292,7 @@ impl AsyncWrite for ComponentStdioWriter {
 
 impl std::io::Write for ComponentStdioWriter {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        spin_telemetry::log::app_log_to_tracing_event(buf);
+        spin_telemetry::logs::handle_app_log(buf);
 
         match &mut self.inner {
             ComponentStdioWriterInner::Inherit => {

crates/telemetry/src/logs.rs

@@ -16,6 +16,7 @@ services:
       - jaeger
       - prometheus
       - tempo
+      - loki
 
   jaeger:
     image: jaegertracing/all-in-one:latest
@@ -59,3 +60,12 @@ services:
     depends_on:
       - prometheus
       - tempo
+
+  loki:
+    image: grafana/loki:2.9.8
+    command:
+      - '-config.file=/etc/loki/config.yaml'
+    volumes:
+      - ./loki.yaml:/etc/loki/config.yaml
+    ports:
+      - 3100:3100

crates/trigger/src/stdio.rs

@@ -13,4 +13,12 @@ datasources:
     access: proxy
     version: 1
     editable: false
-    uid: tempo
+    uid: tempo
+  - name: Loki
+    type: loki
+    url: http://loki:3100
+    access: proxy
+    version: 1
+    editable: false
+    uid: loki
+    basicAuth: false