diff options
author | Albin Suresh <albin.suresh@softwareag.com> | 2022-05-27 16:59:53 +0530 |
---|---|---|
committer | Albin Suresh <albin.suresh@softwareag.com> | 2022-05-27 16:59:53 +0530 |
commit | a10c6c119899a8a9b36884e4e93e3f6bd873d793 (patch) | |
tree | 40bfb9f047545b7a4ece09e05f6cba2dab41bb35 | |
parent | e3775e430d3109081d3926ab4e7b13b05e1c2741 (diff) |
Fix watchdog health check with timestamp validation
-rw-r--r-- | Cargo.lock | 1 | ||||
-rw-r--r-- | crates/core/tedge_agent/src/agent.rs | 4 | ||||
-rw-r--r-- | crates/core/tedge_mapper/src/collectd/monitor.rs | 4 | ||||
-rw-r--r-- | crates/core/tedge_mapper/src/core/mapper.rs | 7 | ||||
-rw-r--r-- | crates/core/tedge_watchdog/Cargo.toml | 3 | ||||
-rw-r--r-- | crates/core/tedge_watchdog/src/systemd_watchdog.rs | 56 |
6 files changed, 57 insertions, 18 deletions
@@ -3064,6 +3064,7 @@ dependencies = [ "tedge_config", "tedge_utils", "thiserror", + "time", "tokio", "tracing", ] diff --git a/crates/core/tedge_agent/src/agent.rs b/crates/core/tedge_agent/src/agent.rs index b7d946b0..18a1617c 100644 --- a/crates/core/tedge_agent/src/agent.rs +++ b/crates/core/tedge_agent/src/agent.rs @@ -27,6 +27,7 @@ use tedge_config::{ TEdgeConfigLocation, TmpPathSetting, DEFAULT_LOG_PATH, DEFAULT_RUN_PATH, }; use tedge_utils::file::create_directory_with_user_group; +use time::OffsetDateTime; use tokio::sync::Mutex; use tracing::{debug, error, info, instrument, warn}; @@ -296,7 +297,8 @@ impl SmAgent { topic if self.config.request_topics_health.accept_topic(topic) => { let health_status = json!({ "status": "up", - "pid": process::id() + "pid": process::id(), + "time": OffsetDateTime::now_utc().unix_timestamp() }) .to_string(); let health_message = diff --git a/crates/core/tedge_mapper/src/collectd/monitor.rs b/crates/core/tedge_mapper/src/collectd/monitor.rs index f927f887..61de30a5 100644 --- a/crates/core/tedge_mapper/src/collectd/monitor.rs +++ b/crates/core/tedge_mapper/src/collectd/monitor.rs @@ -3,6 +3,7 @@ use std::process; use batcher::{BatchConfigBuilder, BatchDriver, BatchDriverInput, BatchDriverOutput, Batcher}; use mqtt_channel::{Connection, Message, QoS, SinkExt, StreamExt, Topic, TopicFilter}; use serde_json::json; +use time::OffsetDateTime; use tracing::{error, info, instrument}; use super::{batcher::MessageBatch, collectd::CollectdMessage, error::DeviceMonitorError}; @@ -109,7 +110,8 @@ impl DeviceMonitor { if health_check_topics.accept(&message) { let health_status = json!({ "status": "up", - "pid": process::id() + "pid": process::id(), + "time": OffsetDateTime::now_utc().unix_timestamp() }) .to_string(); let health_message = Message::new(&health_status_topic, health_status); diff --git a/crates/core/tedge_mapper/src/core/mapper.rs b/crates/core/tedge_mapper/src/core/mapper.rs index 9309a31e..58a212a1 100644 --- a/crates/core/tedge_mapper/src/core/mapper.rs +++ b/crates/core/tedge_mapper/src/core/mapper.rs @@ -7,6 +7,7 @@ use mqtt_channel::{ UnboundedSender, }; use serde_json::json; +use time::OffsetDateTime; use tracing::{error, info, instrument}; const SYNC_WINDOW: Duration = Duration::from_secs(3); @@ -133,7 +134,8 @@ impl Mapper { if self.health_check_topics.accept(&message) { let health_status = json!({ "status": "up", - "pid": process::id() + "pid": process::id(), + "time": OffsetDateTime::now_utc().unix_timestamp() }) .to_string(); let health_message = Message::new(&self.health_status_topic, health_status); @@ -242,7 +244,7 @@ mod tests { let common_health_check_topic = "tedge/health-check"; let health_status = broker .wait_for_response_on_publish( - &common_health_check_topic, + common_health_check_topic, "", &health_topic, Duration::from_secs(1), @@ -252,6 +254,7 @@ mod tests { let health_status: Value = serde_json::from_str(health_status.as_str())?; assert_json_include!(actual: &health_status, expected: json!({"status": "up"})); assert!(health_status["pid"].is_number()); + assert!(health_status["time"].is_number()); Ok(()) } diff --git a/crates/core/tedge_watchdog/Cargo.toml b/crates/core/tedge_watchdog/Cargo.toml index 7374eb5c..da3647c0 100644 --- a/crates/core/tedge_watchdog/Cargo.toml +++ b/crates/core/tedge_watchdog/Cargo.toml @@ -28,5 +28,6 @@ freedesktop_entry_parser = "1.3.0" tedge_config = { path = "../../common/tedge_config" } tedge_utils = { path = "../../common/tedge_utils", features = ["logging"] } thiserror ="1.0.30" -tokio = { version = "1.12", features = ["sync", "time"] } +time = { version = "0.3", features = ["formatting", "serde-well-known"] } +tokio = { version = "1.12", features = ["sync", "time", "rt-multi-thread"] } tracing = { version = "0.1", features = ["attributes", "log"] } diff --git a/crates/core/tedge_watchdog/src/systemd_watchdog.rs b/crates/core/tedge_watchdog/src/systemd_watchdog.rs index 3fec6840..5ee6b916 100644 --- a/crates/core/tedge_watchdog/src/systemd_watchdog.rs +++ b/crates/core/tedge_watchdog/src/systemd_watchdog.rs @@ -1,5 +1,6 @@ use crate::error::WatchdogError; use freedesktop_entry_parser::parse_entry; +use futures::channel::mpsc; use futures::stream::FuturesUnordered; use futures::StreamExt; use mqtt_channel::{Config, Message, PubChannel, Topic}; @@ -14,12 +15,14 @@ use tedge_config::{ ConfigRepository, ConfigSettingAccessor, MqttBindAddressSetting, MqttPortSetting, TEdgeConfigLocation, }; +use time::OffsetDateTime; use tracing::{debug, error, info, warn}; -#[derive(Serialize, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] pub struct HealthStatus { status: String, pid: u32, + time: i64, } pub async fn start_watchdog(tedge_config_dir: PathBuf) -> Result<(), anyhow::Error> { @@ -90,20 +93,22 @@ async fn monitor_tedge_service( let start = Instant::now(); - match tokio::time::timeout(tokio::time::Duration::from_secs(interval), received.next()) - .await + let request_timestamp = OffsetDateTime::now_utc().unix_timestamp(); + match tokio::time::timeout( + tokio::time::Duration::from_secs(interval), + get_latest_health_status_message(request_timestamp, &mut received), + ) + .await { - Ok(Some(msg)) => { - let message = msg.payload_str()?; - - let p: HealthStatus = serde_json::from_str(message)?; - - debug!("Sending notification for {} with pid: {}", name, p.pid); - notify_systemd(p.pid, "WATCHDOG=1")?; + Ok(health_status) => { + debug!( + "Sending notification for {} with pid: {}", + name, health_status.pid + ); + notify_systemd(health_status.pid, "WATCHDOG=1")?; } - Ok(None) => {} - Err(elapsed) => { - warn!("The {name} failed with {elapsed}"); + Err(_) => { + warn!("No health check response received from {name} in time"); } } @@ -114,6 +119,31 @@ async fn monitor_tedge_service( } } +async fn get_latest_health_status_message( + request_timestamp: i64, + messages: &mut mpsc::UnboundedReceiver<Message>, +) -> HealthStatus { + loop { + if let Some(message) = messages.next().await { + if let Ok(message) = message.payload_str() { + debug!("Health response received: {}", message); + if let Ok(health_status) = serde_json::from_str::<HealthStatus>(message) { + if health_status.time >= request_timestamp { + return health_status; + } else { + debug!( + "Ignoring stale health response: {:?} older than request time: {}", + health_status, request_timestamp + ); + } + } else { + error!("Invalid health response received: {}", message); + } + } + } + } +} + fn get_mqtt_config( tedge_config_location: TEdgeConfigLocation, client_id: &str, |