decode and flatten kinesis log before ingesting to parseable

server/src/handlers.rs

@@ -24,9 +24,14 @@ const PREFIX_META: &str = "x-p-meta-";
 const STREAM_NAME_HEADER_KEY: &str = "x-p-stream";
 const AUTHORIZATION_KEY: &str = "authorization";
 const KINESIS_COMMON_ATTRIBUTES_KEY: &str = "x-amz-firehose-common-attributes";
+const LOG_SOURCE_KEY: &str = "x-p-log-source";
 const SEPARATOR: char = '^';
 
 const OIDC_SCOPE: &str = "openid profile email";
 const COOKIE_AGE_DAYS: usize = 7;
 const SESSION_COOKIE_NAME: &str = "session";
 const USER_COOKIE_NAME: &str = "username";
+
+// constants for Log Source values for known sources
+const LOG_SOURCE_KINESIS: &str = "kinesis";
+const LOG_SOURCE_OTEL: &str = "otel";

server/src/handlers/http.rs

@@ -40,6 +40,7 @@ use self::middleware::{DisAllowRootUser, RouteExt};
 mod about;
 mod health_check;
 mod ingest;
+mod kinesis;
 mod llm;
 mod logstream;
 mod middleware;

server/src/handlers/http/ingest.rs

@@ -16,23 +16,25 @@
  *
  */
 
-use std::collections::HashMap;
-use std::sync::Arc;
-
-use actix_web::http::header::ContentType;
-use actix_web::{HttpRequest, HttpResponse};
+use actix_web::{http::header::ContentType, HttpRequest, HttpResponse};
 use arrow_schema::Field;
 use bytes::Bytes;
 use http::StatusCode;
 use serde_json::Value;
+use std::collections::{BTreeMap, HashMap};
+use std::sync::Arc;
 
 use crate::event::error::EventError;
 use crate::event::format::EventFormat;
 use crate::event::{self, format};
-use crate::handlers::{PREFIX_META, PREFIX_TAGS, SEPARATOR, STREAM_NAME_HEADER_KEY};
+use crate::handlers::{
+    LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, LOG_SOURCE_OTEL, PREFIX_META, PREFIX_TAGS, SEPARATOR,
+    STREAM_NAME_HEADER_KEY,
+};
 use crate::metadata::STREAM_INFO;
 use crate::utils::header_parsing::{collect_labelled_headers, ParseHeaderError};
 
+use super::kinesis;
 use super::logstream::error::CreateStreamError;
 
 // Handler for POST /api/v1/ingest
@@ -46,19 +48,48 @@ pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostE
     {
         let stream_name = stream_name.to_str().unwrap().to_owned();
         create_stream_if_not_exists(&stream_name).await?;
-        push_logs(stream_name, req, body).await?;
+
+        flatten_and_push_logs(req, body, stream_name).await?;
         Ok(HttpResponse::Ok().finish())
     } else {
         Err(PostError::Header(ParseHeaderError::MissingStreamName))
     }
 }
 
+async fn flatten_and_push_logs(
+    req: HttpRequest,
+    body: Bytes,
+    stream_name: String,
+) -> Result<(), PostError> {
+    //flatten logs
+    if let Some((_, log_source)) = req.headers().iter().find(|&(key, _)| key == LOG_SOURCE_KEY) {
+        let mut json: Vec<BTreeMap<String, Value>> = Vec::new();
+        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::warn!("Unknown log source: {}", log_source);
+                push_logs(stream_name.to_string(), req.clone(), body).await?;
+            }
+        }
+        for record in json.iter_mut() {
+            let body: Bytes = serde_json::to_vec(record).unwrap().into();
+            push_logs(stream_name.to_string(), req.clone(), body).await?;
+        }
+    } else {
+        push_logs(stream_name.to_string(), req, body).await?;
+    }
+    Ok(())
+}
+
 // Handler for POST /api/v1/logstream/{logstream}
 // only ingests events into the specified logstream
 // fails if the logstream does not exist
 pub async fn post_event(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
-    push_logs(stream_name, req, body).await?;
+
+    flatten_and_push_logs(req, body, stream_name).await?;
     Ok(HttpResponse::Ok().finish())
 }
 

server/src/handlers/http/kinesis.rs

@@ -0,0 +1,66 @@
+/*
+ * Parseable Server (C) 2022 - 2023 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use base64::{engine::general_purpose::STANDARD, Engine as _};
+use bytes::Bytes;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::collections::BTreeMap;
+use std::str;
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Message {
+    #[serde(rename = "records")]
+    records: Vec<Data>,
+    #[serde(rename = "requestId")]
+    request_id: String,
+    timestamp: u64,
+}
+#[derive(Serialize, Deserialize, Debug)]
+struct Data {
+    #[serde(rename = "data")]
+    data: String,
+}
+
+pub fn flatten_kinesis_logs(body: &Bytes) -> Vec<BTreeMap<String, Value>> {
+    let body_str = std::str::from_utf8(body).unwrap();
+    let message: Message = serde_json::from_str(body_str).unwrap();
+    let mut vec_kinesis_json: Vec<BTreeMap<String, Value>> = Vec::new();
+
+    for record in message.records.iter() {
+        let bytes = STANDARD.decode(record.data.clone()).unwrap();
+        let json_string: String = String::from_utf8(bytes).unwrap();
+        let json: serde_json::Value = serde_json::from_str(&json_string).unwrap();
+        let mut kinesis_json: BTreeMap<String, Value> = match serde_json::from_value(json) {
+            Ok(value) => value,
+            Err(error) => panic!("Failed to deserialize JSON: {}", error),
+        };
+
+        kinesis_json.insert(
+            "requestId".to_owned(),
+            Value::String(message.request_id.clone()),
+        );
+        kinesis_json.insert(
+            "timestamp".to_owned(),
+            Value::String(message.timestamp.to_string()),
+        );
+
+        vec_kinesis_json.push(kinesis_json);
+    }
+    vec_kinesis_json
+}

server/src/handlers/http/middleware.rs

@@ -27,7 +27,10 @@ use actix_web::{
 };
 use futures_util::future::LocalBoxFuture;
 
-use crate::handlers::{AUTHORIZATION_KEY, KINESIS_COMMON_ATTRIBUTES_KEY, STREAM_NAME_HEADER_KEY};
+use crate::handlers::{
+    AUTHORIZATION_KEY, KINESIS_COMMON_ATTRIBUTES_KEY, LOG_SOURCE_KEY, LOG_SOURCE_KINESIS,
+    STREAM_NAME_HEADER_KEY,
+};
 use crate::{
     option::CONFIG,
     rbac::Users,
@@ -149,6 +152,10 @@ where
                 header::HeaderValue::from_str(&message.common_attributes.x_p_stream.clone())
                     .unwrap(),
             );
+            req.headers_mut().insert(
+                HeaderName::from_static(LOG_SOURCE_KEY),
+                header::HeaderValue::from_static(LOG_SOURCE_KINESIS),
+            );
         }
 
         /* ## Section end */