add open telemetry logger and tracer

Author: KSDaemon

rust/cubestore/cubestore/src/telemetry/mod.rs

@@ -1,3 +1,4 @@
+pub mod otel_tracing;
 pub mod tracing;
 use crate::CubeError;
 use async_trait::async_trait;
@@ -10,6 +11,7 @@ use futures::{Sink, StreamExt};
 use futures_timer::Delay;
 use log::{Level, Log, Metadata, Record};
 use nanoid::nanoid;
+pub use otel_tracing::OpenTelemetryLogger;
 use reqwest::header::HeaderMap;
 use serde_json::{Map, Number, Value};
 use std::collections::{HashMap, HashSet};

rust/cubestore/cubestore/src/telemetry/otel_tracing.rs

@@ -0,0 +1,118 @@
+use log::{Level, Log, Metadata, Record};
+use opentelemetry::logs::{LogRecord, Logger, LoggerProvider};
+use opentelemetry::trace::{TracerProvider, SpanKind, Tracer};
+use opentelemetry_sdk::logs::{LogRecord as LogRecordSDK, Logger as LoggerSDK};
+use opentelemetry_sdk::trace::Tracer as TracerSDK;
+use lazy_static::lazy_static;
+use opentelemetry::global::ObjectSafeSpan;
+use opentelemetry::{KeyValue, StringValue};
+use opentelemetry::logs::AnyValue::String;
+
+const OTEL_SERVICE_NAME: &str = "cubestore";
+
+lazy_static! {
+    pub static ref OT_TRACER: TracerSDK = init_tracing().unwrap();
+    pub static ref OT_LOGGER: LoggerSDK = init_logging().unwrap();
+}
+
+pub fn init_tracing() -> Result<TracerSDK, Box<dyn std::error::Error>> {
+    let otlp_exporter = opentelemetry_otlp::new_exporter()
+        .http()
+        .with_http_client(reqwest::Client::new());
+    let tracer_provider = opentelemetry_otlp::new_pipeline()
+        .tracing()
+        .with_exporter(otlp_exporter)
+        .install_batch(opentelemetry_sdk::runtime::Tokio)?;
+
+    let tracer =
+        tracer_provider.tracer_builder(OTEL_SERVICE_NAME).build();
+    Ok(tracer)
+}
+
+pub fn init_logging() -> Result<LoggerSDK, Box<dyn std::error::Error>> {
+    let otlp_exporter = opentelemetry_otlp::new_exporter()
+        .http()
+        .with_http_client(reqwest::Client::new());
+    let logger_provider = opentelemetry_otlp::new_pipeline()
+        .logging()
+        .with_exporter(otlp_exporter)
+        .install_batch(opentelemetry_sdk::runtime::Tokio)?;
+
+    let logger =
+        logger_provider.logger_builder(OTEL_SERVICE_NAME).build();
+    Ok(logger)
+}
+
+pub struct OpenTelemetryLogger {
+    logger: Box<dyn Log>,
+}
+
+impl OpenTelemetryLogger {
+    pub fn new(logger: Box<dyn Log>) -> Self {
+        Self { logger }
+    }
+}
+
+impl Log for OpenTelemetryLogger {
+    fn enabled<'a>(&self, metadata: &Metadata<'a>) -> bool {
+        self.logger.enabled(metadata)
+    }
+
+    fn log<'a>(&self, record: &Record<'a>) {
+        if !self.enabled(record.metadata()) {
+            return;
+        }
+        self.logger.log(&record);
+
+        match record.metadata().level() {
+            Level::Error => {
+                // Log error messages using OpenTelemetry logger
+                let logger = &*OT_LOGGER;
+                emit_log(record, logger);
+            }
+            Level::Warn => {
+                // Log warning messages using OpenTelemetry logger
+                let logger = &*OT_LOGGER;
+                emit_log(record, logger);
+            }
+            Level::Info => {}
+            Level::Debug => {
+                // Create an OpenTelemetry trace for Debug level
+                let tracer = &*OT_TRACER;
+                create_log_trace(record, tracer);
+            }
+            Level::Trace => {
+                // Create an OpenTelemetry trace for Trace level
+                let tracer = &*OT_TRACER;
+                create_log_trace(record, tracer);
+            }
+        }
+    }
+
+    fn flush(&self) {
+        self.logger.flush()
+    }
+}
+
+fn emit_log(record: &Record, logger: &LoggerSDK) {
+    let mut rec = LogRecordSDK::default();
+
+    rec.set_target(record.target().to_string());
+    rec.set_severity_text(record.level().as_str());
+    rec.set_body(String(StringValue::from(record.args().to_string())));
+
+    logger.emit(rec);
+}
+
+fn create_log_trace(record: &Record, tracer: &TracerSDK) {
+    let mut span = tracer
+        .span_builder(format!("{} ({})", record.module_path().unwrap_or(record.target()), record.file().unwrap_or("-")))
+        .with_kind(SpanKind::Server)
+        .with_attributes([
+            KeyValue::new("level", record.level().as_str()),
+            KeyValue::new("target", record.target().to_string()),
+            KeyValue::new("message", record.args().to_string()),
+        ])
+        .start(tracer);
+    span.end();
+}

rust/cubestore/cubestore/src/util/logger.rs

@@ -1,4 +1,4 @@
-use crate::telemetry::ReportingLogger;
+use crate::telemetry::{OpenTelemetryLogger, ReportingLogger};
 use log::{Level, Log, Metadata, Record};
 use simple_logger::SimpleLogger;
 use std::env;
@@ -30,7 +30,12 @@ pub fn init_cube_logger(enable_telemetry: bool) {
     }
     let mut logger: Box<dyn Log> = Box::new(ContextLogger::new(ctx, logger));
     if enable_telemetry {
-        logger = Box::new(ReportingLogger::new(logger))
+        // No need to introduce special env vars, let's use the de facto standard ones
+        if let Ok(_) = env::var("OTEL_EXPORTER_OTLP_ENDPOINT") {
+            logger = Box::new(OpenTelemetryLogger::new(logger));
+        } else {
+            logger = Box::new(ReportingLogger::new(logger)); 
+        }
     }
 
     log::set_boxed_logger(logger).expect("Failed to initialize logger");