graph: Warn on log entry parse failures instead of silently skipping

Author: fordN

graph/src/components/log_store/elasticsearch.rs

@@ -2,6 +2,7 @@ use async_trait::async_trait;
 use reqwest::Client;
 use serde::Deserialize;
 use serde_json::json;
+use slog::{warn, Logger};
 use std::collections::HashMap;
 use std::time::Duration;
 
@@ -17,11 +18,13 @@ pub struct ElasticsearchLogStore {
     client: Client,
     index: String,
     timeout: Duration,
+    logger: Logger,
 }
 
 impl ElasticsearchLogStore {
     pub fn new(config: ElasticLoggingConfig, index: String, timeout: Duration) -> Self {
         Self {
+            logger: crate::log::logger(false),
             endpoint: config.endpoint,
             username: config.username,
             password: config.password,
@@ -146,8 +149,21 @@ impl ElasticsearchLogStore {
     }
 
     fn parse_log_entry(&self, source: ElasticsearchLogDocument) -> Option<LogEntry> {
-        let level = source.level.parse().ok()?;
-        let subgraph_id = DeploymentHash::new(&source.subgraph_id).ok()?;
+        let level = match source.level.parse() {
+            Ok(l) => l,
+            Err(_) => {
+                warn!(self.logger, "Invalid log level in Elasticsearch entry"; "level" => &source.level);
+                return None;
+            }
+        };
+
+        let subgraph_id = match DeploymentHash::new(&source.subgraph_id) {
+            Ok(id) => id,
+            Err(_) => {
+                warn!(self.logger, "Invalid subgraph ID in Elasticsearch entry"; "subgraph_id" => &source.subgraph_id);
+                return None;
+            }
+        };
 
         // Convert arguments HashMap to Vec<(String, String)>
         let arguments: Vec<(String, String)> = source.arguments.into_iter().collect();

graph/src/components/log_store/file.rs

@@ -1,5 +1,6 @@
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
+use slog::{warn, Logger};
 use std::cmp::Reverse;
 use std::collections::BinaryHeap;
 use std::fs::File;
@@ -13,6 +14,7 @@ use super::{LogEntry, LogMeta, LogQuery, LogStore, LogStoreError};
 pub struct FileLogStore {
     directory: PathBuf,
     retention_hours: u32,
+    logger: Logger,
 }
 
 impl FileLogStore {
@@ -24,6 +26,7 @@ impl FileLogStore {
         let store = Self {
             directory,
             retention_hours,
+            logger: crate::log::logger(false),
         };
 
         // Run cleanup on startup for all existing log files
@@ -53,10 +56,29 @@ impl FileLogStore {
 
     /// Parse a JSON line into a LogEntry
     fn parse_line(&self, line: &str) -> Option<LogEntry> {
-        let doc: FileLogDocument = serde_json::from_str(line).ok()?;
+        let doc: FileLogDocument = match serde_json::from_str(line) {
+            Ok(doc) => doc,
+            Err(e) => {
+                warn!(self.logger, "Failed to parse log line"; "error" => e.to_string());
+                return None;
+            }
+        };
 
-        let level = doc.level.parse().ok()?;
-        let subgraph_id = DeploymentHash::new(&doc.subgraph_id).ok()?;
+        let level = match doc.level.parse() {
+            Ok(l) => l,
+            Err(_) => {
+                warn!(self.logger, "Invalid log level"; "level" => &doc.level);
+                return None;
+            }
+        };
+
+        let subgraph_id = match DeploymentHash::new(&doc.subgraph_id) {
+            Ok(id) => id,
+            Err(_) => {
+                warn!(self.logger, "Invalid subgraph ID"; "subgraph_id" => &doc.subgraph_id);
+                return None;
+            }
+        };
 
         Some(LogEntry {
             id: doc.id,

graph/src/components/log_store/loki.rs

@@ -1,6 +1,7 @@
 use async_trait::async_trait;
 use reqwest::Client;
 use serde::Deserialize;
+use slog::{warn, Logger};
 use std::collections::HashMap;
 use std::time::Duration;
 
@@ -14,6 +15,7 @@ pub struct LokiLogStore {
     username: Option<String>,
     password: Option<String>,
     client: Client,
+    logger: Logger,
 }
 
 impl LokiLogStore {
@@ -34,6 +36,7 @@ impl LokiLogStore {
             username,
             password,
             client,
+            logger: crate::log::logger(false),
         })
     }
 
@@ -146,10 +149,29 @@ impl LokiLogStore {
     ) -> Option<LogEntry> {
         // value is [timestamp_ns, log_line]
         // We expect the log line to be JSON with our log entry structure
-        let log_data: LokiLogDocument = serde_json::from_str(&value.1).ok()?;
+        let log_data: LokiLogDocument = match serde_json::from_str(&value.1) {
+            Ok(doc) => doc,
+            Err(e) => {
+                warn!(self.logger, "Failed to parse Loki log entry"; "error" => e.to_string());
+                return None;
+            }
+        };
+
+        let level = match log_data.level.parse() {
+            Ok(l) => l,
+            Err(_) => {
+                warn!(self.logger, "Invalid log level in Loki entry"; "level" => &log_data.level);
+                return None;
+            }
+        };
 
-        let level = log_data.level.parse().ok()?;
-        let subgraph_id = DeploymentHash::new(&log_data.subgraph_id).ok()?;
+        let subgraph_id = match DeploymentHash::new(&log_data.subgraph_id) {
+            Ok(id) => id,
+            Err(_) => {
+                warn!(self.logger, "Invalid subgraph ID in Loki entry"; "subgraph_id" => &log_data.subgraph_id);
+                return None;
+            }
+        };
 
         Some(LogEntry {
             id: log_data.id,