refactor

Author: nikhilsinhaparseable

src/handlers/http/ingest.rs

@@ -498,7 +498,7 @@ impl actix_web::ResponseError for PostError {
             PostError::MissingTimePartition(_) => StatusCode::BAD_REQUEST,
             PostError::KnownFormat(_) => StatusCode::BAD_REQUEST,
             PostError::IncorrectLogFormat(_) => StatusCode::BAD_REQUEST,
-            PostError::OtelError(_) => StatusCode::EXPECTATION_FAILED,
+            PostError::OtelError(_) => StatusCode::BAD_REQUEST,
         }
     }
 

src/otel/logs.rs

@@ -37,7 +37,7 @@ pub const OTEL_LOG_KNOWN_FIELD_LIST: [&str; 16] = [
     "scope_log_schema_url",
     "scope_dropped_attributes_count",
     "resource_dropped_attributes_count",
-    "resource_schema_url",
+    "schema_url",
     "time_unix_nano",
     "observed_time_unix_nano",
     "severity_number",
@@ -178,8 +178,8 @@ pub fn flatten_otel_logs(message: &LogsData) -> Result<Vec<Value>, OtelError> {
             resource_logs_json.extend(resource_log_json.clone());
 
             let attribute_count = resource_logs_json
-                .keys()
-                .filter(|key| !known_fields.contains(key.as_str()))
+                .iter()
+                .filter(|(key, _)| !known_fields.contains(key.as_str()))
                 .count();
             // Check if the number of attributes exceeds the allowed limit
             if attribute_count > PARSEABLE.options.otel_attributes_allowed_limit {

src/otel/metrics.rs

@@ -32,7 +32,7 @@ use super::otel_utils::{
     convert_epoch_nano_to_timestamp, insert_attributes, insert_number_if_some,
 };
 
-pub const OTEL_METRICS_KNOWN_FIELD_LIST: [&str; 35] = [
+pub const OTEL_METRICS_KNOWN_FIELD_LIST: [&str; 34] = [
     "metric_name",
     "metric_description",
     "metric_unit",
@@ -63,11 +63,10 @@ pub const OTEL_METRICS_KNOWN_FIELD_LIST: [&str; 35] = [
     "metric_type",
     "scope_name",
     "scope_version",
-    "scope_metrics_schema_url",
+    "scope_schema_url",
     "scope_dropped_attributes_count",
     "resource_dropped_attributes_count",
     "resource_schema_url",
-    "resource_metrics_schema_url",
 ];
 
 /// otel metrics event has json array for exemplar
@@ -544,7 +543,7 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Result<Vec<Value>, OtelErro
             }
 
             scope_metrics_json.insert(
-                "scope_metrics_schema_url".to_string(),
+                "scope_schema_url".to_string(),
                 Value::String(scope_metric.schema_url.clone()),
             );
 
@@ -556,7 +555,7 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Result<Vec<Value>, OtelErro
         }
 
         resource_metrics_json.insert(
-            "resource_metrics_schema_url".to_string(),
+            "resource_schema_url".to_string(),
             Value::String(record.schema_url.clone()),
         );
 
@@ -566,8 +565,8 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Result<Vec<Value>, OtelErro
             }
 
             let attribute_count = resource_metric_json
-                .keys()
-                .filter(|key| !known_fields.contains(key.as_str()))
+                .iter()
+                .filter(|(key, _)| !known_fields.contains(key.as_str()))
                 .count();
 
             // Check if the number of attributes exceeds the allowed limit

src/otel/otel_utils.rs

@@ -20,6 +20,7 @@ use chrono::DateTime;
 use opentelemetry_proto::tonic::common::v1::{
     any_value::Value as OtelValue, AnyValue, ArrayValue, KeyValue, KeyValueList,
 };
+use serde::Serialize;
 use serde_json::{Map, Value};
 
 // Value can be one of types - String, Bool, Int, Double, ArrayValue, AnyValue, KeyValueList, Byte
@@ -149,7 +150,7 @@ pub fn value_to_string(value: serde_json::Value) -> String {
     }
 }
 
-pub fn flatten_attributes(attributes: &Vec<KeyValue>) -> Map<String, Value> {
+pub fn flatten_attributes(attributes: &[KeyValue]) -> Map<String, Value> {
     let mut attributes_json: Map<String, Value> = Map::new();
     for attribute in attributes {
         let key = &attribute.key;
@@ -182,7 +183,7 @@ pub fn insert_bool_if_some(map: &mut Map<String, Value>, key: &str, option: &Opt
     }
 }
 
-pub fn insert_attributes(map: &mut Map<String, Value>, attributes: &Vec<KeyValue>) {
+pub fn insert_attributes(map: &mut Map<String, Value>, attributes: &[KeyValue]) {
     let attributes_json = flatten_attributes(attributes);
     for (key, value) in attributes_json {
         map.insert(key, value);
@@ -194,7 +195,7 @@ pub fn convert_epoch_nano_to_timestamp(epoch_ns: i64) -> String {
     dt.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()
 }
 
-#[derive(Debug, thiserror::Error)]
+#[derive(Debug, thiserror::Error, Serialize)]
 pub enum OtelError {
     #[error("Ingestion failed because the attributes count {0} exceeded the threshold of {1}")]
     AttributeCountExceeded(usize, usize),

src/otel/traces.rs

@@ -32,7 +32,7 @@ use crate::parseable::PARSEABLE;
 use super::otel_utils::convert_epoch_nano_to_timestamp;
 use super::otel_utils::insert_attributes;
 
-pub const OTEL_TRACES_KNOWN_FIELD_LIST: [&str; 31] = [
+pub const OTEL_TRACES_KNOWN_FIELD_LIST: [&str; 30] = [
     "scope_name",
     "scope_version",
     "scope_schema_url",
@@ -43,7 +43,6 @@ pub const OTEL_TRACES_KNOWN_FIELD_LIST: [&str; 31] = [
     "span_span_id",
     "span_name",
     "span_parent_span_id",
-    "flags",
     "name",
     "span_kind",
     "span_kind_description",
@@ -96,7 +95,7 @@ fn flatten_scope_span(scope_span: &ScopeSpans) -> Vec<Map<String, Value>> {
 
     for span_json in &mut vec_scope_span_json {
         span_json.insert(
-            "schema_url".to_string(),
+            "scope_schema_url".to_string(),
             Value::String(scope_span.schema_url.clone()),
         );
     }
@@ -127,7 +126,7 @@ pub fn flatten_otel_traces(message: &TracesData) -> Result<Vec<Value>, OtelError
         }
 
         resource_span_json.insert(
-            "schema_url".to_string(),
+            "resource_schema_url".to_string(),
             Value::String(record.schema_url.clone()),
         );
 
@@ -137,8 +136,8 @@ pub fn flatten_otel_traces(message: &TracesData) -> Result<Vec<Value>, OtelError
             }
 
             let attribute_count = resource_spans_json
-                .keys()
-                .filter(|key| !known_fields.contains(key.as_str()))
+                .iter()
+                .filter(|(key, _)| !known_fields.contains(key.as_str()))
                 .count();
 
             // Check if the number of attributes exceeds the allowed limit