hyperhive/swarm-logs/src/query.rs

//! The one request this binary makes: a LogsQL query against the store's
//! machine route on the gateway.
//!
//! The route is `^~ /select/logsql/`, added by `swarm-victorialogs.nix`
//! specifically so a machine caller is answered with a **401 it cannot
//! mistake for an empty result**. The operator's `/` route ends in
//! `error_page 401 =302` and hands an unauthenticated caller authelia's login
//! page as a 200 with an HTML body — which is why [`run`] below reports a
//! non-200 with its body rather than printing nothing and exiting 0. That
//! failure mode is the reason the route exists; reproducing it in the client
//! would be the whole bug again one layer up.

use std::io::{BufRead as _, BufReader, Write};

use anyhow::{Context as _, Result, bail};

use crate::auth::Config;

/// How the matched records are written to stdout.
#[derive(Clone, Copy, Debug, PartialEq, Eq, clap::ValueEnum)]
pub enum Format {
    /// One log message per line — the `_msg` field and nothing else.
    ///
    /// The default because the ask this CLI answers was to *pipe and grep*
    /// rather than hand-roll curl, and a grep pattern is written against the
    /// message. A record with no `_msg` is passed through as its own JSON
    /// object rather than dropped: silently losing a row would make an
    /// incomplete answer indistinguishable from a complete one.
    Message,
    /// The store's response body, unmodified, one JSON object per line.
    ///
    /// For everything [`Format::Message`] leaves out — `_time`, `_stream`
    /// and whatever fields the record carries — and for feeding `jq`.
    Json,
}

/// Run `query`, writing matched records to `out`.
///
/// Streams rather than collecting: the route applies no scoping, so a loose
/// query matches the whole swarm's logs and buffering the answer to print it
/// is an out-of-memory waiting for a bad `LogsQL`.
pub fn run(
    cfg: &Config,
    logsql: &str,
    limit: Option<u64>,
    format: Format,
    out: &mut impl Write,
) -> Result<()> {
    // Minted per invocation, with the query URL as the audience and the
    // `authelia.bearer.authz` scope. All three are load-bearing: authelia
    // refuses a token carrying no audience at the authz endpoint the
    // gateway's `auth_request` calls, and the audience it checks is the URL
    // being requested — so the string sent here and the string requested
    // below must be one binding, which is why `Config` holds exactly one.
    //
    // The scope has to be ASKED for, not merely registered: the agent client
    // is granted `authelia.bearer.authz` by the `agentClients` entry in
    // `swarm-authelia.nix`, but an OAuth2 server issues no scope the client
    // never requested, and a scopeless token is refused at the authz endpoint
    // exactly as an unauthenticated one is — a bare nginx 401 with nothing in
    // it that names the scope. `swarm-otel.nix` records the same failure
    // against the collector's client, on the same string.
    let token = swarm_queue_client::mint_token_for_blocking(
        &cfg.queue,
        Some(&cfg.query_url),
        Some("authelia.bearer.authz"),
    )
    .map_err(|e| anyhow::anyhow!("{}", swarm_queue_client::chain(&e)))
    .context("minting an access token for the swarm log store")?;

    let http = build_http_client(cfg)?;

    let mut form = vec![("query", logsql.to_owned())];
    if let Some(limit) = limit {
        form.push(("limit", limit.to_string()));
    }

    let response = http
        .post(&cfg.query_url)
        .bearer_auth(&token)
        .form(&form)
        .send()
        .with_context(|| format!("querying the swarm log store at {}", cfg.query_url))?;

    let status = response.status();
    if !status.is_success() {
        // Surfaced, not swallowed. A 401 here is the ordinary shape of "this
        // client's identity is not authorised for this route", and the body
        // is the half that says which — nginx answers its own HTML for an
        // `auth_request` denial, the store answers JSON for a bad LogsQL.
        // Truncated because the login-page case is a full HTML document and
        // an operator needs the first screen of it, not all of it.
        let body = response.text().unwrap_or_default();
        let body = body.trim();
        let shown: String = body.chars().take(2000).collect();
        bail!(
            "the swarm log store refused the query ({status}): {shown}{}",
            if shown.len() < body.chars().count() {
                " […]"
            } else {
                ""
            }
        );
    }

    let mut reader = BufReader::new(response);
    let mut line = String::new();
    loop {
        line.clear();
        let read = reader
            .read_line(&mut line)
            .context("reading the query response from the swarm log store")?;
        if read == 0 {
            break;
        }
        let line = line.trim_end_matches(['\r', '\n']);
        if line.is_empty() {
            continue;
        }
        writeln!(out, "{}", render(line, format)).context("writing to stdout")?;
    }
    Ok(())
}

/// Render one NDJSON record from the store as the caller asked for it.
///
/// Pure over the line so the projection is testable without a store: what
/// goes wrong here is a record shape, not a request.
fn render(line: &str, format: Format) -> String {
    match format {
        Format::Json => line.to_owned(),
        Format::Message => serde_json::from_str::<serde_json::Value>(line)
            .ok()
            .and_then(|v| v.get("_msg")?.as_str().map(str::to_owned))
            // Anything without a string `_msg` — including a line that is not
            // JSON at all — falls back to the line itself. See
            // [`Format::Message`]: dropping it would make a partial answer
            // look like a whole one.
            .unwrap_or_else(|| line.to_owned()),
    }
}

/// The HTTP client for the query itself.
///
/// Separate from the one `swarm-queue-client` builds for the token endpoint
/// because they are two hosts — `auth.<swarm>` and `logs.<swarm>` — and the
/// CA-trust knob is declared per endpoint. Same optional anchor, same reason:
/// without it a swarm using its own CA fails at TLS with
/// `invalid peer certificate: UnknownIssuer`, having never looked at the
/// anchor sitting on the host.
///
/// No timeout, deliberately, where the token client has one: a query over a
/// long retention window legitimately takes longer than any number that would
/// be right for a token request, and a client-side cut would truncate a
/// correct answer into a shorter one with no marker saying so.
fn build_http_client(cfg: &Config) -> Result<reqwest::blocking::Client> {
    let mut builder = reqwest::blocking::Client::builder();
    if let Some(path) = &cfg.queue.ca_file {
        let pem = std::fs::read(path).with_context(|| {
            format!(
                "reading the log-store CA certificate from {}",
                path.display()
            )
        })?;
        let cert = reqwest::Certificate::from_pem(&pem).with_context(|| {
            format!(
                "parsing the log-store CA certificate from {} as PEM",
                path.display()
            )
        })?;
        builder = builder.add_root_certificate(cert);
    }
    builder
        .build()
        .context("building the log-store HTTP client")
}

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

    #[test]
    fn json_passes_the_record_through_unmodified() {
        let line = r#"{"_time":"2026-09-16T14:14:43Z","_msg":"hello","_stream":"{}"}"#;
        assert_eq!(render(line, Format::Json), line);
    }

    #[test]
    fn message_projects_just_the_log_line() {
        let line = r#"{"_time":"2026-09-16T14:14:43Z","_msg":"hello there"}"#;
        assert_eq!(render(line, Format::Message), "hello there");
    }

    /// The grep case this CLI exists for: a marker inside `_msg` has to
    /// survive the projection intact, punctuation and all.
    #[test]
    fn message_keeps_a_marker_intact() {
        let line = r#"{"_msg":"probe marker-1789568083-6970 emitted"}"#;
        assert!(render(line, Format::Message).contains("marker-1789568083-6970"));
    }

    /// A record the projection cannot understand is still a match, and a
    /// caller counting lines must not get a different answer than one reading
    /// them. Dropping it would make an incomplete answer look complete.
    #[test]
    fn a_record_without_a_message_falls_back_to_the_whole_line() {
        let line = r#"{"_time":"2026-09-16T14:14:43Z","_stream":"{}"}"#;
        assert_eq!(render(line, Format::Message), line);
    }

    #[test]
    fn a_non_json_line_falls_back_to_itself() {
        assert_eq!(
            render("not json at all", Format::Message),
            "not json at all"
        );
    }

    /// `_msg` holding a non-string is the same "shape I cannot project" case
    /// as a missing one, and must take the same fallback rather than
    /// rendering Rust's debug form of a number.
    #[test]
    fn a_non_string_message_falls_back_to_the_whole_line() {
        let line = r#"{"_msg":42}"#;
        assert_eq!(render(line, Format::Message), line);
    }
}