feat: add log stuuport to fake-collector (#152)

* Add FakeLogsService; Split trace to new file

Added a FakeLogsService that works in a similar manner to the
FakeTraceService.

I also split the tracing functionality out into separate file/module for
better code organization now that there's more than a single fake
service

* Remove println!

* Remove another println!

Author: krlohnes

fake-opentelemetry-collector/Cargo.toml

@@ -14,7 +14,7 @@ license.workspace = true
 futures = "0.3"
 hex = "0.4"
 opentelemetry = { workspace = true }
-opentelemetry-otlp = { workspace = true, features = ["grpc-tonic", "trace"] }
+opentelemetry-otlp = { workspace = true, features = ["grpc-tonic", "trace", "logs"] }
 opentelemetry-proto = { workspace = true, features = ["gen-tonic"] }
 # need tokio runtime to run smoke tests.
 opentelemetry_sdk = { workspace = true, features = [

fake-opentelemetry-collector/src/common.rs

@@ -0,0 +1,10 @@
+use std::collections::BTreeMap;
+
+pub(crate) fn cnv_attributes(
+    attributes: &[opentelemetry_proto::tonic::common::v1::KeyValue],
+) -> BTreeMap<String, String> {
+    attributes
+        .iter()
+        .map(|kv| (kv.key.to_string(), format!("{:?}", kv.value)))
+        .collect::<BTreeMap<String, String>>()
+}

fake-opentelemetry-collector/src/lib.rs

@@ -1,163 +1,26 @@
-//! based on https://github.com/open-telemetry/opentelemetry-rust/blob/main/opentelemetry-otlp/tests/smoke.rs
+mod common;
+mod logs;
+mod trace;
+pub use logs::ExportedLog;
+pub use trace::ExportedSpan;
+
+use logs::*;
+use trace::*;
+
+use std::net::SocketAddr;
+
 use futures::StreamExt;
-use opentelemetry_proto::tonic::collector::trace::v1::{
-    trace_service_server::{TraceService, TraceServiceServer},
-    ExportTraceServiceRequest, ExportTraceServiceResponse,
-};
-use serde::Serialize;
-use std::collections::BTreeMap;
+use opentelemetry_otlp::WithExportConfig;
+use opentelemetry_proto::tonic::collector::logs::v1::logs_service_server::LogsServiceServer;
+use opentelemetry_proto::tonic::collector::trace::v1::trace_service_server::TraceServiceServer;
 use std::sync::mpsc;
-use std::{net::SocketAddr, sync::Mutex};
 use tokio_stream::wrappers::TcpListenerStream;
 use tracing::debug;
 
-//pub type ExportedSpan = opentelemetry_proto::tonic::trace::v1::Span;
-
-/// opentelemetry_proto::tonic::trace::v1::Span is no compatible with serde::Serialize
-/// and to be able to test with insta,... it's needed (Debug is not enough to be able to filter unstable value,...)
-#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
-pub struct ExportedSpan {
-    pub trace_id: String,
-    pub span_id: String,
-    pub trace_state: String,
-    pub parent_span_id: String,
-    pub name: String,
-    pub kind: String, //SpanKind,
-    pub start_time_unix_nano: u64,
-    pub end_time_unix_nano: u64,
-    pub attributes: BTreeMap<String, String>,
-    pub dropped_attributes_count: u32,
-    pub events: Vec<Event>,
-    pub dropped_events_count: u32,
-    pub links: Vec<Link>,
-    pub dropped_links_count: u32,
-    pub status: Option<Status>,
-}
-
-impl From<opentelemetry_proto::tonic::trace::v1::Span> for ExportedSpan {
-    fn from(value: opentelemetry_proto::tonic::trace::v1::Span) -> Self {
-        Self {
-            trace_id: hex::encode(&value.trace_id),
-            span_id: hex::encode(&value.span_id),
-            trace_state: value.trace_state.clone(),
-            parent_span_id: hex::encode(&value.parent_span_id),
-            name: value.name.clone(),
-            kind: value.kind().as_str_name().to_owned(),
-            start_time_unix_nano: value.start_time_unix_nano,
-            end_time_unix_nano: value.end_time_unix_nano,
-            attributes: cnv_attributes(&value.attributes),
-            dropped_attributes_count: value.dropped_attributes_count,
-            events: value.events.iter().map(Event::from).collect(),
-            dropped_events_count: value.dropped_events_count,
-            links: value.links.iter().map(Link::from).collect(),
-            dropped_links_count: value.dropped_links_count,
-            status: value.status.map(Status::from),
-        }
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Serialize)]
-pub struct Status {
-    message: String,
-    code: String,
-}
-
-impl From<opentelemetry_proto::tonic::trace::v1::Status> for Status {
-    fn from(value: opentelemetry_proto::tonic::trace::v1::Status) -> Self {
-        Self {
-            message: value.message.clone(),
-            code: value.code().as_str_name().to_string(),
-        }
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
-pub struct Link {
-    pub trace_id: String,
-    pub span_id: String,
-    pub trace_state: String,
-    pub attributes: BTreeMap<String, String>,
-    pub dropped_attributes_count: u32,
-}
-
-impl From<&opentelemetry_proto::tonic::trace::v1::span::Link> for Link {
-    fn from(value: &opentelemetry_proto::tonic::trace::v1::span::Link) -> Self {
-        Self {
-            trace_id: hex::encode(&value.trace_id),
-            span_id: hex::encode(&value.span_id),
-            trace_state: value.trace_state.clone(),
-            attributes: cnv_attributes(&value.attributes),
-            dropped_attributes_count: value.dropped_attributes_count,
-        }
-    }
-}
-
-fn cnv_attributes(
-    attributes: &[opentelemetry_proto::tonic::common::v1::KeyValue],
-) -> BTreeMap<String, String> {
-    attributes
-        .iter()
-        .map(|kv| (kv.key.to_string(), format!("{:?}", kv.value)))
-        .collect::<BTreeMap<String, String>>()
-    // v.sort_by_key(|kv| kv.0.clone());
-    // v
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
-pub struct Event {
-    time_unix_nano: u64,
-    name: String,
-    attributes: BTreeMap<String, String>,
-    dropped_attributes_count: u32,
-}
-
-impl From<&opentelemetry_proto::tonic::trace::v1::span::Event> for Event {
-    fn from(value: &opentelemetry_proto::tonic::trace::v1::span::Event) -> Self {
-        Self {
-            time_unix_nano: value.time_unix_nano,
-            name: value.name.clone(),
-            attributes: cnv_attributes(&value.attributes),
-            dropped_attributes_count: value.dropped_attributes_count,
-        }
-    }
-}
-
-struct FakeTraceService {
-    tx: Mutex<mpsc::SyncSender<ExportedSpan>>,
-}
-
-impl FakeTraceService {
-    pub fn new(tx: mpsc::SyncSender<ExportedSpan>) -> Self {
-        Self { tx: Mutex::new(tx) }
-    }
-}
-
-#[tonic::async_trait]
-impl TraceService for FakeTraceService {
-    async fn export(
-        &self,
-        request: tonic::Request<ExportTraceServiceRequest>,
-    ) -> Result<tonic::Response<ExportTraceServiceResponse>, tonic::Status> {
-        debug!("Sending request into channel...");
-        request
-            .into_inner()
-            .resource_spans
-            .into_iter()
-            .flat_map(|rs| rs.scope_spans)
-            .flat_map(|ss| ss.spans)
-            .map(ExportedSpan::from)
-            .for_each(|es| {
-                self.tx.lock().unwrap().send(es).expect("Channel full");
-            });
-        Ok(tonic::Response::new(ExportTraceServiceResponse {
-            partial_success: None,
-        }))
-    }
-}
-
 pub struct FakeCollectorServer {
     address: SocketAddr,
     req_rx: mpsc::Receiver<ExportedSpan>,
+    log_rx: mpsc::Receiver<ExportedLog>,
     handle: tokio::task::JoinHandle<()>,
 }
 
@@ -174,20 +37,23 @@ impl FakeCollectorServer {
         });
 
         let (req_tx, req_rx) = mpsc::sync_channel::<ExportedSpan>(1024);
-        let service = TraceServiceServer::new(FakeTraceService::new(req_tx));
+        let (log_tx, log_rx) = mpsc::sync_channel::<ExportedLog>(1024);
+        let trace_service = TraceServiceServer::new(FakeTraceService::new(req_tx));
+        let logs_service = LogsServiceServer::new(FakeLogsService::new(log_tx));
         let handle = tokio::task::spawn(async move {
             debug!("start FakeCollectorServer http://{addr}"); //Devskim: ignore DS137138)
             tonic::transport::Server::builder()
-                .add_service(service)
+                .add_service(trace_service)
+                .add_service(logs_service)
                 .serve_with_incoming(stream)
-                // .serve(addr)
                 .await
                 .expect("Server failed");
             debug!("stop FakeCollectorServer");
         });
         Ok(Self {
             address: addr,
             req_rx,
+            log_rx,
             handle,
         })
     }
@@ -204,13 +70,16 @@ impl FakeCollectorServer {
         std::iter::from_fn(|| self.req_rx.try_recv().ok()).collect::<Vec<_>>()
     }
 
+    pub fn exported_logs(&self) -> Vec<ExportedLog> {
+        std::iter::from_fn(|| self.log_rx.try_recv().ok()).collect::<Vec<_>>()
+    }
+
     pub fn abort(self) {
         self.handle.abort()
     }
 }
 
 pub async fn setup_tracer(fake_server: &FakeCollectorServer) -> opentelemetry_sdk::trace::Tracer {
-    use opentelemetry_otlp::WithExportConfig;
     // if the environment variable is set (in test or in caller), `with_endpoint` value is ignored
     std::env::remove_var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT");
     opentelemetry_otlp::new_pipeline()
@@ -221,14 +90,29 @@ pub async fn setup_tracer(fake_server: &FakeCollectorServer) -> opentelemetry_sd
                 .with_endpoint(fake_server.endpoint()),
         )
         .install_batch(opentelemetry_sdk::runtime::Tokio)
-        .expect("failed to install")
+        .expect("failed to install tracer")
+}
+
+pub async fn setup_logger(
+    fake_server: &FakeCollectorServer,
+) -> opentelemetry_sdk::logs::LoggerProvider {
+    opentelemetry_otlp::new_pipeline()
+        .logging()
+        .with_exporter(
+            opentelemetry_otlp::new_exporter()
+                .tonic()
+                .with_endpoint(fake_server.endpoint()),
+        )
+        .install_simple() //Install simple so we don't have to wait for batching in tests
+        .expect("failed to install logging")
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    //use opentelemetry::{KeyValue, Value};
+
     use opentelemetry::global::shutdown_tracer_provider;
+    use opentelemetry::logs::{LogRecord, Logger, LoggerProvider, Severity};
     use opentelemetry::trace::{Span, SpanKind, Tracer};
 
     #[tokio::test(flavor = "multi_thread")]
@@ -245,7 +129,6 @@ mod tests {
             .start(&tracer);
         span.add_event("my-test-event", vec![]);
         span.end();
-
         shutdown_tracer_provider();
 
         let otel_spans = fake_collector.exported_spans();
@@ -272,4 +155,35 @@ mod tests {
             }),
         });
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_fake_logger_and_collector() {
+        let fake_collector = FakeCollectorServer::start()
+            .await
+            .expect("fake collector setup and started");
+
+        let logger_provider = setup_logger(&fake_collector).await;
+        let logger = logger_provider.logger("test");
+        let mut record = logger.create_log_record();
+        record.set_body("This is information".into());
+        record.set_severity_number(Severity::Info);
+        record.set_severity_text("info".into());
+        logger.emit(record);
+
+        let otel_logs = fake_collector.exported_logs();
+        insta::assert_yaml_snapshot!(otel_logs, {
+            "[].trace_id" => insta::dynamic_redaction(|value, _path| {
+                assert2::let_assert!(Some(trace_id) = value.as_str());
+                format!("[trace_id:lg{}]", trace_id.len())
+            }),
+            "[].span_id" => insta::dynamic_redaction(|value, _path| {
+                assert2::let_assert!(Some(span_id) = value.as_str());
+                format!("[span_id:lg{}]", span_id.len())
+            }),
+            "[].observed_time_unix_nano" => "[timestamp]",
+            "[].severity_number" => 9,
+            "[].severity_text" => "info",
+            "[].body" => "AnyValue { value: Some(StringValue(\"This is information\")) }",
+        });
+    }
 }

fake-opentelemetry-collector/src/logs.rs

@@ -0,0 +1,68 @@
+use crate::common::cnv_attributes;
+use opentelemetry_proto::tonic::collector::logs::v1::{
+    logs_service_server::LogsService, ExportLogsServiceRequest, ExportLogsServiceResponse,
+};
+use opentelemetry_proto::tonic::common::v1::AnyValue;
+use serde::Serialize;
+use std::collections::BTreeMap;
+use std::sync::{mpsc, Mutex};
+
+/// This is created to flatten the log record to make it more compatible with insta for testing
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct ExportedLog {
+    pub trace_id: String,
+    pub span_id: String,
+    pub observed_time_unix_nano: u64,
+    pub severity_number: i32,
+    pub severity_text: String,
+    pub body: Option<String>,
+    pub attributes: BTreeMap<String, String>,
+    pub dropped_attributes_count: u32,
+    pub flags: u32,
+}
+
+impl From<opentelemetry_proto::tonic::logs::v1::LogRecord> for ExportedLog {
+    fn from(value: opentelemetry_proto::tonic::logs::v1::LogRecord) -> Self {
+        Self {
+            trace_id: hex::encode(value.trace_id),
+            span_id: hex::encode(value.span_id),
+            observed_time_unix_nano: value.observed_time_unix_nano,
+            severity_number: value.severity_number,
+            severity_text: value.severity_text,
+            body: value.body.map(|value| format!("{:?}", value)),
+            attributes: cnv_attributes(&value.attributes),
+            dropped_attributes_count: value.dropped_attributes_count,
+            flags: value.flags,
+        }
+    }
+}
+
+pub(crate) struct FakeLogsService {
+    tx: Mutex<mpsc::SyncSender<ExportedLog>>,
+}
+
+impl FakeLogsService {
+    pub fn new(tx: mpsc::SyncSender<ExportedLog>) -> Self {
+        Self { tx: Mutex::new(tx) }
+    }
+}
+
+#[tonic::async_trait]
+impl LogsService for FakeLogsService {
+    async fn export(
+        &self,
+        request: tonic::Request<ExportLogsServiceRequest>,
+    ) -> Result<tonic::Response<ExportLogsServiceResponse>, tonic::Status> {
+        request
+            .into_inner()
+            .resource_logs
+            .into_iter()
+            .flat_map(|rl| rl.scope_logs)
+            .flat_map(|sl| sl.log_records)
+            .map(ExportedLog::from)
+            .for_each(|el| self.tx.lock().unwrap().send(el).unwrap());
+        Ok(tonic::Response::new(ExportLogsServiceResponse {
+            partial_success: None,
+        }))
+    }
+}