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.
This commit is contained in:
Richie committed
1 parent
3960254b27
commit
438b4e09e6
9 files changed
+1365
No files matched your search
@@ -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
|
||||
@@ -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);
|
||||
@@ -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;
|
||||
};
|
||||
}
|
||||
@@ -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<String>,
|
||||
pub last_cursor: Option<String>,
|
||||
pub last_journal_timestamp: Option<DateTime<Utc>>,
|
||||
pub last_commit_time: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
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<Option<SourceStatus>, 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<T, ApiError>
|
||||
where
|
||||
T: Decode<'r, Postgres> + Type<Postgres>,
|
||||
{
|
||||
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<Option<String>, 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<Utc>,
|
||||
client_ip: Option<String>,
|
||||
client_port: Option<i32>,
|
||||
bytes_sent: i64,
|
||||
retries: i32,
|
||||
active_connections: Option<i32>,
|
||||
frontend_connections: Option<i32>,
|
||||
backend_connections: Option<i32>,
|
||||
server_connections: Option<i32>,
|
||||
server_queue: Option<i32>,
|
||||
backend_queue: Option<i32>,
|
||||
}
|
||||
|
||||
fn prepare_events(batch: &LogBatch) -> Result<Vec<PreparedEvent<'_>>, 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::<IpAddr>().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::<Postgres>::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<Utc>,
|
||||
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<DateTime<Utc>>,
|
||||
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<u32>) -> Result<Option<i32>, 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");
|
||||
}
|
||||
}
|
||||
@@ -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<Self> {
|
||||
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<String>,
|
||||
last_cursor: Option<String>,
|
||||
last_journal_timestamp: Option<DateTime<Utc>>,
|
||||
last_commit_time: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
#[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<usize>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
last_cursor: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct ApiError {
|
||||
status: StatusCode,
|
||||
message: String,
|
||||
record_index: Option<usize>,
|
||||
last_cursor: Option<String>,
|
||||
}
|
||||
|
||||
impl ApiError {
|
||||
fn new(status: StatusCode, message: impl Into<String>) -> 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<String>) -> 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::<SocketAddr>(),
|
||||
)
|
||||
.await
|
||||
.context("ingestion server failed")
|
||||
}
|
||||
|
||||
async fn get_status(
|
||||
State(state): State<AppState>,
|
||||
Path(machine_id): Path<String>,
|
||||
headers: HeaderMap,
|
||||
) -> Result<axum::Json<StatusResponse>, 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<AppState>,
|
||||
ConnectInfo(peer): ConnectInfo<SocketAddr>,
|
||||
headers: HeaderMap,
|
||||
body: Bytes,
|
||||
) -> Result<axum::Json<CommitResponse>, 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<Vec<u8>, 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<CommitResponse, ApiError> {
|
||||
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<DateTime<Utc>, 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
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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<usize> {
|
||||
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::<Utc>::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::<IpAddr>().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<T>(index: usize, reason: &str) -> Result<T, ValidationError> {
|
||||
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));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user