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<()> {
    let http = build_http_client()?;

    let mut form = vec![("query", logsql.to_owned())];
    if let Some(limit) = limit {
        form.push(("limit", limit.to_string()));
    }
    let send = || -> Result<reqwest::blocking::Response> {
        let bearer = cfg
            .bearer()
            .context("reading this agent's credential for the swarm log store")?;
        http.post(&cfg.query_url)
            .bearer_auth(bearer)
            .form(&form)
            .send()
            .with_context(|| format!("querying the swarm log store at {}", cfg.query_url))
    };

    let mut response = send()?;
    // A 401 can be a secret re-minted between the store read and the check,
    // so it is read again and the query sent once more.
    if response.status() == reqwest::StatusCode::UNAUTHORIZED {
        response = send()?;
    }

    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, trusting the container's own roots.
///
/// No timeout, deliberately: a query over a long retention window legitimately
/// takes long, and a client-side cut would truncate a correct answer into a
/// shorter one with no marker saying so.
fn build_http_client() -> Result<reqwest::blocking::Client> {
    reqwest::blocking::Client::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);
    }
}