From 5fb721346f28fd5594c667b9fa521d70824d7df1 Mon Sep 17 00:00:00 2001 From: Richie Cahill Date: Fri, 16 Oct 2026 20:55:23 -0400 Subject: [PATCH] feat(haproxy): add initial implementation for HAProxy log ingestion --- rust/haproxy_ingest_proto/Cargo.toml | 17 ++++ rust/haproxy_ingest_proto/README.md | 96 +++++++++++++++++++ rust/haproxy_ingest_proto/build.rs | 11 +++ .../proto/haproxy_ingest.proto | 71 ++++++++++++++ rust/haproxy_ingest_proto/src/lib.rs | 12 +++ 5 files changed, 207 insertions(+) create mode 100644 rust/haproxy_ingest_proto/Cargo.toml create mode 100644 rust/haproxy_ingest_proto/README.md create mode 100644 rust/haproxy_ingest_proto/build.rs create mode 100644 rust/haproxy_ingest_proto/proto/haproxy_ingest.proto create mode 100644 rust/haproxy_ingest_proto/src/lib.rs diff --git a/rust/haproxy_ingest_proto/Cargo.toml b/rust/haproxy_ingest_proto/Cargo.toml new file mode 100644 index 0000000..9da864d --- /dev/null +++ b/rust/haproxy_ingest_proto/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "haproxy-ingest-proto" +version = "0.1.0" +edition = "2024" +description = "Shared protocol types and limits for HAProxy ingestion" +license = "MIT" +build = "build.rs" + +[lints] +workspace = true + +[dependencies] +prost.workspace = true + +[build-dependencies] +prost-build.workspace = true +protoc-bin-vendored.workspace = true diff --git a/rust/haproxy_ingest_proto/README.md b/rust/haproxy_ingest_proto/README.md new file mode 100644 index 0000000..8ce317d --- /dev/null +++ b/rust/haproxy_ingest_proto/README.md @@ -0,0 +1,96 @@ +# HAProxy cloud log ingestion + +This workspace is split into three crates: + +- `haproxy-ingest-proto` owns the shared Protobuf schema and protocol limits. +- `haproxy-log-shipper` is a short-lived client launched by a systemd timer on + `portal-1`. It reads only local `haproxy.service` journal records and drains + them in cursor order. +- `haproxy-ingest-server` is the private HTTP API on Jeeves. It authenticates + the source, commits records to PostgreSQL, and advances the journal cursor + atomically. + +The wire contract is in `proto/haproxy_ingest.proto`; the initial PostgreSQL +schema is in `../haproxy_ingest_server/migrations/001_initial.sql`. + +## One-time setup + +The shipper endpoint is supplied by the `haproxy-log-shipper` service as +`http://jeeves.taile39c31.ts.net:8444`; the environment file only needs the +shared token. Ingestion uses HTTP over Tailscale and requires no certificates. + +Create one random token and install the same value on the shipper and server. +A VM rebuild with a new `/etc/machine-id` is still treated as a new log source. + +On `portal-1`, create `/var/lib/haproxy-log-shipper/environment`: + +```text +HAPROXY_INGEST_TOKEN= +``` + +Set the file ownership and mode to `root:root` and `0600`. + +On Jeeves, create +`/zfs/storage/secrets/services/haproxy-ingest-server`: + +```text +HAPROXY_INGEST_TOKEN= +``` + +Set this file to `root:root` and `0600`. systemd loads it before switching the +API to its unprivileged service account. + +A suitable token can be generated without writing it to shell history: + +```console +openssl rand -hex 32 +``` + +Deploy Jeeves before `portal-1`, then verify: + +```console +systemctl status haproxy-ingest-server.service +systemctl status haproxy-log-shipper.timer +journalctl -u haproxy-log-shipper.service +``` + +The server listens over IPv4 on TCP 8444, which Jeeves' firewall permits only +through `tailscale0`. Both service sandboxes disable IPv6. Tailscale encrypts +traffic between the hosts. The bearer token authorizes access to the ingestion API; +the machine ID identifies a log source but is not an authorization boundary. + +## Protocol behavior + +- Maximum sender batch: 1,000 records or 20 MiB serialized Protobuf. +- Maximum receiver request: 24 MiB compressed, 20 MiB decompressed, 10,000 records. +- Compression: one Zstandard frame over the complete Protobuf body. +- Request timeouts: 3 seconds to connect and 10 seconds total. +- Retry delays: 1, 2, and 4 seconds plus bounded jitter. +- `409`: fetch the authoritative server cursor and rebuild. +- `413`: fail because a sender batch within the configured limits was rejected. +- `422`: replace only the reported offending record with a durable rejection. +- Ambiguous failures: query the status endpoint before retrying. + +Journal retention on `portal-1` is persistent, bounded to 512 MiB, and retained +for up to 14 days. If an acknowledged cursor has already been vacuumed, the +shipper records a `JournalGap` before resuming from the oldest retained matching +entry. + +## Database inspection + +The API owns the `haproxy_ingest` database and creates daily +`haproxy_logs_YYYYMMDD` partitions before using chunked SQLx inserts. +Useful checks include: + +```sql +SELECT machine_id, hostname, last_journal_timestamp, last_commit_time +FROM ingest_source_state; + +SELECT journal_timestamp, client_ip, http_method, request_path, status_code +FROM haproxy_logs +ORDER BY journal_timestamp DESC +LIMIT 100; + +SELECT * FROM ingest_journal_gaps ORDER BY detected_at DESC; +SELECT * FROM ingest_rejected_events ORDER BY ingest_time DESC; +``` diff --git a/rust/haproxy_ingest_proto/build.rs b/rust/haproxy_ingest_proto/build.rs new file mode 100644 index 0000000..5cc39d7 --- /dev/null +++ b/rust/haproxy_ingest_proto/build.rs @@ -0,0 +1,11 @@ +fn main() -> Result<(), Box> { + let protoc = protoc_bin_vendored::protoc_bin_path()?; + unsafe { + std::env::set_var("PROTOC", protoc); + } + let mut config = prost_build::Config::new(); + config.type_attribute(".haproxy.ingest.v1.LogBatch", "#[derive(Eq)]"); + config.compile_protos(&["proto/haproxy_ingest.proto"], &["proto"])?; + println!("cargo:rerun-if-changed=proto/haproxy_ingest.proto"); + Ok(()) +} diff --git a/rust/haproxy_ingest_proto/proto/haproxy_ingest.proto b/rust/haproxy_ingest_proto/proto/haproxy_ingest.proto new file mode 100644 index 0000000..43ab1f7 --- /dev/null +++ b/rust/haproxy_ingest_proto/proto/haproxy_ingest.proto @@ -0,0 +1,71 @@ +syntax = "proto3"; + +package haproxy.ingest.v1; + +message LogBatch { + uint32 schema_version = 1; + string machine_id = 2; + string hostname = 3; + optional string previous_cursor = 4; + string last_cursor = 5; + int64 first_journal_timestamp_us = 6; + int64 last_journal_timestamp_us = 7; + repeated LogRecord records = 8; +} + +message LogRecord { + int64 journal_timestamp_us = 1; + + oneof payload { + HAProxyEvent event = 2; + RejectedEvent rejected = 3; + JournalGap gap = 4; + } +} + +message HAProxyEvent { + reserved 22; + + optional string client_ip = 1; + optional uint32 client_port = 2; + optional string frontend_name = 3; + optional string backend_name = 4; + optional string server_name = 5; + optional string http_method = 6; + optional string request_path = 7; + optional string http_version = 8; + int32 status_code = 9; + uint64 bytes_sent = 10; + int32 time_request_ms = 11; + int32 time_wait_ms = 12; + int32 time_connect_ms = 13; + int32 time_response_ms = 14; + int32 time_total_ms = 15; + optional string termination_state = 16; + uint32 retries = 17; + optional string tls_sni = 18; + optional string tls_version = 19; + optional string tls_cipher = 20; + optional bool tls_resumed = 21; + repeated string normalization_flags = 23; + optional uint32 active_connections = 24; + optional uint32 frontend_connections = 25; + optional uint32 backend_connections = 26; + optional uint32 server_connections = 27; + optional uint32 server_queue = 28; + optional uint32 backend_queue = 29; +} + +message RejectedEvent { + string reason = 1; + bytes message_sha256 = 2; + optional string raw_message_prefix = 3; + bool raw_message_truncated = 4; +} + +message JournalGap { + string expected_cursor = 1; + string first_available_cursor = 2; + optional int64 first_available_timestamp_us = 3; + string reason = 4; +} diff --git a/rust/haproxy_ingest_proto/src/lib.rs b/rust/haproxy_ingest_proto/src/lib.rs new file mode 100644 index 0000000..2d94e19 --- /dev/null +++ b/rust/haproxy_ingest_proto/src/lib.rs @@ -0,0 +1,12 @@ +pub mod proto { + #![allow(clippy::large_enum_variant)] + + include!(concat!(env!("OUT_DIR"), "/haproxy.ingest.v1.rs")); +} + +pub const SCHEMA_VERSION: u32 = 1; +pub const MAX_BATCH_RECORDS: usize = 1_000; +pub const MAX_BATCH_PROTO_BYTES: usize = 20 * 1024 * 1024; +pub const MAX_COMPRESSED_BYTES: usize = 24 * 1024 * 1024; +pub const MAX_DECOMPRESSED_BYTES: usize = MAX_BATCH_PROTO_BYTES; +pub const SERVER_MAX_RECORDS: usize = 10_000;