attributes in otel metrics

Author: nikhilsinhaparseable

src/otel/metrics.rs

@@ -24,7 +24,8 @@ use opentelemetry_proto::tonic::metrics::v1::{
 use serde_json::{Map, Value};
 
 use super::otel_utils::{
-    convert_epoch_nano_to_timestamp, insert_attributes, insert_number_if_some,
+    convert_epoch_nano_to_timestamp, fetch_attributes_string, insert_attributes,
+    insert_number_if_some, merge_attributes_in_json,
 };
 
 pub const OTEL_METRICS_KNOWN_FIELD_LIST: [&str; 5] = [
@@ -39,18 +40,16 @@ pub const OTEL_METRICS_KNOWN_FIELD_LIST: [&str; 5] = [
 /// this function flatten the exemplar json array
 /// and returns a `Map` of the exemplar json
 /// this function is reused in all json objects that have exemplar
-fn flatten_exemplar(
-    exemplars: &[Exemplar],
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_exemplar(exemplars: &[Exemplar]) -> Vec<Map<String, Value>> {
     exemplars
         .iter()
         .map(|exemplar| {
             let mut exemplar_json = Map::new();
+            let mut other_attributes = Map::new();
             insert_attributes(
                 &mut exemplar_json,
                 &exemplar.filtered_attributes,
-                other_attributes,
+                &mut other_attributes,
             );
             exemplar_json.insert(
                 "exemplar_time_unix_nano".to_string(),
@@ -84,6 +83,13 @@ fn flatten_exemplar(
                     }
                 }
             }
+            if !other_attributes.is_empty() {
+                let other_attributes = fetch_attributes_string(&other_attributes);
+                exemplar_json.insert(
+                    "other_attributes".to_string(),
+                    Value::String(other_attributes),
+                );
+            }
             exemplar_json
         })
         .collect()
@@ -93,18 +99,16 @@ fn flatten_exemplar(
 /// this function flatten the number data points json array
 /// and returns a `Vec` of `Map` of the flattened json
 /// this function is reused in all json objects that have number data points
-fn flatten_number_data_points(
-    data_points: &[NumberDataPoint],
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_number_data_points(data_points: &[NumberDataPoint]) -> Vec<Map<String, Value>> {
     data_points
         .iter()
         .map(|data_point| {
             let mut data_point_json = Map::new();
+            let mut other_attributes = Map::new();
             insert_attributes(
                 &mut data_point_json,
                 &data_point.attributes,
-                other_attributes,
+                &mut other_attributes,
             );
             data_point_json.insert(
                 "start_time_unix_nano".to_string(),
@@ -119,11 +123,19 @@ fn flatten_number_data_points(
                 )),
             );
 
-            let exemplar_json = flatten_exemplar(&data_point.exemplars, other_attributes);
-            for exemplar in exemplar_json {
-                for (key, value) in exemplar {
-                    data_point_json.insert(key, value);
+            let mut exemplar_json = flatten_exemplar(&data_point.exemplars);
+            if !exemplar_json.is_empty() {
+                merge_attributes_in_json(other_attributes, &mut exemplar_json);
+                for exemplar in exemplar_json {
+                    for (key, value) in exemplar {
+                        data_point_json.insert(key, value);
+                    }
                 }
+            } else {
+                data_point_json.insert(
+                    "other_attributes".to_string(),
+                    Value::String(fetch_attributes_string(&other_attributes)),
+                );
             }
 
             data_point_json.extend(flatten_data_point_flags(data_point.flags));
@@ -154,12 +166,10 @@ fn flatten_number_data_points(
 /// each gauge object has json array for data points
 /// this function flatten the gauge json object
 /// and returns a `Vec` of `Map` for each data point
-fn flatten_gauge(
-    gauge: &Gauge,
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_gauge(gauge: &Gauge) -> Vec<Map<String, Value>> {
     let mut vec_gauge_json = Vec::new();
-    let data_points_json = flatten_number_data_points(&gauge.data_points, other_attributes);
+    let data_points_json = flatten_number_data_points(&gauge.data_points);
+
     for data_point_json in data_points_json {
         let mut gauge_json = Map::new();
         for (key, value) in &data_point_json {
@@ -174,9 +184,9 @@ fn flatten_gauge(
 /// each sum object has json array for data points
 /// this function flatten the sum json object
 /// and returns a `Vec` of `Map` for each data point
-fn flatten_sum(sum: &Sum, other_attributes: &mut Map<String, Value>) -> Vec<Map<String, Value>> {
+fn flatten_sum(sum: &Sum) -> Vec<Map<String, Value>> {
     let mut vec_sum_json = Vec::new();
-    let data_points_json = flatten_number_data_points(&sum.data_points, other_attributes);
+    let data_points_json = flatten_number_data_points(&sum.data_points);
     for data_point_json in data_points_json {
         let mut sum_json = Map::new();
         for (key, value) in &data_point_json {
@@ -199,17 +209,15 @@ fn flatten_sum(sum: &Sum, other_attributes: &mut Map<String, Value>) -> Vec<Map<
 /// each histogram object has json array for data points
 /// this function flatten the histogram json object
 /// and returns a `Vec` of `Map` for each data point
-fn flatten_histogram(
-    histogram: &Histogram,
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_histogram(histogram: &Histogram) -> Vec<Map<String, Value>> {
     let mut data_points_json = Vec::new();
     for data_point in &histogram.data_points {
         let mut data_point_json = Map::new();
+        let mut other_attributes = Map::new();
         insert_attributes(
             &mut data_point_json,
             &data_point.attributes,
-            other_attributes,
+            &mut other_attributes,
         );
         data_point_json.insert(
             "start_time_unix_nano".to_string(),
@@ -254,11 +262,19 @@ fn flatten_histogram(
             "data_point_explicit_bounds".to_string(),
             data_point_explicit_bounds,
         );
-        let exemplar_json = flatten_exemplar(&data_point.exemplars, other_attributes);
-        for exemplar in exemplar_json {
-            for (key, value) in exemplar {
-                data_point_json.insert(key, value);
+        let mut exemplar_json = flatten_exemplar(&data_point.exemplars);
+        if !exemplar_json.is_empty() {
+            merge_attributes_in_json(other_attributes, &mut exemplar_json);
+            for exemplar in exemplar_json {
+                for (key, value) in exemplar {
+                    data_point_json.insert(key, value);
+                }
             }
+        } else {
+            data_point_json.insert(
+                "other_attributes".to_string(),
+                Value::String(fetch_attributes_string(&other_attributes)),
+            );
         }
 
         data_point_json.extend(flatten_data_point_flags(data_point.flags));
@@ -301,17 +317,15 @@ fn flatten_buckets(bucket: &Buckets) -> Map<String, Value> {
 /// each exponential histogram object has json array for data points
 /// this function flatten the exponential histogram json object
 /// and returns a `Vec` of `Map` for each data point
-fn flatten_exp_histogram(
-    exp_histogram: &ExponentialHistogram,
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_exp_histogram(exp_histogram: &ExponentialHistogram) -> Vec<Map<String, Value>> {
     let mut data_points_json = Vec::new();
     for data_point in &exp_histogram.data_points {
         let mut data_point_json = Map::new();
+        let mut other_attributes = Map::new();
         insert_attributes(
             &mut data_point_json,
             &data_point.attributes,
-            other_attributes,
+            &mut other_attributes,
         );
         data_point_json.insert(
             "start_time_unix_nano".to_string(),
@@ -350,11 +364,19 @@ fn flatten_exp_histogram(
                 data_point_json.insert(format!("negative_{}", key), value);
             }
         }
-        let exemplar_json = flatten_exemplar(&data_point.exemplars, other_attributes);
-        for exemplar in exemplar_json {
-            for (key, value) in exemplar {
-                data_point_json.insert(key, value);
+        let mut exemplar_json = flatten_exemplar(&data_point.exemplars);
+        if !exemplar_json.is_empty() {
+            merge_attributes_in_json(other_attributes, &mut exemplar_json);
+            for exemplar in exemplar_json {
+                for (key, value) in exemplar {
+                    data_point_json.insert(key, value);
+                }
             }
+        } else {
+            data_point_json.insert(
+                "other_attributes".to_string(),
+                Value::String(fetch_attributes_string(&other_attributes)),
+            );
         }
 
         data_points_json.push(data_point_json);
@@ -375,17 +397,15 @@ fn flatten_exp_histogram(
 /// each summary object has json array for data points
 /// this function flatten the summary json object
 /// and returns a `Vec` of `Map` for each data point
-fn flatten_summary(
-    summary: &Summary,
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+fn flatten_summary(summary: &Summary) -> Vec<Map<String, Value>> {
     let mut data_points_json = Vec::new();
     for data_point in &summary.data_points {
         let mut data_point_json = Map::new();
+        let mut other_attributes = Map::new();
         insert_attributes(
             &mut data_point_json,
             &data_point.attributes,
-            other_attributes,
+            &mut other_attributes,
         );
         data_point_json.insert(
             "start_time_unix_nano".to_string(),
@@ -449,33 +469,31 @@ fn flatten_summary(
 /// this function flatten the metric json object
 /// and returns a `Vec` of `Map` of the flattened json
 /// this function is called recursively for each metric record object in the otel metrics event
-pub fn flatten_metrics_record(
-    metrics_record: &Metric,
-    other_attributes: &mut Map<String, Value>,
-) -> Vec<Map<String, Value>> {
+pub fn flatten_metrics_record(metrics_record: &Metric) -> Vec<Map<String, Value>> {
     let mut data_points_json = Vec::new();
     let mut metric_json = Map::new();
+    let mut other_attributes = Map::new();
     let mut metric_type = String::default();
     match &metrics_record.data {
         Some(metric::Data::Gauge(gauge)) => {
             metric_type = "gauge".to_string();
-            data_points_json.extend(flatten_gauge(gauge, other_attributes));
+            data_points_json.extend(flatten_gauge(gauge));
         }
         Some(metric::Data::Sum(sum)) => {
             metric_type = "sum".to_string();
-            data_points_json.extend(flatten_sum(sum, other_attributes));
+            data_points_json.extend(flatten_sum(sum));
         }
         Some(metric::Data::Histogram(histogram)) => {
             metric_type = "histogram".to_string();
-            data_points_json.extend(flatten_histogram(histogram, other_attributes));
+            data_points_json.extend(flatten_histogram(histogram));
         }
         Some(metric::Data::ExponentialHistogram(exp_histogram)) => {
             metric_type = "exponential_histogram".to_string();
-            data_points_json.extend(flatten_exp_histogram(exp_histogram, other_attributes));
+            data_points_json.extend(flatten_exp_histogram(exp_histogram));
         }
         Some(metric::Data::Summary(summary)) => {
             metric_type = "summary".to_string();
-            data_points_json.extend(flatten_summary(summary, other_attributes));
+            data_points_json.extend(flatten_summary(summary));
         }
         None => {}
     }
@@ -492,7 +510,11 @@ pub fn flatten_metrics_record(
         Value::String(metrics_record.unit.clone()),
     );
     metric_json.insert("metric_type".to_string(), Value::String(metric_type));
-    insert_attributes(&mut metric_json, &metrics_record.metadata, other_attributes);
+    insert_attributes(
+        &mut metric_json,
+        &metrics_record.metadata,
+        &mut other_attributes,
+    );
     for data_point_json in &mut data_points_json {
         for (key, value) in &metric_json {
             data_point_json.insert(key.clone(), value.clone());
@@ -501,16 +523,17 @@ pub fn flatten_metrics_record(
     if data_points_json.is_empty() {
         data_points_json.push(metric_json);
     }
+    merge_attributes_in_json(other_attributes, &mut data_points_json);
     data_points_json
 }
 
 /// this function performs the custom flattening of the otel metrics
 /// and returns a `Vec` of `Value::Object` of the flattened json
 pub fn flatten_otel_metrics(message: MetricsData) -> Vec<Value> {
     let mut vec_otel_json = Vec::new();
-    let mut other_attributes = Map::new();
     for record in &message.resource_metrics {
         let mut resource_metrics_json = Map::new();
+        let mut other_attributes = Map::new();
         if let Some(resource) = &record.resource {
             insert_attributes(
                 &mut resource_metrics_json,
@@ -525,11 +548,9 @@ 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();
             for metrics_record in &scope_metric.metrics {
-                vec_scope_metrics_json.extend(flatten_metrics_record(
-                    metrics_record,
-                    &mut other_attributes,
-                ));
+                vec_scope_metrics_json.extend(flatten_metrics_record(metrics_record));
             }
             if let Some(scope) = &scope_metric.scope {
                 scope_metrics_json
@@ -558,6 +579,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);
         }
         resource_metrics_json.insert(
             "resource_metrics_schema_url".to_string(),
@@ -568,22 +590,9 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Vec<Value> {
                 resource_metric_json.insert(key.clone(), value.clone());
             }
         }
+        merge_attributes_in_json(other_attributes, &mut vec_scope_metrics_json);
         vec_otel_json.extend(vec_scope_metrics_json);
     }
-    // Add common attributes as one attribute in stringified array to each metric 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 metric_record_json in &mut vec_otel_json {
-        metric_record_json.insert(
-            "other_attributes".to_string(),
-            Value::String(other_attributes.clone()),
-        );
-    }
     vec_otel_json.into_iter().map(Value::Object).collect()
 }
 

src/otel/otel_utils.rs

@@ -45,7 +45,13 @@ pub fn collect_json_from_value(key: &String, value: OtelValue) -> Map<String, Va
         OtelValue::ArrayValue(array_val) => {
             let json_array_value = collect_json_from_array_value(array_val);
             // Convert the array to a JSON string
-            let json_array_string = serde_json::to_string(&json_array_value).unwrap();
+            let json_array_string = match serde_json::to_string(&json_array_value) {
+                Ok(s) => s,
+                Err(e) => {
+                    tracing::warn!("failed to serialise array value: {e}");
+                    String::default()
+                }
+            };
             // Insert the array into the result map
             value_json.insert(key.to_string(), Value::String(json_array_string));
         }
@@ -105,9 +111,12 @@ fn collect_json_from_key_value_list(key_value_list: KeyValueList) -> Map<String,
     let mut kv_list_json: Map<String, Value> = Map::new();
     for key_value in key_value_list.values {
         if let Some(val) = key_value.value {
-            let val = val.value.unwrap();
-            let json_value = collect_json_from_value(&key_value.key, val);
-            kv_list_json.extend(json_value);
+            if let Some(val) = val.value {
+                let json_value = collect_json_from_value(&key_value.key, val);
+                kv_list_json.extend(json_value);
+            } else {
+                tracing::warn!("Key '{}' has no value in key-value list", key_value.key);
+            }
         }
     }
     kv_list_json
@@ -206,7 +215,7 @@ pub fn merge_attributes_in_json(
                 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}");
+                let other_attributes = format!("{other_attributes} {attributes}");
                 json.insert(
                     "other_attributes".to_string(),
                     Value::String(other_attributes),