feat: enable OTEL traces

add custom flattening for otel traces
use endpoints
`/v1/logs` for OTEL logs
`/v1/metrics` for OTEL metrics
`/v1/traces` for OTEL traces

add custom header X-P-Log-Source when using endpint `api/v1/ingest`
`otel-logs` for OTEL logs
`otel-metrics` for OTEL metrics
`otel-traces` for OTEL traces

Author: nikhilsinhaparseable

src/handlers/http/ingest.rs

@@ -20,14 +20,14 @@ use super::logstream::error::{CreateStreamError, StreamError};
 use super::modal::utils::ingest_utils::{flatten_and_push_logs, push_logs};
 use super::users::dashboards::DashboardError;
 use super::users::filters::FiltersError;
-use super::{otel_logs, otel_metrics};
+use super::{otel_logs, otel_metrics, otel_traces};
 use crate::event::{
     self,
     error::EventError,
     format::{self, EventFormat},
 };
 use crate::handlers::http::modal::utils::logstream_utils::create_stream_and_schema_from_storage;
-use crate::handlers::{LOG_SOURCE_KEY, LOG_SOURCE_OTEL, STREAM_NAME_HEADER_KEY};
+use crate::handlers::STREAM_NAME_HEADER_KEY;
 use crate::metadata::error::stream_info::MetadataError;
 use crate::metadata::STREAM_INFO;
 use crate::option::{Mode, CONFIG};
@@ -118,23 +118,11 @@ pub async fn handle_otel_logs_ingestion(
         let stream_name = stream_name.to_str().unwrap().to_owned();
         create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
 
-        //flatten logs
-        if let Some((_, log_source)) = req.headers().iter().find(|&(key, _)| key == LOG_SOURCE_KEY)
-        {
-            let log_source: String = log_source.to_str().unwrap().to_owned();
-            if log_source == LOG_SOURCE_OTEL {
-                let mut json = otel_logs::flatten_otel_logs(&body);
-                for record in json.iter_mut() {
-                    let body: Bytes = serde_json::to_vec(record).unwrap().into();
-                    push_logs(stream_name.to_string(), req.clone(), body).await?;
-                }
-            } else {
-                return Err(PostError::CustomError("Unknown log source".to_string()));
-            }
-        } else {
-            return Err(PostError::CustomError(
-                "log source key header is missing".to_string(),
-            ));
+        //custom flattening required for otel logs
+        let mut json = otel_logs::flatten_otel_logs(&body);
+        for record in json.iter_mut() {
+            let body: Bytes = serde_json::to_vec(record).unwrap().into();
+            push_logs(stream_name.to_string(), req.clone(), body).await?;
         }
     } else {
         return Err(PostError::Header(ParseHeaderError::MissingStreamName));
@@ -157,23 +145,11 @@ pub async fn handle_otel_metrics_ingestion(
         let stream_name = stream_name.to_str().unwrap().to_owned();
         create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
 
-        //flatten logs
-        if let Some((_, log_source)) = req.headers().iter().find(|&(key, _)| key == LOG_SOURCE_KEY)
-        {
-            let log_source: String = log_source.to_str().unwrap().to_owned();
-            if log_source == LOG_SOURCE_OTEL {
-                let mut json = otel_metrics::flatten_otel_metrics(&body);
-                for record in json.iter_mut() {
-                    let body: Bytes = serde_json::to_vec(record).unwrap().into();
-                    push_logs(stream_name.to_string(), req.clone(), body).await?;
-                }
-            } else {
-                return Err(PostError::CustomError("Unknown log source".to_string()));
-            }
-        } else {
-            return Err(PostError::CustomError(
-                "log source key header is missing".to_string(),
-            ));
+        //custom flattening required for otel metrics
+        let mut json = otel_metrics::flatten_otel_metrics(&body);
+        for record in json.iter_mut() {
+            let body: Bytes = serde_json::to_vec(record).unwrap().into();
+            push_logs(stream_name.to_string(), req.clone(), body).await?;
         }
     } else {
         return Err(PostError::Header(ParseHeaderError::MissingStreamName));
@@ -195,7 +171,13 @@ pub async fn handle_otel_traces_ingestion(
     {
         let stream_name = stream_name.to_str().unwrap().to_owned();
         create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
-        push_logs(stream_name.to_string(), req.clone(), body).await?;
+
+        //custom flattening required for otel traces
+        let mut json = otel_traces::flatten_otel_traces(&body);
+        for record in json.iter_mut() {
+            let body: Bytes = serde_json::to_vec(record).unwrap().into();
+            push_logs(stream_name.to_string(), req.clone(), body).await?;
+        }
     } else {
         return Err(PostError::Header(ParseHeaderError::MissingStreamName));
     }

src/handlers/http/mod.rs

@@ -40,6 +40,7 @@ pub mod oidc;
 pub mod otel;
 pub mod otel_logs;
 pub mod otel_metrics;
+pub mod otel_traces;
 pub mod query;
 pub mod rbac;
 pub mod role;

src/handlers/http/modal/utils/ingest_utils.rs

@@ -33,8 +33,9 @@ use crate::{
         format::{self, EventFormat},
     },
     handlers::{
-        http::{ingest::PostError, kinesis, otel_logs},
-        LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, LOG_SOURCE_OTEL, PREFIX_META, PREFIX_TAGS, SEPARATOR,
+        http::{ingest::PostError, kinesis, otel_logs, otel_metrics, otel_traces},
+        LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, LOG_SOURCE_OTEL_LOGS, LOG_SOURCE_OTEL_METRICS,
+        LOG_SOURCE_OTEL_TRACES, PREFIX_META, PREFIX_TAGS, SEPARATOR,
     },
     metadata::STREAM_INFO,
     storage::StreamType,
@@ -52,9 +53,21 @@ pub async fn flatten_and_push_logs(
         let log_source: String = log_source.to_str().unwrap().to_owned();
         match log_source.as_str() {
             LOG_SOURCE_KINESIS => json = kinesis::flatten_kinesis_logs(&body),
-            LOG_SOURCE_OTEL => {
+
+            //custom flattening required for otel logs
+            LOG_SOURCE_OTEL_LOGS => {
                 json = otel_logs::flatten_otel_logs(&body);
             }
+
+            //custom flattening required for otel metrics
+            LOG_SOURCE_OTEL_METRICS => {
+                json = otel_metrics::flatten_otel_metrics(&body);
+            }
+
+            //custom flattening required for otel traces
+            LOG_SOURCE_OTEL_TRACES => {
+                json = otel_traces::flatten_otel_traces(&body);
+            }
             _ => {
                 tracing::warn!("Unknown log source: {}", log_source);
                 push_logs(stream_name.to_string(), req.clone(), body).await?;

src/handlers/http/otel/opentelemetry.proto.trace.v1.rs

@@ -86,21 +86,21 @@ pub struct Span {
     /// is zero-length and thus is also invalid).
     ///
     /// This field is required.
-    pub trace_id: Option<Vec<u8>>,
+    pub trace_id: Option<String>,
     /// A unique identifier for a span within a trace, assigned when the span
     /// is created. The ID is an 8-byte array. An ID with all zeroes OR of length
     /// other than 8 bytes is considered invalid (empty string in OTLP/JSON
     /// is zero-length and thus is also invalid).
     ///
     /// This field is required.
-    pub span_id: Option<Vec<u8>>,
+    pub span_id: Option<String>,
     /// trace_state conveys information about request position in multiple distributed tracing graphs.
     /// It is a trace_state in w3c-trace-context format: <https://www.w3.org/TR/trace-context/#tracestate-header>
     /// See also <https://github.com/w3c/distributed-tracing> for more details about this field.
     pub trace_state: Option<String>,
     /// The `span_id` of this span's parent span. If this is a root span, then this
     /// field must be empty. The ID is an 8-byte array.
-    pub parent_span_id: Option<Vec<u8>>,
+    pub parent_span_id: Option<String>,
     /// Flags, a bit field.
     ///
     /// Bits 0-7 (8 least significant bits) are the trace flags as defined in W3C Trace
@@ -145,14 +145,14 @@ pub struct Span {
     /// Value is UNIX Epoch time in nanoseconds since 00:00:00 UTC on 1 January 1970.
     ///
     /// This field is semantically required and it is expected that end_time >= start_time.
-    pub start_time_unix_nano: Option<u64>,
+    pub start_time_unix_nano: Option<String>,
     /// end_time_unix_nano is the end time of the span. On the client side, this is the time
     /// kept by the local machine where the span execution ends. On the server side, this
     /// is the time when the server application handler stops running.
     /// Value is UNIX Epoch time in nanoseconds since 00:00:00 UTC on 1 January 1970.
     ///
     /// This field is semantically required and it is expected that end_time >= start_time.
-    pub end_time_unix_nano: Option<u64>,
+    pub end_time_unix_nano: Option<String>,
     /// attributes is a collection of key/value pairs. Note, global attributes
     /// like server name can be set using the resource API. Examples of attributes:
     ///
@@ -213,9 +213,9 @@ pub mod span {
     pub struct Link {
         /// A unique identifier of a trace that this linked span is part of. The ID is a
         /// 16-byte array.
-        pub trace_id: Option<Vec<u8>>,
+        pub trace_id: Option<String>,
         /// A unique identifier for the linked span. The ID is an 8-byte array.
-        pub span_id: Option<Vec<u8>>,
+        pub span_id: Option<String>,
         /// The trace_state associated with the link.
         pub trace_state: Option<String>,
         /// attributes is a collection of attribute key/value pairs on the link.

src/handlers/http/otel_logs.rs

@@ -28,6 +28,7 @@ use serde_json::Value;
 use super::otel::collect_json_from_values;
 use super::otel::flatten_attributes;
 
+//flatten log record and all its attributes to create one record for each log record
 pub fn flatten_log_record(log_record: &LogRecord) -> BTreeMap<String, Value> {
     let mut log_record_json: BTreeMap<String, Value> = BTreeMap::new();
     if log_record.time_unix_nano.is_some() {
@@ -121,6 +122,7 @@ pub fn flatten_log_record(log_record: &LogRecord) -> BTreeMap<String, Value> {
     log_record_json
 }
 
+//flatten otel logs
 pub fn flatten_otel_logs(body: &Bytes) -> Vec<BTreeMap<String, Value>> {
     let mut vec_otel_json: Vec<BTreeMap<String, Value>> = Vec::new();
     let body_str = std::str::from_utf8(body).unwrap();
@@ -132,6 +134,7 @@ pub fn flatten_otel_logs(body: &Bytes) -> Vec<BTreeMap<String, Value>> {
             let mut resource_log_json: BTreeMap<String, Value> = BTreeMap::new();
 
             if let Some(resource) = record.resource.as_ref() {
+                //flatten resource attributes to create multiple key value pairs (headers) for each log record
                 if let Some(attributes) = resource.attributes.as_ref() {
                     let attributes_json = flatten_attributes(attributes);
                     for key in attributes_json.keys() {
@@ -151,27 +154,28 @@ pub fn flatten_otel_logs(body: &Bytes) -> Vec<BTreeMap<String, Value>> {
 
             if let Some(scope_logs) = record.scope_logs.as_ref() {
                 let mut vec_scope_log_json: Vec<BTreeMap<String, Value>> = Vec::new();
+                //create flattened record for each scope log
                 for scope_log in scope_logs.iter() {
                     let mut scope_log_json: BTreeMap<String, Value> = BTreeMap::new();
                     if scope_log.scope.is_some() {
                         let instrumentation_scope = scope_log.scope.as_ref().unwrap();
                         if instrumentation_scope.name.is_some() {
                             scope_log_json.insert(
-                                "instrumentation_scope_name".to_string(),
+                                "scope_name".to_string(),
                                 Value::String(
                                     instrumentation_scope.name.as_ref().unwrap().to_string(),
                                 ),
                             );
                         }
                         if instrumentation_scope.version.is_some() {
                             scope_log_json.insert(
-                                "instrumentation_scope_version".to_string(),
+                                "scope_version".to_string(),
                                 Value::String(
                                     instrumentation_scope.version.as_ref().unwrap().to_string(),
                                 ),
                             );
                         }
-
+                        //flatten instrumentation scope attributes to create multiple key value pairs (headers) for each log record
                         if let Some(attributes) = instrumentation_scope.attributes.as_ref() {
                             let attributes_json = flatten_attributes(attributes);
                             for key in attributes_json.keys() {
@@ -182,7 +186,7 @@ pub fn flatten_otel_logs(body: &Bytes) -> Vec<BTreeMap<String, Value>> {
 
                         if instrumentation_scope.dropped_attributes_count.is_some() {
                             scope_log_json.insert(
-                                "instrumentation_scope_dropped_attributes_count".to_string(),
+                                "scope_dropped_attributes_count".to_string(),
                                 Value::Number(serde_json::Number::from(
                                     instrumentation_scope.dropped_attributes_count.unwrap(),
                                 )),
@@ -196,6 +200,7 @@ pub fn flatten_otel_logs(body: &Bytes) -> Vec<BTreeMap<String, Value>> {
                         );
                     }
 
+                    //create flattened record for each log record
                     for log_record in scope_log.log_records.iter() {
                         let log_record_json = flatten_log_record(log_record);