path: root/src/api/stats/prometheus.rs

//! /admin/stats in the Prometheus text exposition format (version 0.0.4), for a
//! scraper asking for it. Adds per-peer series (the figures /peers shows), labelled
//! with the peer's public key.

use std::fmt::Write;

use tokio::time::Instant;

use super::StatsResponse;
use crate::locks::MutexExt;
use crate::state::{AppState, Incidents, Liveness};
use crate::transport::Destination;
use crate::transport::stats::{Reachability, TransportStats};
use crate::transport::traffic::{ByKind, Counter, PeerTraffic};

pub(super) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";

/// Whether the client prefers this format to JSON: Prometheus asks for OpenMetrics or
/// plain text; browsers and curl don't name either.
pub(super) fn wanted(accept: Option<&str>) -> bool {
    accept.is_some_and(|accept| {
        accept.split(',').any(|item| {
            let media = item.split(';').next().unwrap_or("").trim();
            media == "application/openmetrics-text" || media == "text/plain"
        })
    })
}

#[derive(Default)]
struct Out(String);

impl Out {
    fn family(&mut self, name: &str, kind: &str, help: &str) {
        let _ = writeln!(self.0, "# HELP continuum_{name} {help}");
        let _ = writeln!(self.0, "# TYPE continuum_{name} {kind}");
    }

    fn sample(&mut self, name: &str, labels: &[(&str, &str)], value: f64) {
        let _ = write!(self.0, "continuum_{name}");
        if !labels.is_empty() {
            self.0.push('{');
            for (i, (key, value)) in labels.iter().enumerate() {
                if i > 0 {
                    self.0.push(',');
                }
                let _ = write!(self.0, "{key}=\"{}\"", escape(value));
            }
            self.0.push('}');
        }
        let _ = writeln!(self.0, " {value}");
    }

    /// A family with one unlabelled sample.
    fn single(&mut self, name: &str, kind: &str, help: &str, value: f64) {
        self.family(name, kind, help);
        self.sample(name, &[], value);
    }
}

const DISK_COUNTERS: usize = 7;

/// Name, help and value of each counter of one kind of file.
fn disk_counters(c: &super::DiskStats) -> [(&'static str, &'static str, u64); DISK_COUNTERS] {
    [
        (
            "disk_reads_total",
            "Reads of this node's own files.",
            c.reads,
        ),
        (
            "disk_read_bytes_total",
            "Bytes read from them.",
            c.read_bytes,
        ),
        ("disk_writes_total", "Whole files written.", c.writes),
        (
            "disk_written_bytes_total",
            "Bytes of them.",
            c.written_bytes,
        ),
        (
            "disk_appends_total",
            "Appends to this node's own files.",
            c.appends,
        ),
        (
            "disk_appended_bytes_total",
            "Bytes appended.",
            c.appended_bytes,
        ),
        (
            "disk_errors_total",
            "Failed reads, writes and appends.",
            c.errors,
        ),
    ]
}

fn escape(value: &str) -> String {
    value
        .replace('\\', "\\\\")
        .replace('"', "\\\"")
        .replace('\n', "\\n")
}

/// The transport of an address.
fn scheme(address: &str) -> &str {
    address.split_once(':').map_or("", |(scheme, _)| scheme)
}

fn flag(value: bool) -> f64 {
    if value { 1.0 } else { 0.0 }
}

/// What we show per peer.
struct Peer {
    key: String,
    alias: Option<String>,
    neighbour: bool,
    dormant: bool,
    last_seen_secs: Option<f64>,
    last_sync_secs: Option<f64>,
    diverged_secs: Option<f64>,
    traffic: PeerTraffic,
    incidents: Incidents,
    throughput: Option<f64>,
    destination: Option<Destination>,
}

async fn peers(state: &AppState) -> Vec<Peer> {
    let now = Instant::now();
    let snapshot = state.peers_snapshot().await;
    let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc);
    let mut peers = Vec::with_capacity(snapshot.len());
    for (key, peer_state) in snapshot {
        let neighbour = crate::swarm::topology::is_neighbour(state, &key);
        let throughput = state
            .throughput
            .lock_or_recover()
            .get(&key)
            .map(|estimate| estimate.bytes_per_sec);
        let traffic = state.traffic.peer(&key).unwrap_or_default();
        let peer_state = peer_state.lock().await;
        peers.push(Peer {
            neighbour,
            dormant: matches!(peer_state.liveness, Liveness::Dormant { .. }),
            last_seen_secs: peer_state
                .last_seen
                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
            last_sync_secs: peer_state
                .last_sync_rx
                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
            diverged_secs: peer_state
                .diverged_since
                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
            traffic,
            incidents: peer_state.incidents,
            throughput,
            alias: aliases.remove(&key),
            destination: state.transport.destination(&key),
            key,
        });
    }
    peers.sort_by(|a, b| a.key.cmp(&b.key));
    peers
}

/// Every (direction, kind, counter) of `traffic`.
fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Counter)> {
    let by_kind = |direction: &'static str, by: &ByKind| {
        [
            (direction, "hello", by.hello),
            (direction, "hello_reply", by.hello_reply),
            (direction, "sync", by.sync),
            (direction, "status", by.status),
            (direction, "status_reply", by.status_reply),
            (direction, "blob", by.blob),
            (direction, "identity", by.identity),
            (direction, "paths", by.paths),
        ]
    };
    let mut series = Vec::new();
    series.extend(by_kind("sent", &traffic.sent));
    series.extend(by_kind("received", &traffic.received));
    series.extend(by_kind("refused", &traffic.refused));
    series
}

/// The transports' figures (`TransportStats`).
fn transports(out: &mut Out, stats: &TransportStats) {
    if let Some(identity) = &stats.identity {
        out.family(
            "identity_proven_nodes",
            "gauge",
            "Nodes proven at an address, by transport.",
        );
        for (transport, nodes) in &identity.proven {
            out.sample(
                "identity_proven_nodes",
                &[("transport", transport)],
                *nodes as f64,
            );
        }
        out.single(
            "identity_hinted_nodes",
            "gauge",
            "Nodes known only from hints, proven nowhere.",
            identity.hinted as f64,
        );
        let counts = &identity.counts;
        for (name, help, value) in [
            (
                "identity_proofs_rejected_total",
                "Identity proofs that didn't hold.",
                counts.proofs_rejected,
            ),
            (
                "identity_address_changes_total",
                "Nodes that proved themselves at another address on a transport than before.",
                counts.address_changes,
            ),
            (
                "identity_dropped_unproven_total",
                "Messages dropped because no node proved itself at their address.",
                counts.dropped_unproven,
            ),
            (
                "identity_refused_unproven_total",
                "Sends refused because the peer hadn't proven where it is yet.",
                counts.refused_unproven,
            ),
            (
                "identity_seeds_unanswered_total",
                "Seeds no node answered at in time.",
                counts.seeds_unanswered,
            ),
        ] {
            out.single(name, "counter", help, value as f64);
        }
    }
    if let Some(paths) = &stats.paths {
        let counts = &paths.counts;
        out.family(
            "paths_probes_total",
            "counter",
            "PINGs to paths of peers with several, by result.",
        );
        for (result, count) in [
            ("sent", counts.probes_sent),
            ("answered", counts.probes_answered),
            ("unanswered", counts.probes_unanswered),
        ] {
            out.sample("paths_probes_total", &[("result", result)], count as f64);
        }
        for (name, help, value) in [
            (
                "paths_send_failures_total",
                "Sends that failed on a path, each tried at the next.",
                counts.send_failures,
            ),
            (
                "paths_failovers_total",
                "Sends that went over another path after one failed.",
                counts.failovers,
            ),
            (
                "paths_switches_total",
                "Changes of the path in use for a faster one.",
                counts.switches,
            ),
        ] {
            out.single(name, "counter", help, value as f64);
        }
    }
    if let Some(libp2p) = &stats.libp2p {
        out.family(
            "libp2p_reachability",
            "gauge",
            "What AutoNAT says: 1 for the current status.",
        );
        for status in Reachability::ALL {
            out.sample(
                "libp2p_reachability",
                &[("status", status.as_str())],
                flag(libp2p.reachability == status),
            );
        }
        out.family(
            "libp2p_connections",
            "gauge",
            "Open libp2p connections, direct or through a relay.",
        );
        for (kind, count) in [
            ("direct", libp2p.connections.direct),
            ("relayed", libp2p.connections.relayed),
        ] {
            out.sample("libp2p_connections", &[("kind", kind)], count as f64);
        }
        let slots = &libp2p.relay_slots;
        out.single(
            "libp2p_relay_slots",
            "gauge",
            "Slots held on relays, to be reachable through them.",
            slots.held as f64,
        );
        out.single(
            "libp2p_relays_paused",
            "gauge",
            "Relays picked that never granted a slot, not picked again for a while.",
            slots.paused as f64,
        );
        out.family(
            "libp2p_relay_slot_events_total",
            "counter",
            "Relay slots granted, lost after being granted, and never granted.",
        );
        for (event, count) in [
            ("granted", slots.granted),
            ("lost", slots.lost),
            ("not_granted", slots.not_granted),
        ] {
            out.sample(
                "libp2p_relay_slot_events_total",
                &[("event", event)],
                count as f64,
            );
        }
        out.family(
            "libp2p_hole_punches_total",
            "counter",
            "Direct connections tried over a relayed one (DCUtR), by result.",
        );
        for (result, count) in [
            ("succeeded", libp2p.hole_punches.succeeded),
            ("failed", libp2p.hole_punches.failed),
        ] {
            out.sample(
                "libp2p_hole_punches_total",
                &[("result", result)],
                count as f64,
            );
        }
        let failures = &libp2p.dial_failures;
        out.family(
            "libp2p_dial_failures_total",
            "counter",
            "libp2p connections that couldn't be made, by cause.",
        );
        for (cause, count) in [
            ("unreachable", failures.unreachable),
            ("wrong_peer_id", failures.wrong_peer_id),
            ("no_addresses", failures.no_addresses),
            ("other", failures.other),
        ] {
            out.sample(
                "libp2p_dial_failures_total",
                &[("cause", cause)],
                count as f64,
            );
        }
    }
}

/// A per-peer incident counter: its name, its help and how to read it.
type IncidentSeries = (&'static str, &'static str, fn(&Incidents) -> f64);

pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
    let mut out = Out::default();

    // Names for joining (`* on(peer) group_left(alias) continuum_peer_info`), kept out
    // of the other series so a rename doesn't start new ones.
    out.family("info", "gauge", "This node's node ID and alias (always 1).");
    out.sample(
        "info",
        &[
            ("node_id", &stats.node_id),
            ("alias", stats.alias.as_deref().unwrap_or("")),
        ],
        1.0,
    );

    out.single(
        "uptime_seconds",
        "gauge",
        "Seconds since this node started.",
        stats.uptime_secs as f64,
    );
    out.single(
        "ready",
        "gauge",
        "1 once the node holds the swarm's document.",
        flag(stats.ready),
    );
    out.single(
        "epoch_generation",
        "gauge",
        "Generation of the local epoch (remakes so far).",
        state.local_epoch().generation as f64,
    );

    let document = &stats.document;
    out.single(
        "document_heads",
        "gauge",
        "Heads of the local document.",
        document.heads as f64,
    );
    out.single(
        "document_max_heads",
        "gauge",
        "Most heads after applying a sync message, since start.",
        document.max_heads as f64,
    );
    if let Some(bytes) = document.saved_bytes {
        out.single(
            "document_saved_bytes",
            "gauge",
            "Size of the document as saved (compressed); measured in the background.",
            bytes as f64,
        );
    }

    out.family(
        "document_lock_wait_seconds",
        "histogram",
        "Time waited for the document's lock, by mode (read, write).",
    );
    let lock_wait = &document.lock_wait;
    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
        let mut cumulative = 0;
        for (le_us, count) in waits.buckets() {
            cumulative += count;
            let le = match le_us {
                Some(us) => (us as f64 / 1e6).to_string(),
                None => "+Inf".to_owned(),
            };
            out.sample(
                "document_lock_wait_seconds_bucket",
                &[("mode", mode), ("le", &le)],
                cumulative as f64,
            );
        }
        out.sample(
            "document_lock_wait_seconds_sum",
            &[("mode", mode)],
            waits.sum_us as f64 / 1e6,
        );
        out.sample(
            "document_lock_wait_seconds_count",
            &[("mode", mode)],
            waits.count as f64,
        );
    }
    out.family(
        "document_lock_wait_max_seconds",
        "gauge",
        "Longest wait for the document's lock since start, by mode.",
    );
    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
        out.sample(
            "document_lock_wait_max_seconds",
            &[("mode", mode)],
            waits.max_us as f64 / 1e6,
        );
    }

    out.family(
        "document_lock_wait_seconds",
        "histogram",
        "Time waited for the document's lock, by mode (read, write).",
    );
    let lock_wait = &document.lock_wait;
    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
        let mut cumulative = 0;
        for (le_us, count) in waits.buckets() {
            cumulative += count;
            let le = match le_us {
                Some(us) => (us as f64 / 1e6).to_string(),
                None => "+Inf".to_owned(),
            };
            out.sample(
                "document_lock_wait_seconds_bucket",
                &[("mode", mode), ("le", &le)],
                cumulative as f64,
            );
        }
        out.sample(
            "document_lock_wait_seconds_sum",
            &[("mode", mode)],
            waits.sum_us as f64 / 1e6,
        );
        out.sample(
            "document_lock_wait_seconds_count",
            &[("mode", mode)],
            waits.count as f64,
        );
    }
    out.family(
        "document_lock_wait_max_seconds",
        "gauge",
        "Longest wait for the document's lock since start, by mode.",
    );
    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
        out.sample(
            "document_lock_wait_max_seconds",
            &[("mode", mode)],
            waits.max_us as f64 / 1e6,
        );
    }

    let catch_up = &stats.catch_up;
    out.single(
        "catch_up_sources_total",
        "counter",
        "Catch-up sources chosen (more than one: switched).",
        catch_up.sources as f64,
    );
    if let Some(secs) = catch_up.done_after_secs {
        out.single(
            "catch_up_duration_seconds",
            "gauge",
            "Seconds from start until the initial sync was done.",
            secs,
        );
    }
    if let Some(bytes) = catch_up.sync_bytes {
        out.single(
            "catch_up_sync_bytes",
            "gauge",
            "Sync payload received until the initial sync was done.",
            bytes as f64,
        );
    }

    let remakes = &stats.remakes;
    out.single(
        "remakes_undecided",
        "gauge",
        "Remakes of the local epoch not decided yet (expired ones included).",
        remakes.undecided as f64,
    );
    out.single(
        "remakes_committed",
        "gauge",
        "Committed remakes of the local epoch: 1 while switching; more is a double commit.",
        remakes.committed as f64,
    );
    out.single(
        "rejoining",
        "gauge",
        "1 while syncing a newer epoch from scratch.",
        flag(remakes.rejoining),
    );
    out.single(
        "catch_up_attempts",
        "gauge",
        "Attempts at catching up with a newer epoch (0 when not catching up).",
        remakes.catch_up_attempts as f64,
    );
    out.single(
        "quarantined",
        "gauge",
        "Quarantined change sets (in memory only).",
        remakes.quarantined as f64,
    );
    if let Some(secs) = remakes.oldest_quarantine_secs {
        out.single(
            "oldest_quarantine_seconds",
            "gauge",
            "Age of the oldest quarantined change set.",
            secs as f64,
        );
    }
    out.single(
        "archived_epochs",
        "gauge",
        "Old epochs' documents kept in memory.",
        remakes.archived_epochs as f64,
    );

    if let Some(secs) = stats.sync.last_from_neighbour_secs {
        out.single(
            "last_sync_from_neighbour_seconds",
            "gauge",
            "Seconds since a sync message from any neighbour was applied.",
            secs as f64,
        );
    }
    out.single(
        "largest_sync_received_bytes",
        "gauge",
        "The largest sync message received since start; larger transfers come as deltas.",
        stats.sync.largest_received_bytes as f64,
    );
    out.single(
        "diverged_peers",
        "gauge",
        "Peers whose heads differ from ours while sync is idle.",
        stats.sync.diverged_peers as f64,
    );
    let relay = &stats.relay;
    out.single(
        "announce_rounds_total",
        "counter",
        "Announcement rounds (our new heads to the neighbours).",
        relay.announce_rounds as f64,
    );
    out.single(
        "pulls_total",
        "counter",
        "Pulls of changes we lacked, started.",
        relay.pulls as f64,
    );
    out.single(
        "pulls_queued_total",
        "counter",
        "Announcements of heads we lacked that waited for the pull in progress.",
        relay.pulls_queued as f64,
    );
    out.single(
        "blob_requests_dropped_total",
        "counter",
        "Blob requests not answered: too many were being answered.",
        relay.blob_requests_dropped as f64,
    );

    let topology = &stats.topology;
    out.single(
        "neighbours_target",
        "gauge",
        "Target number of neighbours (k).",
        topology.k as f64,
    );
    if let Some(view) = &topology.view {
        out.single(
            "members",
            "gauge",
            "Registered members, this node included.",
            view.members as f64,
        );
        out.single(
            "reachable_members",
            "gauge",
            "Members other than this node not judged unreachable.",
            view.reachable as f64,
        );
        out.single(
            "stable_members",
            "gauge",
            "Members with the stable tier.",
            view.stable as f64,
        );
        out.single(
            "neighbours",
            "gauge",
            "Current neighbours.",
            view.neighbours as f64,
        );
        out.single(
            "stable_neighbours",
            "gauge",
            "Current neighbours with the stable tier.",
            view.stable_neighbours as f64,
        );
        out.single(
            "full_mesh",
            "gauge",
            "1 while every reachable member is a neighbour.",
            flag(view.full_mesh),
        );
    }

    let stability = &stats.stability;
    out.single(
        "stable",
        "gauge",
        "1 while this node's published tier is stable.",
        flag(stability.tier == "stable"),
    );
    out.single(
        "connected_ratio",
        "gauge",
        "Fraction of the observed time (up to a day) with a live peer.",
        stability.connected_fraction,
    );
    out.single(
        "connectivity_observed_seconds",
        "gauge",
        "Time the connected ratio is based on.",
        stability.observed_secs as f64,
    );

    out.single(
        "stale_hello_replies_total",
        "counter",
        "HELLO_REPLYs to an earlier run of this node, ignored.",
        stats.stale_hello_replies as f64,
    );

    for counter in 0..DISK_COUNTERS {
        for (i, (kind, counts)) in stats.disk.iter().enumerate() {
            let (name, help, value) = disk_counters(counts)[counter];
            if i == 0 {
                out.family(name, "counter", help);
            }
            out.sample(name, &[("file", *kind)], value as f64);
        }
    }

    let propagation = &stats.propagation;
    out.family(
        "propagation_delay_seconds",
        "histogram",
        "Time from a change's commit elsewhere until applied here (includes clock skew).",
    );
    let mut cumulative = 0;
    for bucket in &propagation.buckets {
        cumulative += bucket.count;
        let le = match bucket.le_ms {
            Some(ms) => (ms as f64 / 1000.0).to_string(),
            None => "+Inf".to_owned(),
        };
        out.sample(
            "propagation_delay_seconds_bucket",
            &[("le", &le)],
            cumulative as f64,
        );
    }
    out.sample(
        "propagation_delay_seconds_sum",
        &[],
        propagation.sum_ms as f64 / 1000.0,
    );
    out.sample(
        "propagation_delay_seconds_count",
        &[],
        propagation.count as f64,
    );
    out.single(
        "propagation_clock_ahead_total",
        "counter",
        "Changes stamped later than this node's clock on arrival.",
        propagation.ahead as f64,
    );

    // Node totals, including peers no longer listed.
    let totals = traffic_series(&stats.traffic);
    out.family(
        "messages_total",
        "counter",
        "Messages by direction and kind, all peers.",
    );
    for (direction, kind, counter) in &totals {
        out.sample(
            "messages_total",
            &[("direction", direction), ("kind", kind)],
            counter.messages as f64,
        );
    }
    out.family(
        "payload_bytes_total",
        "counter",
        "Payload bytes by direction and kind, all peers.",
    );
    for (direction, kind, counter) in &totals {
        out.sample(
            "payload_bytes_total",
            &[("direction", direction), ("kind", kind)],
            counter.bytes as f64,
        );
    }

    transports(&mut out, &stats.transports);

    let peers = peers(state).await;

    out.family(
        "peer_info",
        "gauge",
        "A peer's alias from the member list (always 1).",
    );
    for peer in &peers {
        out.sample(
            "peer_info",
            &[
                ("peer", &peer.key),
                ("alias", peer.alias.as_deref().unwrap_or("")),
            ],
            1.0,
        );
    }

    out.family(
        "peer_destination_info",
        "gauge",
        "Where messages to the peer go: its transport, whether it proved itself there, and the connection (always 1).",
    );
    for peer in &peers {
        if let Some(destination) = &peer.destination {
            let transport = scheme(&destination.address);
            out.sample(
                "peer_destination_info",
                &[
                    ("peer", &peer.key),
                    ("transport", transport),
                    ("proven", if destination.proven { "true" } else { "false" }),
                    (
                        "connection",
                        destination.connection.map_or("", |c| c.as_str()),
                    ),
                ],
                1.0,
            );
        }
    }

    out.family(
        "peer_path_rtt_seconds",
        "gauge",
        "Smoothed round-trip time of the PINGs on each path to a peer with several.",
    );
    for peer in &peers {
        for path in peer.destination.iter().flat_map(|d| &d.paths) {
            if let Some(ms) = path.rtt_ms {
                out.sample(
                    "peer_path_rtt_seconds",
                    &[("peer", &peer.key), ("transport", scheme(&path.address))],
                    ms / 1000.0,
                );
            }
        }
    }
    out.family(
        "peer_path_down",
        "gauge",
        "1 while a path to the peer is down (PINGs unanswered, or a send failed).",
    );
    for peer in &peers {
        for path in peer.destination.iter().flat_map(|d| &d.paths) {
            out.sample(
                "peer_path_down",
                &[("peer", &peer.key), ("transport", scheme(&path.address))],
                flag(path.down),
            );
        }
    }

    out.family("peer_neighbour", "gauge", "1 if the peer is a neighbour.");
    for peer in &peers {
        out.sample(
            "peer_neighbour",
            &[("peer", &peer.key)],
            flag(peer.neighbour),
        );
    }
    out.family("peer_dormant", "gauge", "1 while the peer is dormant.");
    for peer in &peers {
        out.sample("peer_dormant", &[("peer", &peer.key)], flag(peer.dormant));
    }
    out.family(
        "peer_last_seen_seconds",
        "gauge",
        "Seconds since the peer's last message.",
    );
    for peer in &peers {
        if let Some(secs) = peer.last_seen_secs {
            out.sample("peer_last_seen_seconds", &[("peer", &peer.key)], secs);
        }
    }
    out.family(
        "peer_last_sync_seconds",
        "gauge",
        "Seconds since a sync message from the peer was applied (or the epoch changed).",
    );
    for peer in &peers {
        if let Some(secs) = peer.last_sync_secs {
            out.sample("peer_last_sync_seconds", &[("peer", &peer.key)], secs);
        }
    }
    out.family(
        "peer_diverged_seconds",
        "gauge",
        "Seconds the peer's heads have differed from ours while sync was idle.",
    );
    for peer in &peers {
        if let Some(secs) = peer.diverged_secs {
            out.sample("peer_diverged_seconds", &[("peer", &peer.key)], secs);
        }
    }
    out.family(
        "peer_sync_throughput_bytes_per_second",
        "gauge",
        "Smoothed sync throughput in large exchanges.",
    );
    for peer in &peers {
        if let Some(throughput) = peer.throughput {
            out.sample(
                "peer_sync_throughput_bytes_per_second",
                &[("peer", &peer.key)],
                throughput,
            );
        }
    }

    out.family(
        "peer_messages_total",
        "counter",
        "Messages by direction and kind.",
    );
    for peer in &peers {
        for (direction, kind, counter) in traffic_series(&peer.traffic) {
            out.sample(
                "peer_messages_total",
                &[
                    ("peer", &peer.key),
                    ("direction", direction),
                    ("kind", kind),
                ],
                counter.messages as f64,
            );
        }
    }
    out.family(
        "peer_payload_bytes_total",
        "counter",
        "Payload bytes by direction and kind.",
    );
    for peer in &peers {
        for (direction, kind, counter) in traffic_series(&peer.traffic) {
            out.sample(
                "peer_payload_bytes_total",
                &[
                    ("peer", &peer.key),
                    ("direction", direction),
                    ("kind", kind),
                ],
                counter.bytes as f64,
            );
        }
    }

    let incidents: [IncidentSeries; 4] = [
        (
            "peer_reply_timeouts_total",
            "Sync messages unanswered until the reply timeout.",
            |i| i.reply_timeouts as f64,
        ),
        (
            "peer_heals_total",
            "Re-handshakes because documents differed while sync was idle.",
            |i| i.heals as f64,
        ),
        (
            "peer_dormancies_total",
            "Times the peer went dormant.",
            |i| i.dormancies as f64,
        ),
        (
            "peer_dormant_seconds_total",
            "Time spent dormant, up to the last return.",
            |i| i.dormant_ms as f64 / 1000.0,
        ),
    ];
    for (name, help, value) in incidents {
        out.family(name, "counter", help);
        for peer in &peers {
            out.sample(name, &[("peer", &peer.key)], value(&peer.incidents));
        }
    }

    out.0
}

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

    #[test]
    fn negotiation() {
        let prometheus = "application/openmetrics-text;version=1.0.0,application/openmetrics-text;version=0.0.1;q=0.75,text/plain;version=0.0.4;q=0.5,*/*;q=0.1";
        assert!(wanted(Some(prometheus)));
        assert!(wanted(Some("text/plain")));
        assert!(!wanted(Some("*/*")));
        assert!(!wanted(Some("application/json")));
        assert!(!wanted(Some("text/html,application/xhtml+xml,*/*;q=0.8")));
        assert!(!wanted(None));
    }

    #[test]
    fn sample_lines() {
        let mut out = Out::default();
        out.single("x_total", "counter", "An x.", 3.0);
        out.sample("y", &[("peer", "a\"b"), ("kind", "sync")], 0.25);
        assert_eq!(
            out.0,
            "# HELP continuum_x_total An x.\n# TYPE continuum_x_total counter\ncontinuum_x_total 3\ncontinuum_y{peer=\"a\\\"b\",kind=\"sync\"} 0.25\n"
        );
    }
}