add protobuf support for otel metrics and traces

Author: nikhilsinhaparseable

src/handlers/http/ingest.rs

@@ -18,7 +18,7 @@
 
 use std::collections::{HashMap, HashSet};
 
-use actix_web::web::{Json, Path};
+use actix_web::web::{self, Json, Path};
 use actix_web::{HttpRequest, HttpResponse, http::header::ContentType};
 use arrow_array::RecordBatch;
 use bytes::Bytes;
@@ -33,20 +33,16 @@ use crate::handlers::http::modal::utils::ingest_utils::push_logs;
 use crate::handlers::{EXTRACT_LOG_KEY, LOG_SOURCE_KEY, STREAM_NAME_HEADER_KEY};
 use crate::metadata::SchemaVersion;
 use crate::option::Mode;
-use crate::otel::logs::{flatten_otel_protobuf, OTEL_LOG_KNOWN_FIELD_LIST};
-use crate::otel::metrics::OTEL_METRICS_KNOWN_FIELD_LIST;
-use crate::otel::traces::OTEL_TRACES_KNOWN_FIELD_LIST;
+use crate::otel::logs::{OTEL_LOG_KNOWN_FIELD_LIST, flatten_otel_protobuf};
+use crate::otel::metrics::{OTEL_METRICS_KNOWN_FIELD_LIST, flatten_otel_metrics_protobuf};
+use crate::otel::traces::{OTEL_TRACES_KNOWN_FIELD_LIST, flatten_otel_traces_protobuf};
 use crate::parseable::{PARSEABLE, StreamNotFound};
 use crate::storage::{ObjectStorageError, StreamType};
 use crate::utils::header_parsing::ParseHeaderError;
 use crate::utils::json::{flatten::JsonFlattenError, strict::StrictValue};
-use actix_web::web::{self, Json, Path};
-use actix_web::{http::header::ContentType, HttpRequest, HttpResponse};
-use arrow_array::RecordBatch;
-use bytes::Bytes;
-use chrono::Utc;
-use http::StatusCode;
 use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
+use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
+use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
 use prost::Message;
 
 use super::logstream::error::{CreateStreamError, StreamError};
@@ -229,14 +225,25 @@ pub async fn handle_otel_logs_ingestion(
             }
 
             if content_type == "application/x-protobuf" {
+                const MAX_PROTOBUF_SIZE: usize = 10 * 1024 * 1024; // 10MB limit
+                if body.len() > MAX_PROTOBUF_SIZE {
+                    return Err(PostError::Invalid(anyhow::anyhow!(
+                        "Protobuf message size {} exceeds maximum allowed size of {} bytes",
+                        body.len(),
+                        MAX_PROTOBUF_SIZE
+                    )));
+                }
                 match ExportLogsServiceRequest::decode(body) {
                     Ok(json) => {
                         for record in flatten_otel_protobuf(&json) {
                             push_logs(&stream_name, record, &log_source, &p_custom_fields).await?;
                         }
                     }
                     Err(e) => {
-                        return Err(PostError::Invalid(e.into()));
+                        return Err(PostError::Invalid(anyhow::anyhow!(
+                            "Failed to decode protobuf message: {}",
+                            e
+                        )));
                     }
                 }
             }
@@ -254,7 +261,7 @@ pub async fn handle_otel_logs_ingestion(
 // creates if stream does not exist
 pub async fn handle_otel_metrics_ingestion(
     req: HttpRequest,
-    Json(json): Json<StrictValue>,
+    body: web::Bytes,
 ) -> Result<HttpResponse, PostError> {
     let Some(stream_name) = req.headers().get(STREAM_NAME_HEADER_KEY) else {
         return Err(PostError::Header(ParseHeaderError::MissingStreamName));
@@ -303,13 +310,35 @@ pub async fn handle_otel_metrics_ingestion(
 
     let p_custom_fields = get_custom_fields_from_header(&req);
 
-    flatten_and_push_logs(
-        json.into_inner(),
-        &stream_name,
-        &log_source,
-        &p_custom_fields,
-    )
-    .await?;
+    match req.headers().get("Content-Type") {
+        Some(content_type) => {
+            if content_type == "application/json" {
+                flatten_and_push_logs(
+                    serde_json::from_slice(&body)?,
+                    &stream_name,
+                    &log_source,
+                    &p_custom_fields,
+                )
+                .await?;
+            }
+
+            if content_type == "application/x-protobuf" {
+                match ExportMetricsServiceRequest::decode(body) {
+                    Ok(json) => {
+                        for record in flatten_otel_metrics_protobuf(&json) {
+                            push_logs(&stream_name, record, &log_source, &p_custom_fields).await?;
+                        }
+                    }
+                    Err(e) => {
+                        return Err(PostError::Invalid(e.into()));
+                    }
+                }
+            }
+        }
+        None => {
+            return Err(PostError::Header(ParseHeaderError::InvalidValue));
+        }
+    }
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -319,7 +348,7 @@ pub async fn handle_otel_metrics_ingestion(
 // creates if stream does not exist
 pub async fn handle_otel_traces_ingestion(
     req: HttpRequest,
-    Json(json): Json<StrictValue>,
+    body: web::Bytes,
 ) -> Result<HttpResponse, PostError> {
     let Some(stream_name) = req.headers().get(STREAM_NAME_HEADER_KEY) else {
         return Err(PostError::Header(ParseHeaderError::MissingStreamName));
@@ -369,13 +398,35 @@ pub async fn handle_otel_traces_ingestion(
 
     let p_custom_fields = get_custom_fields_from_header(&req);
 
-    flatten_and_push_logs(
-        json.into_inner(),
-        &stream_name,
-        &log_source,
-        &p_custom_fields,
-    )
-    .await?;
+    match req.headers().get("Content-Type") {
+        Some(content_type) => {
+            if content_type == "application/json" {
+                flatten_and_push_logs(
+                    serde_json::from_slice(&body)?,
+                    &stream_name,
+                    &log_source,
+                    &p_custom_fields,
+                )
+                .await?;
+            }
+
+            if content_type == "application/x-protobuf" {
+                match ExportTraceServiceRequest::decode(body) {
+                    Ok(json) => {
+                        for record in flatten_otel_traces_protobuf(&json) {
+                            push_logs(&stream_name, record, &log_source, &p_custom_fields).await?;
+                        }
+                    }
+                    Err(e) => {
+                        return Err(PostError::Invalid(e.into()));
+                    }
+                }
+            }
+        }
+        None => {
+            return Err(PostError::Header(ParseHeaderError::InvalidValue));
+        }
+    }
 
     Ok(HttpResponse::Ok().finish())
 }

src/otel/metrics.rs

@@ -15,6 +15,7 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  */
+use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
 use opentelemetry_proto::tonic::metrics::v1::number_data_point::Value as NumberDataPointValue;
 use opentelemetry_proto::tonic::metrics::v1::{
     Exemplar, ExponentialHistogram, Gauge, Histogram, Metric, MetricsData, NumberDataPoint, Sum,
@@ -601,3 +602,60 @@ fn flatten_data_point_flags(flags: u32) -> Map<String, Value> {
     );
     data_point_flags_json
 }
+
+/// Flattens OpenTelemetry metrics from protobuf format
+pub fn flatten_otel_metrics_protobuf(message: &ExportMetricsServiceRequest) -> Vec<Value> {
+    let mut vec_otel_json = Vec::new();
+    for resource_metrics in &message.resource_metrics {
+        let mut resource_metrics_json = Map::new();
+        if let Some(resource) = &resource_metrics.resource {
+            insert_attributes(&mut resource_metrics_json, &resource.attributes);
+            resource_metrics_json.insert(
+                "resource_dropped_attributes_count".to_string(),
+                Value::Number(resource.dropped_attributes_count.into()),
+            );
+        }
+
+        let mut vec_resource_metrics_json = Vec::new();
+        for scope_metrics in &resource_metrics.scope_metrics {
+            // scope-level metadata
+            let mut scope_metrics_json = Map::new();
+            if let Some(scope) = &scope_metrics.scope {
+                scope_metrics_json
+                    .insert("scope_name".to_string(), Value::String(scope.name.clone()));
+                scope_metrics_json.insert(
+                    "scope_version".to_string(),
+                    Value::String(scope.version.clone()),
+                );
+                insert_attributes(&mut scope_metrics_json, &scope.attributes);
+                scope_metrics_json.insert(
+                    "scope_dropped_attributes_count".to_string(),
+                    Value::Number(scope.dropped_attributes_count.into()),
+                );
+            }
+            scope_metrics_json.insert(
+                "scope_schema_url".to_string(),
+                Value::String(scope_metrics.schema_url.clone()),
+            );
+
+            for metric in &scope_metrics.metrics {
+                vec_resource_metrics_json.extend(flatten_metrics_record(metric));
+            }
+
+            for resource_metrics_json_item in &mut vec_resource_metrics_json {
+                resource_metrics_json_item.extend(scope_metrics_json.clone());
+            }
+        }
+
+        resource_metrics_json.insert(
+            "resource_schema_url".to_string(),
+            Value::String(resource_metrics.schema_url.clone()),
+        );
+
+        for resource_metrics_json_item in &mut vec_resource_metrics_json {
+            resource_metrics_json_item.extend(resource_metrics_json.clone());
+            vec_otel_json.push(Value::Object(resource_metrics_json_item.clone()));
+        }
+    }
+    vec_otel_json
+}

src/otel/traces.rs

@@ -15,6 +15,7 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  */
+use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
 use opentelemetry_proto::tonic::trace::v1::ScopeSpans;
 use opentelemetry_proto::tonic::trace::v1::Span;
 use opentelemetry_proto::tonic::trace::v1::Status;
@@ -932,3 +933,65 @@ mod tests {
         }
     }
 }
+
+/// Flattens OpenTelemetry traces from protobuf format
+pub fn flatten_otel_traces_protobuf(message: &ExportTraceServiceRequest) -> Vec<Value> {
+    let mut vec_otel_json = Vec::new();
+    for resource_spans in &message.resource_spans {
+        let mut resource_spans_json = Map::new();
+        if let Some(resource) = &resource_spans.resource {
+            insert_attributes(&mut resource_spans_json, &resource.attributes);
+            resource_spans_json.insert(
+                "resource_dropped_attributes_count".to_string(),
+                Value::Number(resource.dropped_attributes_count.into()),
+            );
+        }
+
+        let mut vec_resource_spans_json: Vec<Map<String, Value>> = Vec::new();
+        for scope_spans in &resource_spans.scope_spans {
+            // 1) collect all span JSON under this scope
+            let mut scope_spans_json = Vec::new();
+            for span in &scope_spans.spans {
+                scope_spans_json.extend(flatten_span_record(span));
+            }
+
+            // 2) build a Map of scope‐level fields
+            let mut scope_span_json = Map::new();
+            if let Some(scope) = &scope_spans.scope {
+                scope_span_json.insert("scope_name".to_string(), Value::String(scope.name.clone()));
+                scope_span_json.insert(
+                    "scope_version".to_string(),
+                    Value::String(scope.version.clone()),
+                );
+                insert_attributes(&mut scope_span_json, &scope.attributes);
+                scope_span_json.insert(
+                    "scope_dropped_attributes_count".to_string(),
+                    Value::Number(scope.dropped_attributes_count.into()),
+                );
+            }
+            scope_span_json.insert(
+                "scope_schema_url".to_string(),
+                Value::String(scope_spans.schema_url.clone()),
+            );
+
+            // 3) merge scope fields into each span record
+            for span_json in &mut scope_spans_json {
+                span_json.extend(scope_span_json.clone());
+            }
+
+            // 4) append to the resource‐level accumulator
+            vec_resource_spans_json.extend(scope_spans_json);
+        }
+
+        resource_spans_json.insert(
+            "resource_schema_url".to_string(),
+            Value::String(resource_spans.schema_url.clone()),
+        );
+
+        for resource_spans_json_item in &mut vec_resource_spans_json {
+            resource_spans_json_item.extend(resource_spans_json.clone());
+            vec_otel_json.push(Value::Object(resource_spans_json_item.clone()));
+        }
+    }
+    vec_otel_json
+}