refactor and changes to logs and traces attributes handling

Author: nikhilsinhaparseable

src/otel/logs.rs

@@ -25,7 +25,9 @@ use serde_json::Value;
 
 use super::otel_utils::collect_json_from_values;
 use super::otel_utils::convert_epoch_nano_to_timestamp;
+use super::otel_utils::fetch_attributes_string;
 use super::otel_utils::insert_attributes;
+use super::otel_utils::merge_attributes_in_json;
 
 pub const OTEL_LOG_KNOWN_FIELD_LIST: [&str; 6] = [
     "time_unix_nano",
@@ -106,13 +108,7 @@ pub fn flatten_log_record(log_record: &LogRecord) -> Map<String, Value> {
 
     // Add the `other_attributes` to the log record json
     if !other_attributes.is_empty() {
-        let other_attributes = match serde_json::to_string(&other_attributes) {
-            Ok(s) => s,
-            Err(e) => {
-                tracing::warn!("failed to serialise OTEL other_attributes: {e}");
-                String::default()
-            }
-        };
+        let other_attributes = fetch_attributes_string(&other_attributes);
         log_record_json.insert(
             "other_attributes".to_string(),
             Value::String(other_attributes),
@@ -156,33 +152,8 @@ fn flatten_scope_log(scope_log: &ScopeLogs) -> Vec<Map<String, Value>> {
         vec_scope_log_json.push(combined_json);
     }
 
-    if !other_attributes.is_empty() {
-        let scope_other_attributes = match serde_json::to_string(&other_attributes) {
-            Ok(s) => s,
-            Err(e) => {
-                tracing::warn!("failed to serialise OTEL other_attributes: {e}");
-                String::default()
-            }
-        };
-        // append scope_other_attributes to each log record json
-        for scope_log_json in &mut vec_scope_log_json {
-            // fetch the other_attributes from the scope log json
-            if let Some(other_attributes) = scope_log_json.get("other_attributes") {
-                let other_attributes = other_attributes.as_str().unwrap_or_default();
-                // append the other_attributes to the scope log json
-                let other_attributes = format!("{other_attributes}, {scope_other_attributes}");
-                scope_log_json.insert(
-                    "other_attributes".to_string(),
-                    Value::String(other_attributes),
-                );
-            } else {
-                scope_log_json.insert(
-                    "other_attributes".to_string(),
-                    Value::String(scope_other_attributes.clone()),
-                );
-            }
-        }
-    }
+    // Add the `other_attributes` to the scope log json
+    merge_attributes_in_json(other_attributes, &mut vec_scope_log_json);
 
     vec_scope_log_json
 }
@@ -220,34 +191,8 @@ pub fn flatten_otel_logs(message: &LogsData) -> Vec<Value> {
             resource_logs_json.extend(resource_log_json.clone());
         }
 
-        if !other_attributes.is_empty() {
-            let resource_other_attributes = match serde_json::to_string(&other_attributes) {
-                Ok(s) => s,
-                Err(e) => {
-                    tracing::warn!("failed to serialise OTEL other_attributes: {e}");
-                    String::default()
-                }
-            };
-            // append scope_other_attributes to each log record json
-            for resource_logs_json in &mut vec_resource_logs_json {
-                // fetch the other_attributes from the scope log json
-                if let Some(other_attributes) = resource_logs_json.get("other_attributes") {
-                    let other_attributes = other_attributes.as_str().unwrap_or_default();
-                    // append the other_attributes to the scope log json
-                    let other_attributes =
-                        format!("{other_attributes}, {resource_other_attributes}");
-                    resource_logs_json.insert(
-                        "other_attributes".to_string(),
-                        Value::String(other_attributes),
-                    );
-                } else {
-                    resource_logs_json.insert(
-                        "other_attributes".to_string(),
-                        Value::String(resource_other_attributes.clone()),
-                    );
-                }
-            }
-        }
+        // Add the `other_attributes` to the resource log json
+        merge_attributes_in_json(other_attributes, &mut vec_resource_logs_json);
 
         vec_otel_json.extend(vec_resource_logs_json);
     }

src/otel/otel_utils.rs

@@ -194,3 +194,51 @@ pub fn convert_epoch_nano_to_timestamp(epoch_ns: i64) -> String {
     let dt = DateTime::from_timestamp_nanos(epoch_ns).naive_utc();
     dt.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()
 }
+
+pub fn merge_attributes_in_json(
+    attributes: Map<String, Value>,
+    vec_json: &mut Vec<Map<String, Value>>,
+) {
+    if !attributes.is_empty() {
+        let attributes = fetch_attributes_string(&attributes);
+        for json in vec_json {
+            if json.contains_key("other_attributes") {
+                let other_attributes = json.get_mut("other_attributes").unwrap();
+                let other_attributes = other_attributes.as_str().unwrap_or_default();
+                // append the other_attributes to the scope log json
+                let other_attributes = format!("{other_attributes}, {attributes}");
+                json.insert(
+                    "other_attributes".to_string(),
+                    Value::String(other_attributes),
+                );
+            } else {
+                json.insert(
+                    "other_attributes".to_string(),
+                    Value::String(attributes.clone()),
+                );
+            }
+        }
+    }
+}
+
+pub fn fetch_attributes_from_json(json_arr: &Vec<Map<String, Value>>) -> String {
+    let mut combined_attributes = String::default();
+    for json in json_arr {
+        if let Some(other_attributes) = json.get("other_attributes") {
+            if let Some(other_attributes) = other_attributes.as_str() {
+                combined_attributes.push_str(other_attributes);
+            }
+        }
+    }
+    combined_attributes
+}
+
+pub fn fetch_attributes_string(attributes: &Map<String, Value>) -> String {
+    match serde_json::to_string(attributes) {
+        Ok(s) => s,
+        Err(e) => {
+            tracing::warn!("failed to serialise OTEL other_attributes: {e}");
+            String::default()
+        }
+    }
+}

src/otel/traces.rs

@@ -25,7 +25,10 @@ use opentelemetry_proto::tonic::trace::v1::TracesData;
 use serde_json::{Map, Value};
 
 use super::otel_utils::convert_epoch_nano_to_timestamp;
+use super::otel_utils::fetch_attributes_from_json;
+use super::otel_utils::fetch_attributes_string;
 use super::otel_utils::insert_attributes;
+use super::otel_utils::merge_attributes_in_json;
 
 pub const OTEL_TRACES_KNOWN_FIELD_LIST: [&str; 15] = [
     "span_trace_id",
@@ -46,15 +49,12 @@ pub const OTEL_TRACES_KNOWN_FIELD_LIST: [&str; 15] = [
 ];
 /// this function flattens the `ScopeSpans` object
 /// and returns a `Vec` of `Map` of the flattened json
-fn flatten_scope_span(
-    scope_span: &ScopeSpans,
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_scope_span(scope_span: &ScopeSpans) -> Vec<Map<String, Value>> {
     let mut vec_scope_span_json = Vec::new();
     let mut scope_span_json = Map::new();
-
+    let mut other_attributes = Map::new();
     for span in &scope_span.spans {
-        let span_record_json = flatten_span_record(span, other_attributes);
+        let span_record_json = flatten_span_record(span);
         vec_scope_span_json.extend(span_record_json);
     }
 
@@ -64,7 +64,11 @@ fn flatten_scope_span(
             "scope_version".to_string(),
             Value::String(scope.version.clone()),
         );
-        insert_attributes(&mut scope_span_json, &scope.attributes, other_attributes);
+        insert_attributes(
+            &mut scope_span_json,
+            &scope.attributes,
+            &mut other_attributes,
+        );
         scope_span_json.insert(
             "scope_dropped_attributes_count".to_string(),
             Value::Number(scope.dropped_attributes_count.into()),
@@ -83,6 +87,8 @@ fn flatten_scope_span(
             Value::String(scope_span.schema_url.clone()),
         );
     }
+    // Add the `other_attributes` to the scope span json
+    merge_attributes_in_json(other_attributes.clone(), &mut vec_scope_span_json);
 
     vec_scope_span_json
 }
@@ -109,7 +115,7 @@ pub fn flatten_otel_traces(message: &TracesData) -> Vec<Value> {
 
         let mut vec_resource_spans_json = Vec::new();
         for scope_span in &record.scope_spans {
-            let scope_span_json = flatten_scope_span(scope_span, &mut other_attributes);
+            let scope_span_json = flatten_scope_span(scope_span);
             vec_resource_spans_json.extend(scope_span_json);
         }
 
@@ -123,49 +129,43 @@ pub fn flatten_otel_traces(message: &TracesData) -> Vec<Value> {
                 resource_spans_json.insert(key.clone(), value.clone());
             }
         }
-
+        // Add the `other_attributes` to the resource span json
+        merge_attributes_in_json(other_attributes.clone(), &mut vec_resource_spans_json);
         vec_otel_json.extend(vec_resource_spans_json);
     }
-    // Add common attributes as one attribute in stringified array to each span record
-    let other_attributes = match serde_json::to_string(&other_attributes) {
-        Ok(s) => s,
-        Err(e) => {
-            tracing::warn!("failed to serialise OTEL other_attributes: {e}");
-            String::default()
-        }
-    };
-    for span_record_json in &mut vec_otel_json {
-        span_record_json.insert(
-            "other_attributes".to_string(),
-            Value::String(other_attributes.clone()),
-        );
-    }
+
     vec_otel_json.into_iter().map(Value::Object).collect()
 }
 
 /// otel traces has json array of events
 /// this function flattens the `Event` object
 /// and returns a `Vec` of `Map` of the flattened json
-fn flatten_events(
-    events: &[Event],
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_events(events: &[Event]) -> Vec<Map<String, Value>> {
     events
         .iter()
         .map(|event| {
             let mut event_json = Map::new();
+            let mut other_attributes = Map::new();
             event_json.insert(
                 "event_time_unix_nano".to_string(),
                 Value::String(
                     convert_epoch_nano_to_timestamp(event.time_unix_nano as i64).to_string(),
                 ),
             );
             event_json.insert("event_name".to_string(), Value::String(event.name.clone()));
-            insert_attributes(&mut event_json, &event.attributes, other_attributes);
+            insert_attributes(&mut event_json, &event.attributes, &mut other_attributes);
             event_json.insert(
                 "event_dropped_attributes_count".to_string(),
                 Value::Number(event.dropped_attributes_count.into()),
             );
+
+            if !other_attributes.is_empty() {
+                let other_attributes = fetch_attributes_string(&other_attributes);
+                event_json.insert(
+                    "other_attributes".to_string(),
+                    Value::String(other_attributes),
+                );
+            }
             event_json
         })
         .collect()
@@ -174,14 +174,12 @@ fn flatten_events(
 /// otel traces has json array of links
 /// this function flattens the `Link` object
 /// and returns a `Vec` of `Map` of the flattened json
-fn flatten_links(
-    links: &[Link],
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_links(links: &[Link]) -> Vec<Map<String, Value>> {
     links
         .iter()
         .map(|link| {
             let mut link_json = Map::new();
+            let mut other_attributes = Map::new();
             link_json.insert(
                 "link_span_id".to_string(),
                 Value::String(hex::encode(&link.span_id)),
@@ -191,11 +189,19 @@ fn flatten_links(
                 Value::String(hex::encode(&link.trace_id)),
             );
 
-            insert_attributes(&mut link_json, &link.attributes, other_attributes);
+            insert_attributes(&mut link_json, &link.attributes, &mut other_attributes);
             link_json.insert(
                 "link_dropped_attributes_count".to_string(),
                 Value::Number(link.dropped_attributes_count.into()),
             );
+
+            if !other_attributes.is_empty() {
+                let other_attributes = fetch_attributes_string(&other_attributes);
+                link_json.insert(
+                    "other_attributes".to_string(),
+                    Value::String(other_attributes),
+                );
+            }
             link_json
         })
         .collect()
@@ -278,12 +284,9 @@ fn flatten_kind(kind: i32) -> Map<String, Value> {
 /// this function flattens the `Span` object
 /// and returns a `Vec` of `Map` of the flattened json
 /// this function is called recursively for each span record object in the otel traces event
-fn flatten_span_record(
-    span_record: &Span,
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_span_record(span_record: &Span) -> Vec<Map<String, Value>> {
     let mut span_records_json = Vec::new();
-
+    let mut other_attributes = Map::new();
     let mut span_record_json = Map::new();
     span_record_json.insert(
         "span_trace_id".to_string(),
@@ -322,18 +325,41 @@ fn flatten_span_record(
     insert_attributes(
         &mut span_record_json,
         &span_record.attributes,
-        other_attributes,
+        &mut other_attributes,
     );
     span_record_json.insert(
         "span_dropped_attributes_count".to_string(),
         Value::Number(span_record.dropped_attributes_count.into()),
     );
-    span_records_json.extend(flatten_events(&span_record.events, other_attributes));
+    let events_json = flatten_events(&span_record.events);
+    // fetch all other_attributes from the events_json
+    let events_other_attributes = fetch_attributes_from_json(&events_json);
+    span_records_json.extend(events_json);
     span_record_json.insert(
         "span_dropped_events_count".to_string(),
         Value::Number(span_record.dropped_events_count.into()),
     );
-    span_records_json.extend(flatten_links(&span_record.links, other_attributes));
+    let links_json = flatten_links(&span_record.links);
+    // fetch all other_attributes from the links_json
+    let links_other_attributes = fetch_attributes_from_json(&links_json);
+    span_records_json.extend(links_json);
+    if !other_attributes.is_empty() {
+        let other_attributes = fetch_attributes_string(&other_attributes);
+        span_record_json.insert(
+            "other_attributes".to_string(),
+            Value::String(format!(
+                "{other_attributes} {events_other_attributes} {links_other_attributes}"
+            )),
+        );
+    } else {
+        span_record_json.insert(
+            "other_attributes".to_string(),
+            Value::String(format!(
+                "{events_other_attributes} {links_other_attributes}"
+            )),
+        );
+    }
+
     span_record_json.insert(
         "span_dropped_links_count".to_string(),
         Value::Number(span_record.dropped_links_count.into()),