feat: support OpenTelemetry log flattening (#657)

fixes: #312

Author: nikhilsinhaparseable

Cargo.lock

@@ -103,6 +103,8 @@ http-auth-basic = "0.3.3"
 serde_repr = "0.1.17"
 hashlru = { version = "0.11.0", features = ["serde"] }
 path-clean = "1.0.1"
+prost = "0.12.3"
+
 
 [build-dependencies]
 cargo_toml = "0.15"
@@ -112,6 +114,7 @@ ureq = "2.6"
 vergen = { version = "8.1", features = ["build", "git", "cargo", "gitcl"] }
 zip = { version = "0.6", default_features = false, features = ["deflate"] }
 url = "2.4.0"
+prost-build = "0.12.3"
 
 [dev-dependencies]
 maplit = "1.0"

server/Cargo.toml

@@ -45,6 +45,7 @@ mod llm;
 mod logstream;
 mod middleware;
 mod oidc;
+mod otel;
 mod query;
 mod rbac;
 mod role;

server/src/handlers/http.rs

@@ -34,8 +34,8 @@ use crate::handlers::{
 use crate::metadata::STREAM_INFO;
 use crate::utils::header_parsing::{collect_labelled_headers, ParseHeaderError};
 
-use super::kinesis;
 use super::logstream::error::CreateStreamError;
+use super::{kinesis, otel};
 
 // Handler for POST /api/v1/ingest
 // ingests events by extracting stream name from header
@@ -67,7 +67,7 @@ async fn flatten_and_push_logs(
         let log_source: String = log_source.to_str().unwrap().to_owned();
         match log_source.as_str() {
             LOG_SOURCE_KINESIS => json = kinesis::flatten_kinesis_logs(&body),
-            LOG_SOURCE_OTEL => {}
+            LOG_SOURCE_OTEL => json = otel::flatten_otel_logs(&body),
             _ => {
                 log::warn!("Unknown log source: {}", log_source);
                 push_logs(stream_name.to_string(), req.clone(), body).await?;