refactor and store other attributes as map

Author: nikhilsinhaparseable

src/otel/metrics.rs

@@ -131,7 +131,7 @@ fn flatten_number_data_points(data_points: &[NumberDataPoint]) -> Vec<Map<String
                         data_point_json.insert(key, value);
                     }
                 }
-            } else {
+            } else if !other_attributes.is_empty() {
                 data_point_json.insert(
                     "other_attributes".to_string(),
                     Value::String(fetch_attributes_string(&other_attributes)),
@@ -270,7 +270,7 @@ fn flatten_histogram(histogram: &Histogram) -> Vec<Map<String, Value>> {
                     data_point_json.insert(key, value);
                 }
             }
-        } else {
+        } else if !other_attributes.is_empty() {
             data_point_json.insert(
                 "other_attributes".to_string(),
                 Value::String(fetch_attributes_string(&other_attributes)),
@@ -372,7 +372,7 @@ fn flatten_exp_histogram(exp_histogram: &ExponentialHistogram) -> Vec<Map<String
                     data_point_json.insert(key, value);
                 }
             }
-        } else {
+        } else if !other_attributes.is_empty() {
             data_point_json.insert(
                 "other_attributes".to_string(),
                 Value::String(fetch_attributes_string(&other_attributes)),
@@ -459,6 +459,14 @@ fn flatten_summary(summary: &Summary) -> Vec<Map<String, Value>> {
                     .collect(),
             ),
         );
+
+        if !other_attributes.is_empty() {
+            data_point_json.insert(
+                "other_attributes".to_string(),
+                Value::String(fetch_attributes_string(&other_attributes)),
+            );
+        }
+
         data_points_json.push(data_point_json);
     }
     data_points_json
@@ -548,7 +556,7 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Vec<Value> {
         let mut vec_scope_metrics_json = Vec::new();
         for scope_metric in &record.scope_metrics {
             let mut scope_metrics_json = Map::new();
-            let mut other_attributes = Map::new();
+            let mut scope_other_attributes = Map::new();
             for metrics_record in &scope_metric.metrics {
                 vec_scope_metrics_json.extend(flatten_metrics_record(metrics_record));
             }
@@ -562,7 +570,7 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Vec<Value> {
                 insert_attributes(
                     &mut scope_metrics_json,
                     &scope.attributes,
-                    &mut other_attributes,
+                    &mut scope_other_attributes,
                 );
                 scope_metrics_json.insert(
                     "scope_dropped_attributes_count".to_string(),
@@ -579,7 +587,7 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Vec<Value> {
                     scope_metric_json.insert(key.clone(), value.clone());
                 }
             }
-            merge_attributes_in_json(other_attributes, &mut vec_scope_metrics_json);
+            merge_attributes_in_json(scope_other_attributes, &mut vec_scope_metrics_json);
         }
         resource_metrics_json.insert(
             "resource_metrics_schema_url".to_string(),

src/otel/otel_utils.rs

@@ -204,44 +204,66 @@ pub fn convert_epoch_nano_to_timestamp(epoch_ns: i64) -> String {
     dt.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()
 }
 
+/// fetch `other_attributes` from array of JSON objects
+/// merge them with the provided attributes
+/// and return the merged array of JSON object
 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()),
-                );
+            if let Some(other_attrs) = json.get("other_attributes") {
+                if let Value::String(attrs_str) = other_attrs {
+                    if let Ok(mut existing_attrs) =
+                        serde_json::from_str::<Map<String, Value>>(attrs_str)
+                    {
+                        for (key, value) in attributes.clone() {
+                            existing_attrs.insert(key, value);
+                        }
+                        if let Ok(merged_str) = serde_json::to_string(&existing_attrs) {
+                            json.insert("other_attributes".to_string(), Value::String(merged_str));
+                        }
+                    } else if let Ok(attrs_str) = serde_json::to_string(&attributes) {
+                        json.insert("other_attributes".to_string(), Value::String(attrs_str));
+                    }
+                } else if let Value::Object(existing_attrs) = other_attrs {
+                    let mut merged_attrs = existing_attrs.clone();
+                    for (key, value) in attributes.clone() {
+                        merged_attrs.insert(key, value);
+                    }
+                    if let Ok(merged_str) = serde_json::to_string(&merged_attrs) {
+                        json.insert("other_attributes".to_string(), Value::String(merged_str));
+                    }
+                }
+            } else if let Ok(attrs_str) = serde_json::to_string(&attributes) {
+                json.insert("other_attributes".to_string(), Value::String(attrs_str));
             }
         }
     }
 }
 
-pub fn fetch_attributes_from_json(json_arr: &Vec<Map<String, Value>>) -> String {
-    let mut combined_attributes = String::default();
+/// fetch `other_attributes` from array of JSON objects
+/// and merge them into a single map
+/// and return the merged map
+pub fn fetch_attributes_from_json(json_arr: &Vec<Map<String, Value>>) -> Map<String, Value> {
+    let mut merged_attributes = Map::new();
+
     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);
+        if let Some(Value::String(attrs_str)) = json.get("other_attributes") {
+            if let Ok(attrs) = serde_json::from_str::<Map<String, Value>>(attrs_str) {
+                for (key, value) in attrs {
+                    merged_attributes.insert(key, value);
+                }
             }
         }
     }
-    combined_attributes
+    merged_attributes
 }
 
+/// convert attributes map to a string
+/// and return the string
+/// if serialisation fails, return an empty string
 pub fn fetch_attributes_string(attributes: &Map<String, Value>) -> String {
     match serde_json::to_string(attributes) {
         Ok(s) => s,

src/otel/traces.rs

@@ -343,25 +343,19 @@ fn flatten_span_record(span_record: &Span) -> Vec<Map<String, Value>> {
     // 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 mut other_attributes = fetch_attributes_string(&other_attributes);
-        if !events_other_attributes.is_empty() {
-            other_attributes.push_str(&events_other_attributes);
+    // merge all other_attributes from the events_json and links_json
+    if !other_attributes.is_empty()
+        || !events_other_attributes.is_empty()
+        || !links_other_attributes.is_empty()
+    {
+        for (key, value) in &events_other_attributes {
+            other_attributes.insert(key.clone(), value.clone());
         }
-        if !links_other_attributes.is_empty() {
-            other_attributes.push_str(&links_other_attributes);
+        for (key, value) in &links_other_attributes {
+            other_attributes.insert(key.clone(), value.clone());
         }
-        span_record_json.insert(
-            "other_attributes".to_string(),
-            Value::String(other_attributes),
-        );
-    } else {
-        span_record_json.insert(
-            "other_attributes".to_string(),
-            Value::String(format!(
-                "{events_other_attributes} {links_other_attributes}"
-            )),
-        );
+        let attrs_str = fetch_attributes_string(&other_attributes);
+        span_record_json.insert("other_attributes".to_string(), Value::String(attrs_str));
     }
     span_record_json.insert(
         "span_dropped_links_count".to_string(),