1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
//! Runtime-liveness heartbeat for the systemd service watchdog.
//!
//! When the unit sets `WatchdogSec=`, systemd passes `WATCHDOG_USEC` and
//! `NOTIFY_SOCKET` to the daemon and restarts it if no `WATCHDOG=1` arrives
//! within that period. The heartbeat is a task on the tokio runtime, so it
//! proves that a runtime worker is free to run work, not only that the process
//! exists. A task that holds every worker without awaiting stops the heartbeat,
//! and systemd then restarts the node instead of reporting it `active` while it
//! serves nothing.
//!
//! The node has no setting for this. The unit's `WatchdogSec=` turns it on and
//! sets the period. Without `WATCHDOG_USEC` the heartbeat does not start.
use std::ffi::OsString;
use std::time::Duration;
/// Where to send the heartbeat, and how often.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Heartbeat {
/// The `NOTIFY_SOCKET` address: a filesystem path, or an abstract socket
/// name that starts with `@`.
pub socket: OsString,
/// Half the watchdog period, as `sd_watchdog_enabled(3)` recommends.
pub interval: Duration,
}
impl Heartbeat {
/// The heartbeat systemd asks for through the environment, or `None` when
/// it asks for none.
///
/// `WATCHDOG_PID`, when set, names the process the watchdog is for. A
/// process with another PID does not send the heartbeat.
#[must_use]
pub fn from_env_vars(
notify_socket: Option<OsString>,
watchdog_usec: Option<&str>,
watchdog_pid: Option<&str>,
own_pid: u32,
) -> Option<Self> {
let socket = notify_socket.filter(|s| !s.is_empty())?;
let usec: u64 = watchdog_usec?.trim().parse().ok()?;
if usec == 0 {
return None;
}
if let Some(pid) = watchdog_pid
&& pid.trim().parse::<u32>().ok()? != own_pid
{
return None;
}
let interval = Duration::from_micros(usec / 2).max(Duration::from_millis(1));
Some(Self { socket, interval })
}
/// The heartbeat systemd asks this process for.
#[must_use]
pub fn from_env() -> Option<Self> {
Self::from_env_vars(
std::env::var_os("NOTIFY_SOCKET"),
std::env::var("WATCHDOG_USEC").ok().as_deref(),
std::env::var("WATCHDOG_PID").ok().as_deref(),
std::process::id(),
)
}
}
/// A datagram socket bound to the `NOTIFY_SOCKET` address, built once and
/// reused for every heartbeat.
#[cfg(unix)]
#[derive(Debug)]
pub struct Notifier {
sock: std::os::unix::net::UnixDatagram,
addr: std::os::unix::net::SocketAddr,
}
#[cfg(unix)]
impl Notifier {
/// A notifier for the `NOTIFY_SOCKET` address `socket`: a filesystem path,
/// or an abstract socket name that starts with `@`.
///
/// # Errors
///
/// Fails when the socket cannot be created or the address is invalid.
pub fn new(socket: &std::ffi::OsStr) -> std::io::Result<Self> {
use std::os::unix::ffi::OsStrExt;
use std::os::unix::net::{SocketAddr, UnixDatagram};
let addr = match socket.as_bytes().strip_prefix(b"@") {
Some(name) => abstract_addr(name)?,
None => SocketAddr::from_pathname(socket)?,
};
let sock = UnixDatagram::unbound()?;
// A full receive queue must not block a runtime worker.
sock.set_nonblocking(true)?;
Ok(Self { sock, addr })
}
/// Send one `WATCHDOG=1` datagram.
///
/// # Errors
///
/// Fails when the datagram cannot be sent.
pub fn notify(&self) -> std::io::Result<()> {
self.sock.send_to_addr(b"WATCHDOG=1", &self.addr).map(drop)
}
}
#[cfg(target_os = "linux")]
fn abstract_addr(name: &[u8]) -> std::io::Result<std::os::unix::net::SocketAddr> {
use std::os::linux::net::SocketAddrExt;
std::os::unix::net::SocketAddr::from_abstract_name(name)
}
#[cfg(all(unix, not(target_os = "linux")))]
fn abstract_addr(_name: &[u8]) -> std::io::Result<std::os::unix::net::SocketAddr> {
Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"abstract NOTIFY_SOCKET names exist only on Linux",
))
}
/// Start the heartbeat on the current tokio runtime when systemd asks for one.
///
/// Call it from inside the runtime before any other work, so the heartbeat also
/// covers bring-up and shutdown drain. The task runs until the runtime stops.
pub fn spawn() {
#[cfg(unix)]
if let Some(heartbeat) = Heartbeat::from_env() {
tokio::spawn(run(heartbeat));
}
}
#[cfg(unix)]
async fn run(heartbeat: Heartbeat) {
let mut ticker = tokio::time::interval(heartbeat.interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut notifier = None;
let mut warned = false;
loop {
ticker.tick().await;
// A notifier that failed to build is built again on the next tick.
let sent = match ¬ifier {
Some(n) => Notifier::notify(n),
None => Notifier::new(&heartbeat.socket).and_then(|n| {
let sent = n.notify();
notifier = Some(n);
sent
}),
};
match sent {
Ok(()) => warned = false,
Err(error) if !warned => {
warned = true;
tracing::warn!(%error, "systemd watchdog heartbeat failed");
}
Err(_) => {}
}
}
}
#[cfg(test)]
mod tests;