fix: don't accept unknown log format

Author: Devdutt Shenoi

src/event/format/known_schema.rs

@@ -22,7 +22,7 @@ use once_cell::sync::Lazy;
 use regex::Regex;
 use serde::{Deserialize, Deserializer};
 use serde_json::{Map, Value};
-use tracing::{error, warn};
+use tracing::error;
 
 /// Predefined JSON with known textual logging formats
 const FORMATS_JSON: &str = include_str!("../../../resources/formats.json");
@@ -32,8 +32,12 @@ pub static KNOWN_SCHEMA_LIST: Lazy<EventProcessor> =
     Lazy::new(|| EventProcessor::new(FORMATS_JSON));
 
 #[derive(Debug, thiserror::Error)]
-#[error("Event is not in the expected text/JSON format for {0}")]
-pub struct Unacceptable(String);
+pub enum Error {
+    #[error("Event is not in the expected text/JSON format for {0}")]
+    Unacceptable(String),
+    #[error("Unknown log format: {0}")]
+    Unknown(String),
+}
 
 /// Deserializes a string pattern into a compiled Regex
 /// NOTE: we only warn if the pattern doesn't compile
@@ -178,10 +182,9 @@ impl EventProcessor {
         json: &mut Value,
         log_source: &str,
         extract_log: Option<&str>,
-    ) -> Result<HashSet<String>, Unacceptable> {
+    ) -> Result<HashSet<String>, Error> {
         let Some(schema) = self.schema_definitions.get(log_source) else {
-            warn!("Unknown log format: {log_source}");
-            return Ok(HashSet::new());
+            return Err(Error::Unknown(log_source.to_owned()));
         };
 
         let mut fields = HashSet::new();
@@ -194,15 +197,15 @@ impl EventProcessor {
                     if let Some(known_fields) = schema.check_or_extract(event, extract_log) {
                         fields.extend(known_fields);
                     } else {
-                        return Err(Unacceptable(log_source.to_owned()));
+                        return Err(Error::Unacceptable(log_source.to_owned()));
                     }
                 }
             }
             Value::Object(event) => {
                 if let Some(known_fields) = schema.check_or_extract(event, extract_log) {
                     return Ok(known_fields);
                 } else {
-                    return Err(Unacceptable(log_source.to_owned()));
+                    return Err(Error::Unacceptable(log_source.to_owned()));
                 }
             }
             _ => unreachable!("We don't accept events of the form: {json}"),

src/handlers/http/ingest.rs

@@ -28,7 +28,7 @@ use serde_json::Value;
 
 use crate::event;
 use crate::event::error::EventError;
-use crate::event::format::known_schema::{Unacceptable, KNOWN_SCHEMA_LIST};
+use crate::event::format::known_schema::{self, KNOWN_SCHEMA_LIST};
 use crate::event::format::{self, EventFormat, LogSource, LogSourceEntry};
 use crate::handlers::{EXTRACT_LOG_KEY, LOG_SOURCE_KEY, STREAM_NAME_HEADER_KEY};
 use crate::metadata::SchemaVersion;
@@ -49,7 +49,10 @@ use super::users::filters::FiltersError;
 // Handler for POST /api/v1/ingest
 // ingests events by extracting stream name from header
 // creates if stream does not exist
-pub async fn ingest(req: HttpRequest, Json(mut json): Json<Value>) -> Result<HttpResponse, PostError> {
+pub async fn ingest(
+    req: HttpRequest,
+    Json(mut json): Json<Value>,
+) -> Result<HttpResponse, PostError> {
     let Some(stream_name) = req.headers().get(STREAM_NAME_HEADER_KEY) else {
         return Err(PostError::Header(ParseHeaderError::MissingStreamName));
     };
@@ -363,7 +366,7 @@ pub enum PostError {
     #[error("Missing field for time partition in json: {0}")]
     MissingTimePartition(String),
     #[error("{0}")]
-    KnownFormat(#[from] Unacceptable),
+    KnownFormat(#[from] known_schema::Error),
 }
 
 impl actix_web::ResponseError for PostError {