From 438b4e09e6ee2dbb6fa4e085787c4b94455b99be Mon Sep 17 00:00:00 2001 From: Richie Cahill Date: Fri, 16 Oct 2026 21:12:37 -0400 Subject: [PATCH] feat: add HAProxy log ingestion server - Introduced a new Rust-based service for ingesting HAProxy logs, including the necessary database schema and API endpoints. - Implemented database interactions for managing source status, inserting events, and handling ingestion batches. - Added validation for incoming log batches to ensure data integrity. - Configured systemd service for the HAProxy ingestion server with appropriate security settings. - Updated firewall rules to allow traffic on the ingestion server's port (8444). - Created necessary PostgreSQL database and user configurations for the ingestion service. --- rust/haproxy_ingest_server/Cargo.toml | 29 + .../migrations/001_initial.sql | 102 ++++ rust/haproxy_ingest_server/package.nix | 19 + rust/haproxy_ingest_server/src/db.rs | 376 +++++++++++++ rust/haproxy_ingest_server/src/main.rs | 530 ++++++++++++++++++ rust/haproxy_ingest_server/src/validation.rs | 214 +++++++ systems/jeeves/networking/firewall.nix | 2 + systems/jeeves/services/haproxy_ingest.nix | 82 +++ systems/jeeves/services/postgress.nix | 11 + 9 files changed, 1365 insertions(+) create mode 100644 rust/haproxy_ingest_server/Cargo.toml create mode 100644 rust/haproxy_ingest_server/migrations/001_initial.sql create mode 100644 rust/haproxy_ingest_server/package.nix create mode 100644 rust/haproxy_ingest_server/src/db.rs create mode 100644 rust/haproxy_ingest_server/src/main.rs create mode 100644 rust/haproxy_ingest_server/src/validation.rs create mode 100644 systems/jeeves/services/haproxy_ingest.nix diff --git a/rust/haproxy_ingest_server/Cargo.toml b/rust/haproxy_ingest_server/Cargo.toml new file mode 100644 index 0000000..e38b36a --- /dev/null +++ b/rust/haproxy_ingest_server/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "haproxy-ingest-server" +version = "0.1.0" +edition = "2024" +description = "HAProxy log ingestion API" +license = "MIT" + +[lints] +workspace = true + +[dependencies] +anyhow.workspace = true +axum.workspace = true +bytes.workspace = true +chrono = { workspace = true, features = ["serde"] } +clap.workspace = true +haproxy-ingest-proto = { path = "../haproxy_ingest_proto" } +hex.workspace = true +prost.workspace = true +serde.workspace = true +sha2.workspace = true +sqlx.workspace = true +subtle.workspace = true +thiserror.workspace = true +tokio.workspace = true +tower-http.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true +zstd.workspace = true diff --git a/rust/haproxy_ingest_server/migrations/001_initial.sql b/rust/haproxy_ingest_server/migrations/001_initial.sql new file mode 100644 index 0000000..8b11b27 --- /dev/null +++ b/rust/haproxy_ingest_server/migrations/001_initial.sql @@ -0,0 +1,102 @@ +CREATE TABLE IF NOT EXISTS ingest_source_state ( + machine_id TEXT PRIMARY KEY, + hostname TEXT NOT NULL, + last_cursor TEXT, + last_journal_timestamp TIMESTAMPTZ, + last_source_ip INET, + last_commit_time TIMESTAMPTZ +); + +CREATE TABLE IF NOT EXISTS haproxy_logs ( + id BIGINT GENERATED ALWAYS AS IDENTITY, + machine_id TEXT NOT NULL, + hostname TEXT NOT NULL, + source_ip INET NOT NULL, + journal_timestamp TIMESTAMPTZ NOT NULL, + request_time TIMESTAMPTZ, + client_ip INET, + client_port INTEGER, + frontend_name TEXT, + backend_name TEXT, + server_name TEXT, + http_method TEXT, + request_path TEXT, + http_version TEXT, + status_code INTEGER NOT NULL, + bytes_sent BIGINT NOT NULL, + time_request_ms INTEGER NOT NULL, + time_wait_ms INTEGER NOT NULL, + time_connect_ms INTEGER NOT NULL, + time_response_ms INTEGER NOT NULL, + time_total_ms INTEGER NOT NULL, + termination_state TEXT, + retries INTEGER NOT NULL, + tls_sni TEXT, + tls_version TEXT, + tls_cipher TEXT, + tls_resumed BOOLEAN, + normalization_flags TEXT[] NOT NULL DEFAULT '{}', + active_connections INTEGER, + frontend_connections INTEGER, + backend_connections INTEGER, + server_connections INTEGER, + server_queue INTEGER, + backend_queue INTEGER, + ingest_time TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (journal_timestamp, id) +) PARTITION BY RANGE (journal_timestamp); + +CREATE INDEX IF NOT EXISTS haproxy_logs_machine_time_idx +ON haproxy_logs (machine_id, journal_timestamp); + +CREATE INDEX IF NOT EXISTS haproxy_logs_status_time_idx +ON haproxy_logs (status_code, journal_timestamp); + +CREATE TABLE IF NOT EXISTS ingest_rejected_events ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + machine_id TEXT NOT NULL, + hostname TEXT NOT NULL, + journal_timestamp TIMESTAMPTZ NOT NULL, + reason TEXT NOT NULL, + message_sha256 BYTEA NOT NULL, + raw_message_prefix TEXT, + raw_message_truncated BOOLEAN NOT NULL, + source_ip INET NOT NULL, + ingest_time TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS ingest_rejected_events_machine_time_idx +ON ingest_rejected_events (machine_id, journal_timestamp); + +CREATE TABLE IF NOT EXISTS ingest_journal_gaps ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + machine_id TEXT NOT NULL, + hostname TEXT NOT NULL, + expected_cursor TEXT NOT NULL, + first_available_cursor TEXT NOT NULL, + first_available_timestamp TIMESTAMPTZ, + reason TEXT NOT NULL, + source_ip INET NOT NULL, + detected_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS ingest_journal_gaps_machine_time_idx +ON ingest_journal_gaps (machine_id, detected_at); + +CREATE TABLE IF NOT EXISTS ingest_batches ( + batch_id TEXT PRIMARY KEY, + machine_id TEXT NOT NULL, + hostname TEXT NOT NULL, + source_ip INET NOT NULL, + previous_cursor TEXT, + last_cursor TEXT NOT NULL, + first_journal_timestamp TIMESTAMPTZ, + last_journal_timestamp TIMESTAMPTZ, + record_count INTEGER NOT NULL, + compressed_bytes BIGINT NOT NULL, + uncompressed_bytes BIGINT NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS ingest_batches_machine_time_idx +ON ingest_batches (machine_id, received_at); diff --git a/rust/haproxy_ingest_server/package.nix b/rust/haproxy_ingest_server/package.nix new file mode 100644 index 0000000..3ed38b7 --- /dev/null +++ b/rust/haproxy_ingest_server/package.nix @@ -0,0 +1,19 @@ +{ + lib, + rustPlatform, +}: +rustPlatform.buildRustPackage { + pname = "haproxy-ingest-server"; + version = "0.1.0"; + src = import ../rust-source.nix { inherit lib; }; + + cargoLock.lockFile = ../../Cargo.lock; + cargoBuildFlags = [ "-p haproxy-ingest-server" ]; + cargoTestFlags = [ "-p haproxy-ingest-server" ]; + + meta = { + description = "HAProxy log ingestion API"; + license = lib.licenses.mit; + platforms = lib.platforms.linux; + }; +} diff --git a/rust/haproxy_ingest_server/src/db.rs b/rust/haproxy_ingest_server/src/db.rs new file mode 100644 index 0000000..efd74f2 --- /dev/null +++ b/rust/haproxy_ingest_server/src/db.rs @@ -0,0 +1,376 @@ +//! `SQLx` statements and Rust-to-PostgreSQL value conversion. + +use std::net::IpAddr; + +use anyhow::{Context, Result}; +use chrono::{DateTime, NaiveDate, Utc}; +use haproxy_ingest_proto::proto::{HaProxyEvent, JournalGap, LogBatch, RejectedEvent, log_record}; +use sha2::{Digest, Sha256}; +use sqlx::postgres::PgRow; +use sqlx::{AssertSqlSafe, Decode, PgPool, Postgres, QueryBuilder, Row, Transaction, Type}; + +use crate::{ApiError, timestamp}; + +const MIGRATION: &str = include_str!("../migrations/001_initial.sql"); +const EVENT_INSERT_CHUNK_SIZE: usize = 1_000; + +pub struct SourceStatus { + pub hostname: Option, + pub last_cursor: Option, + pub last_journal_timestamp: Option>, + pub last_commit_time: Option>, +} + +pub async fn apply_ingestion_schema(pool: &PgPool) -> Result<()> { + sqlx::raw_sql(MIGRATION) + .execute(pool) + .await + .context("failed to apply ingestion schema")?; + Ok(()) +} + +pub async fn select_source_status( + pool: &PgPool, + machine_id: &str, +) -> Result, ApiError> { + let row = sqlx::query( + "SELECT hostname, last_cursor, last_journal_timestamp, last_commit_time \ + FROM ingest_source_state WHERE machine_id = $1", + ) + .bind(machine_id) + .fetch_optional(pool) + .await + .map_err(ApiError::internal)?; + row.map(|row| { + Ok(SourceStatus { + hostname: get(&row, "hostname")?, + last_cursor: get(&row, "last_cursor")?, + last_journal_timestamp: get(&row, "last_journal_timestamp")?, + last_commit_time: get(&row, "last_commit_time")?, + }) + }) + .transpose() +} + +fn get<'r, T>(row: &'r PgRow, column: &str) -> Result +where + T: Decode<'r, Postgres> + Type, +{ + row.try_get(column).map_err(ApiError::internal) +} + +pub async fn set_lock_timeout(transaction: &mut Transaction<'_, Postgres>) -> Result<(), ApiError> { + sqlx::query("SET LOCAL lock_timeout = '5s'") + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +pub async fn set_statement_timeout( + transaction: &mut Transaction<'_, Postgres>, +) -> Result<(), ApiError> { + sqlx::query("SET LOCAL statement_timeout = '8s'") + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +pub async fn insert_source_state( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, +) -> Result<(), ApiError> { + sqlx::query( + "INSERT INTO ingest_source_state (machine_id, hostname) VALUES ($1, $2) \ + ON CONFLICT (machine_id) DO NOTHING", + ) + .bind(&batch.machine_id) + .bind(&batch.hostname) + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +pub async fn select_committed_cursor( + transaction: &mut Transaction<'_, Postgres>, + machine_id: &str, +) -> Result, ApiError> { + sqlx::query_scalar( + "SELECT last_cursor FROM ingest_source_state WHERE machine_id = $1 FOR UPDATE", + ) + .bind(machine_id) + .fetch_one(&mut **transaction) + .await + .map_err(ApiError::internal) +} + +pub async fn create_daily_partition( + transaction: &mut Transaction<'_, Postgres>, + date: NaiveDate, + next: NaiveDate, +) -> Result<(), ApiError> { + let name = partition_name(date); + let statement = format!( + "CREATE TABLE IF NOT EXISTS {name} PARTITION OF haproxy_logs \ + FOR VALUES FROM ('{date}') TO ('{next}')" + ); + sqlx::query(AssertSqlSafe(statement.as_str())) + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +fn partition_name(date: NaiveDate) -> String { + format!("haproxy_logs_{}", date.format("%Y%m%d")) +} + +struct PreparedEvent<'a> { + event: &'a HaProxyEvent, + journal_timestamp: DateTime, + client_ip: Option, + client_port: Option, + bytes_sent: i64, + retries: i32, + active_connections: Option, + frontend_connections: Option, + backend_connections: Option, + server_connections: Option, + server_queue: Option, + backend_queue: Option, +} + +fn prepare_events(batch: &LogBatch) -> Result>, ApiError> { + let mut prepared = Vec::new(); + for record in &batch.records { + let Some(log_record::Payload::Event(event)) = &record.payload else { + continue; + }; + prepared.push(PreparedEvent { + event, + journal_timestamp: timestamp(record.journal_timestamp_us)?, + client_ip: event + .client_ip + .as_deref() + .and_then(|value| value.parse::().ok()) + .map(|value| value.to_string()), + client_port: event + .client_port + .map(i32::try_from) + .transpose() + .map_err(ApiError::internal)?, + bytes_sent: i64::try_from(event.bytes_sent).map_err(ApiError::internal)?, + retries: i32::try_from(event.retries).map_err(ApiError::internal)?, + active_connections: sql_i32(event.active_connections).map_err(ApiError::internal)?, + frontend_connections: sql_i32(event.frontend_connections) + .map_err(ApiError::internal)?, + backend_connections: sql_i32(event.backend_connections).map_err(ApiError::internal)?, + server_connections: sql_i32(event.server_connections).map_err(ApiError::internal)?, + server_queue: sql_i32(event.server_queue).map_err(ApiError::internal)?, + backend_queue: sql_i32(event.backend_queue).map_err(ApiError::internal)?, + }); + } + Ok(prepared) +} + +#[allow(clippy::too_many_lines)] +pub async fn insert_events( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, + source_ip: IpAddr, +) -> Result<(), ApiError> { + let prepared = prepare_events(batch)?; + let source_ip = source_ip.to_string(); + for chunk in prepared.chunks(EVENT_INSERT_CHUNK_SIZE) { + let mut query = QueryBuilder::::new( + "INSERT INTO haproxy_logs (machine_id,hostname,source_ip,journal_timestamp,client_ip,\ + client_port,frontend_name,backend_name,server_name,http_method,request_path,http_version,\ + status_code,bytes_sent,time_request_ms,time_wait_ms,time_connect_ms,time_response_ms,\ + time_total_ms,termination_state,retries,tls_sni,tls_version,tls_cipher,tls_resumed,\ + normalization_flags,active_connections,frontend_connections,\ + backend_connections,server_connections,server_queue,backend_queue) ", + ); + query.push_values(chunk, |mut values, row| { + values + .push_bind(&batch.machine_id) + .push_bind(&batch.hostname) + .push_bind(source_ip.as_str()) + .push_unseparated("::inet") + .push_bind(row.journal_timestamp) + .push_bind(row.client_ip.as_deref()) + .push_unseparated("::inet") + .push_bind(row.client_port) + .push_bind(row.event.frontend_name.as_deref()) + .push_bind(row.event.backend_name.as_deref()) + .push_bind(row.event.server_name.as_deref()) + .push_bind(row.event.http_method.as_deref()) + .push_bind(row.event.request_path.as_deref()) + .push_bind(row.event.http_version.as_deref()) + .push_bind(row.event.status_code) + .push_bind(row.bytes_sent) + .push_bind(row.event.time_request_ms) + .push_bind(row.event.time_wait_ms) + .push_bind(row.event.time_connect_ms) + .push_bind(row.event.time_response_ms) + .push_bind(row.event.time_total_ms) + .push_bind(row.event.termination_state.as_deref()) + .push_bind(row.retries) + .push_bind(row.event.tls_sni.as_deref()) + .push_bind(row.event.tls_version.as_deref()) + .push_bind(row.event.tls_cipher.as_deref()) + .push_bind(row.event.tls_resumed) + .push_bind(row.event.normalization_flags.as_slice()) + .push_bind(row.active_connections) + .push_bind(row.frontend_connections) + .push_bind(row.backend_connections) + .push_bind(row.server_connections) + .push_bind(row.server_queue) + .push_bind(row.backend_queue); + }); + query + .build() + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + } + Ok(()) +} + +pub async fn insert_rejected_event( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, + source_ip: IpAddr, + journal_timestamp: DateTime, + rejected: &RejectedEvent, +) -> Result<(), ApiError> { + sqlx::query( + "INSERT INTO ingest_rejected_events (machine_id,hostname,journal_timestamp,reason,\ + message_sha256,raw_message_prefix,raw_message_truncated,source_ip) \ + VALUES ($1,$2,$3,$4,$5,$6,$7,$8::inet)", + ) + .bind(&batch.machine_id) + .bind(&batch.hostname) + .bind(journal_timestamp) + .bind(&rejected.reason) + .bind(&rejected.message_sha256) + .bind(&rejected.raw_message_prefix) + .bind(rejected.raw_message_truncated) + .bind(source_ip.to_string()) + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +pub async fn insert_journal_gap( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, + source_ip: IpAddr, + first_available_timestamp: Option>, + gap: &JournalGap, +) -> Result<(), ApiError> { + sqlx::query( + "INSERT INTO ingest_journal_gaps (machine_id,hostname,expected_cursor,\ + first_available_cursor,first_available_timestamp,reason,source_ip) \ + VALUES ($1,$2,$3,$4,$5,$6,$7::inet)", + ) + .bind(&batch.machine_id) + .bind(&batch.hostname) + .bind(&gap.expected_cursor) + .bind(&gap.first_available_cursor) + .bind(first_available_timestamp) + .bind(&gap.reason) + .bind(source_ip.to_string()) + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +pub async fn insert_ingest_batch( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, + source_ip: IpAddr, + compressed_bytes: usize, + uncompressed_bytes: usize, +) -> Result<(), ApiError> { + let first_timestamp = timestamp(batch.first_journal_timestamp_us)?; + let last_timestamp = timestamp(batch.last_journal_timestamp_us)?; + let batch_id = batch_id(batch); + let record_count = i32::try_from(batch.records.len()).map_err(ApiError::internal)?; + let compressed_bytes = i64::try_from(compressed_bytes).map_err(ApiError::internal)?; + let uncompressed_bytes = i64::try_from(uncompressed_bytes).map_err(ApiError::internal)?; + sqlx::query( + "INSERT INTO ingest_batches (batch_id, machine_id, hostname, source_ip, previous_cursor, \ + last_cursor, first_journal_timestamp, last_journal_timestamp, record_count, \ + compressed_bytes, uncompressed_bytes) \ + VALUES ($1,$2,$3,$4::inet,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT (batch_id) DO NOTHING", + ) + .bind(&batch_id) + .bind(&batch.machine_id) + .bind(&batch.hostname) + .bind(source_ip.to_string()) + .bind(&batch.previous_cursor) + .bind(&batch.last_cursor) + .bind(first_timestamp) + .bind(last_timestamp) + .bind(record_count) + .bind(compressed_bytes) + .bind(uncompressed_bytes) + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +pub async fn update_source_state( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, + source_ip: IpAddr, +) -> Result<(), ApiError> { + let last_timestamp = timestamp(batch.last_journal_timestamp_us)?; + sqlx::query( + "UPDATE ingest_source_state SET hostname=$2, last_cursor=$3, \ + last_journal_timestamp=$4, last_source_ip=$5::inet, last_commit_time=now() \ + WHERE machine_id=$1", + ) + .bind(&batch.machine_id) + .bind(&batch.hostname) + .bind(&batch.last_cursor) + .bind(last_timestamp) + .bind(source_ip.to_string()) + .execute(&mut **transaction) + .await + .map_err(ApiError::internal)?; + Ok(()) +} + +fn batch_id(batch: &LogBatch) -> String { + let mut hash = Sha256::new(); + hash.update(batch.machine_id.as_bytes()); + hash.update([0]); + if let Some(previous) = &batch.previous_cursor { + hash.update(previous.as_bytes()); + } + hash.update([0]); + hash.update(batch.last_cursor.as_bytes()); + hex::encode(hash.finalize()) +} + +fn sql_i32(value: Option) -> Result, std::num::TryFromIntError> { + value.map(i32::try_from).transpose() +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn partition_names_are_date_only() { + let date = NaiveDate::from_ymd_opt(2026, 8, 29).unwrap(); + assert_eq!(partition_name(date), "haproxy_logs_20260829"); + } +} diff --git a/rust/haproxy_ingest_server/src/main.rs b/rust/haproxy_ingest_server/src/main.rs new file mode 100644 index 0000000..8f0442a --- /dev/null +++ b/rust/haproxy_ingest_server/src/main.rs @@ -0,0 +1,530 @@ +use std::{ + collections::BTreeSet, + io::Read, + net::{IpAddr, SocketAddr}, +}; + +use anyhow::{Context, Result, bail}; +use axum::{ + Router, + body::Bytes, + extract::{ConnectInfo, DefaultBodyLimit, Path, State}, + http::{HeaderMap, StatusCode, header}, + response::{IntoResponse, Response}, + routing::{get as get_route, post}, +}; +use chrono::{DateTime, Days, Utc}; +use clap::Parser; +use haproxy_ingest_proto::{ + MAX_COMPRESSED_BYTES, MAX_DECOMPRESSED_BYTES, + proto::{LogBatch, log_record}, +}; +use prost::Message; +use serde::Serialize; +use sha2::{Digest, Sha256}; +use sqlx::{PgPool, Postgres, Transaction, postgres::PgPoolOptions}; +use subtle::ConstantTimeEq; +use tower_http::trace::TraceLayer; +use tracing::{info, warn}; +use tracing_subscriber::EnvFilter; + +mod db; +mod validation; + +use validation::{ValidationError, validate_batch}; + +#[derive(Debug, Parser)] +#[command(about, version)] +struct Args { + #[arg(long, env = "HAPROXY_INGEST_LISTEN", default_value = "0.0.0.0:8444")] + listen: SocketAddr, + + #[arg(long, env = "HAPROXY_INGEST_DATABASE_URL", hide_env_values = true)] + database_url: String, + + #[arg(long, env = "HAPROXY_INGEST_TOKEN", hide_env_values = true)] + token: String, +} + +#[derive(Clone)] +struct AppState { + pool: PgPool, + auth: SourceToken, +} + +#[derive(Clone)] +struct SourceToken { + hash: [u8; 32], +} + +impl SourceToken { + fn parse(value: &str) -> Result { + let token = value.trim(); + if token.len() < 32 { + bail!("source token must be at least 32 characters"); + } + Ok(Self { + hash: Sha256::digest(token.as_bytes()).into(), + }) + } + + fn authenticate(&self, headers: &HeaderMap) -> Result<(), ApiError> { + let authorization = headers + .get(header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .ok_or_else(|| ApiError::new(StatusCode::UNAUTHORIZED, "missing bearer token"))?; + let presented: [u8; 32] = Sha256::digest(authorization.as_bytes()).into(); + (presented.ct_eq(&self.hash).unwrap_u8() == 1) + .then_some(()) + .ok_or_else(|| ApiError::new(StatusCode::UNAUTHORIZED, "invalid bearer token")) + } +} + +#[derive(Debug, Serialize)] +struct StatusResponse { + machine_id: String, + hostname: Option, + last_cursor: Option, + last_journal_timestamp: Option>, + last_commit_time: Option>, +} + +#[derive(Debug, Serialize)] +struct CommitResponse { + machine_id: String, + last_cursor: String, + records_committed: usize, + already_committed: bool, +} + +#[derive(Debug, Serialize)] +struct ErrorResponse { + error: String, + #[serde(skip_serializing_if = "Option::is_none")] + record_index: Option, + #[serde(skip_serializing_if = "Option::is_none")] + last_cursor: Option, +} + +#[derive(Debug)] +struct ApiError { + status: StatusCode, + message: String, + record_index: Option, + last_cursor: Option, +} + +impl ApiError { + fn new(status: StatusCode, message: impl Into) -> Self { + Self { + status, + message: message.into(), + record_index: None, + last_cursor: None, + } + } + + fn invalid(error: &ValidationError) -> Self { + Self { + status: StatusCode::UNPROCESSABLE_ENTITY, + record_index: error.record_index(), + message: error.to_string(), + last_cursor: None, + } + } + + fn conflict(last_cursor: Option) -> Self { + Self { + status: StatusCode::CONFLICT, + message: "batch previous_cursor does not match the committed cursor".to_owned(), + record_index: None, + last_cursor, + } + } + + fn internal(error: impl std::fmt::Display) -> Self { + warn!(error = %error, "ingestion request failed"); + Self::new( + StatusCode::SERVICE_UNAVAILABLE, + "ingestion backend unavailable", + ) + } +} + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + ( + self.status, + axum::Json(ErrorResponse { + error: self.message, + record_index: self.record_index, + last_cursor: self.last_cursor, + }), + ) + .into_response() + } +} + +#[derive(Debug, Eq, PartialEq)] +enum CursorDecision { + Accept, + AlreadyCommitted, + Conflict, +} + +#[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(); + let pool = PgPoolOptions::new() + .max_connections(8) + .connect(&args.database_url) + .await + .context("failed to connect to PostgreSQL")?; + + db::apply_ingestion_schema(&pool).await?; + + let state = AppState { + pool, + auth: SourceToken::parse(&args.token)?, + }; + + let app = Router::new() + .route("/v1/logs/status/{machine_id}", get_route(get_status)) + .route("/v1/logs/batch", post(post_batch)) + .layer(DefaultBodyLimit::max(MAX_COMPRESSED_BYTES)) + .layer(TraceLayer::new_for_http()) + .with_state(state); + let listener = tokio::net::TcpListener::bind(args.listen) + .await + .context("failed to bind ingestion HTTP listener")?; + + info!(listen = %args.listen, "HAProxy ingestion API listening over HTTP"); + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .context("ingestion server failed") +} + +async fn get_status( + State(state): State, + Path(machine_id): Path, + headers: HeaderMap, +) -> Result, ApiError> { + state.auth.authenticate(&headers)?; + let status = db::select_source_status(&state.pool, &machine_id).await?; + let response = match status { + None => StatusResponse { + machine_id: machine_id.clone(), + hostname: None, + last_cursor: None, + last_journal_timestamp: None, + last_commit_time: None, + }, + Some(status) => StatusResponse { + machine_id: machine_id.clone(), + hostname: status.hostname, + last_cursor: status.last_cursor, + last_journal_timestamp: status.last_journal_timestamp, + last_commit_time: status.last_commit_time, + }, + }; + Ok(axum::Json(response)) +} + +async fn post_batch( + State(state): State, + ConnectInfo(peer): ConnectInfo, + headers: HeaderMap, + body: Bytes, +) -> Result, ApiError> { + state.auth.authenticate(&headers)?; + check_protobuf(&headers)?; + check_zstd(&headers)?; + let compressed_bytes = body.len(); + let decoded = decompress_bounded(&body)?; + let uncompressed_bytes = decoded.len(); + let batch = LogBatch::decode(decoded.as_slice()).map_err(|error| { + ApiError::new( + StatusCode::BAD_REQUEST, + format!("invalid protobuf: {error}"), + ) + })?; + validate_batch(&batch).map_err(|error| ApiError::invalid(&error))?; + let response = commit_batch( + &state.pool, + &batch, + peer.ip(), + compressed_bytes, + uncompressed_bytes, + ) + .await?; + Ok(axum::Json(response)) +} + +fn check_protobuf(headers: &HeaderMap) -> Result<(), ApiError> { + if headers + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + != Some("application/x-protobuf") + { + return Err(ApiError::new( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "Content-Type must be application/x-protobuf", + )); + } + Ok(()) +} + +fn check_zstd(headers: &HeaderMap) -> Result<(), ApiError> { + if headers + .get(header::CONTENT_ENCODING) + .and_then(|value| value.to_str().ok()) + != Some("zstd") + { + return Err(ApiError::new( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "Content-Encoding must be zstd", + )); + } + Ok(()) +} + +fn decompress_bounded(body: &[u8]) -> Result, ApiError> { + let decoder = zstd::stream::read::Decoder::new(body).map_err(|error| { + ApiError::new( + StatusCode::BAD_REQUEST, + format!("invalid zstd body: {error}"), + ) + })?; + let mut output = Vec::new(); + let read_limit = u64::try_from(MAX_DECOMPRESSED_BYTES) + .map_err(ApiError::internal)? + .saturating_add(1); + decoder + .take(read_limit) + .read_to_end(&mut output) + .map_err(|error| { + ApiError::new( + StatusCode::BAD_REQUEST, + format!("invalid zstd body: {error}"), + ) + })?; + if output.len() > MAX_DECOMPRESSED_BYTES { + return Err(ApiError::new( + StatusCode::PAYLOAD_TOO_LARGE, + "decompressed protobuf exceeds 20 MiB", + )); + } + Ok(output) +} + +async fn commit_batch( + pool: &PgPool, + batch: &LogBatch, + source_ip: IpAddr, + compressed_bytes: usize, + uncompressed_bytes: usize, +) -> Result { + let mut transaction = pool.begin().await.map_err(ApiError::internal)?; + db::set_lock_timeout(&mut transaction).await?; + db::set_statement_timeout(&mut transaction).await?; + db::insert_source_state(&mut transaction, batch).await?; + let committed_cursor = db::select_committed_cursor(&mut transaction, &batch.machine_id).await?; + match classify_cursor_chain( + committed_cursor.as_deref(), + batch.previous_cursor.as_deref(), + &batch.last_cursor, + ) { + CursorDecision::Accept => {} + CursorDecision::AlreadyCommitted => { + transaction.commit().await.map_err(ApiError::internal)?; + return Ok(CommitResponse { + machine_id: batch.machine_id.clone(), + last_cursor: batch.last_cursor.clone(), + records_committed: batch.records.len(), + already_committed: true, + }); + } + CursorDecision::Conflict => return Err(ApiError::conflict(committed_cursor)), + } + + ensure_daily_partitions(&mut transaction, batch).await?; + db::insert_events(&mut transaction, batch, source_ip).await?; + insert_non_event_records(&mut transaction, batch, source_ip).await?; + db::insert_ingest_batch( + &mut transaction, + batch, + source_ip, + compressed_bytes, + uncompressed_bytes, + ) + .await?; + db::update_source_state(&mut transaction, batch, source_ip).await?; + transaction.commit().await.map_err(ApiError::internal)?; + info!( + machine_id = %batch.machine_id, + records = batch.records.len(), + last_cursor = %batch.last_cursor, + "committed HAProxy log batch" + ); + Ok(CommitResponse { + machine_id: batch.machine_id.clone(), + last_cursor: batch.last_cursor.clone(), + records_committed: batch.records.len(), + already_committed: false, + }) +} + +async fn ensure_daily_partitions( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, +) -> Result<(), ApiError> { + let mut dates = BTreeSet::new(); + for record in &batch.records { + if matches!(record.payload, Some(log_record::Payload::Event(_))) { + dates.insert(timestamp(record.journal_timestamp_us)?.date_naive()); + } + } + for date in dates { + let next = date.checked_add_days(Days::new(1)).ok_or_else(|| { + ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "partition date overflow") + })?; + db::create_daily_partition(transaction, date, next).await?; + } + Ok(()) +} + +async fn insert_non_event_records( + transaction: &mut Transaction<'_, Postgres>, + batch: &LogBatch, + source_ip: IpAddr, +) -> Result<(), ApiError> { + for record in &batch.records { + match &record.payload { + Some(log_record::Payload::Rejected(rejected)) => { + let journal_timestamp = timestamp(record.journal_timestamp_us)?; + db::insert_rejected_event( + transaction, + batch, + source_ip, + journal_timestamp, + rejected, + ) + .await?; + } + Some(log_record::Payload::Gap(gap)) => { + let first_available_timestamp = gap + .first_available_timestamp_us + .map(timestamp) + .transpose()?; + db::insert_journal_gap( + transaction, + batch, + source_ip, + first_available_timestamp, + gap, + ) + .await?; + } + Some(log_record::Payload::Event(_)) | None => {} + } + } + Ok(()) +} + +fn timestamp(micros: i64) -> Result, ApiError> { + DateTime::from_timestamp_micros(micros).ok_or_else(|| { + ApiError::new( + StatusCode::UNPROCESSABLE_ENTITY, + "journal timestamp is outside supported range", + ) + }) +} + +fn classify_cursor_chain( + committed: Option<&str>, + previous: Option<&str>, + last: &str, +) -> CursorDecision { + if committed == Some(last) { + CursorDecision::AlreadyCommitted + } else if committed == previous { + CursorDecision::Accept + } else { + CursorDecision::Conflict + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn starts_without_certificate_arguments() { + Args::try_parse_from([ + "haproxy-ingest-server", + "--database-url", + "postgres://localhost/haproxy_ingest", + "--token", + "0123456789abcdef0123456789abcdef", + ]) + .unwrap(); + } + + #[test] + fn shared_token_authenticates_requests() { + let token = SourceToken::parse("0123456789abcdef0123456789abcdef").unwrap(); + let mut headers = HeaderMap::new(); + headers.insert( + header::AUTHORIZATION, + "Bearer 0123456789abcdef0123456789abcdef".parse().unwrap(), + ); + assert!(token.authenticate(&headers).is_ok()); + headers.insert( + header::AUTHORIZATION, + "Bearer fedcba9876543210fedcba9876543210".parse().unwrap(), + ); + assert!(token.authenticate(&headers).is_err()); + } + + #[test] + fn decompression_limit_is_enforced() { + let input = vec![0_u8; MAX_DECOMPRESSED_BYTES + 1]; + let compressed = zstd::stream::encode_all(input.as_slice(), 1).unwrap(); + assert!(matches!( + decompress_bounded(&compressed), + Err(ApiError { + status: StatusCode::PAYLOAD_TOO_LARGE, + .. + }) + )); + } + + #[test] + fn cursor_chain_handles_bootstrap_continuation_duplicate_and_conflict() { + assert_eq!( + classify_cursor_chain(None, None, "c1"), + CursorDecision::Accept + ); + assert_eq!( + classify_cursor_chain(Some("c1"), Some("c1"), "c2"), + CursorDecision::Accept + ); + assert_eq!( + classify_cursor_chain(Some("c2"), Some("c1"), "c2"), + CursorDecision::AlreadyCommitted + ); + assert_eq!( + classify_cursor_chain(Some("other"), Some("c1"), "c2"), + CursorDecision::Conflict + ); + } +} diff --git a/rust/haproxy_ingest_server/src/validation.rs b/rust/haproxy_ingest_server/src/validation.rs new file mode 100644 index 0000000..05712ae --- /dev/null +++ b/rust/haproxy_ingest_server/src/validation.rs @@ -0,0 +1,214 @@ +use std::net::IpAddr; + +use chrono::{DateTime, Utc}; +use haproxy_ingest_proto::{ + SCHEMA_VERSION, SERVER_MAX_RECORDS, + proto::{LogBatch, LogRecord, log_record}, +}; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum ValidationError { + #[error("unsupported schema_version {0}")] + SchemaVersion(u32), + #[error("machine_id must be 32 lowercase hexadecimal characters")] + MachineId, + #[error("hostname is empty or longer than 255 bytes")] + Hostname, + #[error("last_cursor is empty")] + LastCursor, + #[error("batch contains no records or more than {SERVER_MAX_RECORDS} records")] + RecordCount, + #[error("batch journal timestamp bounds are invalid")] + TimestampBounds, + #[error("record {index}: {reason}")] + Record { index: usize, reason: String }, +} + +impl ValidationError { + #[must_use] + pub const fn record_index(&self) -> Option { + match self { + Self::Record { index, .. } => Some(*index), + _ => None, + } + } +} + +/// Validates protocol-level and per-record invariants before database access. +/// +/// # Errors +/// +/// Returns [`ValidationError`] when the batch cannot be committed safely. +pub fn validate_batch(batch: &LogBatch) -> Result<(), ValidationError> { + if batch.schema_version != SCHEMA_VERSION { + return Err(ValidationError::SchemaVersion(batch.schema_version)); + } + if batch.machine_id.len() != 32 + || !batch + .machine_id + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(ValidationError::MachineId); + } + if batch.hostname.is_empty() || batch.hostname.len() > 255 { + return Err(ValidationError::Hostname); + } + if batch.last_cursor.is_empty() { + return Err(ValidationError::LastCursor); + } + if batch.records.is_empty() || batch.records.len() > SERVER_MAX_RECORDS { + return Err(ValidationError::RecordCount); + } + // Journal cursors define record order; wall-clock timestamps may move backward. + if batch + .records + .first() + .map(|record| record.journal_timestamp_us) + != Some(batch.first_journal_timestamp_us) + || batch + .records + .last() + .map(|record| record.journal_timestamp_us) + != Some(batch.last_journal_timestamp_us) + { + return Err(ValidationError::TimestampBounds); + } + + for (index, record) in batch.records.iter().enumerate() { + validate_record(index, record)?; + } + Ok(()) +} + +fn validate_record(index: usize, record: &LogRecord) -> Result<(), ValidationError> { + if DateTime::::from_timestamp_micros(record.journal_timestamp_us).is_none() { + return record_error( + index, + "journal timestamp is outside PostgreSQL's supported range", + ); + } + let Some(payload) = &record.payload else { + return record_error(index, "payload is absent"); + }; + match payload { + log_record::Payload::Event(event) => { + if event.bytes_sent > i64::MAX.cast_unsigned() { + return record_error(index, "bytes_sent exceeds PostgreSQL BIGINT"); + } + if event.retries > i32::MAX.cast_unsigned() + || [ + event.active_connections, + event.frontend_connections, + event.backend_connections, + event.server_connections, + event.server_queue, + event.backend_queue, + ] + .into_iter() + .flatten() + .any(|value| value > i32::MAX.cast_unsigned()) + { + return record_error(index, "counter exceeds PostgreSQL INTEGER"); + } + if event.client_port.is_some_and(|port| port > u16::MAX.into()) { + return record_error(index, "client_port exceeds 65535"); + } + if let Some(ip) = &event.client_ip + && ip.parse::().is_err() + { + return record_error(index, "client_ip is invalid"); + } + } + log_record::Payload::Rejected(rejected) => { + if rejected.reason.is_empty() || rejected.message_sha256.len() != 32 { + return record_error(index, "rejected event reason or SHA-256 is invalid"); + } + } + log_record::Payload::Gap(gap) => { + if gap.expected_cursor.is_empty() + || gap.first_available_cursor.is_empty() + || gap.reason.is_empty() + { + return record_error(index, "journal gap fields are incomplete"); + } + } + } + Ok(()) +} + +fn record_error(index: usize, reason: &str) -> Result { + Err(ValidationError::Record { + index, + reason: reason.to_owned(), + }) +} + +#[cfg(test)] +#[allow(clippy::indexing_slicing, clippy::unwrap_used)] +mod tests { + use haproxy_ingest_proto::proto::{HaProxyEvent, LogRecord, log_record}; + + use super::*; + + fn batch() -> LogBatch { + LogBatch { + schema_version: 1, + machine_id: "0123456789abcdef0123456789abcdef".to_owned(), + hostname: "portal-1".to_owned(), + previous_cursor: None, + last_cursor: "cursor-1".to_owned(), + first_journal_timestamp_us: 1_700_000_000_000_000, + last_journal_timestamp_us: 1_700_000_000_000_000, + records: vec![LogRecord { + journal_timestamp_us: 1_700_000_000_000_000, + payload: Some(log_record::Payload::Event(HaProxyEvent::default())), + }], + } + } + + #[test] + fn accepts_a_valid_batch() { + validate_batch(&batch()).unwrap(); + } + + #[test] + fn accepts_a_batch_with_decreasing_journal_timestamps() { + let mut batch = batch(); + batch.previous_cursor = Some("cursor-0".to_owned()); + batch.last_cursor = "cursor-2".to_owned(); + batch.last_journal_timestamp_us = 1_699_999_999_000_000; + batch.records.push(LogRecord { + journal_timestamp_us: batch.last_journal_timestamp_us, + payload: Some(log_record::Payload::Event(HaProxyEvent::default())), + }); + + validate_batch(&batch).unwrap(); + } + + #[test] + fn rejects_batch_timestamps_that_do_not_match_the_endpoint_records() { + for (first, last) in [ + (1_699_999_999_000_000, 1_700_000_000_000_000), + (1_700_000_000_000_000, 1_700_000_001_000_000), + ] { + let mut batch = batch(); + batch.first_journal_timestamp_us = first; + batch.last_journal_timestamp_us = last; + + assert!(matches!( + validate_batch(&batch), + Err(ValidationError::TimestampBounds) + )); + } + } + + #[test] + fn reports_the_invalid_record_index() { + let mut batch = batch(); + batch.records[0].payload = None; + let error = validate_batch(&batch).unwrap_err(); + assert_eq!(error.record_index(), Some(0)); + } +} diff --git a/systems/jeeves/networking/firewall.nix b/systems/jeeves/networking/firewall.nix index ae2ead0..61ad321 100644 --- a/systems/jeeves/networking/firewall.nix +++ b/systems/jeeves/networking/firewall.nix @@ -6,6 +6,7 @@ let gitea_http = 6443; gitea_ssh = 2223; grafana = 3000; + haproxy_ingest = 8444; jellyfin_http = 8096; jellyswarrm = 3100; nix_binary_cache = 5000; @@ -26,6 +27,7 @@ in interfaces.tailscale0.allowedTCPPorts = [ ports.audiobookshelf ports.gems + ports.haproxy_ingest ports.jellyfin_http ports.jellyswarrm ports.nix_binary_cache diff --git a/systems/jeeves/services/haproxy_ingest.nix b/systems/jeeves/services/haproxy_ingest.nix new file mode 100644 index 0000000..d66f6e7 --- /dev/null +++ b/systems/jeeves/services/haproxy_ingest.nix @@ -0,0 +1,82 @@ +{ + pkgs, + ... +}: +let + vars = import ../vars.nix; + haproxy_ingest_server = pkgs.callPackage ../../../rust/haproxy_ingest_server/package.nix { }; +in +{ + users = { + groups.haproxy-ingest = { }; + users.haproxy-ingest = { + isSystemUser = true; + group = "haproxy-ingest"; + }; + }; + + systemd = { + services = { + haproxy-ingest-server = { + description = "HAProxy log ingestion API"; + documentation = [ "file://${../../../rust/haproxy_ingest_proto/README.md}" ]; + wantedBy = [ "multi-user.target" ]; + after = [ + "network-online.target" + "postgresql.service" + "tailscaled-autoconnect.service" + ]; + wants = [ + "network-online.target" + "postgresql.service" + "tailscaled-autoconnect.service" + ]; + + serviceConfig = { + Type = "simple"; + User = "haproxy-ingest"; + Group = "haproxy-ingest"; + ExecStart = "${haproxy_ingest_server}/bin/haproxy-ingest-server"; + Environment = [ + "HAPROXY_INGEST_DATABASE_URL=postgres://haproxy_ingest@localhost/haproxy_ingest?host=/run/postgresql&sslmode=disable" + "HAPROXY_INGEST_LISTEN=0.0.0.0:8444" + "RUST_LOG=info,tower_http=warn" + ]; + EnvironmentFile = "${vars.secrets}/services/haproxy-ingest-server"; + Restart = "on-failure"; + RestartSec = "5s"; + + 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"; + }; + }; + }; + }; +} diff --git a/systems/jeeves/services/postgress.nix b/systems/jeeves/services/postgress.nix index 9ae0c79..c30fdb2 100644 --- a/systems/jeeves/services/postgress.nix +++ b/systems/jeeves/services/postgress.nix @@ -36,6 +36,9 @@ in # signalbot local signalbot signalbot trust + # HAProxy cloud log ingestion API + local haproxy_ingest haproxy_ingest trust + # math local postgres math trust host postgres math 127.0.0.1/32 trust @@ -97,6 +100,13 @@ in replication = true; }; } + { + name = "haproxy_ingest"; + ensureDBOwnership = true; + ensureClauses = { + login = true; + }; + } { name = "signalbot"; ensureDBOwnership = true; @@ -108,6 +118,7 @@ in ensureDatabases = [ "data_science_dev" "gitea" + "haproxy_ingest" "math" "n8n" "richie"