feat(haproxy): add initial implementation for HAProxy log ingestion

This commit is contained in:
Richie committed 2026-10-16 20:55:23 -04:00
1 parent 876683066c
commit 5fb721346f
5 files changed
+207

No files matched your search

+17
View File
@@ -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
+96
View File
@@ -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=<at-least-32-character-random-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=<the-same-random-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;
```
+11
View File
@@ -0,0 +1,11 @@
fn main() -> Result<(), Box<dyn std::error::Error>> {
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(())
}
@@ -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;
}
+12
View File
@@ -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;