mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
feat(metrics): Prometheus observability + CAKE container contract (#88)
This commit is contained in:
tlongwell-block
@@ -0,0 +1,7 @@
|
||||
target/
|
||||
desktop/
|
||||
.git/
|
||||
.scratch/
|
||||
*.md
|
||||
!README.md
|
||||
.env*
|
||||
Generated
+223
-1
@@ -29,6 +29,18 @@ dependencies = [
|
||||
"cpufeatures",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ahash"
|
||||
version = "0.8.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"once_cell",
|
||||
"version_check",
|
||||
"zerocopy",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "aho-corasick"
|
||||
version = "1.1.4"
|
||||
@@ -199,6 +211,28 @@ dependencies = [
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "aws-lc-rs"
|
||||
version = "1.16.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "94bffc006df10ac2a68c83692d734a465f8ee6c5b384d8545a636f81d858f4bf"
|
||||
dependencies = [
|
||||
"aws-lc-sys",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "aws-lc-sys"
|
||||
version = "0.38.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4321e568ed89bb5a7d291a7f37997c2c0df89809d7b6d12062c81ddb54aa782e"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"cmake",
|
||||
"dunce",
|
||||
"fs_extra",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "aws-region"
|
||||
version = "0.25.5"
|
||||
@@ -456,6 +490,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "aebf35691d1bfb0ac386a69bac2fde4dd276fb618cf8bf4f5318fe285e821bb2"
|
||||
dependencies = [
|
||||
"find-msvc-tools",
|
||||
"jobserver",
|
||||
"libc",
|
||||
"shlex",
|
||||
]
|
||||
|
||||
@@ -571,6 +607,15 @@ version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9"
|
||||
|
||||
[[package]]
|
||||
name = "cmake"
|
||||
version = "0.1.57"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "75443c44cd6b379beb8c5b45d85d0773baf31cce901fe7bb252f4eff3008ef7d"
|
||||
dependencies = [
|
||||
"cc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "color_quant"
|
||||
version = "1.1.0"
|
||||
@@ -903,6 +948,12 @@ version = "0.15.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b"
|
||||
|
||||
[[package]]
|
||||
name = "dunce"
|
||||
version = "1.0.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
|
||||
|
||||
[[package]]
|
||||
name = "dyn-clone"
|
||||
version = "1.0.20"
|
||||
@@ -1016,6 +1067,12 @@ version = "0.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
|
||||
|
||||
[[package]]
|
||||
name = "foldhash"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
|
||||
|
||||
[[package]]
|
||||
name = "foreign-types"
|
||||
version = "0.3.2"
|
||||
@@ -1040,6 +1097,12 @@ dependencies = [
|
||||
"percent-encoding",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fs_extra"
|
||||
version = "1.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
|
||||
|
||||
[[package]]
|
||||
name = "futures"
|
||||
version = "0.3.32"
|
||||
@@ -1199,6 +1262,25 @@ dependencies = [
|
||||
"weezl",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "h2"
|
||||
version = "0.4.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54"
|
||||
dependencies = [
|
||||
"atomic-waker",
|
||||
"bytes",
|
||||
"fnv",
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
"http 1.4.0",
|
||||
"indexmap",
|
||||
"slab",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.14.5"
|
||||
@@ -1213,7 +1295,7 @@ checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
|
||||
dependencies = [
|
||||
"allocator-api2",
|
||||
"equivalent",
|
||||
"foldhash",
|
||||
"foldhash 0.1.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1221,6 +1303,9 @@ name = "hashbrown"
|
||||
version = "0.16.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
|
||||
dependencies = [
|
||||
"foldhash 0.2.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hashlink"
|
||||
@@ -1391,6 +1476,7 @@ dependencies = [
|
||||
"bytes",
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"h2",
|
||||
"http 1.4.0",
|
||||
"http-body 1.0.1",
|
||||
"httparse",
|
||||
@@ -1413,6 +1499,7 @@ dependencies = [
|
||||
"hyper 1.8.1",
|
||||
"hyper-util",
|
||||
"rustls",
|
||||
"rustls-native-certs",
|
||||
"rustls-pki-types",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
@@ -1708,6 +1795,16 @@ version = "1.0.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2"
|
||||
|
||||
[[package]]
|
||||
name = "jobserver"
|
||||
version = "0.1.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33"
|
||||
dependencies = [
|
||||
"getrandom 0.3.4",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "js-sys"
|
||||
version = "0.3.91"
|
||||
@@ -1863,6 +1960,54 @@ version = "2.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
|
||||
|
||||
[[package]]
|
||||
name = "metrics"
|
||||
version = "0.24.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5d5312e9ba3771cfa961b585728215e3d972c950a3eed9252aa093d6301277e8"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"portable-atomic",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "metrics-exporter-prometheus"
|
||||
version = "0.18.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3589659543c04c7dc5526ec858591015b87cd8746583b51b48ef4353f99dbcda"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"http-body-util",
|
||||
"hyper 1.8.1",
|
||||
"hyper-rustls",
|
||||
"hyper-util",
|
||||
"indexmap",
|
||||
"ipnet",
|
||||
"metrics",
|
||||
"metrics-util",
|
||||
"quanta",
|
||||
"rustls",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "metrics-util"
|
||||
version = "0.20.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cdfb1365fea27e6dd9dc1dbc19f570198bc86914533ad639dae939635f096be4"
|
||||
dependencies = [
|
||||
"crossbeam-epoch",
|
||||
"crossbeam-utils",
|
||||
"hashbrown 0.16.1",
|
||||
"metrics",
|
||||
"quanta",
|
||||
"rand 0.9.2",
|
||||
"rand_xoshiro",
|
||||
"sketches-ddsketch",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mime"
|
||||
version = "0.3.17"
|
||||
@@ -2361,6 +2506,21 @@ version = "0.1.28"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b5a041e753da8b807c9255f28de81879c78c876392ff2469cde94799b2896b9d"
|
||||
|
||||
[[package]]
|
||||
name = "quanta"
|
||||
version = "0.12.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
"libc",
|
||||
"once_cell",
|
||||
"raw-cpuid",
|
||||
"wasi",
|
||||
"web-sys",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quick-error"
|
||||
version = "2.0.1"
|
||||
@@ -2512,6 +2672,24 @@ dependencies = [
|
||||
"getrandom 0.3.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand_xoshiro"
|
||||
version = "0.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f703f4665700daf5512dcca5f43afa6af89f09db47fb56be587f80636bda2d41"
|
||||
dependencies = [
|
||||
"rand_core 0.9.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "raw-cpuid"
|
||||
version = "11.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186"
|
||||
dependencies = [
|
||||
"bitflags",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redis"
|
||||
version = "0.27.6"
|
||||
@@ -2772,6 +2950,7 @@ version = "0.23.37"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4"
|
||||
dependencies = [
|
||||
"aws-lc-rs",
|
||||
"once_cell",
|
||||
"ring",
|
||||
"rustls-pki-types",
|
||||
@@ -2780,6 +2959,18 @@ dependencies = [
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustls-native-certs"
|
||||
version = "0.8.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63"
|
||||
dependencies = [
|
||||
"openssl-probe",
|
||||
"rustls-pki-types",
|
||||
"schannel",
|
||||
"security-framework",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustls-pki-types"
|
||||
version = "1.14.0"
|
||||
@@ -2796,6 +2987,7 @@ version = "0.103.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d7df23109aa6c1567d1c575b9952556388da57401e4ace1d15f79eedad0d8f53"
|
||||
dependencies = [
|
||||
"aws-lc-rs",
|
||||
"ring",
|
||||
"rustls-pki-types",
|
||||
"untrusted",
|
||||
@@ -3123,6 +3315,12 @@ dependencies = [
|
||||
"time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sketches-ddsketch"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0c6f73aeb92d671e0cc4dca167e59b2deb6387c375391bc99ee743f326994a2b"
|
||||
|
||||
[[package]]
|
||||
name = "slab"
|
||||
version = "0.4.12"
|
||||
@@ -3422,6 +3620,8 @@ dependencies = [
|
||||
"futures-util",
|
||||
"hex",
|
||||
"infer",
|
||||
"metrics",
|
||||
"metrics-exporter-prometheus",
|
||||
"moka",
|
||||
"nostr",
|
||||
"redis",
|
||||
@@ -4528,6 +4728,28 @@ dependencies = [
|
||||
"wasite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "winapi"
|
||||
version = "0.3.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
|
||||
dependencies = [
|
||||
"winapi-i686-pc-windows-gnu",
|
||||
"winapi-x86_64-pc-windows-gnu",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "winapi-i686-pc-windows-gnu"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
|
||||
|
||||
[[package]]
|
||||
name = "winapi-x86_64-pc-windows-gnu"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
|
||||
|
||||
[[package]]
|
||||
name = "windows-core"
|
||||
version = "0.62.2"
|
||||
|
||||
@@ -57,6 +57,8 @@ cron = "0.12"
|
||||
# Observability
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
|
||||
metrics = "0.24"
|
||||
metrics-exporter-prometheus = "0.18"
|
||||
|
||||
# Error handling
|
||||
thiserror = "2"
|
||||
|
||||
+34
@@ -0,0 +1,34 @@
|
||||
# ── Build stage ──────────────────────────────────────────────
|
||||
# Hard-code --platform to prevent exec format error on ARM Macs.
|
||||
FROM --platform=linux/amd64 rust:1.88-bookworm AS builder
|
||||
WORKDIR /build
|
||||
COPY . .
|
||||
RUN cargo build --release -p sprout-relay \
|
||||
&& strip target/release/sprout-relay
|
||||
|
||||
# ── Runtime stage ────────────────────────────────────────────
|
||||
FROM --platform=linux/amd64 debian:bookworm-slim
|
||||
|
||||
# CAKE: non-root UID 1000 (numeric, not username)
|
||||
RUN groupadd -g 1000 sprout && useradd -u 1000 -g sprout -m sprout
|
||||
|
||||
# CAKE: writable dirs
|
||||
RUN mkdir -p /cache /tmp && chown sprout:sprout /cache /tmp
|
||||
|
||||
# socat for Istio abstract→file socket bridge
|
||||
RUN apt-get update && apt-get install -y --no-install-recommends \
|
||||
ca-certificates socat && rm -rf /var/lib/apt/lists/*
|
||||
|
||||
COPY --from=builder /build/target/release/sprout-relay /code/sprout-relay
|
||||
COPY script/start /code/start
|
||||
RUN chmod +x /code/start
|
||||
|
||||
# CAKE: required Envoy env vars (overridden at runtime by CAKE).
|
||||
ENV ENVOY_ADMIN_SOCKET_PATH="@envoy-admin.sock" \
|
||||
ENVOY_INGRESS_PORT="20001" \
|
||||
ENVOY_HTTP_EGRESS_SOCKET_PATH="@egress.sock" \
|
||||
ENVOY_DATADOG_PORT="3030" \
|
||||
CASH_FRAMEWORK="rust"
|
||||
|
||||
USER 1000
|
||||
ENTRYPOINT ["/code/start"]
|
||||
@@ -47,6 +47,8 @@ sha2 = { workspace = true }
|
||||
hex = { workspace = true }
|
||||
url = { workspace = true }
|
||||
moka = { workspace = true }
|
||||
metrics = { workspace = true }
|
||||
metrics-exporter-prometheus = { workspace = true }
|
||||
|
||||
[features]
|
||||
dev = ["sprout-auth/dev"]
|
||||
|
||||
@@ -117,6 +117,13 @@ pub async fn upload_blob(
|
||||
)
|
||||
.await?;
|
||||
|
||||
// Normalize MIME to a known set to bound label cardinality.
|
||||
let mime_label = match descriptor.mime_type.as_str() {
|
||||
"image/jpeg" | "image/png" | "image/gif" | "image/webp" => &descriptor.mime_type,
|
||||
_ => "other",
|
||||
};
|
||||
metrics::counter!("sprout_media_uploads_total", "mime" => mime_label.to_owned()).increment(1);
|
||||
|
||||
// Fire-and-forget audit — never block the response on audit I/O.
|
||||
let audit = state.audit.clone();
|
||||
let desc = descriptor.clone();
|
||||
|
||||
@@ -48,6 +48,12 @@ pub struct Config {
|
||||
/// Optional Unix Domain Socket path. When set, the relay also listens on this
|
||||
/// UDS for traffic (e.g. service mesh sidecar). Health probes still use TCP.
|
||||
pub uds_path: Option<String>,
|
||||
/// TCP port for the health-only router (`/_liveness`, `/_readiness`, `/_status`).
|
||||
/// Separate from the app router so K8s probes bypass Istio and auth middleware.
|
||||
pub health_port: u16,
|
||||
/// TCP port for the Prometheus metrics exporter (`GET /metrics`).
|
||||
pub metrics_port: u16,
|
||||
|
||||
/// When true, NIP-42 pubkey-only authentication (no JWT or API token) is
|
||||
/// restricted to pubkeys in the `pubkey_allowlist` table. Users with valid
|
||||
/// API tokens or Okta JWTs bypass the allowlist entirely.
|
||||
@@ -137,6 +143,16 @@ impl Config {
|
||||
.map(|s| s.trim().to_string())
|
||||
.filter(|s| !s.is_empty());
|
||||
|
||||
let health_port = std::env::var("SPROUT_HEALTH_PORT")
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(8080);
|
||||
|
||||
let metrics_port = std::env::var("SPROUT_METRICS_PORT")
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(9102);
|
||||
|
||||
let media = sprout_media::MediaConfig {
|
||||
s3_endpoint: std::env::var("SPROUT_S3_ENDPOINT")
|
||||
.unwrap_or_else(|_| "http://localhost:9000".to_string()),
|
||||
@@ -193,6 +209,8 @@ impl Config {
|
||||
cors_origins,
|
||||
relay_private_key,
|
||||
uds_path,
|
||||
health_port,
|
||||
metrics_port,
|
||||
pubkey_allowlist_enabled,
|
||||
media,
|
||||
})
|
||||
|
||||
@@ -82,6 +82,7 @@ impl ConnectionState {
|
||||
let count = self.backpressure_count.fetch_add(1, Ordering::Relaxed) + 1;
|
||||
if count >= SLOW_CLIENT_GRACE_LIMIT {
|
||||
warn!(conn_id = %self.conn_id, count, "sustained backpressure — closing slow client");
|
||||
metrics::counter!("sprout_ws_backpressure_disconnects_total").increment(1);
|
||||
self.cancel.cancel();
|
||||
} else {
|
||||
warn!(conn_id = %self.conn_id, count, grace = SLOW_CLIENT_GRACE_LIMIT, "send buffer full — grace {count}/{SLOW_CLIENT_GRACE_LIMIT}");
|
||||
@@ -134,6 +135,7 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
|
||||
});
|
||||
|
||||
info!(conn_id = %conn_id, addr = %addr, "WebSocket connection established");
|
||||
metrics::counter!("sprout_ws_connections_total").increment(1);
|
||||
|
||||
let challenge_msg = RelayMessage::auth_challenge(&challenge);
|
||||
if tx
|
||||
@@ -145,6 +147,10 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
|
||||
return;
|
||||
}
|
||||
|
||||
// Gauge incremented AFTER challenge send succeeds — early disconnects
|
||||
// don't leak. Decremented in the cleanup path below.
|
||||
metrics::gauge!("sprout_ws_connections_active").increment(1.0);
|
||||
|
||||
// Register after challenge succeeds — avoids leaked entries on early disconnect.
|
||||
state.conn_manager.register(
|
||||
conn_id,
|
||||
@@ -181,6 +187,7 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
|
||||
|
||||
state.sub_registry.remove_connection(conn.conn_id);
|
||||
state.conn_manager.deregister(conn.conn_id);
|
||||
metrics::gauge!("sprout_ws_connections_active").decrement(1.0);
|
||||
info!(conn_id = %conn_id, addr = %addr, "WebSocket connection closed");
|
||||
|
||||
drop(permit);
|
||||
|
||||
@@ -52,6 +52,8 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
|
||||
}
|
||||
});
|
||||
|
||||
metrics::counter!("sprout_auth_attempts_total", "method" => if auth_token.as_ref().is_some_and(|t| t.starts_with("sprout_")) { "api_token" } else { "nip42" }).increment(1);
|
||||
|
||||
if let Some(ref token) = auth_token {
|
||||
if token.starts_with("sprout_") {
|
||||
// ── API token path ──────────────────────────────────────────────
|
||||
@@ -119,6 +121,7 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(conn_id = %conn_id, error = %e, "API token verification failed");
|
||||
metrics::counter!("sprout_auth_failures_total", "reason" => "api_token_invalid").increment(1);
|
||||
*conn.auth_state.write().await = AuthState::Failed;
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
@@ -154,6 +157,8 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
|
||||
};
|
||||
if !allowed {
|
||||
warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), "pubkey not in allowlist");
|
||||
metrics::counter!("sprout_auth_failures_total", "reason" => "allowlist_denied")
|
||||
.increment(1);
|
||||
*conn.auth_state.write().await = AuthState::Failed;
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
@@ -169,6 +174,8 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(conn_id = %conn_id, error = %e, "NIP-42 auth failed");
|
||||
metrics::counter!("sprout_auth_failures_total", "reason" => "nip42_invalid")
|
||||
.increment(1);
|
||||
*conn.auth_state.write().await = AuthState::Failed;
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
|
||||
@@ -25,6 +25,34 @@ use crate::connection::{AuthState, ConnectionState};
|
||||
use crate::protocol::RelayMessage;
|
||||
use crate::state::AppState;
|
||||
|
||||
/// Increment the rejection counter with a bounded reason label.
|
||||
fn reject(reason: &'static str) {
|
||||
metrics::counter!("sprout_events_rejected_total", "reason" => reason).increment(1);
|
||||
}
|
||||
|
||||
/// Bound the `kind` label to prevent cardinality explosion from arbitrary Nostr kinds.
|
||||
/// Known Sprout kind ranges pass through; everything else becomes "other".
|
||||
/// See `sprout_core::kind` for the authoritative registry.
|
||||
fn bounded_kind_label(kind: u32) -> String {
|
||||
match kind {
|
||||
0..=9 | 1059 | 1063 => kind.to_string(), // NIP-01 core + gift-wrap + file-meta + stream msg
|
||||
9000..=9022 | 9100 | 9110 | 9900 => kind.to_string(), // NIP-29 admin + system
|
||||
20000..=29999 => kind.to_string(), // Ephemeral (presence, typing) + AUTH (22242)
|
||||
30023 | 39000..=39003 => kind.to_string(), // NIP-23 + NIP-29 addressable state
|
||||
40002..=40100 => kind.to_string(), // Stream messaging + canvas + system msg
|
||||
41001..=41003 => kind.to_string(), // DMs
|
||||
42001..=42003 => kind.to_string(), // Topics
|
||||
43001..=43006 => kind.to_string(), // Agent jobs
|
||||
44001..=44004 | 44100..=44101 => kind.to_string(), // Subscriptions + notifications
|
||||
45001..=45003 => kind.to_string(), // Forum
|
||||
46001..=46012 => kind.to_string(), // Workflow engine
|
||||
47001..=47003 => kind.to_string(), // User groups
|
||||
48001..=48005 | 48100..=48105 => kind.to_string(), // Admin + huddle
|
||||
49001 => kind.to_string(), // Media upload
|
||||
_ => "other".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Publish a stored event to subscribers and kick off async side effects.
|
||||
pub(crate) async fn dispatch_persistent_event(
|
||||
state: &Arc<AppState>,
|
||||
@@ -76,8 +104,16 @@ pub(crate) async fn dispatch_persistent_event(
|
||||
let search = Arc::clone(&state.search);
|
||||
let stored_for_search = stored_event.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = search.index_event(&stored_for_search).await {
|
||||
error!(event_id = %stored_for_search.event.id.to_hex(), "Search index failed: {e}");
|
||||
let t = std::time::Instant::now();
|
||||
match search.index_event(&stored_for_search).await {
|
||||
Ok(()) => {
|
||||
metrics::histogram!("sprout_search_index_seconds")
|
||||
.record(t.elapsed().as_secs_f64());
|
||||
}
|
||||
Err(e) => {
|
||||
metrics::counter!("sprout_search_index_errors_total").increment(1);
|
||||
error!(event_id = %stored_for_search.event.id.to_hex(), "Search index failed: {e}");
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -95,8 +131,11 @@ pub(crate) async fn dispatch_persistent_event(
|
||||
channel_id: audit_channel_id,
|
||||
metadata: serde_json::Value::Null,
|
||||
};
|
||||
let t = std::time::Instant::now();
|
||||
if let Err(e) = audit.log(entry).await {
|
||||
error!(event_id = %audit_event_id, "Audit log failed: {e}");
|
||||
} else {
|
||||
metrics::histogram!("sprout_audit_log_seconds").record(t.elapsed().as_secs_f64());
|
||||
}
|
||||
});
|
||||
|
||||
@@ -115,9 +154,13 @@ pub(crate) async fn dispatch_persistent_event(
|
||||
{
|
||||
let workflow_engine = Arc::clone(&state.workflow_engine);
|
||||
let workflow_event = stored_event.clone();
|
||||
let trigger_kind = kind_u32.to_string();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = workflow_engine.on_event(&workflow_event).await {
|
||||
tracing::error!(event_id = ?workflow_event.event.id, "Workflow trigger failed: {e}");
|
||||
} else {
|
||||
metrics::counter!("sprout_workflow_runs_total", "trigger" => trigger_kind)
|
||||
.increment(1);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -127,9 +170,12 @@ pub(crate) async fn dispatch_persistent_event(
|
||||
|
||||
/// Handle an EVENT message: authenticate, verify, store, fan-out, index, and audit the event.
|
||||
pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<AppState>) {
|
||||
let start = std::time::Instant::now();
|
||||
let event_id_hex = event.id.to_hex();
|
||||
let kind_u32 = event_kind_u32(&event);
|
||||
let kind_str = bounded_kind_label(kind_u32);
|
||||
debug!(event_id = %event_id_hex, kind = kind_u32, "EVENT");
|
||||
metrics::counter!("sprout_events_received_total", "kind" => kind_str.clone()).increment(1);
|
||||
|
||||
let (conn_id, pubkey_hex, pubkey_bytes, auth_pubkey, has_proxy_scope) = {
|
||||
let auth = conn.auth_state.read().await;
|
||||
@@ -155,6 +201,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
)
|
||||
}
|
||||
_ => {
|
||||
reject("auth");
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
false,
|
||||
@@ -172,6 +219,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
// via NIP-42 auth for rate limiting and abuse prevention.
|
||||
let is_gift_wrap = kind_u32 == KIND_GIFT_WRAP;
|
||||
if event.pubkey != auth_pubkey && !has_proxy_scope && !is_gift_wrap {
|
||||
reject("invalid");
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
false,
|
||||
@@ -189,6 +237,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
}
|
||||
|
||||
if kind_u32 == KIND_AUTH {
|
||||
reject("invalid");
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
false,
|
||||
@@ -199,6 +248,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
|
||||
// Membership notification events are relay-signed only — reject client submissions.
|
||||
if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION || kind_u32 == KIND_MEMBER_REMOVED_NOTIFICATION {
|
||||
reject("invalid");
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
false,
|
||||
@@ -227,6 +277,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
match verify_result {
|
||||
Ok(Ok(())) => {}
|
||||
Ok(Err(e)) => {
|
||||
reject("invalid");
|
||||
warn!(conn_id = %conn_id, event_id = %event_id_hex, "Verification failed: {e}");
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
@@ -307,6 +358,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
};
|
||||
|
||||
if requires_h_channel_scope(kind_u32) && channel_id.is_none() {
|
||||
reject("invalid");
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
false,
|
||||
@@ -447,6 +499,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
reject("error");
|
||||
error!(conn_id = %conn_id, event_id = %event_id_hex, "DB insert failed: {e}");
|
||||
conn.send(RelayMessage::ok(
|
||||
&event_id_hex,
|
||||
@@ -473,6 +526,10 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
|
||||
let fan_out = dispatch_persistent_event(&state, &stored_event, kind_u32, &pubkey_hex).await;
|
||||
|
||||
metrics::counter!("sprout_events_stored_total", "kind" => kind_str).increment(1);
|
||||
metrics::histogram!("sprout_event_processing_seconds").record(start.elapsed().as_secs_f64());
|
||||
metrics::histogram!("sprout_fanout_recipients").record(fan_out as f64);
|
||||
|
||||
conn.send(RelayMessage::ok(&event_id_hex, true, ""));
|
||||
|
||||
info!(
|
||||
|
||||
@@ -12,6 +12,8 @@ pub mod connection;
|
||||
pub mod error;
|
||||
/// WebSocket message handlers for NIP-01 client commands.
|
||||
pub mod handlers;
|
||||
/// Prometheus metrics: recorder, upkeep, HTTP middleware.
|
||||
pub mod metrics;
|
||||
/// NIP-11 relay information document.
|
||||
pub mod nip11;
|
||||
/// NIP-01 client/relay message parsing.
|
||||
|
||||
+122
-23
@@ -1,3 +1,4 @@
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::sync::Arc;
|
||||
|
||||
use tracing::{error, info};
|
||||
@@ -9,13 +10,17 @@ use sprout_db::{Db, DbConfig};
|
||||
use sprout_pubsub::PubSubManager;
|
||||
use sprout_search::{SearchConfig, SearchService};
|
||||
|
||||
use sprout_relay::{config::Config, router::build_router, state::AppState};
|
||||
use sprout_relay::config::Config;
|
||||
use sprout_relay::metrics as relay_metrics;
|
||||
use sprout_relay::router::{build_health_router, build_router};
|
||||
use sprout_relay::state::AppState;
|
||||
use sprout_workflow::WorkflowEngine;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
// JSON-only structured logs — simple, machine-parseable, CAKE-compatible.
|
||||
tracing_subscriber::registry()
|
||||
.with(fmt::layer())
|
||||
.with(fmt::layer().json().flatten_event(true))
|
||||
.with(EnvFilter::from_default_env().add_directive("sprout_relay=info".parse()?))
|
||||
.init();
|
||||
|
||||
@@ -25,7 +30,20 @@ async fn main() -> anyhow::Result<()> {
|
||||
error!("Invalid configuration: {e}");
|
||||
anyhow::anyhow!("Configuration error: {e}")
|
||||
})?;
|
||||
info!(bind_addr = %config.bind_addr, relay_url = %config.relay_url, "Config loaded");
|
||||
info!(
|
||||
bind_addr = %config.bind_addr,
|
||||
relay_url = %config.relay_url,
|
||||
health_port = config.health_port,
|
||||
metrics_port = config.metrics_port,
|
||||
"Config loaded"
|
||||
);
|
||||
|
||||
// ── Metrics recorder (Prometheus exporter on :9102) ──────────────────────
|
||||
relay_metrics::install(config.metrics_port);
|
||||
info!(
|
||||
port = config.metrics_port,
|
||||
"Prometheus metrics exporter started"
|
||||
);
|
||||
|
||||
let db_config = DbConfig {
|
||||
database_url: config.database_url.clone(),
|
||||
@@ -153,6 +171,7 @@ async fn main() -> anyhow::Result<()> {
|
||||
}
|
||||
|
||||
let matches = state_for_sub.sub_registry.fan_out(&stored);
|
||||
metrics::counter!("sprout_multinode_fanout_total").increment(1);
|
||||
if matches.is_empty() {
|
||||
continue;
|
||||
}
|
||||
@@ -182,6 +201,7 @@ async fn main() -> anyhow::Result<()> {
|
||||
}
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
||||
metrics::counter!("sprout_multinode_fanout_lag_total").increment(n);
|
||||
tracing::warn!("Multi-node fan-out lagged by {n} messages");
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
|
||||
@@ -194,21 +214,68 @@ async fn main() -> anyhow::Result<()> {
|
||||
}
|
||||
|
||||
let router = build_router(Arc::clone(&state));
|
||||
let health_router = build_health_router(Arc::clone(&state));
|
||||
|
||||
serve_dual(router, &config).await
|
||||
serve(router, health_router, Arc::clone(&state)).await
|
||||
}
|
||||
|
||||
/// Bind TCP (always) and optionally UDS, then run both concurrently.
|
||||
/// If either listener exits, the function returns and the process exits.
|
||||
async fn serve_dual(router: axum::Router, config: &Config) -> anyhow::Result<()> {
|
||||
/// Bind all listeners and run with graceful shutdown.
|
||||
///
|
||||
/// ```text
|
||||
/// ┌─────────────────────────────────────────────────────────┐
|
||||
/// │ Listener 1: TCP SPROUT_BIND_ADDR:3000 (app router) │
|
||||
/// │ Listener 2: UDS SPROUT_UDS_PATH (app, optional)│
|
||||
/// │ Listener 3: TCP 0.0.0.0:8080 (health only) │
|
||||
/// │ Listener 4: TCP 0.0.0.0:9102 (metrics, via │
|
||||
/// │ PrometheusBuilder — already bound) │
|
||||
/// │ │
|
||||
/// │ SIGTERM → shutting_down=true → readiness 503 │
|
||||
/// │ → graceful drain (30s) → exit │
|
||||
/// └─────────────────────────────────────────────────────────┘
|
||||
/// ```
|
||||
async fn serve(
|
||||
router: axum::Router,
|
||||
health_router: axum::Router,
|
||||
state: Arc<AppState>,
|
||||
) -> anyhow::Result<()> {
|
||||
let config = &state.config;
|
||||
|
||||
// ── Health listener (port 8080) ──────────────────────────────────────────
|
||||
let health_listener = tokio::net::TcpListener::bind(("0.0.0.0", config.health_port))
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to bind health port {}: {e}", config.health_port))?;
|
||||
info!(port = config.health_port, "Health probe listener started");
|
||||
tokio::spawn(async move {
|
||||
axum::serve(health_listener, health_router).await.ok();
|
||||
});
|
||||
|
||||
// ── Shutdown coordination ────────────────────────────────────────────────
|
||||
let (shutdown_tx, _) = tokio::sync::watch::channel(false);
|
||||
let shutdown_flag = Arc::clone(&state.shutting_down);
|
||||
let tx = shutdown_tx.clone();
|
||||
tokio::spawn(async move {
|
||||
shutdown_signal().await;
|
||||
shutdown_flag.store(true, Ordering::Relaxed);
|
||||
info!("Shutdown signal received — readiness now returns 503");
|
||||
// 5s grace: let K8s stop routing new traffic before we close listeners.
|
||||
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
|
||||
info!("Starting graceful drain (30s timeout)");
|
||||
let _ = tx.send(true);
|
||||
// Hard timeout: force exit if connections don't drain within 30s.
|
||||
tokio::time::sleep(std::time::Duration::from_secs(30)).await;
|
||||
tracing::error!("Drain timeout exceeded — forcing exit");
|
||||
std::process::exit(1);
|
||||
});
|
||||
|
||||
// ── App listener (TCP) ───────────────────────────────────────────────────
|
||||
let tcp_listener = tokio::net::TcpListener::bind(&config.bind_addr)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to bind {}: {e}", config.bind_addr))?;
|
||||
info!(addr = %config.bind_addr, "sprout-relay TCP listening");
|
||||
|
||||
// ── App listener (UDS, optional) ─────────────────────────────────────────
|
||||
#[cfg(unix)]
|
||||
if let Some(ref uds_path) = config.uds_path {
|
||||
// Only remove stale socket files — refuse to delete non-socket paths.
|
||||
use std::os::unix::fs::FileTypeExt as _;
|
||||
match std::fs::symlink_metadata(uds_path) {
|
||||
Ok(meta) if meta.file_type().is_socket() => {
|
||||
@@ -216,42 +283,74 @@ async fn serve_dual(router: axum::Router, config: &Config) -> anyhow::Result<()>
|
||||
}
|
||||
Ok(_) => {
|
||||
return Err(anyhow::anyhow!(
|
||||
"SPROUT_UDS_PATH {uds_path} exists but is not a socket — refusing to delete"
|
||||
"SPROUT_UDS_PATH {uds_path} exists but is not a socket"
|
||||
));
|
||||
}
|
||||
Err(_) => {} // Path doesn't exist — fine, we'll create it
|
||||
Err(_) => {}
|
||||
}
|
||||
let uds_listener = tokio::net::UnixListener::bind(uds_path)
|
||||
.map_err(|e| anyhow::anyhow!("Failed to bind UDS {uds_path}: {e}"))?;
|
||||
info!(path = %uds_path, "sprout-relay UDS listening");
|
||||
|
||||
let router_uds = router.clone();
|
||||
tokio::select! {
|
||||
r = axum::serve(
|
||||
tcp_listener,
|
||||
router.into_make_service_with_connect_info::<std::net::SocketAddr>(),
|
||||
) => r.map_err(|e| anyhow::anyhow!("TCP server error: {e}")),
|
||||
r = axum::serve(uds_listener, router_uds.into_make_service()) =>
|
||||
r.map_err(|e| anyhow::anyhow!("UDS server error: {e}")),
|
||||
}?;
|
||||
let mut uds_rx = shutdown_tx.subscribe();
|
||||
let uds_handle = tokio::spawn(async move {
|
||||
axum::serve(uds_listener, router_uds.into_make_service())
|
||||
.with_graceful_shutdown(async move {
|
||||
uds_rx.changed().await.ok();
|
||||
})
|
||||
.await
|
||||
.ok();
|
||||
});
|
||||
|
||||
let mut tcp_rx = shutdown_tx.subscribe();
|
||||
axum::serve(
|
||||
tcp_listener,
|
||||
router.into_make_service_with_connect_info::<std::net::SocketAddr>(),
|
||||
)
|
||||
.with_graceful_shutdown(async move {
|
||||
tcp_rx.changed().await.ok();
|
||||
})
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("TCP server error: {e}"))?;
|
||||
|
||||
uds_handle.abort();
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
if config.uds_path.is_some() {
|
||||
tracing::warn!(
|
||||
"SPROUT_UDS_PATH is set but UDS is not supported on this platform — \
|
||||
falling back to TCP only"
|
||||
);
|
||||
tracing::warn!("SPROUT_UDS_PATH set but UDS not supported on this platform");
|
||||
}
|
||||
|
||||
// TCP-only path (non-unix, or unix without uds_path set).
|
||||
// TCP-only path.
|
||||
let mut tcp_rx = shutdown_tx.subscribe();
|
||||
axum::serve(
|
||||
tcp_listener,
|
||||
router.into_make_service_with_connect_info::<std::net::SocketAddr>(),
|
||||
)
|
||||
.with_graceful_shutdown(async move {
|
||||
tcp_rx.changed().await.ok();
|
||||
})
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("Server error: {e}"))?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Wait for SIGTERM (Unix) or Ctrl+C.
|
||||
async fn shutdown_signal() {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use tokio::signal::unix::{signal, SignalKind};
|
||||
let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler");
|
||||
tokio::select! {
|
||||
_ = tokio::signal::ctrl_c() => {},
|
||||
_ = sigterm.recv() => {},
|
||||
}
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
tokio::signal::ctrl_c().await.ok();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
//! Prometheus metrics: recorder setup, upkeep task, and HTTP middleware.
|
||||
//!
|
||||
//! ```text
|
||||
//! ┌──────────────────────────────────────────────────────────┐
|
||||
//! │ metrics-rs facade (metrics::counter!, histogram!, etc.) │
|
||||
//! │ ↓ │
|
||||
//! │ PrometheusBuilder → HTTP listener on :9102 │
|
||||
//! │ ↓ │
|
||||
//! │ GET /metrics → Prometheus text format │
|
||||
//! └──────────────────────────────────────────────────────────┘
|
||||
//! ```
|
||||
//!
|
||||
//! Framework metrics (`http_requests_total`, `http_request_latency_ms`) are
|
||||
//! recorded by [`track_metrics`] middleware on the app router. Sprout-specific
|
||||
//! metrics are recorded inline at their call sites.
|
||||
|
||||
use std::time::Instant;
|
||||
|
||||
use axum::{
|
||||
extract::{MatchedPath, Request},
|
||||
middleware::Next,
|
||||
response::Response,
|
||||
};
|
||||
use metrics_exporter_prometheus::{Matcher, PrometheusBuilder};
|
||||
|
||||
/// HTTP latency buckets (milliseconds) — only for `http_request_latency_ms`.
|
||||
const LATENCY_BUCKETS_MS: [f64; 11] = [
|
||||
5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0, 10000.0,
|
||||
];
|
||||
|
||||
/// Seconds-scale buckets for internal processing histograms (event, search, audit).
|
||||
const DURATION_BUCKETS_S: [f64; 10] = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0];
|
||||
|
||||
/// Integer-count buckets for fan-out recipient histograms.
|
||||
const FANOUT_BUCKETS: [f64; 9] = [0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0, 1000.0];
|
||||
|
||||
/// Install the global metrics recorder and spawn the Prometheus HTTP exporter.
|
||||
///
|
||||
/// `build()` returns the recorder + exporter future and internally spawns
|
||||
/// the upkeep task, so no separate upkeep call is needed.
|
||||
///
|
||||
/// Must be called from within a Tokio runtime.
|
||||
/// Panics if a recorder is already installed or the port is in use.
|
||||
pub fn install(port: u16) {
|
||||
let (recorder, exporter) = PrometheusBuilder::new()
|
||||
.with_http_listener(([0, 0, 0, 0], port))
|
||||
// Per-metric buckets: ms for HTTP latency, seconds for internal processing.
|
||||
.set_buckets_for_metric(
|
||||
Matcher::Full("http_request_latency_ms".to_owned()),
|
||||
&LATENCY_BUCKETS_MS,
|
||||
)
|
||||
.expect("valid ms bucket boundaries")
|
||||
.set_buckets_for_metric(Matcher::Suffix("_seconds".to_owned()), &DURATION_BUCKETS_S)
|
||||
.expect("valid seconds bucket boundaries")
|
||||
.set_buckets_for_metric(
|
||||
Matcher::Full("sprout_fanout_recipients".to_owned()),
|
||||
&FANOUT_BUCKETS,
|
||||
)
|
||||
.expect("valid fanout bucket boundaries")
|
||||
.build()
|
||||
.expect("metrics exporter must build exactly once");
|
||||
|
||||
metrics::set_global_recorder(recorder).expect("global recorder must be set exactly once");
|
||||
tokio::spawn(exporter);
|
||||
}
|
||||
|
||||
/// Axum middleware that records CAKE framework HTTP metrics.
|
||||
///
|
||||
/// Emits:
|
||||
/// - `http_requests_total{code, caller, action}` — counter
|
||||
/// - `http_request_latency_ms{code, caller, action}` — histogram
|
||||
///
|
||||
/// Skips health/metrics paths (`/_*`, `/health`) to avoid polluting dashboards.
|
||||
///
|
||||
/// Labels:
|
||||
/// - `code`: exact HTTP status code (e.g. "200", "404")
|
||||
/// - `caller`: upstream service from Istio `x-envoy-downstream-service-cluster` header
|
||||
/// - `action`: matched route pattern (e.g. `/api/channels/{channel_id}`)
|
||||
pub async fn track_metrics(req: Request, next: Next) -> Response {
|
||||
// Use the route pattern (e.g. "/api/channels/{channel_id}"), NOT the raw URI.
|
||||
// Falling back to raw URI on 404s would create unbounded cardinality from scanners.
|
||||
let path = req
|
||||
.extensions()
|
||||
.get::<MatchedPath>()
|
||||
.map(|p| p.as_str().to_owned());
|
||||
|
||||
// Skip health probes, metrics endpoint, and unmatched paths (404 scanners).
|
||||
match path.as_deref() {
|
||||
Some(p) if p.starts_with("/_") || p == "/health" || p == "/metrics" => {
|
||||
return next.run(req).await;
|
||||
}
|
||||
None => {
|
||||
// No matched route — 404/scanner traffic. Skip to avoid cardinality bomb.
|
||||
return next.run(req).await;
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
let action = path.unwrap(); // safe: None case returned above
|
||||
|
||||
// Caller from Istio header. In CAKE, this is set by the mesh (trusted).
|
||||
// On the public TCP listener it's client-controlled, so validate format:
|
||||
// only accept short alphanumeric-with-hyphens service names.
|
||||
let caller = req
|
||||
.headers()
|
||||
.get("x-envoy-downstream-service-cluster")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.filter(|s| {
|
||||
s.len() <= 64
|
||||
&& s.bytes()
|
||||
.all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
|
||||
})
|
||||
.unwrap_or("unknown")
|
||||
.to_owned();
|
||||
|
||||
let start = Instant::now();
|
||||
let response = next.run(req).await;
|
||||
let status = response.status().as_u16().to_string();
|
||||
let latency_ms = start.elapsed().as_secs_f64() * 1000.0;
|
||||
|
||||
let labels = [("code", status), ("caller", caller), ("action", action)];
|
||||
metrics::counter!("http_requests_total", &labels).increment(1);
|
||||
metrics::histogram!("http_request_latency_ms", &labels).record(latency_ms);
|
||||
|
||||
response
|
||||
}
|
||||
@@ -1,10 +1,12 @@
|
||||
//! axum router — WebSocket, NIP-11, NIP-05, health.
|
||||
//! axum routers — app (WebSocket + REST), health (K8s probes), metrics (Prometheus).
|
||||
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::{
|
||||
extract::{ConnectInfo, FromRequest, State, WebSocketUpgrade},
|
||||
http::{HeaderMap, StatusCode},
|
||||
middleware,
|
||||
response::{IntoResponse, Json},
|
||||
routing::{delete, get, post, put},
|
||||
Router,
|
||||
@@ -17,6 +19,7 @@ use tower_http::trace::TraceLayer;
|
||||
use crate::api;
|
||||
use crate::api::tokens;
|
||||
use crate::connection::handle_connection;
|
||||
use crate::metrics::track_metrics;
|
||||
use crate::nip11::{relay_info_handler, RelayInfo};
|
||||
use crate::state::AppState;
|
||||
|
||||
@@ -46,6 +49,8 @@ pub fn build_router(state: Arc<AppState>) -> Router {
|
||||
.route("/", get(nip11_or_ws_handler))
|
||||
.route("/info", get(relay_info_handler))
|
||||
.route("/.well-known/nostr.json", get(api::nip05::nostr_nip05))
|
||||
// Health endpoints remain on the app router for backward compat (local dev).
|
||||
// In CAKE, probes hit the dedicated health port (8080) instead.
|
||||
.route("/health", get(health_handler))
|
||||
.route("/_liveness", get(liveness_handler))
|
||||
.route("/_readiness", get(readiness_handler))
|
||||
@@ -171,13 +176,26 @@ pub fn build_router(state: Arc<AppState>) -> Router {
|
||||
.with_state(state.clone());
|
||||
|
||||
// Merge — each sub-router carries its own body limit.
|
||||
// TraceLayer and CORS are applied once over the combined router.
|
||||
// Metrics → Trace → CORS applied once over the combined router.
|
||||
api_router
|
||||
.merge(media_router)
|
||||
.layer(middleware::from_fn(track_metrics))
|
||||
.layer(TraceLayer::new_for_http())
|
||||
.layer(build_cors_layer(&state.config.cors_origins))
|
||||
}
|
||||
|
||||
/// Build the health-only router for K8s probes (port 8080 in CAKE).
|
||||
///
|
||||
/// No metrics middleware, no auth, no CORS, no body limit.
|
||||
/// Separate from the app router so probes bypass Istio and don't pollute metrics.
|
||||
pub fn build_health_router(state: Arc<AppState>) -> Router {
|
||||
Router::new()
|
||||
.route("/_liveness", get(liveness_handler))
|
||||
.route("/_readiness", get(readiness_handler))
|
||||
.route("/_status", get(status_handler))
|
||||
.with_state(state)
|
||||
}
|
||||
|
||||
/// Content-negotiated: NIP-11 JSON for plain HTTP, WebSocket upgrade otherwise.
|
||||
///
|
||||
/// Uses `axum::extract::Request` to manually attempt WS upgrade, so non-WS
|
||||
@@ -229,17 +247,22 @@ async fn liveness_handler() -> impl IntoResponse {
|
||||
(StatusCode::OK, "ok")
|
||||
}
|
||||
|
||||
/// Readiness probe — checks MySQL and Redis connectivity.
|
||||
/// Readiness probe — checks shutdown flag, MySQL, and Redis connectivity.
|
||||
///
|
||||
/// Returns 200 `{"status":"ready"}` when both are reachable, or
|
||||
/// 503 `{"status":"not_ready","mysql":bool,"redis":bool}` otherwise.
|
||||
///
|
||||
/// Note: Redis pool exhaustion (all connections in use) also returns 503.
|
||||
/// This is intentionally conservative — if we can't acquire a connection,
|
||||
/// the pod shouldn't receive traffic.
|
||||
/// Returns 503 immediately during graceful shutdown (SIGTERM received).
|
||||
/// Otherwise returns 200 when both backends are reachable, or 503 with details.
|
||||
async fn readiness_handler(State(state): State<Arc<AppState>>) -> impl IntoResponse {
|
||||
use std::time::Duration;
|
||||
|
||||
// CAKE: readiness must return 503 after graceful shutdown begins.
|
||||
if state.shutting_down.load(Ordering::Relaxed) {
|
||||
return (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
Json(json!({"status": "shutting_down"})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
|
||||
let check = async {
|
||||
let (mysql_ok, redis_ok) = tokio::join!(state.db.ping(), async {
|
||||
state.redis_pool.get().await.is_ok()
|
||||
@@ -262,6 +285,16 @@ async fn readiness_handler(State(state): State<Arc<AppState>>) -> impl IntoRespo
|
||||
}
|
||||
}
|
||||
|
||||
/// Status endpoint — service name, version, uptime. Optional per CAKE contract.
|
||||
async fn status_handler(State(state): State<Arc<AppState>>) -> impl IntoResponse {
|
||||
let uptime_secs = state.started_at.elapsed().as_secs();
|
||||
Json(json!({
|
||||
"service": "sprout-relay",
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
"uptime_seconds": uptime_secs,
|
||||
}))
|
||||
}
|
||||
|
||||
/// Build a CORS layer from the configured origins list.
|
||||
///
|
||||
/// If `cors_origins` is empty (dev default), returns a permissive layer.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Shared application state — Arc-wrapped, shared across all connections.
|
||||
|
||||
use std::sync::atomic::{AtomicU8, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
@@ -87,6 +87,7 @@ impl ConnectionManager {
|
||||
let count = conn.backpressure_count.fetch_add(1, Ordering::Relaxed) + 1;
|
||||
if count >= SLOW_CLIENT_GRACE_LIMIT {
|
||||
tracing::warn!(conn_id = %conn_id, count, "fan-out: sustained backpressure — cancelling slow client");
|
||||
metrics::counter!("sprout_ws_backpressure_disconnects_total").increment(1);
|
||||
conn.cancel.cancel();
|
||||
} else {
|
||||
tracing::warn!(conn_id = %conn_id, count, grace = SLOW_CLIENT_GRACE_LIMIT, "fan-out: send buffer full — grace {count}/{SLOW_CLIENT_GRACE_LIMIT}");
|
||||
@@ -152,6 +153,10 @@ pub struct AppState {
|
||||
pub local_event_ids: Arc<moka::sync::Cache<[u8; 32], ()>>,
|
||||
/// Media storage client (S3/MinIO).
|
||||
pub media_storage: Arc<MediaStorage>,
|
||||
/// Set to `true` on SIGTERM — readiness probe returns 503.
|
||||
pub shutting_down: Arc<AtomicBool>,
|
||||
/// Process start time — used by `/_status` endpoint.
|
||||
pub started_at: Instant,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
@@ -194,6 +199,8 @@ impl AppState {
|
||||
.build(),
|
||||
),
|
||||
media_storage: Arc::new(media_storage),
|
||||
shutting_down: Arc::new(AtomicBool::new(false)),
|
||||
started_at: Instant::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -55,6 +55,7 @@ impl SubscriptionRegistry {
|
||||
.entry(conn_id)
|
||||
.or_default()
|
||||
.insert(sub_id.clone(), (filters.clone(), channel_id));
|
||||
metrics::gauge!("sprout_subscriptions_active").increment(1.0);
|
||||
|
||||
if let Some(ch_id) = channel_id {
|
||||
match extract_kinds_from_filters(&filters) {
|
||||
@@ -93,6 +94,7 @@ impl SubscriptionRegistry {
|
||||
if let Some(mut conn_subs) = self.subs.get_mut(&conn_id) {
|
||||
if let Some((filters, channel_id)) = conn_subs.remove(sub_id) {
|
||||
self.remove_from_index(conn_id, sub_id, &filters, channel_id);
|
||||
metrics::gauge!("sprout_subscriptions_active").decrement(1.0);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -100,9 +102,11 @@ impl SubscriptionRegistry {
|
||||
/// Remove all subscriptions for a connection and clean up index entries.
|
||||
pub fn remove_connection(&self, conn_id: ConnId) {
|
||||
if let Some((_, conn_subs)) = self.subs.remove(&conn_id) {
|
||||
let count = conn_subs.len();
|
||||
for (sub_id, (filters, channel_id)) in &conn_subs {
|
||||
self.remove_from_index(conn_id, sub_id, filters, *channel_id);
|
||||
}
|
||||
metrics::gauge!("sprout_subscriptions_active").decrement(count as f64);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -174,6 +174,27 @@ services:
|
||||
com.sprout.env: "dev"
|
||||
restart: "no"
|
||||
|
||||
prometheus:
|
||||
image: prom/prometheus:latest
|
||||
container_name: sprout-prometheus
|
||||
ports:
|
||||
- "9090:9090"
|
||||
volumes:
|
||||
- ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
|
||||
- prometheus-data:/prometheus
|
||||
networks:
|
||||
- sprout-net
|
||||
extra_hosts:
|
||||
- "host.docker.internal:host-gateway"
|
||||
deploy:
|
||||
resources:
|
||||
limits:
|
||||
memory: 128m
|
||||
labels:
|
||||
com.sprout.service: "prometheus"
|
||||
com.sprout.env: "dev"
|
||||
restart: unless-stopped
|
||||
|
||||
volumes:
|
||||
mysql-data:
|
||||
name: sprout-mysql-data
|
||||
@@ -187,6 +208,10 @@ volumes:
|
||||
name: sprout-minio-data
|
||||
labels:
|
||||
com.sprout.volume: "minio"
|
||||
prometheus-data:
|
||||
name: sprout-prometheus-data
|
||||
labels:
|
||||
com.sprout.volume: "prometheus"
|
||||
|
||||
networks:
|
||||
sprout-net:
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
# Local dev Prometheus config — scrapes sprout-relay metrics.
|
||||
# The relay runs on the host (not in Docker), so we use host.docker.internal.
|
||||
# Default metrics port is 9102; override with SPROUT_METRICS_PORT.
|
||||
global:
|
||||
scrape_interval: 5s
|
||||
|
||||
scrape_configs:
|
||||
- job_name: sprout-relay
|
||||
static_configs:
|
||||
- targets: ["host.docker.internal:9102"]
|
||||
Executable
+26
@@ -0,0 +1,26 @@
|
||||
#!/bin/sh
|
||||
# CAKE entrypoint — bridge Istio abstract socket to file socket, then exec relay.
|
||||
#
|
||||
# Traffic flow:
|
||||
# Client → Envoy → @istio-proxy.sock (abstract) → socat → /tmp/local.sock → Sprout
|
||||
# K8s probes → 0.0.0.0:8080 → Sprout (direct, no Istio)
|
||||
# Prometheus → 0.0.0.0:9102/metrics → Sprout
|
||||
set -eu
|
||||
|
||||
SIDECAR_SOCKET="${SIDECAR_LISTENER_BIND:-@istio-proxy.sock}"
|
||||
LOCAL_SOCKET="/tmp/local.sock"
|
||||
|
||||
# Clean up stale socket file from previous run.
|
||||
rm -f "${LOCAL_SOCKET}"
|
||||
|
||||
# Bridge: Envoy writes to abstract socket → socat → app reads from file socket.
|
||||
# Only start if SIDECAR_LISTENER_BIND is set (i.e., running in CAKE).
|
||||
if [ -n "${SIDECAR_LISTENER_BIND:-}" ]; then
|
||||
socat "ABSTRACT-LISTEN:${SIDECAR_SOCKET#@},fork" "UNIX-CONNECT:${LOCAL_SOCKET}" &
|
||||
export SPROUT_UDS_PATH="${LOCAL_SOCKET}"
|
||||
fi
|
||||
|
||||
export SPROUT_HEALTH_PORT="${SPROUT_HEALTH_PORT:-8080}"
|
||||
export SPROUT_METRICS_PORT="${SPROUT_METRICS_PORT:-9102}"
|
||||
|
||||
exec /code/sprout-relay
|
||||
Reference in New Issue
Block a user