feat: add custom field p_format_verified (#1303)

nikhilsinhaparseable · web-flow · commit a90ed13b322e · 2025-04-26T22:40:57.000+05:30
if server successfully verifies the format specified in the `p_format` header, 
it adds `p_format_verified=true` to the event. Otherwise, if schema detection 
fails, add `p_format_verified=false` to the event
diff --git a/resources/formats.json b/resources/formats.json
@@ -72,7 +72,7 @@
     "name": "alb_log",
     "regex": [
       {
-        "pattern": "^(?<type>http|https|h2|ws|wss) (?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}Z) (?<elb>[^ ]+) (?<client_ip>[\\w\\.:]+):(?<client_port>\\d+) (?<target_ip>[\\w\\.:]+):(?<target_port>\\d+) (?<request_processing_time>[-\\d\\.]+) (?<target_processing_time>[-\\d\\.]+) (?<response_processing_time>[-\\d\\.]+) (?<elb_status_code>\\d+|-) (?<target_status_code>\\d+|-) (?<received_bytes>\\d+) (?<sent_bytes>\\d+) (?<cs_method>POST|GET|PUT|DELETE|HEAD|OPTIONS|CONNECT|TRACE|PATCH) (?<cs_uri_whole>[^ ]+) (?<cs_version>[^ ]+) (?<user_agent>[^ \\(]+) (?:\\([^\\)]+\\))? (?<ssl_cipher>[\\w-]+) (?<ssl_protocol>[\\w\\.-]+) (?<target_group_arn>[^ ]+) (?<trace_id>[^ ]+) (?<domain_name>[^ ]+) (?<chosen_cert_arn>[^ ]+) (?<action_executed>\\d+) (?<request_creation_time>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}Z) (?<redirect_url>[^ ]+) (?<redirect_proto>[^ ]+) (?<redirect_port>[^ ]+) (?<target_ip_port>[\\d\\.:]+) (?<target_status_desc>\\d+|-) (?<classification>[^ ]+) (?<classification_reason>[^ ]+) (?<track_id>[^ ]+)$",
+        "pattern": "^(?<type>http|https|h2|ws|wss) (?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}Z) (?<elb>[^ ]+) (?<client_ip>[\\w\\.:]+):(?<client_port>\\d+) (?<target_ip>[\\w\\.:]+):(?<target_port>\\d+) (?<request_processing_time>[-\\d\\.]+) (?<target_processing_time>[-\\d\\.]+) (?<response_processing_time>[-\\d\\.]+) (?<elb_status_code>\\d+|-) (?<target_status_code>\\d+|-) (?<received_bytes>\\d+) (?<sent_bytes>\\d+) (?<cs_method>POST|GET|PUT|DELETE|HEAD|OPTIONS|CONNECT|TRACE|PATCH) (?<cs_uri_whole>[^ ]+) (?<cs_version>[^ ]+) (?<user_agent>.+?) (?<ssl_cipher>[^ ]+) (?<ssl_protocol>[^ ]+) (?<target_group_arn>[^ ]+) (?<trace_id>[^ ]+) (?<domain_name>[^ ]+) (?<chosen_cert_arn>[^ ]+) (?<action_executed>\\d+) (?<request_creation_time>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}Z) (?<redirect_url>[^ ]+) (?<redirect_proto>[^ ]+) (?<redirect_port>[^ ]+) (?<target_ip_port>[\\d\\.:]+) (?<target_status_desc>\\d+|-) (?<classification>[^ ]+) (?<classification_reason>[^ ]+) (?<track_id>[^ ]+)$",
         "fields": [
           "type",
           "timestamp",
@@ -109,6 +109,54 @@
           "classification_reason",
           "track_id"
         ]
+      },
+      {
+        "pattern": "^(?<type>http|https|h2|ws|wss) (?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}Z) (?<elb>[^ ]+) (?<client_ip>[^:]+):(?<client_port>\\d+) (?<target_ip>[^ ]+) (?<request_processing_time>[^ ]+) (?<target_processing_time>[^ ]+) (?<response_processing_time>[^ ]+) (?<elb_status_code>[^ ]+) (?<target_status_code>[^ ]+) (?<received_bytes>[^ ]+) (?<sent_bytes>[^ ]+) (?<cs_method>[^ ]+) (?<cs_uri_whole>[^ ]+) (?<cs_version>[^ ]+) (?<user_agent>.*?) (?<ssl_cipher>[^ ]+) (?<ssl_protocol>[^ ]+) (?<target_group_arn>[^ ]+) (?<trace_id>[^ ]+) (?<domain_name>[^ ]+) (?<chosen_cert_arn>[^ ]+) (?<action_executed>[^ ]+) (?<request_creation_time>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}Z|[^ ]+) (?<redirect_url>[^ ]+) (?<redirect_proto>[^ ]+) (?<redirect_port>[^ ]+) (?<target_ip_port>[^ ]+|[^ ]*) (?<target_status_desc>[^ ]+|[^ ]*) (?<classification>[^ ]+|[^ ]*) (?<classification_reason>[^ ]+|[^ ]*) (?<track_id>TID_[a-f0-9]+)$",
+        "fields": [
+          "type",
+          "timestamp",
+          "elb",
+          "client_ip",
+          "client_port",
+          "target_ip",
+          "request_processing_time",
+          "target_processing_time",
+          "response_processing_time",
+          "elb_status_code",
+          "target_status_code",
+          "received_bytes",
+          "sent_bytes",
+          "cs_method",
+          "cs_uri_whole",
+          "cs_version",
+          "user_agent",
+          "ssl_cipher",
+          "ssl_protocol",
+          "target_group_arn",
+          "trace_id",
+          "domain_name",
+          "chosen_cert_arn",
+          "action_executed",
+          "request_creation_time",
+          "redirect_url",
+          "redirect_proto",
+          "redirect_port",
+          "target_ip_port",
+          "target_status_desc",
+          "classification",
+          "classification_reason",
+          "track_id"
+        ]
+      },
+      {
+        "pattern": "^(?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}Z) (?<client_ip>[\\d\\.]+) (?<client_port>\\d+) (?<target_port>\\d+) (- ){7}(?<track_id>TID_[a-f0-9]+)$",
+        "fields": [
+          "timestamp",
+          "client_ip",
+          "client_port",
+          "target_port",
+          "track_id"
+        ]
       }
     ]
   },
diff --git a/src/event/format/known_schema.rs b/src/event/format/known_schema.rs
@@ -24,6 +24,8 @@ use serde::{Deserialize, Deserializer};
 use serde_json::{Map, Value};
 use tracing::error;
 
+use crate::event::FORMAT_VERIFY_KEY;
+
 /// Predefined JSON with known textual logging formats
 const FORMATS_JSON: &str = include_str!("../../../resources/formats.json");
 
@@ -120,11 +122,21 @@ impl SchemaDefinition {
                 }
             }
 
+            // add `P_FORMAT_VERIFY_KEY` to the object
+            obj.insert(
+                FORMAT_VERIFY_KEY.to_string(),
+                Value::String("true".to_string()),
+            );
+
             obj.extend(extracted_fields);
 
             return Some(format.fields.clone());
         }
-
+        // add `P_FORMAT_VERIFY_KEY` to the object
+        obj.insert(
+            FORMAT_VERIFY_KEY.to_string(),
+            Value::String("false".to_string()),
+        );
         None
     }
 }
@@ -180,6 +192,7 @@ impl EventProcessor {
     pub fn extract_from_inline_log(
         &self,
         json: &mut Value,
+        p_custom_fields: &mut HashMap<String, String>,
         log_source: &str,
         extract_log: Option<&str>,
     ) -> Result<HashSet<String>, Error> {
@@ -197,15 +210,17 @@ impl EventProcessor {
                     if let Some(known_fields) = schema.check_or_extract(event, extract_log) {
                         fields.extend(known_fields);
                     } else {
-                        return Err(Error::Unacceptable(log_source.to_owned()));
+                        // add `P_FORMAT_VERIFY_KEY` to the object
+                        p_custom_fields.insert(FORMAT_VERIFY_KEY.to_string(), "false".to_string());
                     }
                 }
             }
             Value::Object(event) => {
                 if let Some(known_fields) = schema.check_or_extract(event, extract_log) {
                     return Ok(known_fields);
                 } else {
-                    return Err(Error::Unacceptable(log_source.to_owned()));
+                    // add `P_FORMAT_VERIFY_KEY` to the object
+                    p_custom_fields.insert(FORMAT_VERIFY_KEY.to_string(), "false".to_string());
                 }
             }
             _ => unreachable!("We don't accept events of the form: {json}"),
diff --git a/src/event/mod.rs b/src/event/mod.rs
@@ -38,6 +38,7 @@ pub const DEFAULT_TIMESTAMP_KEY: &str = "p_timestamp";
 pub const USER_AGENT_KEY: &str = "p_user_agent";
 pub const SOURCE_IP_KEY: &str = "p_src_ip";
 pub const FORMAT_KEY: &str = "p_format";
+pub const FORMAT_VERIFY_KEY: &str = "p_format_verified";
 
 #[derive(Clone)]
 pub struct Event {
diff --git a/src/handlers/http/ingest.rs b/src/handlers/http/ingest.rs
@@ -81,10 +81,15 @@ pub async fn ingest(
         return Err(PostError::OtelNotSupported);
     }
 
+    let mut p_custom_fields = get_custom_fields_from_header(&req);
+
     let fields = match &log_source {
-        LogSource::Custom(src) => {
-            KNOWN_SCHEMA_LIST.extract_from_inline_log(&mut json, src, extract_log)?
-        }
+        LogSource::Custom(src) => KNOWN_SCHEMA_LIST.extract_from_inline_log(
+            &mut json,
+            &mut p_custom_fields,
+            src,
+            extract_log,
+        )?,
         _ => HashSet::new(),
     };
 
@@ -114,7 +119,7 @@ pub async fn ingest(
     PARSEABLE
         .add_update_log_source(&stream_name, log_source_entry)
         .await?;
-    let p_custom_fields = get_custom_fields_from_header(req);
+
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields).await?;
 
     Ok(HttpResponse::Ok().finish())
@@ -198,7 +203,7 @@ pub async fn handle_otel_logs_ingestion(
         .add_update_log_source(&stream_name, log_source_entry)
         .await?;
 
-    let p_custom_fields = get_custom_fields_from_header(req);
+    let p_custom_fields = get_custom_fields_from_header(&req);
 
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields).await?;
 
@@ -256,7 +261,7 @@ pub async fn handle_otel_metrics_ingestion(
         .add_update_log_source(&stream_name, log_source_entry)
         .await?;
 
-    let p_custom_fields = get_custom_fields_from_header(req);
+    let p_custom_fields = get_custom_fields_from_header(&req);
 
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields).await?;
 
@@ -315,7 +320,7 @@ pub async fn handle_otel_traces_ingestion(
         .add_update_log_source(&stream_name, log_source_entry)
         .await?;
 
-    let p_custom_fields = get_custom_fields_from_header(req);
+    let p_custom_fields = get_custom_fields_from_header(&req);
 
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields).await?;
 
@@ -363,13 +368,18 @@ pub async fn post_event(
         .headers()
         .get(EXTRACT_LOG_KEY)
         .and_then(|h| h.to_str().ok());
-
+    let mut p_custom_fields = get_custom_fields_from_header(&req);
     match &log_source {
         LogSource::OtelLogs | LogSource::OtelMetrics | LogSource::OtelTraces => {
             return Err(PostError::OtelNotSupported)
         }
         LogSource::Custom(src) => {
-            KNOWN_SCHEMA_LIST.extract_from_inline_log(&mut json, src, extract_log)?;
+            KNOWN_SCHEMA_LIST.extract_from_inline_log(
+                &mut json,
+                &mut p_custom_fields,
+                src,
+                extract_log,
+            )?;
         }
         _ => {}
     }
@@ -386,7 +396,7 @@ pub async fn post_event(
             })
             .ok_or(PostError::IncorrectLogFormat(stream_name.clone()))?;
     }
-    let p_custom_fields = get_custom_fields_from_header(req);
+
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields).await?;
 
     Ok(HttpResponse::Ok().finish())
diff --git a/src/handlers/http/modal/utils/ingest_utils.rs b/src/handlers/http/modal/utils/ingest_utils.rs
@@ -145,7 +145,7 @@ async fn push_logs(
     Ok(())
 }
 
-pub fn get_custom_fields_from_header(req: HttpRequest) -> HashMap<String, String> {
+pub fn get_custom_fields_from_header(req: &HttpRequest) -> HashMap<String, String> {
     let user_agent = req
         .headers()
         .get(USER_AGENT)
@@ -217,7 +217,7 @@ mod tests {
             .insert_header(("x-p-environment", "dev"))
             .to_http_request();
 
-        let custom_fields = get_custom_fields_from_header(req);
+        let custom_fields = get_custom_fields_from_header(&req);
 
         assert_eq!(custom_fields.get(USER_AGENT_KEY).unwrap(), "TestUserAgent");
         assert_eq!(custom_fields.get("environment").unwrap(), "dev");
@@ -230,7 +230,7 @@ mod tests {
             .insert_header((STREAM_NAME_HEADER_KEY, "teststream"))
             .to_http_request();
 
-        let custom_fields = get_custom_fields_from_header(req);
+        let custom_fields = get_custom_fields_from_header(&req);
 
         assert_eq!(custom_fields.get(USER_AGENT_KEY).unwrap(), "TestUserAgent");
         assert!(!custom_fields.contains_key(STREAM_NAME_HEADER_KEY));
@@ -243,7 +243,7 @@ mod tests {
             .insert_header((LOG_SOURCE_KEY, "otel-logs"))
             .to_http_request();
 
-        let custom_fields = get_custom_fields_from_header(req);
+        let custom_fields = get_custom_fields_from_header(&req);
 
         assert_eq!(custom_fields.get(USER_AGENT_KEY).unwrap(), "TestUserAgent");
         assert_eq!(custom_fields.get(FORMAT_KEY).unwrap(), "otel-logs");
@@ -255,7 +255,7 @@ mod tests {
             .insert_header(("x-p-", "empty"))
             .to_http_request();
 
-        let custom_fields = get_custom_fields_from_header(req);
+        let custom_fields = get_custom_fields_from_header(&req);
 
         assert_eq!(custom_fields.len(), 2);
         assert_eq!(custom_fields.get(USER_AGENT_KEY).unwrap(), "");

Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,8 @@ use serde::{Deserialize, Deserializer};`
`24`	`24`	`use serde_json::{Map, Value};`
`25`	`25`	`use tracing::error;`
`26`	`26`
	`27`	`+use crate::event::FORMAT_VERIFY_KEY;`
	`28`	`+`
`27`	`29`	`/// Predefined JSON with known textual logging formats`
`28`	`30`	`const FORMATS_JSON: &str = include_str!("../../../resources/formats.json");`
`29`	`31`
`@@ -120,11 +122,21 @@ impl SchemaDefinition {`
`120`	`122`	`}`
`121`	`123`	`}`
`122`	`124`
	`125`	+ // add `P_FORMAT_VERIFY_KEY` to the object
	`126`	`+ obj.insert(`
	`127`	`+ FORMAT_VERIFY_KEY.to_string(),`
	`128`	`+ Value::String("true".to_string()),`
	`129`	`+ );`
	`130`	`+`
`123`	`131`	`obj.extend(extracted_fields);`
`124`	`132`
`125`	`133`	`return Some(format.fields.clone());`
`126`	`134`	`}`
`127`		`-`
	`135`	+ // add `P_FORMAT_VERIFY_KEY` to the object
	`136`	`+ obj.insert(`
	`137`	`+ FORMAT_VERIFY_KEY.to_string(),`
	`138`	`+ Value::String("false".to_string()),`
	`139`	`+ );`
`128`	`140`	`None`
`129`	`141`	`}`
`130`	`142`	`}`
`@@ -180,6 +192,7 @@ impl EventProcessor {`
`180`	`192`	`pub fn extract_from_inline_log(`
`181`	`193`	`&self,`
`182`	`194`	`json: &mut Value,`
	`195`	`+ p_custom_fields: &mut HashMap<String, String>,`
`183`	`196`	`log_source: &str,`
`184`	`197`	`extract_log: Option<&str>,`
`185`	`198`	`) -> Result<HashSet<String>, Error> {`
`@@ -197,15 +210,17 @@ impl EventProcessor {`
`197`	`210`	`if let Some(known_fields) = schema.check_or_extract(event, extract_log) {`
`198`	`211`	`fields.extend(known_fields);`
`199`	`212`	`} else {`
`200`		`- return Err(Error::Unacceptable(log_source.to_owned()));`
	`213`	+ // add `P_FORMAT_VERIFY_KEY` to the object
	`214`	`+ p_custom_fields.insert(FORMAT_VERIFY_KEY.to_string(), "false".to_string());`
`201`	`215`	`}`
`202`	`216`	`}`
`203`	`217`	`}`
`204`	`218`	`Value::Object(event) => {`
`205`	`219`	`if let Some(known_fields) = schema.check_or_extract(event, extract_log) {`
`206`	`220`	`return Ok(known_fields);`
`207`	`221`	`} else {`
`208`		`- return Err(Error::Unacceptable(log_source.to_owned()));`
	`222`	+ // add `P_FORMAT_VERIFY_KEY` to the object
	`223`	`+ p_custom_fields.insert(FORMAT_VERIFY_KEY.to_string(), "false".to_string());`
`209`	`224`	`}`
`210`	`225`	`}`
`211`	`226`	`_ => unreachable!("We don't accept events of the form: {json}"),`