Merge pull request #383 from beyondessential/claude/source-reachability

Source reachability policy: fold per-source staleness into one reachability check

Author: passcod

.workhorse/specs/monitoring/checks.md

@@ -21,9 +21,19 @@ Canopy-wide checks are Canopy monitoring its own operation (see [SELF](../privat
 A source is a named reporter of checks, identified by a short string.
 Multiple sources may report on the same server, each concerned with part of the system, and each source's reports are independent: a report from one source says nothing about another source's checks.
 
-Two source names are reserved for Canopy itself: `canopy` for conditions Canopy determines on its own (staleness, reachability, backup health, key expiry, self-monitoring), and `manual` for conditions raised by operators.
+Two source names are reserved for Canopy itself: `canopy` for conditions Canopy determines on its own (reachability, backup health, key expiry, self-monitoring), and `manual` for conditions raised by operators.
 Reports arriving over the device API cannot use the reserved names.
 
+### Reachability mode
+
+Each source other than the reserved names carries an operator-set reachability mode, governing how the source's silence bears on its servers' reachability (see "Reachability"):
+
+- `on` — a stale source warns, and all of a server's sources stale is unreachable;
+- `quiet` — a stale source raises no warning, but still counts toward unreachable;
+- `off` — the source is excluded from reachability entirely.
+
+New sources default to `on`. The mode is global to the source, edited by operators alongside the check catalog.
+
 ## Results
 
 A check's result is one of, in decreasing order of urgency:
@@ -102,23 +112,29 @@ A source's report for a server carries that source's complete current set of che
 Canopy trusts the reporter: a check the source previously reported but omits from its current report has recovered, and its state records that.
 Omission by one source never affects another source's checks.
 
-## Source staleness
+## Reachability
+
+Canopy tracks, for each server, the sources expected to report — those that have reported, are not in reachability mode `off`, and whose checks are not all decommissioned — and when each last reported.
+It keeps one `reachability` check per server, under the `canopy` source, reflecting how many expected sources are currently reporting within the server's down threshold:
+
+- **passed** when every expected source is fresh;
+- **warning** when a source in mode `on` is stale but not every expected source is; the stale sources are named in the check's detail, so an operator sees which reporter went quiet. A `quiet` source going stale never raises this warning;
+- **failed** when every expected source is stale — nothing is reaching Canopy — and the server is presented as unreachable. `quiet` and `on` sources count alike here.
 
-A source that has reported on a server is expected to keep reporting, unless all of its checks have been decommissioned.
-When a source's most recent report for a server is older than the server's down threshold, Canopy raises a staleness check for that (server, source) under the `canopy` source, and clears it when the source reports again.
-A server all of whose sources are stale is presented as unreachable.
+A stale source degrades the server rather than silently dropping its checks, so a reporter going quiet is never mistaken for health.
+There is no per-source staleness check; the one reachability check carries the full picture.
 
 ## Liveness and decommissioning
 
-Source staleness is a per-server signal about a reporter that has gone quiet; check liveness is a fleet-wide signal about a (source, check) that has gone away everywhere.
+Reachability is a per-server signal about reporters that have gone quiet; check liveness is a fleet-wide signal about a (source, check) that has gone away everywhere.
 For each catalogued (source, check) Canopy tracks when it was most recently reported on any server.
 
 A (source, check) not reported anywhere for seven days is surfaced to operators as a candidate for decommissioning.
 A (source, check) not reported anywhere for thirty days raises a Canopy-wide warning (see [SELF](../private-server/self-alerts.md)).
 
 Decommissioning is an operator action, never automatic: the candidate list and the Canopy-wide warning surface what has gone away, and an operator decides.
-A decommissioned (source, check) is retired fleet-wide: its state on every server is resolved, recording decommissioning as the reason, and it then contributes to nothing — not health, not incidents, not source staleness.
-A source all of whose checks are decommissioned is no longer expected to report, so its staleness clears.
+A decommissioned (source, check) is retired fleet-wide: its state on every server is resolved, recording decommissioning as the reason, and it then contributes to nothing — not health, not incidents, not reachability.
+A source all of whose checks are decommissioned is no longer an expected source, so it drops out of the reachability signal.
 
 If a decommissioned check is reported again it is treated as newly registered — pending operator review, at the warning ceiling — so a resurrected check never silently resumes a retired policy.
 

crates/commons-types/src/lib.rs

@@ -5,6 +5,7 @@ pub mod device;
 pub mod geo;
 pub mod issue;
 pub mod server;
+pub mod source;
 pub mod status;
 pub mod units;
 pub mod version;

crates/commons-types/src/source.rs

@@ -0,0 +1,64 @@
+//! Operator policy attached to a reporting source (as opposed to a single
+//! check). Currently the reachability mode; the ingest mode follows.
+
+use serde::{Deserialize, Serialize};
+
+/// How a source's silence bears on its servers' reachability.
+///
+/// Stored as text in Postgres, validated as this enum at the edges.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
+#[serde(rename_all = "lowercase")]
+pub enum ReachabilityMode {
+	/// A stale source warns, and all of a server's sources stale is
+	/// unreachable. The default.
+	#[default]
+	On,
+	/// A stale source raises no warning, but still counts toward
+	/// unreachable — so a server whose only (or last) source is quiet still
+	/// reads unreachable when it goes silent.
+	Quiet,
+	/// The source is excluded from reachability entirely.
+	Off,
+}
+
+#[derive(Debug, Clone, Copy, thiserror::Error)]
+#[error("invalid reachability mode; expected one of: on, quiet, off")]
+pub struct ReachabilityModeFromStringError;
+
+impl std::str::FromStr for ReachabilityMode {
+	type Err = ReachabilityModeFromStringError;
+
+	fn from_str(s: &str) -> Result<Self, Self::Err> {
+		match s.to_ascii_lowercase().as_str() {
+			"on" => Ok(Self::On),
+			"quiet" => Ok(Self::Quiet),
+			"off" => Ok(Self::Off),
+			_ => Err(ReachabilityModeFromStringError),
+		}
+	}
+}
+
+impl TryFrom<String> for ReachabilityMode {
+	type Error = ReachabilityModeFromStringError;
+
+	fn try_from(value: String) -> Result<Self, Self::Error> {
+		value.parse()
+	}
+}
+
+impl std::fmt::Display for ReachabilityMode {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		let s = match self {
+			Self::On => "on",
+			Self::Quiet => "quiet",
+			Self::Off => "off",
+		};
+		write!(f, "{s}")
+	}
+}
+
+impl From<ReachabilityMode> for String {
+	fn from(m: ReachabilityMode) -> Self {
+		m.to_string()
+	}
+}

crates/database/src/check_policies.rs

@@ -209,29 +209,9 @@ impl CheckPolicy {
 			crate::issues::Issue::resolve(db, id, by, ResolvedReason::Decommissioned).await?;
 		}
 
-		// A source with no live checks left is no longer expected to report;
-		// clear any per-server staleness it still carries (source_freshness
-		// now excludes it, so nothing re-files these).
-		let live: i64 = cp::check_policies
-			.filter(cp::source.eq(source))
-			.filter(cp::decommissioned_at.is_null())
-			.count()
-			.get_result(db)
-			.await?;
-		if live == 0 {
-			let stale_ref = format!("{}{}", crate::statuses::STALE_REF_PREFIX, source);
-			let stale_ids: Vec<Uuid> = iss::issues
-				.select(iss::id)
-				.filter(iss::source.eq(crate::statuses::CANOPY_SOURCE))
-				.filter(iss::check_name.eq(&stale_ref))
-				.filter(iss::server_id.is_not_null())
-				.filter(iss::resolved_at.is_null())
-				.load(db)
-				.await?;
-			for id in stale_ids {
-				crate::issues::Issue::resolve(db, id, by, ResolvedReason::Decommissioned).await?;
-			}
-		}
+		// A source whose checks are all decommissioned drops out of
+		// source_freshness, so it stops counting toward reachability with no
+		// further action here.
 		Ok(())
 	}
 

crates/database/src/lib.rs

@@ -27,6 +27,7 @@ pub mod server_groups;
 pub mod servers;
 pub mod silenced_refs;
 pub mod slack_outbox;
+pub mod source_policies;
 pub mod sql_playground_history;
 pub mod stability;
 pub mod statuses;

crates/database/src/schema.rs

@@ -545,6 +545,15 @@ diesel::table! {
 	}
 }
 
+diesel::table! {
+	source_policies (source) {
+		source -> Text,
+		reachability -> Text,
+		created_at -> Timestamptz,
+		updated_at -> Timestamptz,
+	}
+}
+
 diesel::table! {
 	statuses (id, created_at) {
 		id -> Uuid,

crates/database/src/source_policies.rs

@@ -0,0 +1,120 @@
+//! Operator policy per reporting source. Currently the reachability mode
+//! (how the source's silence bears on its servers' reachability); the
+//! ingest mode follows in a later change. Absent rows mean the defaults.
+
+use std::collections::HashMap;
+
+use commons_errors::{AppError, Result};
+use commons_types::source::ReachabilityMode;
+use diesel::prelude::*;
+use diesel::sql_types;
+use diesel_async::{AsyncPgConnection, RunQueryDsl};
+use jiff::Timestamp;
+
+/// A reporting source with its reachability policy and fleet-wide
+/// last-seen, for the operator source list. Reserved sources are excluded.
+#[derive(Clone, Debug)]
+pub struct SourceInfo {
+	pub source: String,
+	pub reachability: ReachabilityMode,
+	pub last_seen: Option<Timestamp>,
+}
+
+/// One source's policy row.
+#[derive(Clone, Debug, Queryable, Selectable)]
+#[diesel(table_name = crate::schema::source_policies)]
+#[diesel(check_for_backend(diesel::pg::Pg))]
+pub struct SourcePolicy {
+	pub source: String,
+	#[diesel(deserialize_as = String, serialize_as = String)]
+	pub reachability: ReachabilityMode,
+	#[diesel(deserialize_as = jiff_diesel::Timestamp, serialize_as = jiff_diesel::Timestamp)]
+	pub created_at: Timestamp,
+	#[diesel(deserialize_as = jiff_diesel::Timestamp, serialize_as = jiff_diesel::Timestamp)]
+	pub updated_at: Timestamp,
+}
+
+impl SourcePolicy {
+	/// Every source policy row, ordered by source.
+	pub async fn list(db: &mut AsyncPgConnection) -> Result<Vec<Self>> {
+		use crate::schema::source_policies::dsl;
+		dsl::source_policies
+			.select(Self::as_select())
+			.order(dsl::source.asc())
+			.load(db)
+			.await
+			.map_err(AppError::from)
+	}
+
+	/// Every non-reserved reporting source (those with catalogued checks),
+	/// with its reachability mode (defaulting to `on`) and the most recent
+	/// time any of its checks was seen fleet-wide. Ordered by source.
+	pub async fn list_sources(db: &mut AsyncPgConnection) -> Result<Vec<SourceInfo>> {
+		#[derive(QueryableByName)]
+		struct Row {
+			#[diesel(sql_type = sql_types::Text)]
+			source: String,
+			#[diesel(sql_type = sql_types::Text)]
+			reachability: String,
+			#[diesel(sql_type = sql_types::Nullable<sql_types::Timestamptz>)]
+			last_seen: Option<jiff_diesel::Timestamp>,
+		}
+		let rows: Vec<Row> = diesel::sql_query(
+			"SELECT cp.source, \
+			 coalesce(sp.reachability, 'on') AS reachability, \
+			 max(cp.last_seen) AS last_seen \
+			 FROM check_policies cp \
+			 LEFT JOIN source_policies sp ON sp.source = cp.source \
+			 WHERE cp.source NOT IN ('canopy', 'manual') \
+			 GROUP BY cp.source, sp.reachability \
+			 ORDER BY cp.source",
+		)
+		.load(db)
+		.await?;
+		Ok(rows
+			.into_iter()
+			.map(|r| SourceInfo {
+				source: r.source,
+				reachability: r.reachability.parse().unwrap_or_default(),
+				last_seen: r.last_seen.map(Into::into),
+			})
+			.collect())
+	}
+
+	/// Each source's reachability mode. Sources without a row are absent
+	/// here; callers default them to [`ReachabilityMode::On`].
+	pub async fn modes(db: &mut AsyncPgConnection) -> Result<HashMap<String, ReachabilityMode>> {
+		use crate::schema::source_policies::dsl;
+		let rows: Vec<(String, String)> = dsl::source_policies
+			.select((dsl::source, dsl::reachability))
+			.load(db)
+			.await?;
+		Ok(rows
+			.into_iter()
+			.filter_map(|(source, mode)| mode.parse().ok().map(|m| (source, m)))
+			.collect())
+	}
+
+	/// Set a source's reachability mode, creating its policy row if needed.
+	pub async fn set_reachability(
+		db: &mut AsyncPgConnection,
+		source: &str,
+		mode: ReachabilityMode,
+	) -> Result<()> {
+		use crate::schema::source_policies::dsl;
+		diesel::insert_into(dsl::source_policies)
+			.values((
+				dsl::source.eq(source),
+				dsl::reachability.eq(mode.to_string()),
+			))
+			.on_conflict(dsl::source)
+			.do_update()
+			.set((
+				dsl::reachability.eq(mode.to_string()),
+				dsl::updated_at.eq(jiff_diesel::Timestamp::from(Timestamp::now())),
+			))
+			.execute(db)
+			.await?;
+		Ok(())
+	}
+}