feat(health): add skip_initial_history flag to periodic log collector. False by default

crates/health/example/config.example.toml

@@ -237,6 +237,11 @@ logs_state_file = "/tmp/logs_collector_{machine_id}.json"
 # Skip any Redfish LogService whose odata id contains one of these substrings.
 # Applies when auto-mode downgrades to periodic collection. See [collectors.logs.periodic].
 exclude_services = ["Journal"]
+# When true, on first encounter of a LogService with no saved state, anchor at
+# the current highest entry ID without emitting historical entries. Matches
+# SSE real-time behaviour. Default: false (collect all existing entries on
+# first run). Also read by auto-mode after downgrade to periodic.
+# skip_initial_history = false
 
 # SSE reconnect backoff; read when mode = "sse" and during auto-mode SSE attempts.
 [collectors.logs.sse]
@@ -254,6 +259,11 @@ logs_state_file = "/tmp/logs_collector_{machine_id}.json"
 # self-referential since polling generates Journal entries). Set to [] to collect
 # from every discovered LogService. Also read by auto-mode after downgrade.
 exclude_services = ["Journal"]
+# When true, on first encounter of a LogService with no saved state, anchor at
+# the current highest entry ID without emitting historical entries. Matches
+# SSE real-time behaviour. Default: false (collect all existing entries on
+# first run).
+# skip_initial_history = false
 
 # ==============================================================================
 # TLS: Shared certificate configuration for switch collectors

crates/health/src/collectors/logs/periodic.rs

@@ -46,6 +46,13 @@ pub struct LogsCollectorConfig {
     /// Substrings; a discovered LogService whose odata id contains any of these
     /// is skipped. Empty collects from every service.
     pub exclude_services: Vec<String>,
+
+    /// When true, on the first encounter of a LogService with no saved state,
+    /// anchor at the current highest log entry ID without emitting historical
+    /// entries. Subsequent polls collect only new entries forward, matching
+    /// SSE behaviour. When false (default), all existing entries are collected
+    /// on first encounter.
+    pub skip_initial_history: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -75,6 +82,7 @@ pub struct LogsCollector<B: Bmc> {
     data_sink: Option<Arc<dyn DataSink>>,
     include_diagnostics: bool,
     exclude_services: Vec<String>,
+    skip_initial_history: bool,
 }
 
 impl<B: Bmc + 'static> PeriodicCollector<B> for LogsCollector<B> {
@@ -96,6 +104,7 @@ impl<B: Bmc + 'static> PeriodicCollector<B> for LogsCollector<B> {
             data_sink: config.data_sink,
             include_diagnostics: config.include_diagnostics,
             exclude_services: config.exclude_services,
+            skip_initial_history: config.skip_initial_history,
         })
     }
 
@@ -319,27 +328,54 @@ impl<B: Bmc + 'static> LogsCollector<B> {
                         })
                         .collect()
                 }
-                None => match service.entries().await {
-                    Ok(Some(v)) => {
+                None => {
+                    let all_entries = match service.entries().await {
+                        Ok(Some(v)) => v,
+                        Ok(None) => continue,
+                        Err(error) => {
+                            fetch_failures += 1;
+                            tracing::warn!(
+                                %service_id,
+                                ?error,
+                                "Failed to fetch log entries"
+                            );
+                            continue;
+                        }
+                    };
+
+                    if self.skip_initial_history {
+                        // Anchor at the current highest entry ID without emitting
+                        // historical entries, so the next poll collects only new
+                        // entries. Matches the real-time-only behaviour of SSE.
+                        //
+                        // We always write a sentinel (-1) when the service is
+                        // empty or has no parseable numeric IDs, so the service
+                        // is marked initialised in last_seen_ids. Without this,
+                        // a subsequent poll would re-enter this None arm and
+                        // treat the first real entry as "initial history" and
+                        // discard it. -1 is safe: real Redfish IDs are ≥ 0, so
+                        // the next poll's `id > anchor` filter passes everything.
+                        let anchor_id =
+                            initial_anchor_id(all_entries.iter().map(|e| e.base.id.as_str()));
+                        state
+                            .last_seen_ids
+                            .insert(service.odata_id().clone(), anchor_id);
                         tracing::info!(
                             %service_id,
-                            endpoint=?self.endpoint.addr,
-                            "Last seen id is empty, fetching all entries");
-                        v
-                    }
-                    Ok(None) => {
-                        continue;
-                    }
-                    Err(error) => {
-                        fetch_failures += 1;
-                        tracing::warn!(
-                            %service_id,
-                            ?error,
-                            "Failed to fetch log entries"
+                            anchor_id,
+                            "skip_initial_history: anchored at current log position, \
+                             skipping historical entries"
                         );
                         continue;
                     }
-                },
+
+                    tracing::info!(
+                        %service_id,
+                        endpoint=?self.endpoint.addr,
+                        "Last seen id is empty, fetching all entries"
+                    );
+                    all_entries
+                }
             };
 
             if entries.is_empty() {
@@ -415,6 +451,18 @@ impl<B: Bmc + 'static> LogsCollector<B> {
     }
 }
 
+/// Returns the highest parseable integer ID from `ids`, or -1 when `ids` is
+/// empty or contains no parseable integers.
+///
+/// Used as the `last_seen_ids` anchor when `skip_initial_history = true`. -1
+/// is a safe sentinel because real Redfish entry IDs are non-negative, so a
+/// subsequent poll's `id > anchor` filter passes every entry.
+fn initial_anchor_id<'a>(ids: impl Iterator<Item = &'a str>) -> i32 {
+    ids.filter_map(|id| id.parse::<i32>().ok())
+        .max()
+        .unwrap_or(-1)
+}
+
 /// True if `service_id` contains any of the configured exclude substrings.
 /// An empty `exclude_services` never excludes anything. Matching is a plain
 /// (case-sensitive) substring test against the Redfish LogService odata id.
@@ -428,7 +476,7 @@ fn service_is_excluded(exclude_services: &[String], service_id: &str) -> bool {
 mod tests {
     use carbide_test_support::{Check, check_values};
 
-    use super::service_is_excluded;
+    use super::{initial_anchor_id, service_is_excluded};
 
     const JOURNAL_BMC: &str = "/redfish/v1/Managers/BMC_0/LogServices/Journal";
     const JOURNAL_HGX: &str = "/redfish/v1/Managers/HGX_BMC_0/LogServices/Journal";
@@ -500,4 +548,43 @@ mod tests {
             |(patterns, service_id)| service_is_excluded(&patterns, service_id),
         );
     }
+
+    #[test]
+    fn initial_anchor_id_cases() {
+        check_values(
+            [
+                Check {
+                    scenario: "empty service yields sentinel -1",
+                    input: vec![],
+                    expect: -1,
+                },
+                Check {
+                    scenario: "single numeric id is returned",
+                    input: vec!["42"],
+                    expect: 42,
+                },
+                Check {
+                    scenario: "max of multiple numeric ids is returned",
+                    input: vec!["1", "99", "7"],
+                    expect: 99,
+                },
+                Check {
+                    scenario: "non-parseable ids only yield sentinel -1",
+                    input: vec!["abc", "xyz"],
+                    expect: -1,
+                },
+                Check {
+                    scenario: "mixed parseable and non-parseable uses numeric max",
+                    input: vec!["abc", "5", "xyz", "3"],
+                    expect: 5,
+                },
+                Check {
+                    scenario: "id zero is returned (not confused with sentinel)",
+                    input: vec!["0"],
+                    expect: 0,
+                },
+            ],
+            |ids: Vec<&str>| initial_anchor_id(ids.into_iter()),
+        );
+    }
 }

crates/health/src/config.rs

@@ -1163,6 +1163,14 @@ pub struct PeriodicLogConfig {
     /// discovered LogService.
     #[serde(default)]
     pub exclude_services: Vec<String>,
+
+    /// When true, on the first encounter of a LogService with no saved state,
+    /// anchor at the current highest entry ID without emitting existing entries.
+    /// Subsequent polls collect only new entries, matching SSE real-time
+    /// behaviour. Defaults to false (existing entries are collected on first
+    /// run). Also applies when auto-mode downgrades to periodic collection.
+    #[serde(default)]
+    pub skip_initial_history: bool,
 }
 
 impl Default for PeriodicLogConfig {
@@ -1172,6 +1180,7 @@ impl Default for PeriodicLogConfig {
             state_refresh_interval: Duration::from_secs(1800),
             logs_state_file: "/tmp/logs_collector_{machine_id}.json".to_string(),
             exclude_services: vec!["Journal".to_string()],
+            skip_initial_history: false,
         }
     }
 }

crates/health/src/discovery/spawn.rs

@@ -248,6 +248,7 @@ fn spawn_generic_redfish_collectors(
                     data_sink,
                     include_diagnostics: ctx.logs_include_diagnostics,
                     exclude_services: pcfg.exclude_services.clone(),
+                    skip_initial_history: pcfg.skip_initial_history,
                 },
                 CollectorStartContext {
                     limiter: ctx.limiter.clone(),