hmon: add heartbeat monitor

Add heartbeat monitor HMON.

Author: arkjedrz

Cargo.lock

@@ -6,6 +6,7 @@ members = [
     "src/health_monitoring_lib",
     "examples/rust_supervised_app",
 ]
+default-members = ["src/health_monitoring_lib"]
 
 [workspace.package]
 edition = "2021"
@@ -18,7 +19,6 @@ libc = "0.2.177"
 clap = { version = "4.5.49", features = ["derive"] }
 signal-hook = "0.3.18"
 
-monitor_rs = { path = "src/launch_manager_daemon/health_monitor_lib/rust_bindings" } # Temporary API
 health_monitoring_lib = { path = "src/health_monitoring_lib" }
 score_log = { git = "https://github.com/eclipse-score/baselibs_rust.git", tag = "v0.0.4" }
 score_testing_macros = { git = "https://github.com/eclipse-score/baselibs_rust.git", tag = "v0.0.4" }
@@ -28,3 +28,6 @@ containers = { git = "https://github.com/eclipse-score/baselibs_rust.git", tag =
 [workspace.lints.clippy]
 std_instead_of_core = "warn"
 alloc_instead_of_core = "warn"
+
+[workspace.lints.rust]
+unexpected_cfgs = { level = "warn", check-cfg = ['cfg(loom)'] }

Cargo.toml

@@ -42,6 +42,7 @@ CC_HDRS = [
 rust_library(
     name = "health_monitoring_lib",
     srcs = glob(["rust/**/*.rs"]),
+    crate_features = ["score_supervisor_api_client"],
     proc_macro_deps = PROC_MACRO_DEPS,
     visibility = ["//visibility:public"],
     deps = COMMON_DEPS,
@@ -64,6 +65,7 @@ cc_library(
 rust_static_library(
     name = "health_monitoring_lib_ffi",
     srcs = glob(["rust/**/*.rs"]),
+    crate_features = ["score_supervisor_api_client"],
     crate_name = "health_monitoring_lib",
     proc_macro_deps = [
         "@score_baselibs_rust//src/testing_macros:score_testing_macros",

src/health_monitoring_lib/BUILD

@@ -18,11 +18,15 @@ workspace = true
 score_log.workspace = true
 score_testing_macros.workspace = true
 containers.workspace = true
-monitor_rs.workspace = true
+monitor_rs = { path = "../launch_manager_daemon/health_monitor_lib/rust_bindings", optional = true } # Temporary API
 
 [dev-dependencies]
 stdout_logger.workspace = true
 
+[target.'cfg(loom)'.dependencies]
+loom = { version = "0.7", features = ["checkpoint"] }
+
 [features]
-default = []
+default = ["stub_supervisor_api_client"]
 stub_supervisor_api_client = []
+score_supervisor_api_client = ["monitor_rs"]

src/health_monitoring_lib/Cargo.toml

@@ -15,6 +15,8 @@ use core::fmt::Debug;
 use core::hash::Hash;
 use core::time::Duration;
 use std::sync::Arc;
+use std::time::Instant;
+
 /// Unique identifier for deadlines.
 #[derive(Clone, Copy, Eq)]
 #[repr(C)]
@@ -90,6 +92,7 @@ impl From<&str> for IdentTag {
     }
 }
 
+/// Range of accepted time.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 pub struct TimeRange {
     pub min: Duration,
@@ -103,16 +106,34 @@ impl TimeRange {
     }
 }
 
+/// Get duration as [`u32`].
+pub(crate) fn duration_to_u32(duration: Duration) -> u32 {
+    let millis = duration.as_millis();
+    u32::try_from(millis).expect("Monitor running for too long")
+}
+
+/// Heartbeat monitor error subgroup.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, crate::log::ScoreDebug)]
+pub(crate) enum HeartbeatMonitorEvaluationError {
+    /// Multiple heartbeats were observed in the same epoch.
+    MultipleHeartbeats,
+}
+
 /// Errors that can occur during monitor evaluation.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, crate::log::ScoreDebug)]
 pub(crate) enum MonitorEvaluationError {
     TooEarly,
     TooLate,
+    HeartbeatSpecific(HeartbeatMonitorEvaluationError),
 }
 
 /// Trait for evaluating monitors and reporting errors to be used by HealthMonitor.
 pub(crate) trait MonitorEvaluator {
-    fn evaluate(&self, on_error: &mut dyn FnMut(&IdentTag, MonitorEvaluationError));
+    /// Run monitor evaluation.
+    ///
+    /// - `hmon_starting_point` - starting point of all monitors.
+    /// - `on_error` - error handling, containing tag of failing object and error code.
+    fn evaluate(&self, hmon_starting_point: Instant, on_error: &mut dyn FnMut(&IdentTag, MonitorEvaluationError));
 }
 
 /// Handle to a monitor evaluator, allowing for dynamic dispatch.
@@ -127,8 +148,8 @@ impl MonitorEvalHandle {
 }
 
 impl MonitorEvaluator for MonitorEvalHandle {
-    fn evaluate(&self, on_error: &mut dyn FnMut(&IdentTag, MonitorEvaluationError)) {
-        self.inner.evaluate(on_error)
+    fn evaluate(&self, hmon_starting_point: Instant, on_error: &mut dyn FnMut(&IdentTag, MonitorEvaluationError)) {
+        self.inner.evaluate(hmon_starting_point, on_error)
     }
 }
 

src/health_monitoring_lib/rust/common.rs

@@ -11,7 +11,9 @@
 // SPDX-License-Identifier: Apache-2.0
 // *******************************************************************************
 use super::common::DeadlineTemplate;
-use crate::common::{IdentTag, MonitorEvalHandle, MonitorEvaluationError, MonitorEvaluator, TimeRange};
+use crate::common::{
+    duration_to_u32, IdentTag, MonitorEvalHandle, MonitorEvaluationError, MonitorEvaluator, TimeRange,
+};
 use crate::{
     deadline::{
         common::StateIndex,
@@ -96,7 +98,7 @@ impl DeadlineMonitor {
             inner: Arc::new(DeadlineMonitorInner {
                 deadlines,
                 active_deadlines: active_deadlines.into(),
-                start_time: Instant::now(),
+                monitor_starting_point: Instant::now(),
             }),
         }
     }
@@ -176,7 +178,7 @@ impl Deadline {
     /// Caller must ensure that deadline is not used until it's stopped.
     /// After this call You shall assure there's only a single owner of the `Deadline` instance and it does not call start before stopping.
     pub(super) unsafe fn start_internal(&mut self) -> Result<(), DeadlineError> {
-        let now = self.monitor.now();
+        let now = duration_to_u32(self.monitor.monitor_starting_point.elapsed());
         let max_time = now + self.range.max.as_millis() as u32;
 
         let mut is_broken = false;
@@ -201,7 +203,7 @@ impl Deadline {
     }
 
     pub(super) fn stop_internal(&mut self) {
-        let now = self.monitor.now();
+        let now = duration_to_u32(self.monitor.monitor_starting_point.elapsed());
         let max = self.range.max.as_millis() as u32;
         let min = self.range.min.as_millis() as u32;
 
@@ -243,6 +245,7 @@ impl Deadline {
             (Some(MonitorEvaluationError::TooLate), val) => {
                 error!("Deadline {:?} stopped too late by {} ms", self.tag, val);
             },
+            (Some(MonitorEvaluationError::HeartbeatSpecific(_)), _) => unreachable!(),
             (None, _) => {},
         }
     }
@@ -267,8 +270,9 @@ impl Drop for Deadline {
 }
 
 struct DeadlineMonitorInner {
-    // Start time of the monitor creation to calculate relative timestamps
-    start_time: Instant,
+    /// Monitor starting point.
+    /// Offset is calculated during evaluation in relation to provided health monitor starting point.
+    monitor_starting_point: Instant,
 
     // Templates for deadlines registered in the monitor to create `Deadline` instances.
     deadlines: HashMap<IdentTag, DeadlineTemplate>,
@@ -280,8 +284,8 @@ struct DeadlineMonitorInner {
 }
 
 impl MonitorEvaluator for DeadlineMonitorInner {
-    fn evaluate(&self, on_error: &mut dyn FnMut(&IdentTag, MonitorEvaluationError)) {
-        self.evaluate(on_error);
+    fn evaluate(&self, hmon_starting_point: Instant, on_error: &mut dyn FnMut(&IdentTag, MonitorEvaluationError)) {
+        self.evaluate(hmon_starting_point, on_error);
     }
 }
 
@@ -294,14 +298,7 @@ impl DeadlineMonitorInner {
         }
     }
 
-    fn now(&self) -> u32 {
-        let duration = self.start_time.elapsed();
-        // As u32 can hold up to ~49 days in milliseconds, this should be sufficient for our use case
-        // We still have a room up to 60bits timestamp if needed in future
-        u32::try_from(duration.as_millis()).expect("Monitor running for too long")
-    }
-
-    fn evaluate(&self, mut on_failed: impl FnMut(&IdentTag, MonitorEvaluationError)) {
+    fn evaluate(&self, hmon_starting_point: Instant, mut on_failed: impl FnMut(&IdentTag, MonitorEvaluationError)) {
         for (tag, deadline) in self.active_deadlines.iter() {
             let snapshot = deadline.snapshot();
             if snapshot.is_underrun() {
@@ -316,7 +313,9 @@ impl DeadlineMonitorInner {
                     "Deadline snapshot cannot be both running and stopped"
                 );
 
-                let now = self.now();
+                // Get current timestamp, with offset to HMON time.
+                let offset = hmon_starting_point.duration_since(self.monitor_starting_point);
+                let now = duration_to_u32(self.monitor_starting_point.elapsed() - offset);
                 let expected = snapshot.timestamp_ms();
                 if now > expected {
                     // Deadline missed, report
@@ -395,12 +394,13 @@ mod tests {
         let monitor = create_monitor_with_deadlines();
         let mut deadline = monitor.get_deadline(&IdentTag::from("deadline_long")).unwrap();
         let handle = deadline.start().unwrap();
+        let hmon_starting_point = Instant::now();
 
         std::thread::sleep(core::time::Duration::from_millis(1001)); // Sleep to simulate work within the deadline range
 
         drop(handle); // stop the deadline
 
-        monitor.inner.evaluate(|tag, deadline_failure| {
+        monitor.inner.evaluate(hmon_starting_point, |tag, deadline_failure| {
             panic!(
                 "Deadline {:?} should not have failed or underrun({:?})",
                 tag, deadline_failure
@@ -413,10 +413,11 @@ mod tests {
         let monitor = create_monitor_with_deadlines();
         let mut deadline = monitor.get_deadline(&IdentTag::from("deadline_long")).unwrap();
         let handle = deadline.start().unwrap();
+        let hmon_starting_point = Instant::now();
 
         drop(handle); // stop the deadline
 
-        monitor.inner.evaluate(|tag, deadline_failure| {
+        monitor.inner.evaluate(hmon_starting_point, |tag, deadline_failure| {
             assert_eq!(
                 deadline_failure,
                 MonitorEvaluationError::TooEarly,
@@ -431,10 +432,11 @@ mod tests {
         let monitor = create_monitor_with_deadlines();
         let mut deadline = monitor.get_deadline(&IdentTag::from("deadline_long")).unwrap();
         let handle = deadline.start().unwrap();
+        let hmon_starting_point = Instant::now();
 
         // So deadline stop happens after evaluate, still it should be reported as failed
 
-        monitor.inner.evaluate(|tag, deadline_failure| {
+        monitor.inner.evaluate(hmon_starting_point, |tag, deadline_failure| {
             assert_eq!(
                 deadline_failure,
                 MonitorEvaluationError::TooEarly,
@@ -452,6 +454,7 @@ mod tests {
         let monitor = create_monitor_with_deadlines();
         let mut deadline = monitor.get_deadline(&IdentTag::from("deadline_long")).unwrap();
         let handle = deadline.start().unwrap();
+        let hmon_starting_point = Instant::now();
 
         // So deadline failed, then we start it again so it shall be already expired and also evaluation shall work
         drop(handle); // stop the deadline
@@ -460,7 +463,7 @@ mod tests {
         let handle = deadline.start();
         assert_eq!(handle.err(), Some(DeadlineError::DeadlineAlreadyFailed));
 
-        monitor.inner.evaluate(|tag, deadline_failure| {
+        monitor.inner.evaluate(hmon_starting_point, |tag, deadline_failure| {
             assert_eq!(
                 deadline_failure,
                 MonitorEvaluationError::TooEarly,
@@ -476,10 +479,11 @@ mod tests {
         let monitor = create_monitor_with_deadlines();
         let mut deadline = monitor.get_deadline(&IdentTag::from("deadline_fast")).unwrap();
         let handle = deadline.start().unwrap();
+        let hmon_starting_point = Instant::now();
 
         drop(handle); // stop the deadline
 
-        monitor.inner.evaluate(|tag, deadline_failure| {
+        monitor.inner.evaluate(hmon_starting_point, |tag, deadline_failure| {
             assert_eq!(
                 deadline_failure,
                 MonitorEvaluationError::TooLate,
@@ -493,6 +497,7 @@ mod tests {
     #[test]
     fn monitor_with_multiple_running_deadlines() {
         let monitor = create_monitor_with_multiple_running_deadlines();
+        let hmon_starting_point = Instant::now();
 
         let mut deadline = monitor.get_deadline(&IdentTag::from("deadline_fast1")).unwrap();
         let _handle1 = deadline.start().unwrap();
@@ -507,7 +512,7 @@ mod tests {
 
         let mut cnt = 0;
 
-        monitor.inner.evaluate(|tag, deadline_failure| {
+        monitor.inner.evaluate(hmon_starting_point, |tag, deadline_failure| {
             cnt += 1;
             assert_eq!(
                 deadline_failure,