feat: detect and extract json from log-lines

Dockerfile

@@ -30,6 +30,7 @@ RUN mkdir src && echo "fn main() {}" > src/main.rs && cargo build --release && r
 
 # Build the actual binary
 COPY src ./src
+COPY resources ./resources
 RUN cargo build --release
 
 # final stage

Dockerfile.debug

@@ -29,6 +29,7 @@ RUN mkdir src && echo "fn main() {}" > src/main.rs && cargo build && rm -rf src
 
 # Build the actual binary
 COPY src ./src
+COPY resources ./resources
 RUN cargo build
 
 # final stage

Dockerfile.kafka

@@ -42,6 +42,7 @@ RUN mkdir src && echo "fn main() {}" > src/main.rs && \
 
 # Copy the actual source code
 COPY src ./src
+COPY resources ./resources
 
 # Build the actual binary with kafka feature
 RUN cargo build --release --features kafka

src/event/format/known_schema.rs

@@ -0,0 +1,157 @@
+use std::collections::HashMap;
+
+use once_cell::sync::Lazy;
+use regex::Regex;
+use serde::Deserialize;
+use serde_json::{Map, Value};
+use tracing::{error, warn};
+
+const FORMATS_JSON: &str = include_str!("../../../resources/formats.json");
+
+// Schema definition with pattern matching
+pub static KNOWN_SCHEMA_LIST: Lazy<EventProcessor> = Lazy::new(|| {
+    let mut processor = EventProcessor {
+        schema_definitions: HashMap::new(),
+    };
+
+    // Register known schemas
+    processor.register_schema();
+
+    processor
+});
+
+#[derive(Debug)]
+pub struct SchemaDefinition {
+    pattern: Option<Regex>,
+    field_mappings: Vec<String>, // Maps field names to regex capture groups
+}
+
+impl SchemaDefinition {
+    pub fn extract(&self, event: &str) -> Option<Map<String, Value>> {
+        if let Some(pattern) = &self.pattern {
+            if let Some(captures) = pattern.captures(event) {
+                let mut extracted_fields = Map::new();
+
+                // With named capture groups, you can iterate over the field names
+                for field_name in self.field_mappings.iter() {
+                    if let Some(value) = captures.name(field_name) {
+                        extracted_fields.insert(
+                            field_name.to_owned(),
+                            Value::String(value.as_str().to_string()),
+                        );
+                    }
+                }
+
+                return Some(extracted_fields);
+            }
+        }
+
+        None
+    }
+}
+
+#[derive(Debug, Deserialize)]
+struct Format {
+    name: String,
+    regex: Vec<Pattern>,
+}
+#[derive(Debug, Deserialize)]
+struct Pattern {
+    pattern: Option<String>,
+    fields: Vec<String>,
+}
+
+#[derive(Debug)]
+pub struct EventProcessor {
+    pub schema_definitions: HashMap<String, SchemaDefinition>,
+}
+
+impl EventProcessor {
+    fn register_schema(&mut self) {
+        let json_data: serde_json::Value = serde_json::from_str(FORMATS_JSON).unwrap();
+        let formats: Vec<Format> =
+            serde_json::from_value(json_data).expect("Failed to parse formats.json");
+
+        for format in formats {
+            for pattern in &format.regex {
+                if let Some(pattern_str) = &pattern.pattern {
+                    // Compile the regex pattern
+                    match Regex::new(pattern_str) {
+                        Ok(exp) => {
+                            let field_mappings = pattern
+                                .fields
+                                .iter()
+                                .map(|field| field.to_string())
+                                .collect();
+
+                            self.schema_definitions.insert(
+                                format.name.clone(),
+                                SchemaDefinition {
+                                    pattern: Some(exp),
+                                    field_mappings,
+                                },
+                            );
+                        }
+                        Err(e) => {
+                            error!("Error compiling regex pattern: {e}; Pattern: {pattern_str}");
+                        }
+                    }
+                } else {
+                    let field_mappings = pattern
+                        .fields
+                        .iter()
+                        .map(|field| field.to_string())
+                        .collect();
+
+                    self.schema_definitions.insert(
+                        format.name.clone(),
+                        SchemaDefinition {
+                            pattern: None,
+                            field_mappings,
+                        },
+                    );
+                }
+            }
+        }
+    }
+}
+pub fn extract_from_inline_log(
+    mut json: Value,
+    log_source: &str,
+    extract_log: Option<&str>,
+) -> Value {
+    let Some(schema) = KNOWN_SCHEMA_LIST.schema_definitions.get(log_source) else {
+        warn!("Unknown log format: {log_source}");
+        return json;
+    };
+
+    match &mut json {
+        Value::Array(list) => {
+            for event in list {
+                let Value::Object(event) = event else {
+                    continue;
+                };
+                per_event_extraction(event, schema, extract_log)
+            }
+        }
+        Value::Object(event) => per_event_extraction(event, schema, extract_log),
+        _ => unreachable!("We don't accept events of the form: {json}"),
+    }
+
+    json
+}
+
+pub fn per_event_extraction(
+    obj: &mut Map<String, Value>,
+    schema: &SchemaDefinition,
+    extract_log: Option<&str>,
+) {
+    if let Some(event) = extract_log
+        .and_then(|field| obj.get(field))
+        .and_then(|s| s.as_str())
+    {
+        if let Some(additional) = schema.extract(event) {
+            obj.extend(additional);
+        }
+    }
+}

src/event/format/mod.rs

@@ -39,6 +39,7 @@ use crate::{
 use super::{Event, DEFAULT_TIMESTAMP_KEY};
 
 pub mod json;
+pub mod known_schema;
 
 static TIME_FIELD_NAME_PARTS: [&str; 2] = ["time", "date"];
 type EventSchema = Vec<Arc<Field>>;

src/handlers/http/ingest.rs

@@ -29,7 +29,7 @@ use serde_json::Value;
 use crate::event;
 use crate::event::error::EventError;
 use crate::event::format::{self, EventFormat, LogSource, LogSourceEntry};
-use crate::handlers::{LOG_SOURCE_KEY, STREAM_NAME_HEADER_KEY};
+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::OTEL_LOG_KNOWN_FIELD_LIST;
@@ -65,6 +65,11 @@ pub async fn ingest(req: HttpRequest, Json(json): Json<Value>) -> Result<HttpRes
         .and_then(|h| h.to_str().ok())
         .map_or(LogSource::default(), LogSource::from);
 
+    let extract_log = req
+        .headers()
+        .get(EXTRACT_LOG_KEY)
+        .and_then(|h| h.to_str().ok());
+
     if matches!(
         log_source,
         LogSource::OtelLogs | LogSource::OtelMetrics | LogSource::OtelTraces
@@ -81,7 +86,7 @@ pub async fn ingest(req: HttpRequest, Json(json): Json<Value>) -> Result<HttpRes
         )
         .await?;
 
-    flatten_and_push_logs(json, &stream_name, &log_source).await?;
+    flatten_and_push_logs(json, &stream_name, &log_source, extract_log).await?;
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -144,7 +149,7 @@ pub async fn handle_otel_logs_ingestion(
         )
         .await?;
 
-    flatten_and_push_logs(json, &stream_name, &log_source).await?;
+    flatten_and_push_logs(json, &stream_name, &log_source, None).await?;
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -182,7 +187,7 @@ pub async fn handle_otel_metrics_ingestion(
         )
         .await?;
 
-    flatten_and_push_logs(json, &stream_name, &log_source).await?;
+    flatten_and_push_logs(json, &stream_name, &log_source, None).await?;
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -222,7 +227,7 @@ pub async fn handle_otel_traces_ingestion(
         )
         .await?;
 
-    flatten_and_push_logs(json, &stream_name, &log_source).await?;
+    flatten_and_push_logs(json, &stream_name, &log_source, None).await?;
 
     Ok(HttpResponse::Ok().finish())
 }
@@ -264,14 +269,19 @@ pub async fn post_event(
         .and_then(|h| h.to_str().ok())
         .map_or(LogSource::default(), LogSource::from);
 
+    let extract_log = req
+        .headers()
+        .get(EXTRACT_LOG_KEY)
+        .and_then(|h| h.to_str().ok());
+
     if matches!(
         log_source,
         LogSource::OtelLogs | LogSource::OtelMetrics | LogSource::OtelTraces
     ) {
         return Err(PostError::OtelNotSupported);
     }
 
-    flatten_and_push_logs(json, &stream_name, &log_source).await?;
+    flatten_and_push_logs(json, &stream_name, &log_source, extract_log).await?;
 
     Ok(HttpResponse::Ok().finish())
 }

src/handlers/http/modal/utils/ingest_utils.rs

@@ -23,7 +23,7 @@ use opentelemetry_proto::tonic::{
 use serde_json::Value;
 
 use crate::{
-    event::format::{json, EventFormat, LogSource},
+    event::format::{json, known_schema::extract_from_inline_log, EventFormat, LogSource},
     handlers::http::{
         ingest::PostError,
         kinesis::{flatten_kinesis_logs, Message},
@@ -38,6 +38,7 @@ pub async fn flatten_and_push_logs(
     json: Value,
     stream_name: &str,
     log_source: &LogSource,
+    extract_log: Option<&str>,
 ) -> Result<(), PostError> {
     match log_source {
         LogSource::Kinesis => {
@@ -68,10 +69,13 @@ pub async fn flatten_and_push_logs(
                 push_logs(stream_name, record, log_source).await?;
             }
         }
-        _ => {
+        LogSource::Custom(src) => {
+            let json = extract_from_inline_log(json, src, extract_log);
             push_logs(stream_name, json, log_source).await?;
         }
+        _ => push_logs(stream_name, json, log_source).await?,
     }
+
     Ok(())
 }
 

src/handlers/mod.rs

@@ -22,6 +22,7 @@ pub mod livetail;
 
 pub const STREAM_NAME_HEADER_KEY: &str = "x-p-stream";
 const LOG_SOURCE_KEY: &str = "x-p-log-source";
+const EXTRACT_LOG_KEY: &str = "x-p-extract-log";
 const TIME_PARTITION_KEY: &str = "x-p-time-partition";
 const TIME_PARTITION_LIMIT_KEY: &str = "x-p-time-partition-limit";
 const CUSTOM_PARTITION_KEY: &str = "x-p-custom-partition";