Skip to main content

max / makenotwork

11.0 KB · 287 lines History Blame Raw
1 //! Reading a producer's own observation store.
2 //!
3 //! This is the one place magicmirror renders something other than the
4 //! `ops-status` contract, and the break is deliberate and kept narrow (infra
5 //! `7bafb5dd`). A store is opened **read-only**, queried for exactly the series
6 //! the config names, and never written to. Nothing here learns what a series
7 //! means: the config says that, because the data cannot.
8 //!
9 //! ## Why the config carries the meaning
10 //!
11 //! witchbroom's store is `(series, labels, value, at)` with an `f64`, a shape
12 //! its own module header defends: a new metric is a new string rather than a
13 //! migration, and the accepted cost is that the schema cannot say what a number
14 //! is or what unit it is in. So a viewer has two options. It can render every
15 //! series it finds, unlabelled, which is a generic table browser; or the
16 //! operator can name the series worth watching and say what each one is. The
17 //! second was ruled. A series nobody named is not shown, and that silence is
18 //! the choice, not an oversight.
19 //!
20 //! ## Why it is polled rather than read on draw
21 //!
22 //! `render` is a pure function of the model, the theme and the clock, and it is
23 //! what makes the whole surface testable without a daemon or a database. A
24 //! query inside it would end that. So a store is read on its own slow interval
25 //! into the model, exactly as a source is polled, and the tab draws what was
26 //! last read.
27
28 use std::path::Path;
29
30 use chrono::{DateTime, TimeZone, Utc};
31 use rusqlite::{Connection, OpenFlags};
32 use tokio::sync::mpsc;
33
34 use crate::config::{Config, Store};
35
36 /// One store's read, tagged with which store it came from.
37 pub(crate) struct Update {
38 pub index: usize,
39 pub at: DateTime<Utc>,
40 pub result: Result<Vec<Reading>, String>,
41 }
42
43 /// The newest observation of one series under one label set.
44 ///
45 /// `labels` is the store's own canonical JSON text, carried through verbatim.
46 /// magicmirror does not parse it: the labels are a producer's dimensions and
47 /// what they mean is the producer's business, so they are shown as written
48 /// rather than interpreted into columns this crate would have to invent.
49 #[derive(Debug, Clone, PartialEq)]
50 pub(crate) struct Reading {
51 pub series: String,
52 pub labels: String,
53 pub value: f64,
54 pub at: DateTime<Utc>,
55 }
56
57 /// Spawn one reading task per configured store. Returns the receiving end.
58 pub(crate) fn spawn_all(stores: &[Store]) -> mpsc::Receiver<Update> {
59 let (tx, rx) = mpsc::channel(stores.len().max(1) * 4);
60 for (index, store) in stores.iter().enumerate() {
61 tokio::spawn(read_forever(index, store.clone(), tx.clone()));
62 }
63 rx
64 }
65
66 async fn read_forever(index: usize, store: Store, tx: mpsc::Sender<Update>) {
67 let mut ticker = tokio::time::interval(Config::store_interval(&store));
68 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
69 loop {
70 ticker.tick().await;
71 // The file is opened per read rather than held. A batch producer
72 // replaces its database wholesale often enough (a restore, a prune, a
73 // fresh checkout) that a connection held across hours can end up
74 // reading a file nobody writes to any more, which looks exactly like a
75 // producer that stopped.
76 let store = store.clone();
77 let result = tokio::task::spawn_blocking(move || read(&store))
78 .await
79 .unwrap_or_else(|e| Err(format!("read task: {e}")));
80 if tx
81 .send(Update {
82 index,
83 at: Utc::now(),
84 result,
85 })
86 .await
87 .is_err()
88 {
89 return; // UI is gone
90 }
91 }
92 }
93
94 /// Read every configured series out of one store.
95 ///
96 /// One connection, one query per series, in the order the config names them.
97 /// The order matters: it is the operator's own ranking of what is worth looking
98 /// at, and re-sorting it here would replace a decision with an alphabet.
99 fn read(store: &Store) -> Result<Vec<Reading>, String> {
100 let conn = open(&store.path)?;
101 let mut readings = Vec::new();
102 for series in &store.series {
103 readings.extend(latest_per_labels(&conn, &series.name).map_err(|e| short(&e))?);
104 }
105 Ok(readings)
106 }
107
108 /// Open a store read-only.
109 ///
110 /// `SQLITE_OPEN_READ_ONLY` without `CREATE`, so a path with a typo in it fails
111 /// as "no such file" rather than quietly creating an empty database that then
112 /// reads as a producer which has never recorded anything. That distinction is
113 /// the whole difference between a broken config and a quiet night.
114 fn open(path: &Path) -> Result<Connection, String> {
115 Connection::open_with_flags(
116 path,
117 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI,
118 )
119 .map_err(|e| short(&e))
120 }
121
122 /// The newest value of one series, once per distinct label set.
123 ///
124 /// A bare `value`/`at` beside `MAX(at)` is SQLite's documented bare-column
125 /// behaviour: with a single aggregate the row the aggregate came from is the
126 /// row the bare columns are taken from. It is exactly the "latest per group"
127 /// this needs and costs no window function.
128 fn latest_per_labels(conn: &Connection, series: &str) -> rusqlite::Result<Vec<Reading>> {
129 let mut stmt = conn.prepare(
130 "SELECT labels, value, MAX(at)
131 FROM observations
132 WHERE series = ?1
133 GROUP BY labels
134 ORDER BY labels",
135 )?;
136 let rows = stmt.query_map([series], |row| {
137 Ok(Reading {
138 series: series.to_string(),
139 labels: row.get(0)?,
140 value: row.get(1)?,
141 // Unix seconds, per the producer's schema. A timestamp outside what
142 // chrono can represent is a corrupt row, not a reason to lose the
143 // rest of the series, so it reads as the epoch and shows as very
144 // old rather than taking the read down.
145 at: Utc
146 .timestamp_opt(row.get::<_, i64>(2)?, 0)
147 .single()
148 .unwrap_or_else(|| Utc.timestamp_nanos(0)),
149 })
150 })?;
151 rows.collect()
152 }
153
154 /// A one-line reason, because the store tab has one column for it.
155 fn short(error: &dyn std::error::Error) -> String {
156 let text = error.to_string();
157 match text.split_once('\n') {
158 Some((first, _)) => first.to_string(),
159 None => text,
160 }
161 }
162
163 #[cfg(test)]
164 mod tests {
165 use super::*;
166 use crate::config::Series;
167
168 /// A store file with the producer's schema and the given observations.
169 fn seeded(rows: &[(&str, &str, f64, i64)]) -> (tempfile::TempDir, std::path::PathBuf) {
170 let dir = tempfile::tempdir().unwrap();
171 let path = dir.path().join("observations.db");
172 let conn = Connection::open(&path).unwrap();
173 conn.execute_batch(
174 "CREATE TABLE observations (
175 id INTEGER PRIMARY KEY,
176 series TEXT NOT NULL,
177 labels TEXT NOT NULL,
178 value REAL NOT NULL,
179 at INTEGER NOT NULL
180 );",
181 )
182 .unwrap();
183 for (series, labels, value, at) in rows {
184 conn.execute(
185 "INSERT INTO observations (series, labels, value, at) VALUES (?1, ?2, ?3, ?4)",
186 rusqlite::params![series, labels, value, at],
187 )
188 .unwrap();
189 }
190 (dir, path)
191 }
192
193 fn store(path: &Path, series: &[&str]) -> Store {
194 Store {
195 name: "witchbroom".into(),
196 path: path.to_path_buf(),
197 poll_secs: 60,
198 series: series
199 .iter()
200 .map(|s| Series {
201 name: (*s).to_string(),
202 label: format!("label for {s}"),
203 unit: Some("edges".into()),
204 })
205 .collect(),
206 }
207 }
208
209 #[test]
210 fn a_series_reads_its_newest_value_per_label_set() {
211 let (_dir, path) = seeded(&[
212 ("soak.coverage_edges", r#"{"repo":"a"}"#, 100.0, 1_000),
213 ("soak.coverage_edges", r#"{"repo":"a"}"#, 200.0, 2_000),
214 ("soak.coverage_edges", r#"{"repo":"b"}"#, 50.0, 1_500),
215 ]);
216 let readings = read(&store(&path, &["soak.coverage_edges"])).unwrap();
217 assert_eq!(readings.len(), 2, "one row per label set, not per sample");
218 assert_eq!(readings[0].labels, r#"{"repo":"a"}"#);
219 assert!(
220 (readings[0].value - 200.0).abs() < f64::EPSILON,
221 "the newest, not the first"
222 );
223 assert_eq!(readings[0].at.timestamp(), 2_000);
224 assert!((readings[1].value - 50.0).abs() < f64::EPSILON);
225 }
226
227 #[test]
228 fn a_series_nobody_named_is_not_read() {
229 // The ruling's accepted cost, asserted: the store holds it and the
230 // config does not name it, so it does not exist as far as this goes.
231 let (_dir, path) = seeded(&[
232 ("soak.coverage_edges", "{}", 1.0, 10),
233 ("cache.size_bytes", "{}", 999.0, 10),
234 ]);
235 let readings = read(&store(&path, &["soak.coverage_edges"])).unwrap();
236 assert_eq!(readings.len(), 1);
237 assert_eq!(readings[0].series, "soak.coverage_edges");
238 }
239
240 #[test]
241 fn a_named_series_with_no_observations_reads_as_nothing_not_an_error() {
242 let (_dir, path) = seeded(&[("cache.size_bytes", "{}", 1.0, 10)]);
243 let readings = read(&store(&path, &["soak.coverage_edges"])).unwrap();
244 assert!(readings.is_empty());
245 }
246
247 #[test]
248 fn a_missing_store_is_a_short_reason_rather_than_a_panic() {
249 let dir = tempfile::tempdir().unwrap();
250 let err = read(&store(&dir.path().join("nope.db"), &["x"])).unwrap_err();
251 assert!(!err.is_empty());
252 assert!(!err.contains('\n'), "one line for one column: {err}");
253 }
254
255 #[test]
256 fn reading_never_creates_the_file_it_could_not_find() {
257 // A typo in a path must stay a visible error. Creating an empty
258 // database here would turn it into a producer that reads as having
259 // recorded nothing, which is a very different night.
260 let dir = tempfile::tempdir().unwrap();
261 let path = dir.path().join("nope.db");
262 assert!(read(&store(&path, &["x"])).is_err());
263 assert!(!path.exists(), "a read must not create a store");
264 }
265
266 #[test]
267 fn a_store_missing_the_producers_table_says_so_rather_than_reading_empty() {
268 let dir = tempfile::tempdir().unwrap();
269 let path = dir.path().join("wrong.db");
270 Connection::open(&path)
271 .unwrap()
272 .execute_batch("CREATE TABLE something_else (x INTEGER);")
273 .unwrap();
274 let err = read(&store(&path, &["x"])).unwrap_err();
275 assert!(err.contains("observations"), "{err}");
276 }
277
278 #[test]
279 fn the_series_come_back_in_the_order_the_config_names_them() {
280 // The operator's ranking, not an alphabet.
281 let (_dir, path) = seeded(&[("zzz", "{}", 1.0, 10), ("aaa", "{}", 2.0, 10)]);
282 let readings = read(&store(&path, &["zzz", "aaa"])).unwrap();
283 let names: Vec<&str> = readings.iter().map(|r| r.series.as_str()).collect();
284 assert_eq!(names, vec!["zzz", "aaa"]);
285 }
286 }
287