Observe a cluster

Every node has its own view of the cluster: the members with their states, the members it cannot reach, and more. tellus offers that view as a value that changes over time, and the tellus-cluster-inspect crate serves it over HTTP.

Read the current view

cluster::cluster_state returns a tokio::sync::watch::Receiver of the ClusterState. Its current value is this node's view, and wait_for waits for a view with a given property, as the clustered actors tutorial does. A ClusterState holds:

• members, every member with its address, incarnation and state: up, left or down. This node is among them from the start.
• unreachable, the members that are up but that this node cannot reach, neither directly nor through other members.
• broken_links, direct links one member reports broken while both ends still reach each other through other members. No downing provider acts on such a link: it lasts until it heals or you down one end with cluster::down. Meanwhile, messages between the two ends are queued and asks between them time out.
• foreign_clusters, the addresses that answered as members of another cluster. Two clusters never merge, so a non-empty list is a split for you to resolve.
• cluster, the ID of the cluster this node is a member of, the same on every member.

Follow the transitions

cluster::changes yields the current version of the view first and then every new one, in order. ClusterState::diff compares two versions member by member:

async fn first_node() -> anyhow::Result<()> {
    let addrs = reserved_addrs(2)?;
    start_endpoint(addrs[0])?;
    let mut changes = cluster::changes().context("subscribing to the changes")?;
    let mut previous = cluster::cluster_state()
        .context("reading the cluster state")?
        .borrow()
        .clone();
    cluster::form().context("forming the cluster")?;
    let mut second_node = spawn_second_node(addrs[1], addrs[0])?;

    let name = |addr| {
        if addr == addrs[0] {
            "first node"
        } else {
            "second node"
        }
    };
    let observe = async {
        loop {
            let Change::State(state) = changes.next().await else {
                continue;
            };
            for transition in state.diff(&previous) {
                let member = transition.member().addr();
                println!("{}: {}", name(member), describe(&transition));
                if matches!(transition, Transition::Left(_)) && member == addrs[1] {
                    return;
                }
            }
            previous = state;
        }
    };
    timeout(OBSERVE_TIMEOUT, observe)
        .await
        .context("the second node did not leave within the timeout")?;

    second_node
        .wait()
        .await
        .context("awaiting the second node's exit")?;
    Ok(())
}

fn describe(transition: &Transition) -> &'static str {
    match transition {
        Transition::Up(_) => "up",
        Transition::Left(_) => "left",
        Transition::Down(_) => "down",
        Transition::Forgotten(_) => "forgotten",
        Transition::Unreachable(_) => "unreachable",
        Transition::Reachable(_) => "reachable again",
    }
}

The first node remembers the previous version and prints what changed, while a second node joins, stays a second and leaves. Change::Settled and Change::Gap report the receptionist's verdict and a subscriber that fell behind; the example skips them. After a gap, the next state supersedes everything missed, so a diff against it is still correct.

From a clone of the tellus repository:

cargo run --quiet -p tellus --features cluster-dev --example how_to_observe_a_cluster

second node: up
second node: left

The first node itself is listed up before it forms the cluster, so its own state yields no transition.

Serve the view over HTTP

The tellus-cluster-inspect crate is an axum Router that serves the view:

cargo add tellus-cluster-inspect

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let transport = QuicTransport::dev(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
        .context("dev QUIC transport")?;
    let addr = transport.local_addr().context("local transport address")?;
    cluster::start_endpoint(EndpointConfig::new(addr), transport).context("remoting endpoint")?;
    cluster::form().context("forming the cluster")?;

    let inspect =
        tellus_cluster_inspect::router(InspectConfig::default()).context("inspect router")?;
    let app = Router::new().nest("/inspect", inspect);
    let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, HTTP_PORT)))
        .await
        .context("binding the HTTP port")?;
    tokio::select! {
        served = async { axum::serve(listener, app).await } => served.context("serving HTTP"),
        interrupt = signal::ctrl_c() => interrupt.context("Ctrl-C handler"),
    }
}

Mount it into your own HTTP server. It brings no authentication and binds no port of its own, and it reveals the cluster's topology, so mount it behind whatever protects the rest of your server.

From a clone of the tellus repository:

cargo run --quiet -p tellus-cluster-inspect --example how_to_serve_cluster_state

GET /inspect/cluster answers a snapshot as JSON: the ClusterState, the endpoint's lifecycle (unformed, formed or downed) and the receptionist's settlement. GET /inspect/events streams every change as server-sent events, starting with a hello and the current state. A client that reconnects with the Last-Event-ID it saw last continues right after it, as long as the node still keeps that position.

curl localhost:8080/inspect/cluster
curl -N localhost:8080/inspect/events