Merge remote-tracking branch 'origin/main' into refactor

Author: Devdutt Shenoi

src/handlers/http/ingest.rs

@@ -24,9 +24,6 @@ use arrow_array::RecordBatch;
 use bytes::Bytes;
 use chrono::Utc;
 use http::StatusCode;
-use opentelemetry_proto::tonic::logs::v1::LogsData;
-use opentelemetry_proto::tonic::metrics::v1::MetricsData;
-use opentelemetry_proto::tonic::trace::v1::TracesData;
 use serde_json::Value;
 
 use crate::event;
@@ -35,9 +32,6 @@ use crate::event::format::{json, EventFormat, LogSource};
 use crate::handlers::{LOG_SOURCE_KEY, STREAM_NAME_HEADER_KEY};
 use crate::metadata::SchemaVersion;
 use crate::option::Mode;
-use crate::otel::logs::flatten_otel_logs;
-use crate::otel::metrics::flatten_otel_metrics;
-use crate::otel::traces::flatten_otel_traces;
 use crate::parseable::{StreamNotFound, PARSEABLE};
 use crate::storage::{ObjectStorageError, StreamType};
 use crate::utils::header_parsing::ParseHeaderError;
@@ -69,6 +63,14 @@ pub async fn ingest(req: HttpRequest, Json(json): Json<Value>) -> Result<HttpRes
         .get(LOG_SOURCE_KEY)
         .and_then(|h| h.to_str().ok())
         .map_or(LogSource::default(), LogSource::from);
+
+    if matches!(
+        log_source,
+        LogSource::OtelLogs | LogSource::OtelMetrics | LogSource::OtelTraces
+    ) {
+        return Err(PostError::OtelNotSupported);
+    }
+
     PARSEABLE
         .get_stream(&stream_name)?
         .flatten_and_push_logs(json, &log_source)
@@ -125,13 +127,11 @@ pub async fn handle_otel_logs_ingestion(
     PARSEABLE
         .create_stream_if_not_exists(&stream_name, StreamType::UserDefined, LogSource::OtelLogs)
         .await?;
-    let stream = PARSEABLE.get_stream(&stream_name)?;
 
-    //custom flattening required for otel logs
-    let logs: LogsData = serde_json::from_value(json)?;
-    for record in flatten_otel_logs(&logs) {
-        stream.push_logs(record, &log_source).await?;
-    }
+    PARSEABLE
+        .get_stream(&stream_name)?
+        .flatten_and_push_logs(json, &log_source)
+        .await?;
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -161,13 +161,11 @@ pub async fn handle_otel_metrics_ingestion(
             LogSource::OtelMetrics,
         )
         .await?;
-    let stream = PARSEABLE.get_stream(&stream_name)?;
 
-    //custom flattening required for otel metrics
-    let metrics: MetricsData = serde_json::from_value(json)?;
-    for record in flatten_otel_metrics(metrics) {
-        stream.push_logs(record, &log_source).await?;
-    }
+    PARSEABLE
+        .get_stream(&stream_name)?
+        .flatten_and_push_logs(json, &log_source)
+        .await?;
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -194,13 +192,11 @@ pub async fn handle_otel_traces_ingestion(
     PARSEABLE
         .create_stream_if_not_exists(&stream_name, StreamType::UserDefined, LogSource::OtelTraces)
         .await?;
-    let stream = PARSEABLE.get_stream(&stream_name)?;
 
-    //custom flattening required for otel traces
-    let traces: TracesData = serde_json::from_value(json)?;
-    for record in flatten_otel_traces(&traces) {
-        stream.push_logs(record, &log_source).await?;
-    }
+    PARSEABLE
+        .get_stream(&stream_name)?
+        .flatten_and_push_logs(json, &log_source)
+        .await?;
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -241,6 +237,14 @@ pub async fn post_event(
         .get(LOG_SOURCE_KEY)
         .and_then(|h| h.to_str().ok())
         .map_or(LogSource::default(), LogSource::from);
+
+    if matches!(
+        log_source,
+        LogSource::OtelLogs | LogSource::OtelMetrics | LogSource::OtelTraces
+    ) {
+        return Err(PostError::OtelNotSupported);
+    }
+
     PARSEABLE
         .get_stream(&stream_name)?
         .flatten_and_push_logs(json, &log_source)

src/lib.rs

@@ -17,7 +17,7 @@
  */
 
 pub mod about;
-mod alerts;
+pub mod alerts;
 pub mod analytics;
 pub mod audit;
 pub mod banner;

src/parseable/streams.rs

@@ -32,6 +32,9 @@ use arrow_schema::{ArrowError, Field, Fields, Schema};
 use chrono::{DateTime, NaiveDateTime, Timelike, Utc};
 use derive_more::{Deref, DerefMut};
 use itertools::Itertools;
+use opentelemetry_proto::tonic::{
+    logs::v1::LogsData, metrics::v1::MetricsData, trace::v1::TracesData,
+};
 use parquet::{
     arrow::ArrowWriter,
     basic::Encoding,
@@ -55,6 +58,7 @@ use crate::{
     metadata::{LogStreamMetadata, SchemaVersion},
     metrics,
     option::Mode,
+    otel::{logs::flatten_otel_logs, metrics::flatten_otel_metrics, traces::flatten_otel_traces},
     storage::{
         object_storage::to_bytes, retention::Retention, StreamType, OBJECT_STORE_DATA_GRANULARITY,
     },
@@ -119,14 +123,32 @@ impl Stream {
     ) -> Result<(), PostError> {
         match log_source {
             LogSource::Kinesis => {
+                //custom flattening required for Amazon Kinesis
                 let message: Message = serde_json::from_value(json)?;
-                let json = flatten_kinesis_logs(message);
-                for record in json {
+                for record in flatten_kinesis_logs(message) {
                     self.push_logs(record, &LogSource::default()).await?;
                 }
             }
-            LogSource::OtelLogs | LogSource::OtelMetrics | LogSource::OtelTraces => {
-                return Err(PostError::OtelNotSupported);
+            LogSource::OtelLogs => {
+                //custom flattening required for otel logs
+                let logs: LogsData = serde_json::from_value(json)?;
+                for record in flatten_otel_logs(&logs) {
+                    self.push_logs(record, log_source).await?;
+                }
+            }
+            LogSource::OtelTraces => {
+                //custom flattening required for otel traces
+                let traces: TracesData = serde_json::from_value(json)?;
+                for record in flatten_otel_traces(&traces) {
+                    self.push_logs(record, log_source).await?;
+                }
+            }
+            LogSource::OtelMetrics => {
+                //custom flattening required for otel metrics
+                let metrics: MetricsData = serde_json::from_value(json)?;
+                for record in flatten_otel_metrics(metrics) {
+                    self.push_logs(record, log_source).await?;
+                }
             }
             _ => {
                 self.push_logs(json, log_source).await?;

src/utils/json/flatten.rs

@@ -335,18 +335,11 @@ pub fn has_more_than_four_levels(value: &Value, current_level: usize) -> bool {
 
 // Converts a Vector of values into a `Value::Array`, as long as all of them are objects
 pub fn convert_to_array(flattened: Vec<Value>) -> Result<Value, JsonFlattenError> {
-    let mut result = Vec::new();
-    for item in flattened {
-        let mut map = Map::new();
-        let Some(item) = item.as_object() else {
-            return Err(JsonFlattenError::ExpectedObjectInArray);
-        };
-        for (key, value) in item {
-            map.insert(key.clone(), value.clone());
-        }
-        result.push(Value::Object(map));
+    if flattened.iter().any(|item| !item.is_object()) {
+        return Err(JsonFlattenError::ExpectedObjectInArray);
     }
-    Ok(Value::Array(result))
+
+    Ok(Value::Array(flattened))
 }
 
 #[cfg(test)]