controller gates logs and metrics on the hive capability and mirrors the grant

swarm-controller/src/observability_grant.rs

@@ -0,0 +1,248 @@
+//! The `logs` and `metrics` permissions beyond the queue's permissions bucket:
+//! which hives may be granted them, and the mirror of the grant that the store
+//! gateways read.
+//!
+//! A hive that cannot skip a permission name it does not know would fail to
+//! decode a record that holds one, so the controller grants them only on a hive
+//! whose status snapshot advertises `swarm_wire::hive_status::TOLERANT_PERMISSIONS`.
+//! `swarm-logs-auth` has no queue access, so each grant is also written to
+//! `swarm_secret_client::observability::grant_path`; a leaf that is absent is no
+//! grant.
+
+use std::future::Future;
+
+use anyhow::{Context as _, Result};
+use hive_types::Permission;
+use swarm_secret_client::observability::{Grant, grant_path};
+use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS};
+
+use crate::wanted::WantedWriter;
+
+/// The grant `perms` hold.
+pub(crate) fn grant_of(perms: &[Permission]) -> Grant {
+    Grant {
+        logs: perms.contains(&Permission::Logs),
+        metrics: perms.contains(&Permission::Metrics),
+    }
+}
+
+/// Where a mirrored grant is written. A test stands in for the store.
+pub(crate) trait GrantSink {
+    fn write(&self, agent: &str, grant: Grant) -> impl Future<Output = Result<()>> + Send;
+}
+
+/// The swarm secret store, under the controller's own identity.
+pub(crate) struct Store;
+
+impl GrantSink for Store {
+    async fn write(&self, agent: &str, grant: Grant) -> Result<()> {
+        let path = grant_path(agent)?;
+        crate::store::connect()
+            .await
+            .context("logging in to the swarm secret store")?
+            .write(&path, &grant)
+            .await
+            .with_context(|| format!("writing {path}"))
+    }
+}
+
+/// Bring the mirror in line when `agent`'s permissions go from `before` to
+/// `after`. Nothing is written when neither holds `logs` or `metrics`, which
+/// leaves the leaf absent.
+///
+/// # Errors
+/// The write failed; the responder then keeps the grant it last read, or none.
+pub(crate) async fn mirror(
+    sink: &impl GrantSink,
+    agent: &str,
+    before: &[Permission],
+    after: &[Permission],
+) -> Result<()> {
+    if grant_of(before) == Grant::default() && grant_of(after) == Grant::default() {
+        return Ok(());
+    }
+    sink.write(agent, grant_of(after)).await
+}
+
+/// `Ok` when `hive`'s snapshot advertises tolerant decoding.
+///
+/// # Errors
+/// Naming the hive, when it has published nothing or does not advertise it.
+pub(crate) fn check_capability(
+    hive: &str,
+    snapshot: Option<&serde_json::Value>,
+) -> Result<(), String> {
+    match snapshot {
+        Some(s) if Capabilities::of_snapshot(s).has(TOLERANT_PERMISSIONS) => Ok(()),
+        Some(_) => Err(format!(
+            "hive {hive} does not advertise {TOLERANT_PERMISSIONS}: update it first, \
+             then grant logs or metrics"
+        )),
+        None => Err(format!(
+            "hive {hive} has published no status snapshot, so it is not known to \
+             advertise {TOLERANT_PERMISSIONS}"
+        )),
+    }
+}
+
+/// The hive whose declaration keeps `agent` alive.
+async fn hive_of(wanted: &WantedWriter, hives: &[String], agent: &str) -> Option<String> {
+    for hive in hives {
+        if let Ok(d) = wanted.view(hive).await
+            && crate::agent_renewal::live_agents(d.as_slice()).contains(agent)
+        {
+            return Some(hive.clone());
+        }
+    }
+    None
+}
+
+/// `hive`'s last status snapshot, `None` when it has published none.
+async fn snapshot(client: &async_nats::Client, hive: &str) -> Result<Option<serde_json::Value>> {
+    swarm_queue_client::ensure_connected(client)?;
+    let store = swarm_queue_client::status::open_or_create(client).await?;
+    let Some(entry) = store.get(hive).await? else {
+        return Ok(None);
+    };
+    Ok(serde_json::from_slice(&entry).ok())
+}
+
+/// Whether the hive running `agent` advertises tolerant decoding, waiting
+/// (an error, for a retrying caller) on a hive that has published no snapshot
+/// yet. `false` when no hive declares the agent.
+///
+/// # Errors
+/// The queue could not be read, or the hive has not published a snapshot.
+pub(crate) async fn hive_is_tolerant(
+    client: &async_nats::Client,
+    wanted: &WantedWriter,
+    hives: &[String],
+    agent: &str,
+) -> Result<bool> {
+    let Some(hive) = hive_of(wanted, hives, agent).await else {
+        return Ok(false);
+    };
+    let Some(snapshot) = snapshot(client, &hive).await? else {
+        anyhow::bail!("hive {hive} has published no status snapshot yet");
+    };
+    Ok(check_capability(&hive, Some(&snapshot)).is_ok())
+}
+
+/// `Ok` when the hive running `agent` advertises tolerant decoding, so the
+/// agent's record may name `logs` or `metrics`.
+///
+/// # Errors
+/// The reason: no hive declares the agent, the hive's snapshot is missing or
+/// lacks the capability, or the queue could not be read.
+pub(crate) async fn require_tolerant_hive(
+    client: &async_nats::Client,
+    wanted: &WantedWriter,
+    hives: &[String],
+    agent: &str,
+) -> Result<(), String> {
+    let hive = hive_of(wanted, hives, agent)
+        .await
+        .ok_or_else(|| format!("no hive declares agent {agent}"))?;
+    require_tolerant(client, &hive).await
+}
+
+/// `Ok` when `hive`'s last status snapshot advertises tolerant decoding.
+///
+/// # Errors
+/// The reason, naming the hive.
+pub(crate) async fn require_tolerant(
+    client: &async_nats::Client,
+    hive: &str,
+) -> Result<(), String> {
+    let snapshot = snapshot(client, hive)
+        .await
+        .map_err(|e| format!("reading hive {hive}'s status: {e:#}"))?;
+    check_capability(hive, snapshot.as_ref())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Mutex;
+
+    #[derive(Default)]
+    struct Leaves(Mutex<Vec<(String, Grant)>>);
+
+    impl GrantSink for &Leaves {
+        async fn write(&self, agent: &str, grant: Grant) -> Result<()> {
+            self.0.lock().unwrap().push((agent.to_owned(), grant));
+            Ok(())
+        }
+    }
+
+    struct Down;
+
+    impl GrantSink for Down {
+        async fn write(&self, _: &str, _: Grant) -> Result<()> {
+            anyhow::bail!("store unreachable")
+        }
+    }
+
+    #[tokio::test]
+    async fn a_grant_and_its_revocation_are_written() {
+        let leaves = Leaves::default();
+        let both = [Permission::Messaging, Permission::Logs, Permission::Metrics];
+        mirror(&&leaves, "atlas", &[], &both).await.unwrap();
+        mirror(&&leaves, "atlas", &both, &[Permission::Messaging])
+            .await
+            .unwrap();
+        let written = leaves.0.lock().unwrap().clone();
+        let yes = Grant {
+            logs: true,
+            metrics: true,
+        };
+        assert_eq!(
+            written,
+            [
+                ("atlas".to_owned(), yes),
+                ("atlas".to_owned(), Grant::default())
+            ]
+        );
+    }
+
+    /// An agent that never held either, before or after, gets no leaf; one
+    /// that holds only one of them gets exactly that one.
+    #[tokio::test]
+    async fn only_the_permissions_held_are_mirrored() {
+        let leaves = Leaves::default();
+        let plain = [Permission::Messaging];
+        mirror(&&leaves, "atlas", &plain, &plain).await.unwrap();
+        assert!(leaves.0.lock().unwrap().is_empty());
+        mirror(&&leaves, "atlas", &plain, &[Permission::Metrics])
+            .await
+            .unwrap();
+        let written = leaves.0.lock().unwrap().clone();
+        assert_eq!(written.len(), 1);
+        assert_eq!((written[0].1.logs, written[0].1.metrics), (false, true));
+    }
+
+    #[tokio::test]
+    async fn a_failed_write_is_an_error() {
+        let held = [Permission::Logs];
+        assert!(mirror(&Down, "atlas", &[], &held).await.is_err());
+        // Control: nothing to write is not a failure, whatever the store does.
+        assert!(mirror(&Down, "atlas", &[], &[]).await.is_ok());
+    }
+
+    #[test]
+    fn a_hive_without_the_capability_is_refused_by_name() {
+        let old = serde_json::json!({"status": "ok", "warnings": []});
+        let new = serde_json::json!({"capabilities": [TOLERANT_PERMISSIONS]});
+        let err = check_capability("pr1ma", Some(&old)).unwrap_err();
+        assert!(
+            err.contains("pr1ma") && err.contains(TOLERANT_PERMISSIONS),
+            "{err}"
+        );
+        assert!(
+            check_capability("pr1ma", None)
+                .unwrap_err()
+                .contains("pr1ma")
+        );
+        assert!(check_capability("pr1ma", Some(&new)).is_ok());
+    }
+}