nordisk/holger

holger-estate-front/src/mesh_probe.rs

//! **The front proves the tunnel is the only path** (lane W9): the probe itself is SHARED,
//! Skidbladnir `nordisk_pairing::frontside::mesh_probe` (feature `mesh-probe`; P4c, A28 M11:
//! Gunnar's `gunnar-front` and this front carried the same code). What is Holger's: its plain
//! ports ([`PLAIN_PORTS`]), the probe as a jera `front.mesh_probe` job on the front's own
//! ledger ([`JOBS_REL`]), and its switch (the front starts it only `wg_only`).
//!
//! Once the installer has fed the boxes, it dials each box's UTILITY address on the gRPC control
//! plane, the HTTP door and the replication door — every one must be refused — and each box's
//! control plane on its OVERLAY address, which must answer; one line per dial and one verdict
//! line ([`PASS`] / [`FAIL`]), each at its level and only when it changed. Every
//! [`UNTIL_PASSED`] until the probe has passed once for the fed boxes, then every [`AFTER_PASSED`].

use std::path::Path;
use std::sync::Arc;

use jera::jobs::{ChainQueue, ChainSpec, Priority};
use nordisk_pairing::join::JoinReply;

pub use nordisk_pairing::frontside::mesh_probe::{
    boxes_of, dial, due, levels, probe, verdict, Box3, Seen, Watch, AFTER_PASSED, DIAL, FAIL, PASS,
    REPEAT_FAILURE, UNTIL_PASSED,
};

use crate::FrontVm;

/// The ports a plain-IP connection to a box is tried on: the gRPC control plane, the HTTP door,
/// the replication door.
pub const PLAIN_PORTS: [u16; 3] = [
    crate::CONTROL_PORT,
    crate::HTTP_PORT,
    crate::CONTROL_PORT + 1,
];

/// The ledger kind of one look.
pub const KIND: &str = "front.mesh_probe";
/// Where the front keeps the probe's jobs, under its appliance root.
pub const JOBS_REL: &str = "mesh/jobs";

/// The look of `generation` (one per generation: a resubmit is a no-op).
pub fn spec(generation: i64) -> ChainSpec {
    ChainSpec::new(
        KIND,
        format!("{KIND}-{generation}"),
        KIND,
        generation,
        "holger-estate-front",
        serde_json::json!({ "generation": generation }),
    )
    .with_target("the estate's boxes over the tunnel")
    .with_priority(Priority::Normal)
}

/// **The job**: one look at the boxes the front was fed, its lines said at their level, the
/// verdict as the ledger row's detail.
pub fn run(
    w: &Watch,
    reply: Option<JoinReply>,
    spec: &ChainSpec,
) -> anyhow::Result<serde_json::Value> {
    if spec.kind != KIND {
        anyhow::bail!("no mesh job {:?}", spec.kind);
    }
    let Some(boxes) = reply.as_ref().and_then(|r| boxes_of(r, false)) else {
        return Ok(serde_json::json!({ "verdict": "unfed" }));
    };
    let (lines, ok) = w.look(&boxes, &PLAIN_PORTS, crate::CONTROL_PORT, DIAL);
    for (warn, l) in &lines {
        if *warn {
            tracing::warn!("front: {l}");
        } else {
            tracing::info!("front: {l}");
        }
    }
    Ok(serde_json::json!({ "verdict": if ok { "pass" } else { "fail" }, "boxes": boxes.len() }))
}

/// **Start the probe** on the current runtime: the ledger at `<root>/`[`JOBS_REL`], its
/// ChainQueue's worker, and the ticker that queues one look per [`UNTIL_PASSED`] (then
/// [`AFTER_PASSED`] once passed). `false` when the ledger cannot open (said).
pub fn start(vm: Arc<FrontVm>, root: &Path) -> bool {
    let dir = root.join(JOBS_REL);
    let store = match jera::jobs::JobStore::open(&dir) {
        Ok(s) => s,
        Err(e) => {
            tracing::error!(jobs = %dir.display(), "front: the mesh probe's job ledger cannot open ({e:#}): the tunnel is not probed");
            return false;
        }
    };
    let Ok(handle) = tokio::runtime::Handle::try_current() else {
        return false;
    };
    let q = ChainQueue::new(store.sink());
    let w = Arc::new(Watch::default());
    let (worker, fed) = (Arc::clone(&w), Arc::clone(&vm));
    // A32-14, stays a task: the mesh probe lane's jera worker.
    handle.spawn(Arc::clone(&q).run(move |spec| run(&worker, fed.latest(), &spec)));
    // A32-14, stays a task: the mesh probe lane's ticker (it submits the jera looks).
    handle.spawn(async move {
        let _store = store;
        let mut generation = 0i64;
        loop {
            generation += 1;
            let _ = q.submit(spec(generation));
            tokio::time::sleep(if w.passed() {
                AFTER_PASSED
            } else {
                UNTIL_PASSED
            })
            .await;
        }
    });
    true
}

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

    /// Holger's plain ports: the control plane, the HTTP door, the replication door.
    #[test]
    fn the_plain_ports_are_holgers_three_doors() {
        assert_eq!(PLAIN_PORTS, [50051, 8443, 50052]);
    }

    /// The look is a jera job: an unfed front's look says so and probes nothing; a job of
    /// another kind is refused.
    #[test]
    fn the_job_looks_at_the_fed_boxes_only() {
        let reply = JoinReply {
            estate: "holger.test".into(),
            me: nordisk_pairing::join::Assigned {
                name: "vm1".into(),
                overlay: nordisk_pairing::OVERLAY_FRONT,
            },
            cert_pem: None,
            role: Some(nordisk_pairing::Role::Front),
            primary_overlay: None,
            primary_utility: None,
            peers: vec![],
            feed: Default::default(),
        };
        let w = Watch::default();
        assert_eq!(run(&w, None, &spec(1)).unwrap()["verdict"], "unfed");
        assert_eq!(run(&w, Some(reply), &spec(2)).unwrap()["verdict"], "unfed");
        assert_eq!(
            (spec(3).kind.as_str(), spec(3).priority),
            (KIND, Priority::Normal)
        );
        let other = ChainSpec::new("other", "k", "other", 1, "t", serde_json::json!({}));
        assert!(run(&w, None, &other).is_err());
    }
}