rigtop_core/
nut.rs

1//! UPS readings from Network UPS Tools (NUT): the plain-text protocol of its
2//! `upsd` over TCP (RFC 9271), so that no extra program or privilege is
3//! needed. `upsd` usually listens on 127.0.0.1:3493 of the machine the UPS
4//! is plugged into.
5//!
6//! ```text
7//! LIST UPS                       LIST VAR ups
8//! BEGIN LIST UPS                 BEGIN LIST VAR ups
9//! UPS ups "Test UPS"             VAR ups battery.charge "100"
10//! END LIST UPS                   VAR ups ups.status "OL CHRG"
11//!                                END LIST VAR ups
12//! ```
13
14use std::collections::HashMap;
15use std::io::{BufRead, BufReader, Read, Write};
16use std::net::{TcpStream, ToSocketAddrs};
17use std::time::{Duration, Instant};
18
19use crate::snapshot::UpsStats;
20
21/// NUT's port.
22pub const DEFAULT_PORT: u16 = 3493;
23/// How long to wait for upsd before giving up on this sample.
24const TIMEOUT: Duration = Duration::from_millis(800);
25/// How often to ask upsd; its drivers poll the UPS every 2 seconds by default.
26const POLL_EVERY: Duration = Duration::from_secs(2);
27/// How long to wait before trying again when upsd was never reached (no
28/// NUT on this machine, most often).
29const RETRY_AFTER: Duration = Duration::from_secs(30);
30/// How long to wait when upsd stopped answering after it had.
31const RETRY_LOST_AFTER: Duration = Duration::from_secs(5);
32/// Upper bounds against something else listening at the address.
33const MAX_LINE: u64 = 4096;
34const MAX_LINES: usize = 2000;
35
36/// What a failed request ran into.
37#[derive(Debug)]
38enum RequestError {
39    /// upsd answered with an error (`ERR DATA-STALE`).
40    Nut(String),
41    /// The connection failed or the answer made no sense.
42    Io(String),
43}
44
45/// An open connection to upsd.
46struct Connection {
47    reader: BufReader<TcpStream>,
48    writer: TcpStream,
49}
50
51impl Connection {
52    fn open(address: &str) -> Result<Self, String> {
53        let target = address
54            .to_socket_addrs()
55            .map_err(|err| reason(&err))?
56            .next()
57            .ok_or_else(|| "no address".to_owned())?;
58        let stream = TcpStream::connect_timeout(&target, TIMEOUT).map_err(|err| reason(&err))?;
59        stream
60            .set_read_timeout(Some(TIMEOUT))
61            .and_then(|()| stream.set_write_timeout(Some(TIMEOUT)))
62            .map_err(|err| reason(&err))?;
63        let writer = stream.try_clone().map_err(|err| reason(&err))?;
64        Ok(Self {
65            reader: BufReader::new(stream),
66            writer,
67        })
68    }
69
70    /// Sends `command` (`LIST UPS`) and returns the lines between its
71    /// `BEGIN` and `END` lines.
72    fn request(&mut self, command: &str) -> Result<Vec<String>, RequestError> {
73        self.writer
74            .write_all(format!("{command}\n").as_bytes())
75            .map_err(|err| RequestError::Io(reason(&err)))?;
76        let first = self.line()?;
77        if let Some(error) = first.strip_prefix("ERR ") {
78            return Err(RequestError::Nut(error.trim().to_owned()));
79        }
80        if first != format!("BEGIN {command}") {
81            return Err(RequestError::Io(format!("unexpected answer: {first}")));
82        }
83        let end = format!("END {command}");
84        let mut lines = Vec::new();
85        loop {
86            let line = self.line()?;
87            if line == end {
88                return Ok(lines);
89            }
90            if lines.len() >= MAX_LINES {
91                return Err(RequestError::Io("answer too long".to_owned()));
92            }
93            lines.push(line);
94        }
95    }
96
97    /// One line of the answer, without its line break.
98    fn line(&mut self) -> Result<String, RequestError> {
99        let mut line = String::new();
100        let read = (&mut self.reader)
101            .take(MAX_LINE)
102            .read_line(&mut line)
103            .map_err(|err| RequestError::Io(reason(&err)))?;
104        if read == 0 {
105            return Err(RequestError::Io("upsd closed the connection".to_owned()));
106        }
107        Ok(line.trim_end_matches(['\r', '\n']).to_owned())
108    }
109}
110
111/// Polls upsd, keeping one connection open, and backing off while it is
112/// unreachable.
113pub struct NutReader {
114    /// `host:port`; empty when disabled.
115    address: String,
116    connection: Option<Connection>,
117    /// When to ask upsd again; the last readings are returned until then.
118    next_poll: Option<Instant>,
119    last: Vec<UpsStats>,
120}
121
122impl NutReader {
123    /// Reads from upsd at `address` ("host" or "host:port", IPv6 addresses
124    /// in brackets); an empty address disables it.
125    pub fn new(address: &str) -> Self {
126        Self {
127            address: with_port(address.trim()),
128            connection: None,
129            next_poll: None,
130            last: Vec::new(),
131        }
132    }
133
134    /// The UPSes upsd knows, asked again at most every [`POLL_EVERY`].
135    /// Empty when disabled or when upsd was never reached; UPSes read before
136    /// stay listed, with an error, while upsd does not answer.
137    pub fn read(&mut self, now: Instant) -> Vec<UpsStats> {
138        if self.address.is_empty() || self.next_poll.is_some_and(|at| now < at) {
139            return self.last.clone();
140        }
141        match self.query() {
142            Ok(list) => {
143                self.last = list;
144                self.next_poll = Some(now + POLL_EVERY);
145            }
146            Err(err) => {
147                self.connection = None;
148                if self.last.is_empty() {
149                    self.next_poll = Some(now + RETRY_AFTER);
150                } else {
151                    // Known UPSes stay, without readings: losing upsd means
152                    // losing the shutdown on a power failure too.
153                    self.last = self
154                        .last
155                        .iter()
156                        .map(|ups| UpsStats {
157                            name: ups.name.clone(),
158                            model: ups.model.clone(),
159                            error: Some(format!("upsd not reachable: {err}")),
160                            ..Default::default()
161                        })
162                        .collect();
163                    self.next_poll = Some(now + RETRY_LOST_AFTER);
164                }
165            }
166        }
167        self.last.clone()
168    }
169
170    /// Asks for every UPS and its variables; a connection kept from an
171    /// earlier poll may have been closed by upsd, so a failure on it is
172    /// tried once more on a new one.
173    fn query(&mut self) -> Result<Vec<UpsStats>, String> {
174        let reused = self.connection.is_some();
175        match self.query_once() {
176            Err(_) if reused => {
177                self.connection = None;
178                self.query_once()
179            }
180            result => result,
181        }
182    }
183
184    fn query_once(&mut self) -> Result<Vec<UpsStats>, String> {
185        let connection = match &mut self.connection {
186            Some(connection) => connection,
187            None => self.connection.insert(Connection::open(&self.address)?),
188        };
189        let names = match connection.request("LIST UPS") {
190            Ok(lines) => parse_list_ups(&lines),
191            Err(RequestError::Nut(error)) => return Err(error),
192            Err(RequestError::Io(error)) => return Err(error),
193        };
194        let mut list = Vec::new();
195        for (name, description) in names {
196            let ups = match connection.request(&format!("LIST VAR {name}")) {
197                Ok(lines) => ups_from_vars(&name, &description, &parse_vars(&name, &lines)),
198                // upsd is fine, this UPS is not (its driver lost it).
199                Err(RequestError::Nut(error)) => UpsStats {
200                    name,
201                    model: (!description.is_empty()).then_some(description),
202                    error: Some(error_text(&error)),
203                    ..Default::default()
204                },
205                Err(RequestError::Io(error)) => return Err(error),
206            };
207            list.push(ups);
208        }
209        Ok(list)
210    }
211}
212
213/// An I/O error for an alert: "connection refused", without the "(os
214/// error 111)" that follows it.
215fn reason(err: &std::io::Error) -> String {
216    let text = err.to_string();
217    let text = match text.rfind(" (os error ") {
218        Some(at) => &text[..at],
219        None => &text,
220    };
221    let mut chars = text.chars();
222    match chars.next() {
223        Some(first) => first.to_lowercase().chain(chars).collect(),
224        None => String::new(),
225    }
226}
227
228/// `address` with NUT's port if it has none.
229fn with_port(address: &str) -> String {
230    if address.is_empty() {
231        return String::new();
232    }
233    let has_port = match address.rsplit_once(':') {
234        // "[::1]:3493", "nas:3493"; a bare IPv6 address has more colons.
235        Some((host, port)) => {
236            port.parse::<u16>().is_ok() && (host.ends_with(']') || !host.contains(':'))
237        }
238        None => false,
239    };
240    if has_port {
241        address.to_owned()
242    } else if address.contains(':') && !address.starts_with('[') {
243        format!("[{address}]:{DEFAULT_PORT}")
244    } else {
245        format!("{address}:{DEFAULT_PORT}")
246    }
247}
248
249/// upsd's error names in words: `DATA-STALE` is "data stale".
250fn error_text(error: &str) -> String {
251    error
252        .split_whitespace()
253        .next()
254        .unwrap_or(error)
255        .to_ascii_lowercase()
256        .replace('-', " ")
257}
258
259/// Splits a line into words, a quoted one (`"Rack \"B\""`) being one word
260/// without its quotes and escapes.
261fn words(line: &str) -> Vec<String> {
262    let mut words = Vec::new();
263    let mut chars = line.chars().peekable();
264    while let Some(&c) = chars.peek() {
265        if c == ' ' {
266            chars.next();
267            continue;
268        }
269        let mut word = String::new();
270        if c == '"' {
271            chars.next();
272            while let Some(c) = chars.next() {
273                match c {
274                    '\\' => word.extend(chars.next()),
275                    '"' => break,
276                    c => word.push(c),
277                }
278            }
279        } else {
280            while let Some(&c) = chars.peek() {
281                if c == ' ' {
282                    break;
283                }
284                word.push(c);
285                chars.next();
286            }
287        }
288        words.push(word);
289    }
290    words
291}
292
293/// The UPSes of a `LIST UPS` answer: name and description.
294fn parse_list_ups(lines: &[String]) -> Vec<(String, String)> {
295    lines
296        .iter()
297        .filter_map(|line| match words(line).as_slice() {
298            [kind, name, rest @ ..] if kind == "UPS" => {
299                Some((name.clone(), rest.first().cloned().unwrap_or_default()))
300            }
301            _ => None,
302        })
303        .collect()
304}
305
306/// The variables of a `LIST VAR <ups>` answer.
307fn parse_vars(ups: &str, lines: &[String]) -> HashMap<String, String> {
308    lines
309        .iter()
310        .filter_map(|line| match words(line).as_slice() {
311            [kind, name, var, value] if kind == "VAR" && name == ups => {
312                Some((var.clone(), value.clone()))
313            }
314            _ => None,
315        })
316        .collect()
317}
318
319/// What rigtop shows of a UPS's variables.
320fn ups_from_vars(name: &str, description: &str, vars: &HashMap<String, String>) -> UpsStats {
321    let number = |key: &str| vars.get(key).and_then(|v| v.trim().parse::<f32>().ok());
322    let text = |key: &str| {
323        vars.get(key)
324            .map(|v| v.trim())
325            .filter(|v| !v.is_empty() && !v.eq_ignore_ascii_case("unknown"))
326    };
327    // "APC Back-UPS ES 700"; models often repeat the maker's name.
328    let model = match (
329        text("device.mfr").or(text("ups.mfr")),
330        text("device.model").or(text("ups.model")),
331    ) {
332        (Some(maker), Some(model)) if model.starts_with(maker) => Some(model.to_owned()),
333        (Some(maker), Some(model)) => Some(format!("{maker} {model}")),
334        (None, Some(model)) => Some(model.to_owned()),
335        _ => (!description.is_empty()).then(|| description.to_owned()),
336    };
337    // Many UPSes report only the load and their nominal power.
338    let power_w = number("ups.realpower").or_else(|| {
339        let load = number("ups.load")?;
340        let nominal = number("ups.realpower.nominal")?;
341        Some(load * nominal / 100.0)
342    });
343    UpsStats {
344        name: name.to_owned(),
345        model,
346        status: vars
347            .get("ups.status")
348            .map(|s| s.split_whitespace().map(str::to_owned).collect())
349            .unwrap_or_default(),
350        charge_percent: number("battery.charge"),
351        runtime_secs: number("battery.runtime")
352            .filter(|s| *s >= 0.0)
353            .map(|s| s as u64),
354        load_percent: number("ups.load"),
355        power_w,
356        input_voltage: number("input.voltage"),
357        error: None,
358    }
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364    use std::net::TcpListener;
365    use std::thread;
366
367    fn lines(text: &str) -> Vec<String> {
368        text.lines().map(str::to_owned).collect()
369    }
370
371    #[test]
372    fn parses_ups_lists_and_variables() {
373        assert_eq!(
374            words(r#"UPS rack "Rack \"B\" UPS""#),
375            ["UPS", "rack", r#"Rack "B" UPS"#]
376        );
377        assert_eq!(
378            parse_list_ups(&lines("UPS ups \"Test UPS\"\nUPS rack \"\"")),
379            [
380                ("ups".to_owned(), "Test UPS".to_owned()),
381                ("rack".to_owned(), String::new())
382            ]
383        );
384        let vars = parse_vars(
385            "ups",
386            &lines(
387                "VAR ups battery.charge \"100\"\n\
388                 VAR ups battery.runtime \"1830\"\n\
389                 VAR ups device.mfr \"APC\"\n\
390                 VAR ups device.model \"Back-UPS ES 700\"\n\
391                 VAR ups input.voltage \"230.0\"\n\
392                 VAR ups ups.load \"18\"\n\
393                 VAR ups ups.realpower.nominal \"405\"\n\
394                 VAR ups ups.status \"OL CHRG\"\n\
395                 VAR other ups.status \"OB\"",
396            ),
397        );
398        let ups = ups_from_vars("ups", "Test UPS", &vars);
399        assert_eq!(ups.model.as_deref(), Some("APC Back-UPS ES 700"));
400        assert_eq!(ups.status, ["OL", "CHRG"]);
401        assert_eq!(ups.charge_percent, Some(100.0));
402        assert_eq!(ups.runtime_secs, Some(1830));
403        assert_eq!(ups.load_percent, Some(18.0));
404        // 18% of 405 W.
405        assert!((ups.power_w.unwrap() - 72.9).abs() < 0.01);
406        assert_eq!(ups.input_voltage, Some(230.0));
407
408        // A model that names its maker, real power reported.
409        let vars: HashMap<String, String> = [
410            ("ups.mfr", "CyberPower"),
411            ("ups.model", "CyberPower OR1500"),
412            ("ups.realpower", "260"),
413        ]
414        .into_iter()
415        .map(|(k, v)| (k.to_owned(), v.to_owned()))
416        .collect();
417        let ups = ups_from_vars("rack", "", &vars);
418        assert_eq!(ups.model.as_deref(), Some("CyberPower OR1500"));
419        assert_eq!(ups.power_w, Some(260.0));
420    }
421
422    #[test]
423    fn addresses_get_nut_port() {
424        assert_eq!(with_port(""), "");
425        assert_eq!(with_port("127.0.0.1"), "127.0.0.1:3493");
426        assert_eq!(with_port("nas:4000"), "nas:4000");
427        assert_eq!(with_port("::1"), "[::1]:3493");
428        assert_eq!(with_port("[::1]:4000"), "[::1]:4000");
429        assert_eq!(error_text("DRIVER-NOT-CONNECTED"), "driver not connected");
430        let refused = std::io::Error::from_raw_os_error(111);
431        if cfg!(target_os = "linux") {
432            assert_eq!(reason(&refused), "connection refused");
433        }
434    }
435
436    /// A fake upsd answering `answers` in turn, one per request line, on one
437    /// connection; returns its address.
438    fn fake_upsd(answers: Vec<&'static str>) -> String {
439        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
440        let address = listener.local_addr().unwrap().to_string();
441        thread::spawn(move || {
442            let (stream, _) = listener.accept().unwrap();
443            let mut reader = BufReader::new(stream.try_clone().unwrap());
444            let mut writer = stream;
445            for answer in answers {
446                let mut request = String::new();
447                if reader.read_line(&mut request).unwrap_or(0) == 0 {
448                    return;
449                }
450                writer.write_all(answer.as_bytes()).unwrap();
451            }
452        });
453        address
454    }
455
456    #[test]
457    fn reads_every_ups_and_keeps_a_lost_driver_listed() {
458        let address = fake_upsd(vec![
459            "BEGIN LIST UPS\nUPS ups \"Test UPS\"\nUPS rack \"Rack UPS\"\nEND LIST UPS\n",
460            "BEGIN LIST VAR ups\nVAR ups ups.status \"OB DISCHRG\"\n\
461             VAR ups battery.charge \"72\"\nEND LIST VAR ups\n",
462            "ERR DRIVER-NOT-CONNECTED\n",
463        ]);
464        let mut reader = NutReader::new(&address);
465        let list = reader.read(Instant::now());
466        assert_eq!(list.len(), 2);
467        assert_eq!(list[0].status, ["OB", "DISCHRG"]);
468        assert_eq!(list[0].charge_percent, Some(72.0));
469        assert_eq!(list[1].name, "rack");
470        assert_eq!(list[1].model.as_deref(), Some("Rack UPS"));
471        assert_eq!(list[1].error.as_deref(), Some("driver not connected"));
472
473        // upsd goes away: the UPSes stay listed, without readings.
474        let later = Instant::now() + POLL_EVERY;
475        let list = reader.read(later);
476        assert_eq!(list.len(), 2);
477        assert!(list.iter().all(|u| u.status.is_empty()));
478        assert!(
479            list[0]
480                .error
481                .as_deref()
482                .is_some_and(|e| e.starts_with("upsd not reachable: ")),
483            "{list:?}"
484        );
485    }
486
487    #[test]
488    fn no_upsd_means_no_ups() {
489        // A port nothing listens on: bind one, then free it.
490        let address = TcpListener::bind("127.0.0.1:0")
491            .unwrap()
492            .local_addr()
493            .unwrap()
494            .to_string();
495        let mut reader = NutReader::new(&address);
496        assert!(reader.read(Instant::now()).is_empty());
497        assert!(NutReader::new("").read(Instant::now()).is_empty());
498    }
499}
rigtop_core/nut.rs

rigtop_core/
nut.rs
