hyperhive/swarm-controller/src/observability_grant.rs

//! The `logs` and `metrics` permissions beyond the queue's permissions bucket:
//! which agents may be granted them, and the mirror of the grant that the store
//! gateways read.
//!
//! An agent that cannot skip a permission name it does not know would drop to
//! no permissions on reading a record that holds one, so the controller grants
//! them only when the agent's hive status snapshot and its running harness's
//! turn-state header both advertise
//! `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.
//!
//! **The leaf is what the gateway enforces, the record is what the operator
//! sees**, so a change is ordered to fail towards denying: a revoke writes the
//! leaf first and the record second, a grant the record first and the leaf
//! second. [`reconcile`] makes every leaf equal its record's grant, every
//! [`REMIRROR_EVERY`], for the writes that failed half way.

use std::collections::BTreeMap;
use std::future::Future;
use std::time::Duration;

use anyhow::{Context as _, Result};
use hive_types::Permission;
use swarm_secret_client::SecretStore;
use swarm_secret_client::observability::{Grant, grant_path};
use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS};

use crate::wanted::WantedWriter;

/// How often [`reconcile`] runs, matching the controller's other periodic
/// passes over the swarm's agents (`agent_renewal`).
pub(crate) const REMIRROR_EVERY: Duration = Duration::from_mins(5);

/// The grant `perms` hold.
pub(crate) fn grant_of(perms: &[Permission]) -> Grant {
    Grant {
        logs: perms.contains(&Permission::Logs),
        metrics: perms.contains(&Permission::Metrics),
    }
}

/// Whether `after` holds a grant `before` did not.
pub(crate) fn adds(before: &[Permission], after: &[Permission]) -> bool {
    let (b, a) = (grant_of(before), grant_of(after));
    (a.logs && !b.logs) || (a.metrics && !b.metrics)
}

/// Whether `after` drops a grant `before` held.
pub(crate) fn revokes(before: &[Permission], after: &[Permission]) -> bool {
    adds(after, before)
}

/// Where a mirrored grant is read and written. A test stands in for the store.
pub(crate) trait GrantLeaves {
    /// The leaf, `None` when nothing is stored there.
    fn read(&self, agent: &str) -> impl Future<Output = Result<Option<Grant>>> + Send;
    fn write(&self, agent: &str, grant: Grant) -> impl Future<Output = Result<()>> + Send;
}

/// The swarm secret store, under the controller's own identity, logging in
/// for each call.
pub(crate) struct Store;

impl GrantLeaves for Store {
    async fn read(&self, agent: &str) -> Result<Option<Grant>> {
        Connected(connect().await?).read(agent).await
    }

    async fn write(&self, agent: &str, grant: Grant) -> Result<()> {
        Connected(connect().await?).write(agent, grant).await
    }
}

async fn connect() -> Result<SecretStore> {
    crate::store::connect()
        .await
        .context("logging in to the swarm secret store")
}

/// One login, for a pass that touches many leaves.
pub(crate) struct Connected(pub(crate) SecretStore);

impl Connected {
    pub(crate) async fn login() -> Result<Self> {
        Ok(Self(connect().await?))
    }
}

impl GrantLeaves for Connected {
    async fn read(&self, agent: &str) -> Result<Option<Grant>> {
        let path = grant_path(agent)?;
        self.0
            .read_optional(&path)
            .await
            .with_context(|| format!("reading {path}"))
    }

    async fn write(&self, agent: &str, grant: Grant) -> Result<()> {
        let path = grant_path(agent)?;
        self.0
            .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 gateway then keeps the grant it last read, or none.
pub(crate) async fn mirror(
    leaves: &impl GrantLeaves,
    agent: &str,
    before: &[Permission],
    after: &[Permission],
) -> Result<()> {
    if grant_of(before) == Grant::default() && grant_of(after) == Grant::default() {
        return Ok(());
    }
    leaves.write(agent, grant_of(after)).await
}

/// Make each agent's leaf equal `desired`, the grant its record holds; an
/// absent leaf equals the empty grant. Every correction is logged at `warn`.
/// All agents are tried; `Ok` is the number corrected.
///
/// # Errors
/// How many agents could not be read or written.
pub(crate) async fn reconcile(
    leaves: &impl GrantLeaves,
    desired: &BTreeMap<String, Grant>,
) -> Result<usize> {
    let (mut corrected, mut failed) = (0, 0);
    for (agent, want) in desired {
        let have = match leaves.read(agent).await {
            Ok(leaf) => leaf.unwrap_or_default(),
            Err(e) => {
                tracing::warn!(agent, error = %format!("{e:#}"), "observability grant: reading the leaf failed");
                failed += 1;
                continue;
            }
        };
        if have == *want {
            continue;
        }
        match leaves.write(agent, *want).await {
            Ok(()) => {
                tracing::warn!(
                    agent,
                    ?have,
                    ?want,
                    "observability grant: leaf differed from the record; corrected"
                );
                corrected += 1;
            }
            Err(e) => {
                tracing::warn!(agent, error = %format!("{e:#}"), "observability grant: correcting the leaf failed");
                failed += 1;
            }
        }
    }
    if failed > 0 {
        anyhow::bail!("{failed} agent(s) could not be reconciled");
    }
    Ok(corrected)
}

/// `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 `agent` may be granted `logs` or `metrics`: its hive advertises
/// tolerant decoding and so does the harness it is running (its latest
/// turn-state header, which an agent that is stopped, not yet restarted on a
/// tolerant build, or not reporting yet does not send).
///
/// # Errors
/// The reason: no hive declares the agent, the hive's snapshot is missing or
/// lacks the capability, the harness has not reported it, or the queue could
/// not be read.
pub(crate) async fn require_tolerant_agent(
    client: &async_nats::Client,
    wanted: &WantedWriter,
    hives: &[String],
    reported: Option<&crate::agent_login::Reported>,
    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?;
    if reported.is_some_and(|r| r.advertises(agent, TOLERANT_PERMISSIONS)) {
        return Ok(());
    }
    Err(format!(
        "agent {agent}'s harness has not reported {TOLERANT_PERMISSIONS}: it must be running \
         a build that decodes the new permissions (update it, then wait a minute for its next \
         status)"
    ))
}

/// `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())
}

/// A stand-in for the store's leaves, shared by the tests that mirror.
#[cfg(test)]
pub(crate) mod fake {
    use super::{GrantLeaves, Result};
    use std::collections::BTreeMap;
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicBool, Ordering};
    use swarm_secret_client::observability::Grant;

    /// Leaves by agent. While `failing`, every read and write is an error.
    #[derive(Default)]
    pub(crate) struct Leaves {
        pub(crate) held: Mutex<BTreeMap<String, Grant>>,
        pub(crate) failing: AtomicBool,
    }

    impl Leaves {
        pub(crate) fn fail(&self, on: bool) {
            self.failing.store(on, Ordering::SeqCst);
        }

        pub(crate) fn get(&self, agent: &str) -> Option<Grant> {
            self.held.lock().unwrap().get(agent).copied()
        }

        pub(crate) fn put(&self, agent: &str, grant: Grant) {
            self.held.lock().unwrap().insert(agent.to_owned(), grant);
        }
    }

    impl GrantLeaves for Leaves {
        async fn read(&self, agent: &str) -> Result<Option<Grant>> {
            anyhow::ensure!(!self.failing.load(Ordering::SeqCst), "store unreachable");
            Ok(self.get(agent))
        }

        async fn write(&self, agent: &str, grant: Grant) -> Result<()> {
            anyhow::ensure!(!self.failing.load(Ordering::SeqCst), "store unreachable");
            self.put(agent, grant);
            Ok(())
        }
    }
}

#[cfg(test)]
mod tests {
    use super::fake::Leaves;
    use super::*;

    const YES: Grant = Grant {
        logs: true,
        metrics: true,
    };
    const LOGS: Grant = Grant {
        logs: true,
        metrics: false,
    };
    const BOTH: [Permission; 3] = [Permission::Messaging, Permission::Logs, Permission::Metrics];
    const PLAIN: [Permission; 1] = [Permission::Messaging];

    #[test]
    fn adds_and_revokes_are_the_two_directions() {
        assert!(adds(&PLAIN, &BOTH) && !revokes(&PLAIN, &BOTH));
        assert!(revokes(&BOTH, &PLAIN) && !adds(&BOTH, &PLAIN));
        assert!(
            adds(&[Permission::Logs], &[Permission::Metrics]),
            "swap adds"
        );
        assert!(
            revokes(&[Permission::Logs], &[Permission::Metrics]),
            "and revokes"
        );
        assert!(!adds(&BOTH, &BOTH) && !revokes(&BOTH, &BOTH));
        assert!(!adds(&PLAIN, &PLAIN) && !revokes(&PLAIN, &PLAIN));
    }

    #[tokio::test]
    async fn a_grant_and_its_revocation_are_written() {
        let leaves = Leaves::default();
        mirror(&leaves, "atlas", &[], &BOTH).await.unwrap();
        assert_eq!(leaves.get("atlas"), Some(YES));
        mirror(&leaves, "atlas", &BOTH, &PLAIN).await.unwrap();
        assert_eq!(leaves.get("atlas"), Some(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();
        mirror(&leaves, "atlas", &PLAIN, &PLAIN).await.unwrap();
        assert_eq!(leaves.get("atlas"), None);
        mirror(&leaves, "atlas", &PLAIN, &[Permission::Logs])
            .await
            .unwrap();
        assert_eq!(leaves.get("atlas"), Some(LOGS));
    }

    #[tokio::test]
    async fn a_failed_write_is_an_error() {
        let leaves = Leaves::default();
        leaves.fail(true);
        assert!(mirror(&leaves, "atlas", &[], &BOTH).await.is_err());
        // Control: nothing to write is not a failure, whatever the store does.
        assert!(mirror(&leaves, "atlas", &[], &[]).await.is_ok());
    }

    /// The pass corrects a leaf that is granted where the record is not, one
    /// that is missing or short where the record grants, and leaves the
    /// agreeing ones (and an absent leaf for an empty grant) alone.
    #[tokio::test]
    async fn the_pass_corrects_drift_in_both_directions() {
        let leaves = Leaves::default();
        leaves.put("stale", YES);
        leaves.put("short", LOGS);
        leaves.put("same", YES);
        let desired = BTreeMap::from([
            ("stale".to_owned(), Grant::default()),
            ("short".to_owned(), YES),
            ("missing".to_owned(), YES),
            ("same".to_owned(), YES),
            ("none".to_owned(), Grant::default()),
        ]);
        assert_eq!(reconcile(&leaves, &desired).await.unwrap(), 3);
        assert_eq!(
            leaves.get("stale"),
            Some(Grant::default()),
            "revoke repaired"
        );
        assert_eq!(leaves.get("short"), Some(YES));
        assert_eq!(leaves.get("missing"), Some(YES), "grant repaired");
        assert_eq!(leaves.get("same"), Some(YES));
        assert_eq!(leaves.get("none"), None, "no leaf made for an empty grant");
        assert_eq!(reconcile(&leaves, &desired).await.unwrap(), 0, "converged");
    }

    /// A store that is down is an error and changes nothing; the next pass,
    /// once it is back, fixes it.
    #[tokio::test]
    async fn a_failed_pass_changes_nothing_and_the_next_one_fixes_it() {
        let leaves = Leaves::default();
        leaves.put("stale", YES);
        let desired = BTreeMap::from([
            ("stale".to_owned(), Grant::default()),
            ("missing".to_owned(), YES),
        ]);
        leaves.fail(true);
        assert!(reconcile(&leaves, &desired).await.is_err());
        leaves.fail(false);
        assert_eq!(leaves.get("stale"), Some(YES));
        assert_eq!(leaves.get("missing"), None);
        assert_eq!(reconcile(&leaves, &desired).await.unwrap(), 2);
        assert_eq!(leaves.get("stale"), Some(Grant::default()));
        assert_eq!(leaves.get("missing"), Some(YES));
    }

    #[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());
    }
}