feat(haproxy): implement HAProxy log shipper with initial configuration and dependencies

rust/haproxy_log_shipper/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "haproxy-log-shipper"
+version = "0.1.0"
+edition = "2024"
+description = "Ordered HAProxy journald ingestion client"
+license = "MIT"
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow.workspace = true
+chrono = { workspace = true, features = ["serde"] }
+clap.workspace = true
+garde.workspace = true
+haproxy-ingest-proto = { path = "../haproxy_ingest_proto" }
+prost.workspace = true
+reqwest = { workspace = true, features = ["json"] }
+serde.workspace = true
+serde_json.workspace = true
+sha2.workspace = true
+systemd.workspace = true
+tls.workspace = true
+tokio.workspace = true
+tokio-retry.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+zstd.workspace = true

rust/haproxy_log_shipper/package.nix

@@ -0,0 +1,24 @@
+{
+  lib,
+  pkg-config,
+  rustPlatform,
+  systemd,
+}:
+rustPlatform.buildRustPackage {
+  pname = "haproxy-log-shipper";
+  version = "0.1.0";
+  src = import ../rust-source.nix { inherit lib; };
+
+  cargoLock.lockFile = ../../Cargo.lock;
+  cargoBuildFlags = [ "-p haproxy-log-shipper" ];
+  cargoTestFlags = [ "-p haproxy-log-shipper" ];
+
+  nativeBuildInputs = [ pkg-config ];
+  buildInputs = [ systemd ];
+
+  meta = {
+    description = "Ordered HAProxy journald ingestion client";
+    license = lib.licenses.mit;
+    platforms = lib.platforms.linux;
+  };
+}

rust/haproxy_log_shipper/src/api.rs

@@ -0,0 +1,360 @@
+use std::time::Duration;
+
+use anyhow::{Context, Result, bail};
+use haproxy_ingest_proto::{
+    SCHEMA_VERSION,
+    proto::{LogBatch, LogRecord, log_record},
+};
+use prost::Message;
+use reqwest::{Client, StatusCode, header};
+use serde::Deserialize;
+use tokio_retry::strategy::{ExponentialBackoff, jitter};
+use tracing::{Instrument, info, warn};
+
+use crate::mapper::rejected_event;
+
+const INITIAL_RETRIES: usize = 3;
+
+#[derive(Debug, Deserialize)]
+pub struct StatusResponse {
+    machine_id: String,
+    pub last_cursor: Option<String>,
+    last_journal_timestamp: Option<chrono::DateTime<chrono::Utc>>,
+}
+
+#[derive(Debug, Deserialize)]
+struct CommitResponse {
+    last_cursor: String,
+    records_committed: usize,
+    already_committed: bool,
+}
+
+#[derive(Debug, Deserialize)]
+struct ErrorResponse {
+    error: String,
+    record_index: Option<usize>,
+    last_cursor: Option<String>,
+}
+
+#[derive(Clone, Debug)]
+pub struct PendingRecord {
+    pub record: LogRecord,
+    pub cursor: String,
+    pub raw_message: Option<String>,
+}
+
+#[derive(Debug)]
+pub struct PendingBatch {
+    pub machine_id: String,
+    pub hostname: String,
+    pub previous_cursor: Option<String>,
+    pub records: Vec<PendingRecord>,
+}
+
+impl PendingBatch {
+    fn last_cursor(&self) -> Result<&str> {
+        self.records
+            .last()
+            .map(|record| record.cursor.as_str())
+            .context("pending batch is empty")
+    }
+
+    pub fn to_wire(&self) -> Result<LogBatch> {
+        let first_journal_timestamp_us = self
+            .records
+            .first()
+            .context("pending batch is empty")?
+            .record
+            .journal_timestamp_us;
+        let last_journal_timestamp_us = self
+            .records
+            .last()
+            .context("pending batch is empty")?
+            .record
+            .journal_timestamp_us;
+        Ok(LogBatch {
+            schema_version: SCHEMA_VERSION,
+            machine_id: self.machine_id.clone(),
+            hostname: self.hostname.clone(),
+            previous_cursor: self.previous_cursor.clone(),
+            last_cursor: self.last_cursor()?.to_owned(),
+            first_journal_timestamp_us,
+            last_journal_timestamp_us,
+            records: self
+                .records
+                .iter()
+                .map(|pending| pending.record.clone())
+                .collect(),
+        })
+    }
+
+    fn replace_with_rejection(&mut self, index: usize, reason: &str) -> Result<()> {
+        let pending = self
+            .records
+            .get_mut(index)
+            .with_context(|| format!("receiver returned out-of-range record index {index}"))?;
+        if matches!(
+            pending.record.payload,
+            Some(log_record::Payload::Rejected(_))
+        ) {
+            bail!("receiver rejected an already-isolated RejectedEvent at index {index}");
+        }
+        let raw = pending
+            .raw_message
+            .as_deref()
+            .context("receiver rejected a journal gap marker")?;
+        pending.record.payload = Some(log_record::Payload::Rejected(rejected_event(
+            raw,
+            &format!("receiver validation rejected record: {reason}"),
+        )));
+        Ok(())
+    }
+}
+
+pub enum SendResult {
+    Committed(String),
+    Resync,
+}
+
+enum PostBatchResult {
+    Complete(SendResult),
+    RetryNow,
+    RetryAfterStatusCheck,
+}
+
+enum StatusRequestResult {
+    Complete(StatusResponse),
+    Retry,
+}
+
+pub async fn post_batch_with_retry(
+    client: &Client,
+    endpoint: &str,
+    pending: &mut PendingBatch,
+) -> Result<SendResult> {
+    for (attempt, delay) in retry_schedule().enumerate() {
+        if let Some(delay) = delay {
+            tokio::time::sleep(delay).await;
+        }
+        let last_cursor = loop {
+            let batch = pending.to_wire()?;
+            let last_cursor = batch.last_cursor.clone();
+            let encoded = batch.encode_to_vec();
+            let compressed = zstd::stream::encode_all(encoded.as_slice(), 3)
+                .context("failed to compress protobuf batch")?;
+            let result = post_batch(client, endpoint, pending, compressed, &last_cursor)
+                .instrument(tracing::info_span!("post_batch", attempt))
+                .await?;
+            match result {
+                PostBatchResult::Complete(send_result) => return Ok(send_result),
+                PostBatchResult::RetryNow => {}
+                PostBatchResult::RetryAfterStatusCheck => break last_cursor,
+            }
+        };
+
+        if let Ok(status) = get_status_with_retry(client, endpoint, &pending.machine_id).await {
+            if status.last_cursor.as_deref() == Some(last_cursor.as_str()) {
+                info!(last_cursor, "status check confirmed ambiguous batch commit");
+                return Ok(SendResult::Committed(last_cursor));
+            }
+            if status.last_cursor != pending.previous_cursor {
+                return Ok(SendResult::Resync);
+            }
+        }
+    }
+    bail!("batch failed after initial attempt and {INITIAL_RETRIES} retries")
+}
+
+async fn post_batch(
+    client: &Client,
+    endpoint: &str,
+    pending: &mut PendingBatch,
+    compressed: Vec<u8>,
+    last_cursor: &str,
+) -> Result<PostBatchResult> {
+    let response = client
+        .post(format!("{endpoint}/v1/logs/batch"))
+        .header(header::CONTENT_TYPE, "application/x-protobuf")
+        .header(header::CONTENT_ENCODING, "zstd")
+        .body(compressed)
+        .send()
+        .await;
+
+    match response {
+        Ok(response) if response.status().is_success() => {
+            let committed: CommitResponse = response
+                .json()
+                .await
+                .context("receiver returned an invalid commit response")?;
+            if committed.last_cursor != last_cursor
+                || committed.records_committed != pending.records.len()
+            {
+                bail!("receiver commit response did not match the sent batch");
+            }
+            info!(
+                already_committed = committed.already_committed,
+                "batch acknowledged"
+            );
+            Ok(PostBatchResult::Complete(SendResult::Committed(
+                last_cursor.to_owned(),
+            )))
+        }
+        Ok(response) if response.status() == StatusCode::CONFLICT => {
+            let error = decode_error(response).await;
+            warn!(server_cursor = ?error.last_cursor, error = %error.error, "receiver rejected cursor chain");
+            Ok(PostBatchResult::Complete(SendResult::Resync))
+        }
+        Ok(response) if response.status() == StatusCode::PAYLOAD_TOO_LARGE => {
+            let error = decode_error(response).await;
+            bail!(
+                "receiver rejected a batch within the sender's configured limits as too large: {}",
+                error.error
+            )
+        }
+        Ok(response) if response.status() == StatusCode::UNPROCESSABLE_ENTITY => {
+            let error = decode_error(response).await;
+            let index = error
+                .record_index
+                .context("receiver returned 422 without an offending record index")?;
+            pending.replace_with_rejection(index, &error.error)?;
+            warn!(record_index = index, error = %error.error, "isolated receiver-rejected record");
+            Ok(PostBatchResult::RetryNow)
+        }
+        Ok(response)
+            if matches!(
+                response.status(),
+                StatusCode::UNAUTHORIZED
+                    | StatusCode::FORBIDDEN
+                    | StatusCode::BAD_REQUEST
+                    | StatusCode::UNSUPPORTED_MEDIA_TYPE
+            ) =>
+        {
+            let status = response.status();
+            let error = decode_error(response).await;
+            bail!("permanent receiver error {status}: {}", error.error);
+        }
+        Ok(response) if !response.status().is_server_error() => {
+            let status = response.status();
+            let error = decode_error(response).await;
+            bail!("unexpected receiver response {status}: {}", error.error);
+        }
+        Ok(response) => {
+            warn!(status = %response.status(), "transient receiver failure");
+            Ok(PostBatchResult::RetryAfterStatusCheck)
+        }
+        Err(error) => {
+            warn!(error = %error, "ambiguous batch request failure");
+            Ok(PostBatchResult::RetryAfterStatusCheck)
+        }
+    }
+}
+
+pub async fn get_status_with_retry(
+    client: &Client,
+    endpoint: &str,
+    machine_id: &str,
+) -> Result<StatusResponse> {
+    for (attempt, delay) in retry_schedule().enumerate() {
+        if let Some(delay) = delay {
+            tokio::time::sleep(delay).await;
+        }
+        let result = get_status(client, endpoint, machine_id)
+            .instrument(tracing::info_span!("get_status", attempt))
+            .await?;
+        match result {
+            StatusRequestResult::Complete(status) => return Ok(status),
+            StatusRequestResult::Retry => {}
+        }
+    }
+    bail!("status request failed after initial attempt and {INITIAL_RETRIES} retries")
+}
+
+async fn get_status(
+    client: &Client,
+    endpoint: &str,
+    machine_id: &str,
+) -> Result<StatusRequestResult> {
+    let response = match client
+        .get(format!("{endpoint}/v1/logs/status/{machine_id}"))
+        .send()
+        .await
+    {
+        Ok(response) => response,
+        Err(error) => {
+            warn!(error = %error, "transient receiver status failure");
+            return Ok(StatusRequestResult::Retry);
+        }
+    };
+    if !response.status().is_success() {
+        let status = response.status();
+        let error = decode_error(response).await;
+        if status.is_server_error() {
+            warn!(status = %status, error = %error.error, "transient receiver status failure");
+            return Ok(StatusRequestResult::Retry);
+        }
+        bail!(
+            "receiver status request failed with {status}: {}",
+            error.error
+        );
+    }
+    let status: StatusResponse = response
+        .json()
+        .await
+        .context("receiver returned an invalid status response")?;
+    if status.machine_id != machine_id {
+        bail!("receiver status response contains the wrong machine_id");
+    }
+    info!(
+        last_cursor = ?status.last_cursor,
+        last_journal_timestamp = ?status.last_journal_timestamp,
+        "loaded receiver checkpoint"
+    );
+    Ok(StatusRequestResult::Complete(status))
+}
+
+fn retry_delays() -> impl Iterator<Item = Duration> {
+    // The library raises the base to each attempt, then multiplies by the factor.
+    // 2^n * 500 milliseconds gives delays of 1, 2, and 4 seconds.
+    ExponentialBackoff::from_millis(2)
+        .factor(500)
+        .take(INITIAL_RETRIES)
+}
+
+fn retry_schedule() -> impl Iterator<Item = Option<Duration>> {
+    std::iter::once(None).chain(retry_delays().map(jitter).map(Some))
+}
+
+async fn decode_error(response: reqwest::Response) -> ErrorResponse {
+    response.json().await.unwrap_or_else(|error| ErrorResponse {
+        error: format!("response body was not valid error JSON: {error}"),
+        record_index: None,
+        last_cursor: None,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn retries_use_one_two_and_four_second_delays() {
+        assert_eq!(
+            retry_delays().collect::<Vec<_>>(),
+            [
+                Duration::from_secs(1),
+                Duration::from_secs(2),
+                Duration::from_secs(4),
+            ]
+        );
+    }
+
+    #[test]
+    fn retry_schedule_starts_immediately_and_bounds_jitter() {
+        let mut schedule = retry_schedule();
+        assert_eq!(schedule.next(), Some(None));
+        for limit in retry_delays() {
+            assert!(matches!(schedule.next(), Some(Some(delay)) if delay <= limit));
+        }
+        assert_eq!(schedule.next(), None);
+    }
+}

rust/haproxy_log_shipper/src/main.rs

@@ -0,0 +1,271 @@
+use std::{fs, time::Duration};
+
+use anyhow::{Context, Result, anyhow, bail};
+use clap::Parser;
+use haproxy_ingest_proto::{
+    MAX_BATCH_PROTO_BYTES, MAX_BATCH_RECORDS,
+    proto::{JournalGap, LogRecord, log_record},
+};
+use prost::Message;
+use reqwest::{Client, header};
+use systemd::journal::{Journal, OpenOptions};
+use tls::install_tls_provider;
+use tracing::{info, warn};
+use tracing_subscriber::EnvFilter;
+
+mod api;
+mod mapper;
+
+use api::{PendingBatch, PendingRecord, SendResult, get_status_with_retry, post_batch_with_retry};
+use mapper::map_message;
+
+#[derive(Debug, Parser)]
+#[command(about, version)]
+struct Args {
+    #[arg(long, env = "HAPROXY_INGEST_TOKEN", hide_env_values = true)]
+    token: String,
+
+    #[arg(long, env = "HAPROXY_INGEST_ENDPOINT")]
+    endpoint: String,
+
+    #[arg(long, env = "HAPROXY_INGEST_UNIT", default_value = "haproxy.service")]
+    unit: String,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .json()
+        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()))
+        .init();
+    let args = Args::parse();
+    install_tls_provider()?;
+    let machine_id = get_valid_machine_id()?;
+    let hostname = read_trimmed("/etc/hostname")?;
+    let client = build_client(&args.token)?;
+
+    let mut committed_cursor = get_status_with_retry(&client, &args.endpoint, &machine_id)
+        .await?
+        .last_cursor;
+    loop {
+        let Some(mut batch) =
+            read_batch(&machine_id, &hostname, committed_cursor.clone(), &args.unit)?
+        else {
+            info!(machine_id, "no uncommitted HAProxy journal records");
+            return Ok(());
+        };
+
+        committed_cursor = match post_batch_with_retry(&client, &args.endpoint, &mut batch).await? {
+            SendResult::Committed(cursor) => {
+                info!(
+                    records = batch.records.len(),
+                    last_cursor = %cursor,
+                    "receiver committed HAProxy batch"
+                );
+                Some(cursor)
+            }
+            SendResult::Resync => {
+                let status = get_status_with_retry(&client, &args.endpoint, &machine_id).await?;
+
+                warn!(
+                    server_cursor = ?status.last_cursor,
+                    "cursor conflict; rebuilding from receiver checkpoint"
+                );
+
+                status.last_cursor
+            }
+        };
+    }
+}
+
+fn build_client(token: &str) -> Result<Client> {
+    let authorization = format!("Bearer {token}")
+        .parse()
+        .context("HAPROXY_INGEST_TOKEN contains invalid HTTP header characters")?;
+    Client::builder()
+        .connect_timeout(Duration::from_secs(3))
+        .timeout(Duration::from_secs(10))
+        .default_headers({
+            let mut headers = reqwest::header::HeaderMap::new();
+            headers.insert(header::AUTHORIZATION, authorization);
+            headers
+        })
+        .build()
+        .context("failed to build HTTP client")
+}
+
+fn read_batch(
+    machine_id: &str,
+    hostname: &str,
+    previous_cursor: Option<String>,
+    unit: &str,
+) -> Result<Option<PendingBatch>> {
+    let (mut journal, setup) = setup_journal(machine_id, unit, previous_cursor.as_deref())?;
+    let mut pending_batch = PendingBatch {
+        machine_id: machine_id.to_owned(),
+        hostname: hostname.to_owned(),
+        previous_cursor,
+        records: Vec::new(),
+    };
+    match setup {
+        JournalSetup::Empty => return Ok(None),
+        JournalSetup::Ready => {}
+        JournalSetup::Gap(pending_record) => pending_batch.records.push(*pending_record),
+    }
+    while pending_batch.records.len() < MAX_BATCH_RECORDS {
+        let Some(entry) = journal
+            .next_entry()
+            .context("failed to read journal entry")?
+        else {
+            break;
+        };
+        let cursor = journal.cursor().context("journal entry has no cursor")?;
+        let journal_timestamp_us = i64::try_from(
+            journal
+                .timestamp_usec()
+                .context("journal entry has no timestamp")?,
+        )
+        .context("journal timestamp exceeds int64")?;
+        let message = entry.get("MESSAGE").cloned().unwrap_or_default();
+        pending_batch.records.push(PendingRecord {
+            record: map_message(&message, journal_timestamp_us),
+            cursor,
+            raw_message: Some(message),
+        });
+        if pending_batch.to_wire()?.encoded_len() > MAX_BATCH_PROTO_BYTES {
+            if pending_batch.records.len() == 1 {
+                bail!("single mapped journal record exceeds the 20 MiB protobuf limit");
+            }
+            pending_batch.records.pop();
+            break;
+        }
+    }
+    if pending_batch.records.is_empty() {
+        Ok(None)
+    } else {
+        Ok(Some(pending_batch))
+    }
+}
+
+enum JournalSetup {
+    Empty,
+    Ready,
+    Gap(Box<PendingRecord>),
+}
+
+fn setup_journal(
+    machine_id: &str,
+    unit: &str,
+    previous_cursor: Option<&str>,
+) -> Result<(Journal, JournalSetup)> {
+    let mut journal = OpenOptions::default()
+        .system(true)
+        .local_only(true)
+        .open()
+        .context("failed to open the local system journal")?;
+    journal
+        .match_add("_SYSTEMD_UNIT", unit.as_bytes())
+        .context("failed to filter journal by systemd unit")?;
+    journal
+        .match_add("_MACHINE_ID", machine_id.as_bytes())
+        .context("failed to filter journal by machine ID")?;
+    let setup = position_journal(&mut journal, previous_cursor)?;
+    Ok((journal, setup))
+}
+
+fn position_journal(journal: &mut Journal, previous_cursor: Option<&str>) -> Result<JournalSetup> {
+    let Some(cursor) = previous_cursor else {
+        journal
+            .seek_head()
+            .context("failed to seek to journal head")?;
+        return Ok(JournalSetup::Ready);
+    };
+    let exact = journal.seek_cursor(cursor).is_ok()
+        && journal
+            .next()
+            .context("failed to move to saved journal cursor")?
+            > 0
+        && journal
+            .test_cursor(cursor)
+            .context("failed to test saved journal cursor")?;
+    if exact {
+        return Ok(JournalSetup::Ready);
+    }
+
+    journal
+        .seek_head()
+        .context("failed to seek to retained journal head")?;
+    if journal
+        .next()
+        .context("failed to read retained journal head")?
+        == 0
+    {
+        return Ok(JournalSetup::Empty);
+    }
+    let first_cursor = journal
+        .cursor()
+        .context("retained journal entry has no cursor")?;
+    let first_timestamp = i64::try_from(
+        journal
+            .timestamp_usec()
+            .context("retained journal entry has no timestamp")?,
+    )
+    .context("journal timestamp exceeds int64")?;
+    let gap = PendingRecord {
+        record: LogRecord {
+            journal_timestamp_us: first_timestamp,
+            payload: Some(log_record::Payload::Gap(JournalGap {
+                expected_cursor: cursor.to_owned(),
+                first_available_cursor: first_cursor.clone(),
+                first_available_timestamp_us: Some(first_timestamp),
+                reason: "saved cursor is no longer present in the filtered journal".to_owned(),
+            })),
+        },
+        cursor: first_cursor,
+        raw_message: None,
+    };
+    journal
+        .seek_head()
+        .context("failed to rewind retained journal")?;
+    Ok(JournalSetup::Gap(Box::new(gap)))
+}
+
+fn get_valid_machine_id() -> Result<String> {
+    let machine_id = read_trimmed("/etc/machine-id")?;
+    if machine_id.len() != 32
+        || !machine_id
+            .bytes()
+            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
+    {
+        bail!("/etc/machine-id is not 32 lowercase hexadecimal characters");
+    }
+    Ok(machine_id)
+}
+
+fn read_trimmed(path: &str) -> Result<String> {
+    let value = fs::read_to_string(path).with_context(|| format!("failed to read {path}"))?;
+    let value = value.trim();
+    if value.is_empty() {
+        return Err(anyhow!("{path} is empty"));
+    }
+    Ok(value.to_owned())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn builds_http_client_without_certificate() -> Result<()> {
+        let args = Args::try_parse_from([
+            "haproxy-log-shipper",
+            "--endpoint",
+            "http://127.0.0.1:8444",
+            "--token",
+            "test-token",
+        ])?;
+        install_tls_provider()?;
+        build_client(&args.token)?;
+        Ok(())
+    }
+}

rust/haproxy_log_shipper/src/mapper.rs

@@ -0,0 +1,354 @@
+use std::{fmt::Display, net::IpAddr, str::FromStr};
+
+use anyhow::{Context, Result};
+use garde::{Unvalidated, Valid};
+use haproxy_ingest_proto::proto::{HaProxyEvent, LogRecord, RejectedEvent, log_record};
+use serde::{Deserialize, Deserializer, de::Error as _};
+use sha2::{Digest, Sha256};
+
+const RAW_PREFIX_LIMIT: usize = 4 * 1024;
+
+#[derive(Debug, Deserialize, garde::Validate)]
+#[garde(allow_unvalidated)]
+struct RawHaProxyEvent {
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    client_ip: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 65_535))]
+    client_port: Option<u32>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    frontend_name: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    backend_name: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    server_name: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    http_method: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    request_path: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    http_version: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(required, range(min = 100, max = 599))]
+    status_code: Option<i32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(required, range(max = 9_223_372_036_854_775_807))]
+    bytes_sent: Option<u64>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(required, range(min = -1))]
+    time_request_ms: Option<i32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(required, range(min = -1))]
+    time_wait_ms: Option<i32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(required, range(min = -1))]
+    time_connect_ms: Option<i32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(required, range(min = -1))]
+    time_response_ms: Option<i32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(required, range(min = -1))]
+    time_total_ms: Option<i32>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    termination_state: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 2_147_483_647))]
+    retries: Option<u32>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    tls_sni: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    tls_version: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_string")]
+    tls_cipher: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_optional_bool")]
+    tls_resumed: Option<bool>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 2_147_483_647))]
+    active_connections: Option<u32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 2_147_483_647))]
+    frontend_connections: Option<u32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 2_147_483_647))]
+    backend_connections: Option<u32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 2_147_483_647))]
+    server_connections: Option<u32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 2_147_483_647))]
+    server_queue: Option<u32>,
+    #[serde(default, deserialize_with = "deserialize_optional_number")]
+    #[garde(range(max = 2_147_483_647))]
+    backend_queue: Option<u32>,
+}
+
+#[must_use]
+pub fn map_message(message: &str, journal_timestamp_us: i64) -> LogRecord {
+    let payload = match parse_event(message) {
+        Ok(event) => log_record::Payload::Event(event),
+        Err(error) => log_record::Payload::Rejected(rejected_event(message, &format!("{error:#}"))),
+    };
+    LogRecord {
+        journal_timestamp_us,
+        payload: Some(payload),
+    }
+}
+
+fn parse_event(message: &str) -> Result<HaProxyEvent> {
+    let event: Unvalidated<RawHaProxyEvent> =
+        serde_json::from_str(message).context("invalid HAProxy JSON")?;
+    let event = event.validate().context("invalid HAProxy event")?;
+    normalize_event(event)
+}
+
+fn normalize_event(event: Valid<RawHaProxyEvent>) -> Result<HaProxyEvent> {
+    let event = event.into_inner();
+    let mut normalization_flags = Vec::new();
+    let client_ip = event.client_ip.and_then(|raw| {
+        raw.parse::<IpAddr>().map_or_else(
+            |_| {
+                normalization_flags.push("invalid_client_ip".to_owned());
+                None
+            },
+            |ip| Some(ip.to_string()),
+        )
+    });
+
+    Ok(HaProxyEvent {
+        client_ip,
+        client_port: event.client_port,
+        frontend_name: event.frontend_name,
+        backend_name: event.backend_name,
+        server_name: event.server_name,
+        http_method: event.http_method,
+        request_path: event.request_path,
+        http_version: event.http_version,
+        status_code: event
+            .status_code
+            .context("validated status_code is absent")?,
+        bytes_sent: event.bytes_sent.context("validated bytes_sent is absent")?,
+        time_request_ms: event
+            .time_request_ms
+            .context("validated time_request_ms is absent")?,
+        time_wait_ms: event
+            .time_wait_ms
+            .context("validated time_wait_ms is absent")?,
+        time_connect_ms: event
+            .time_connect_ms
+            .context("validated time_connect_ms is absent")?,
+        time_response_ms: event
+            .time_response_ms
+            .context("validated time_response_ms is absent")?,
+        time_total_ms: event
+            .time_total_ms
+            .context("validated time_total_ms is absent")?,
+        termination_state: event.termination_state,
+        retries: event.retries.unwrap_or_default(),
+        tls_sni: event.tls_sni,
+        tls_version: event.tls_version,
+        tls_cipher: event.tls_cipher,
+        tls_resumed: event.tls_resumed,
+        normalization_flags,
+        active_connections: event.active_connections,
+        frontend_connections: event.frontend_connections,
+        backend_connections: event.backend_connections,
+        server_connections: event.server_connections,
+        server_queue: event.server_queue,
+        backend_queue: event.backend_queue,
+    })
+}
+
+#[must_use]
+pub fn rejected_event(message: &str, reason: &str) -> RejectedEvent {
+    let mut hasher = Sha256::new();
+    hasher.update(message.as_bytes());
+    let prefix = truncate_utf8(message, RAW_PREFIX_LIMIT);
+    RejectedEvent {
+        reason: truncate_utf8(reason, 1_024).to_owned(),
+        message_sha256: hasher.finalize().to_vec(),
+        raw_message_prefix: Some(prefix.to_owned()),
+        raw_message_truncated: prefix.len() != message.len(),
+    }
+}
+
+fn deserialize_optional_string<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    match Option::<String>::deserialize(deserializer)? {
+        Some(value) if is_empty_or_dash(&value) => Ok(None),
+        value => Ok(value),
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum NumberOrString<T> {
+    Number(T),
+    String(String),
+}
+
+fn deserialize_optional_number<'de, D, T>(deserializer: D) -> Result<Option<T>, D::Error>
+where
+    D: Deserializer<'de>,
+    T: Deserialize<'de> + FromStr,
+    T::Err: Display,
+{
+    match Option::<NumberOrString<T>>::deserialize(deserializer)? {
+        None => Ok(None),
+        Some(NumberOrString::Number(value)) => Ok(Some(value)),
+        Some(NumberOrString::String(value)) if is_empty_or_dash(&value) => Ok(None),
+        Some(NumberOrString::String(value)) => value.parse().map(Some).map_err(D::Error::custom),
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum BoolValue {
+    Bool(bool),
+    Integer(i64),
+    String(String),
+}
+
+fn deserialize_optional_bool<'de, D>(deserializer: D) -> Result<Option<bool>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    match Option::<BoolValue>::deserialize(deserializer)? {
+        None => Ok(None),
+        Some(BoolValue::Bool(value)) => Ok(Some(value)),
+        Some(BoolValue::Integer(0)) => Ok(Some(false)),
+        Some(BoolValue::Integer(1)) => Ok(Some(true)),
+        Some(BoolValue::String(value)) if is_empty_or_dash(&value) => Ok(None),
+        Some(BoolValue::String(value)) if matches!(value.as_str(), "0" | "false") => {
+            Ok(Some(false))
+        }
+        Some(BoolValue::String(value)) if matches!(value.as_str(), "1" | "true") => Ok(Some(true)),
+        Some(BoolValue::Integer(_) | BoolValue::String(_)) => {
+            Err(D::Error::custom("expected a boolean, 0, 1, null, or '-'"))
+        }
+    }
+}
+
+fn is_empty_or_dash(value: &str) -> bool {
+    value.is_empty() || value == "-"
+}
+
+fn truncate_utf8(value: &str, max_bytes: usize) -> &str {
+    if value.len() <= max_bytes {
+        return value;
+    }
+    let mut end = max_bytes;
+    while !value.is_char_boundary(end) {
+        end = end.saturating_sub(1);
+    }
+    value.get(..end).unwrap_or_default()
+}
+
+#[cfg(test)]
+#[allow(clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use haproxy_ingest_proto::proto::log_record;
+
+    use super::*;
+
+    fn valid_json(extra: &str) -> String {
+        format!(
+            r#"{{"client_ip":"192.0.2.10","status_code":200,"bytes_sent":12,"time_request_ms":-1,"time_wait_ms":0,"time_connect_ms":3,"time_response_ms":4,"time_total_ms":7{extra}}}"#
+        )
+    }
+
+    #[test]
+    fn preserves_negative_timer_sentinels() {
+        let record = map_message(&valid_json(""), 10);
+        let Some(log_record::Payload::Event(event)) = record.payload else {
+            panic!("expected mapped event");
+        };
+        assert_eq!(event.time_request_ms, -1);
+    }
+
+    #[test]
+    fn serde_parses_the_full_u64_range() {
+        let message =
+            valid_json("").replace("\"bytes_sent\":12", "\"bytes_sent\":18446744073709551615");
+        let event: RawHaProxyEvent = serde_json::from_str(&message).unwrap();
+        assert_eq!(event.bytes_sent, Some(u64::MAX));
+    }
+
+    #[test]
+    fn serde_coerces_haproxy_values() {
+        let message = r#"{
+            "client_port":"443",
+            "frontend_name":"-",
+            "status_code":"200",
+            "bytes_sent":"12",
+            "time_request_ms":"-1",
+            "time_wait_ms":"0",
+            "time_connect_ms":"3",
+            "time_response_ms":"4",
+            "time_total_ms":"7",
+            "retries":"1",
+            "tls_resumed":"true",
+            "server_queue":null
+        }"#;
+        let record = map_message(message, 10);
+        let Some(log_record::Payload::Event(event)) = record.payload else {
+            panic!("expected mapped event");
+        };
+        assert_eq!(event.client_port, Some(443));
+        assert_eq!(event.frontend_name, None);
+        assert_eq!(event.status_code, 200);
+        assert_eq!(event.time_request_ms, -1);
+        assert_eq!(event.retries, 1);
+        assert_eq!(event.tls_resumed, Some(true));
+        assert_eq!(event.server_queue, None);
+    }
+
+    #[test]
+    fn garde_rejects_invalid_semantics() {
+        let messages = [
+            valid_json("").replace("\"status_code\":200", "\"status_code\":99"),
+            valid_json("").replace("\"time_total_ms\":7", "\"time_total_ms\":-2"),
+            valid_json(",\"client_port\":65536"),
+        ];
+        for message in messages {
+            let record = map_message(&message, 10);
+            let Some(log_record::Payload::Rejected(rejected)) = record.payload else {
+                panic!("expected rejected event");
+            };
+            assert!(rejected.reason.contains("invalid HAProxy event"));
+        }
+    }
+
+    #[test]
+    fn invalid_ip_is_non_fatal_and_flagged() {
+        let message = valid_json("").replace("192.0.2.10", "not-an-ip");
+        let record = map_message(&message, 10);
+        let Some(log_record::Payload::Event(event)) = record.payload else {
+            panic!("expected mapped event");
+        };
+        assert_eq!(event.client_ip, None);
+        assert_eq!(event.normalization_flags, ["invalid_client_ip"]);
+    }
+
+    #[test]
+    fn preserves_request_path() {
+        let path = format!("/{}é", "a".repeat(20_000));
+        let message = valid_json(&format!(",\"request_path\":{path:?}"));
+        let record = map_message(&message, 10);
+        let Some(log_record::Payload::Event(event)) = record.payload else {
+            panic!("expected mapped event");
+        };
+        assert_eq!(event.request_path.as_deref(), Some(path.as_str()));
+    }
+
+    #[test]
+    fn malformed_event_becomes_durable_rejection() {
+        let record = map_message("not-json", 10);
+        let Some(log_record::Payload::Rejected(rejected)) = record.payload else {
+            panic!("expected rejected event");
+        };
+        assert_eq!(rejected.message_sha256.len(), 32);
+        assert!(rejected.reason.contains("invalid HAProxy JSON"));
+    }
+}

systems/portal-1/haproxy/default.nix

@@ -1,6 +1,7 @@
 let
   domains = [
     "audiobookshelf"
+    "auth"
     "cache"
     "gitea"
     "gems"
@@ -19,6 +20,8 @@ let
   acmeServices = map (domain: "acme-${domain}.tmmworkshop.com.service") domains;
 in
 {
+  imports = [ ./ingest.nix ];
+
   networking.firewall.allowedTCPPorts = [
     80
     443

systems/portal-1/haproxy/ingest.nix

@@ -0,0 +1,103 @@
+{ pkgs, ... }:
+let
+  haproxy_log_shipper = pkgs.callPackage ../../../rust/haproxy_log_shipper/package.nix { };
+  endpoint = "http://jeeves.taile39c31.ts.net:8444";
+  environment_file = "/var/lib/haproxy-log-shipper/environment";
+in
+{
+  services.journald.settings.Journal = {
+    Storage = "persistent";
+    SystemMaxUse = "512M";
+    MaxRetentionSec = "14day";
+  };
+
+  users = {
+    groups.haproxy-log-shipper = { };
+    users.haproxy-log-shipper = {
+      isSystemUser = true;
+      group = "haproxy-log-shipper";
+      extraGroups = [ "systemd-journal" ];
+    };
+  };
+
+  systemd = {
+    services = {
+      haproxy = {
+        serviceConfig = {
+          StandardOutput = "journal";
+          StandardError = "journal";
+        };
+      };
+
+      haproxy-log-shipper = {
+        description = "Ship ordered HAProxy journal logs to Jeeves";
+        documentation = [ "file://${../../../rust/haproxy_ingest_proto/README.md}" ];
+        after = [
+          "network-online.target"
+          "tailscaled-autoconnect.service"
+        ];
+        wants = [
+          "network-online.target"
+          "tailscaled-autoconnect.service"
+        ];
+
+        serviceConfig = {
+          Type = "oneshot";
+          User = "haproxy-log-shipper";
+          Group = "haproxy-log-shipper";
+          SupplementaryGroups = [ "systemd-journal" ];
+          ExecStart = "${haproxy_log_shipper}/bin/haproxy-log-shipper";
+          Environment = [
+            "HAPROXY_INGEST_ENDPOINT=${endpoint}"
+            "RUST_LOG=info"
+          ];
+          EnvironmentFile = environment_file;
+
+          CapabilityBoundingSet = [ "" ];
+          DeviceAllow = [ "" ];
+          LockPersonality = true;
+          MemoryDenyWriteExecute = true;
+          NoNewPrivileges = true;
+          PrivateDevices = true;
+          PrivateTmp = true;
+          ProtectClock = true;
+          ProtectControlGroups = true;
+          ProtectHome = true;
+          ProtectHostname = true;
+          ProtectKernelLogs = true;
+          ProtectKernelModules = true;
+          ProtectKernelTunables = true;
+          ProtectSystem = "strict";
+          RemoveIPC = true;
+          RestrictAddressFamilies = [
+            "AF_INET"
+            "AF_UNIX"
+          ];
+          RestrictNamespaces = true;
+          RestrictRealtime = true;
+          RestrictSUIDSGID = true;
+          SystemCallArchitectures = "native";
+          SystemCallFilter = [
+            "@system-service"
+            "~@privileged"
+          ];
+          UMask = "0077";
+        };
+      };
+    };
+
+    timers.haproxy-log-shipper = {
+      description = "Run the HAProxy log shipper after each completed catch-up cycle";
+      wantedBy = [ "timers.target" ];
+      timerConfig = {
+        OnBootSec = "30s";
+        OnUnitInactiveSec = "60s";
+        Unit = "haproxy-log-shipper.service";
+      };
+    };
+
+    tmpfiles.rules = [
+      "d /var/lib/haproxy-log-shipper 0755 root root - -"
+    ];
+  };
+}