Skip to main content

max / makenotwork

19.3 KB · 559 lines History Blame Raw
1 //! Read-only MCP tools for orienting on what is live.
2 //!
3 //! These answer the question that otherwise costs an ssh and a curl: what is
4 //! running where, what is wrong with it, and how far behind is it. Nothing here
5 //! promotes, deploys, or closes an incident. Acting stays with Sando and the
6 //! CLI, so a session can read production without being able to change it.
7 //!
8 //! # Local and remote read the same shape
9 //!
10 //! Every tool takes an optional `instance`. Omitted, it reads this machine's
11 //! database directly, which works whether or not a `pom serve` daemon is up
12 //! here. Named, it resolves a configured peer and reads that instance's HTTP
13 //! API over the tailnet, which is the only way to see checks that are local to
14 //! *that* host: systemd units and backup freshness on the production box are
15 //! not observable from here at all.
16 //!
17 //! Both paths land on the same types, `ops_status::Payload` for the status
18 //! tools and [`VersionRow`] for versions, so the formatting below is written
19 //! once and never branches on where the data came from.
20
21 use std::fmt::Write as _;
22
23 use ops_status::{Node, Payload, Status};
24 use schemars::JsonSchema;
25 use serde::Deserialize;
26 use tracing::instrument;
27
28 use crate::api;
29 use crate::error::{PomError, Result};
30 use crate::types::VersionRow;
31 use crate::versions;
32
33 use super::PomServer;
34
35 /// How long to wait on a peer before giving up. A session is waiting on this
36 /// answer, so a hung instance should say so quickly rather than stall the turn.
37 const REMOTE_TIMEOUT_SECS: u64 = 10;
38
39 #[derive(Debug, Deserialize, JsonSchema)]
40 pub struct InstanceParams {
41 /// PoM instance to read: omit for this machine, or name a configured peer
42 /// (e.g. the production instance) to read that host's view.
43 pub instance: Option<String>,
44 }
45
46 #[derive(Debug, Deserialize, JsonSchema)]
47 pub struct TargetInstanceParams {
48 /// Target name, as configured (e.g. "mnw").
49 pub target: String,
50 /// PoM instance to read: omit for this machine, or name a configured peer.
51 pub instance: Option<String>,
52 }
53
54 #[derive(Debug, Deserialize, JsonSchema)]
55 pub struct TrendsParams {
56 /// Target name, as configured (e.g. "mnw").
57 pub target: String,
58 /// Window to report, in hours (default 24).
59 pub hours: Option<u64>,
60 /// Width of each latency bucket, in minutes (default 60).
61 pub bucket_minutes: Option<u64>,
62 /// PoM instance to read: omit for this machine, or name a configured peer.
63 pub instance: Option<String>,
64 }
65
66 /// Which instance a tool call is reading.
67 enum Source {
68 /// This machine's database, read without going through HTTP.
69 Local,
70 /// A configured peer's API.
71 Remote {
72 name: String,
73 base_url: String,
74 token: Option<String>,
75 },
76 }
77
78 impl PomServer {
79 /// Resolve an `instance` parameter to the source to read.
80 ///
81 /// An unknown name lists what is configured rather than just refusing: the
82 /// caller cannot see pom.toml, so the names are otherwise unguessable.
83 fn source(&self, instance: Option<&str>) -> Result<Source> {
84 match instance {
85 None | Some("local" | "") => Ok(Source::Local),
86 Some(name) => match self.config.peers.get(name) {
87 Some(peer) => Ok(Source::Remote {
88 name: name.to_string(),
89 base_url: crate::peer::peer_base_url(&peer.address),
90 token: peer.token.clone(),
91 }),
92 None => {
93 let mut known: Vec<&str> =
94 self.config.peers.keys().map(String::as_str).collect();
95 known.sort_unstable();
96 Err(PomError::Config(format!(
97 "unknown instance: {name}. Configured peers: {}. Omit `instance` to read \
98 this machine.",
99 if known.is_empty() {
100 "none".to_string()
101 } else {
102 known.join(", ")
103 }
104 )))
105 }
106 },
107 }
108 }
109
110 /// GET a path on a remote instance, with the peer's bearer token.
111 async fn get_remote(
112 &self,
113 base_url: &str,
114 token: Option<&str>,
115 path: &str,
116 ) -> Result<serde_json::Value> {
117 let url = format!("{base_url}{path}");
118 let client = crate::tls::https_client_builder()
119 .timeout(std::time::Duration::from_secs(REMOTE_TIMEOUT_SECS))
120 .build()?;
121
122 let mut request = client.get(&url);
123 if let Some(token) = token {
124 request = request.bearer_auth(token);
125 }
126
127 let response = request
128 .send()
129 .await
130 .map_err(|e| PomError::Config(format!("could not reach {url}: {e}")))?;
131
132 let status = response.status();
133 if !status.is_success() {
134 // 401 here means the peer's token is wrong or missing in pom.toml,
135 // which is a config problem on this side and worth naming as one.
136 return Err(PomError::Config(format!(
137 "{url} returned {status}{}",
138 if status.as_u16() == 401 {
139 " (check this peer's token in pom.toml)"
140 } else {
141 ""
142 }
143 )));
144 }
145
146 Ok(response.json().await?)
147 }
148
149 /// The status payload for whichever instance was asked for.
150 async fn payload_for(&self, instance: Option<&str>) -> Result<(String, Payload)> {
151 match self.source(instance)? {
152 Source::Local => Ok((
153 "local".to_string(),
154 api::status_payload(&self.pool, &self.config).await,
155 )),
156 Source::Remote {
157 name,
158 base_url,
159 token,
160 } => {
161 let value = self
162 .get_remote(&base_url, token.as_deref(), "/status.json")
163 .await?;
164 let payload = serde_json::from_value(value)?;
165 Ok((name, payload))
166 }
167 }
168 }
169
170 #[instrument(skip_all)]
171 pub async fn status_table_impl(&self, params: InstanceParams) -> Result<String> {
172 let (instance, payload) = self.payload_for(params.instance.as_deref()).await?;
173 Ok(format_status_table(&instance, &payload))
174 }
175
176 #[instrument(skip_all)]
177 pub async fn target_status_impl(&self, params: TargetInstanceParams) -> Result<String> {
178 let (instance, payload) = self.payload_for(params.instance.as_deref()).await?;
179 let wanted = format!("target:{}", params.target);
180
181 let Some(node) = payload.nodes.iter().find(|n| n.id == wanted) else {
182 let names: Vec<&str> = payload
183 .nodes
184 .iter()
185 .filter_map(|n| n.id.strip_prefix("target:"))
186 .collect();
187 return Ok(format!(
188 "Unknown target: {} on instance {instance}. Known targets: {}",
189 params.target,
190 if names.is_empty() {
191 "none".to_string()
192 } else {
193 names.join(", ")
194 }
195 ));
196 };
197
198 Ok(format_target_detail(&instance, node))
199 }
200
201 #[instrument(skip_all)]
202 pub async fn incidents_impl(&self, params: InstanceParams) -> Result<String> {
203 let (instance, payload) = self.payload_for(params.instance.as_deref()).await?;
204 Ok(format_incidents(&instance, &payload))
205 }
206
207 #[instrument(skip_all)]
208 pub async fn versions_impl(&self, params: InstanceParams) -> Result<String> {
209 let (instance, rows) = match self.source(params.instance.as_deref())? {
210 Source::Local => (
211 "local".to_string(),
212 versions::collect(&self.pool, &self.config).await?,
213 ),
214 Source::Remote {
215 name,
216 base_url,
217 token,
218 } => {
219 let value = self
220 .get_remote(&base_url, token.as_deref(), "/api/versions")
221 .await?;
222 let rows: Vec<VersionRow> = serde_json::from_value(value)?;
223 (name, rows)
224 }
225 };
226
227 let mut out = format!("# Versions on {instance}\n\n");
228 out.push_str(&crate::display::format_versions(&rows));
229 // The count is taken against a checkout on the instance being read, so
230 // a remote answer says what is live there, not how far behind here.
231 if instance != "local" {
232 out.push_str(
233 "\nBEHIND is measured against a checkout on that host, and is blank where it has \
234 no repo.\n",
235 );
236 }
237 Ok(out)
238 }
239
240 #[instrument(skip_all)]
241 pub async fn trends_impl(&self, params: TrendsParams) -> Result<String> {
242 let hours = params.hours.unwrap_or(24);
243 let bucket_minutes = params.bucket_minutes.unwrap_or(60);
244
245 let (instance, trends) = match self.source(params.instance.as_deref())? {
246 Source::Local => {
247 if self.config.get_target(&params.target).is_none() {
248 return Ok(format!("Unknown target: {}", params.target));
249 }
250 (
251 "local".to_string(),
252 api::build_trends(&self.pool, &params.target, hours, bucket_minutes).await,
253 )
254 }
255 Source::Remote {
256 name,
257 base_url,
258 token,
259 } => {
260 // The target name goes into a URL path. Config keys are plain
261 // identifiers, so anything else is rejected here rather than
262 // escaped: a `../` or a `?` would address a different endpoint.
263 if !is_config_name(&params.target) {
264 return Ok(format!("Unusable target name: {:?}", params.target));
265 }
266 let path = format!(
267 "/api/trends/{}?hours={hours}&bucket_minutes={bucket_minutes}",
268 params.target
269 );
270 let value = self.get_remote(&base_url, token.as_deref(), &path).await?;
271 (name, serde_json::from_value(value)?)
272 }
273 };
274
275 Ok(format_trends(&instance, &trends))
276 }
277 }
278
279 /// Every target on one line, worst-first, with the reason it is not green.
280 fn format_status_table(instance: &str, payload: &Payload) -> String {
281 if payload.nodes.is_empty() {
282 return format!("No targets configured on {instance}.\n");
283 }
284
285 let mut nodes: Vec<&Node> = payload.nodes.iter().collect();
286 // Worst first: the point of the table is that the problem is on line one.
287 nodes.sort_by(|a, b| b.status.cmp(&a.status).then(a.id.cmp(&b.id)));
288
289 let rows: Vec<[String; 5]> = nodes
290 .iter()
291 .map(|n| {
292 [
293 n.id.strip_prefix("target:").unwrap_or(&n.id).to_string(),
294 status_word(n.status).to_string(),
295 field_text(n, "version"),
296 field_text(n, "checked"),
297 worst_condition_summary(n),
298 ]
299 })
300 .collect();
301
302 const HEADERS: [&str; 5] = ["TARGET", "STATUS", "VERSION", "CHECKED", "WHY"];
303 let widths: Vec<usize> = (0..HEADERS.len())
304 .map(|i| {
305 rows.iter()
306 .map(|r| r[i].chars().count())
307 .chain(std::iter::once(HEADERS[i].len()))
308 .max()
309 .unwrap_or(0)
310 })
311 .collect();
312
313 let mut out = format!(
314 "# {} on {instance} ({} target{}, worst: {})\n\n",
315 payload.source,
316 nodes.len(),
317 if nodes.len() == 1 { "" } else { "s" },
318 status_word(payload.worst_status()),
319 );
320 write_row(&mut out, &HEADERS.map(String::from), &widths);
321 for row in &rows {
322 write_row(&mut out, row, &widths);
323 }
324 let _ = write!(
325 out,
326 "\nGenerated {}. Use target_status for the full condition list.\n",
327 payload.generated_at.format("%Y-%m-%d %H:%M UTC")
328 );
329 out
330 }
331
332 /// One target in full: its fields, then every condition with its detail.
333 fn format_target_detail(instance: &str, node: &Node) -> String {
334 let name = node.id.strip_prefix("target:").unwrap_or(&node.id);
335 let mut out = format!(
336 "# {name} ({}) on {instance}: {}\n\n",
337 node.label,
338 status_word(node.status)
339 );
340
341 if !node.fields.is_empty() {
342 for field in &node.fields {
343 let _ = writeln!(out, "{}: {}", field.label, value_text(&field.value));
344 }
345 out.push('\n');
346 }
347
348 if node.conditions.is_empty() {
349 out.push_str("No conditions recorded.\n");
350 return out;
351 }
352
353 // Worst first here too, so a failing check is never below three green ones.
354 let mut conditions: Vec<&ops_status::Condition> = node.conditions.iter().collect();
355 conditions.sort_by_key(|c| std::cmp::Reverse(c.status));
356
357 for condition in conditions {
358 let _ = write!(
359 out,
360 "[{}] {}",
361 status_word(condition.status),
362 condition.condition_type
363 );
364 if let Some(since) = condition.since {
365 let _ = write!(out, " (since {})", since.format("%Y-%m-%d %H:%M UTC"));
366 }
367 if let Some(detail) = &condition.detail {
368 let _ = write!(out, ": {}", scrub(detail));
369 }
370 out.push('\n');
371 }
372 out
373 }
374
375 /// Open incidents across every target, and anything else that is not green.
376 fn format_incidents(instance: &str, payload: &Payload) -> String {
377 let mut incidents = Vec::new();
378 let mut other = Vec::new();
379
380 for node in &payload.nodes {
381 let name = node.id.strip_prefix("target:").unwrap_or(&node.id);
382 for condition in &node.conditions {
383 if condition.status == Status::Ok || condition.status == Status::Pending {
384 continue;
385 }
386 let line = format!(
387 "[{}] {name} / {}{}{}",
388 status_word(condition.status),
389 condition.condition_type,
390 condition
391 .since
392 .map(|s| format!(" since {}", s.format("%Y-%m-%d %H:%M UTC")))
393 .unwrap_or_default(),
394 condition
395 .detail
396 .as_ref()
397 .map(|d| format!(": {}", scrub(d)))
398 .unwrap_or_default(),
399 );
400 if condition.condition_type == "incident" {
401 incidents.push(line);
402 } else {
403 other.push(line);
404 }
405 }
406 }
407
408 let mut out = format!("# Open incidents on {instance}\n\n");
409 if incidents.is_empty() {
410 out.push_str("No open incidents.\n");
411 } else {
412 for line in &incidents {
413 let _ = writeln!(out, "{line}");
414 }
415 }
416
417 // A failing check that has not yet opened an incident is still the answer to
418 // "is anything wrong", so it is reported rather than filtered out.
419 if !other.is_empty() {
420 out.push_str("\nOther checks not passing:\n");
421 for line in &other {
422 let _ = writeln!(out, "{line}");
423 }
424 }
425 out
426 }
427
428 /// Latency over the window, against the 7-day baseline.
429 fn format_trends(instance: &str, trends: &api::TrendResponse) -> String {
430 let mut out = format!(
431 "# {} latency on {instance}: last {}h, {}-minute buckets\n\n",
432 trends.target, trends.window_hours, trends.bucket_minutes
433 );
434
435 match &trends.overall {
436 Some(o) => {
437 let _ = writeln!(
438 out,
439 "Window: avg {:.0}ms, p95 {}ms, range {}-{}ms ({} samples)",
440 o.avg_ms, o.p95_ms, o.min_ms, o.max_ms, o.sample_count
441 );
442 }
443 None => out.push_str("Window: no operational checks in this window.\n"),
444 }
445
446 if let Some(b) = &trends.baseline {
447 let _ = writeln!(
448 out,
449 "7d baseline: avg {:.0}ms, p95 {}ms ({} samples)",
450 b.avg_ms, b.p95_ms, b.sample_count
451 );
452 }
453
454 if trends.buckets.is_empty() {
455 return out;
456 }
457
458 out.push_str("\nBUCKET AVG P95 N\n");
459 for bucket in &trends.buckets {
460 let _ = writeln!(
461 out,
462 "{:<18} {:>4.0}ms {:>4}ms {:>4}",
463 bucket.period_start.chars().take(16).collect::<String>(),
464 bucket.avg_ms,
465 bucket.p95_ms,
466 bucket.sample_count
467 );
468 }
469 out
470 }
471
472 /// The wire spelling of a status, which is also the shortest honest label.
473 fn status_word(status: Status) -> &'static str {
474 status.as_str()
475 }
476
477 /// A node field's value as one short string.
478 fn value_text(value: &ops_status::Value) -> String {
479 use ops_status::Value;
480 match value {
481 Value::Text { value } | Value::Ident { value, .. } | Value::Version { value } => {
482 scrub(value)
483 }
484 Value::Path { value } => scrub(value),
485 Value::Instant { value } => value.format("%Y-%m-%d %H:%M UTC").to_string(),
486 Value::Duration { seconds } => format!("{seconds}s"),
487 Value::Progress { value, unit, .. } | Value::Quantity { value, unit } => {
488 format!("{value:.1}{}", unit.as_deref().unwrap_or(""))
489 }
490 Value::State { value } => status_word(*value).to_string(),
491 Value::Link { url, text } => format!("{} <{}>", scrub(text.as_deref().unwrap_or("")), url),
492 }
493 }
494
495 /// A named field's text, or `-` when the node does not carry it.
496 fn field_text(node: &Node, label: &str) -> String {
497 node.fields
498 .iter()
499 .find(|f| f.label == label)
500 .map_or_else(|| "-".to_string(), |f| value_text(&f.value))
501 }
502
503 /// The worst thing said about a node, in a few words. Green nodes get a dash.
504 fn worst_condition_summary(node: &Node) -> String {
505 let worst = node
506 .conditions
507 .iter()
508 .filter(|c| c.status != Status::Ok)
509 .max_by_key(|c| c.status);
510
511 match worst {
512 None => "-".to_string(),
513 Some(c) => {
514 let detail = c
515 .detail
516 .as_deref()
517 .map(|d| format!(": {}", scrub(d)))
518 .unwrap_or_default();
519 let line = format!("{}{detail}", c.condition_type);
520 // One line per target is the whole point of this table, so a long
521 // detail is cut rather than allowed to wrap.
522 if line.chars().count() > 60 {
523 format!("{}...", line.chars().take(57).collect::<String>())
524 } else {
525 line
526 }
527 }
528 }
529 }
530
531 /// Whether a string is shaped like a config key, and so safe to put in a URL
532 /// path segment without escaping.
533 fn is_config_name(s: &str) -> bool {
534 !s.is_empty()
535 && s.len() <= 64
536 && s.chars()
537 .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.')
538 }
539
540 /// Strip terminal control characters from values a monitored target chose.
541 ///
542 /// Same surface as the CLI display sink: these strings reach an operator's
543 /// terminal through the session transcript.
544 fn scrub(s: &str) -> String {
545 s.chars().filter(|c| !c.is_control()).collect()
546 }
547
548 /// Write one table row, every column but the last padded to its width.
549 fn write_row(out: &mut String, cells: &[String; 5], widths: &[usize]) {
550 for (i, cell) in cells.iter().enumerate() {
551 if i + 1 == cells.len() {
552 let _ = writeln!(out, "{cell}");
553 } else {
554 let pad = widths[i].saturating_sub(cell.chars().count());
555 let _ = write!(out, "{cell}{:pad$} ", "", pad = pad);
556 }
557 }
558 }
559