diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a0ace4e1..3252ace78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to RSigma are documented in this file. Each entry correspond ## [Unreleased] +### Delivery failures without a DLQ sink are logged (#615) + +When a sink exhausted its retries and no `--dlq` was set, the daemon counted the dropped result on `rsigma_dlq_events_total` and logged it once, at `debug`, so every later dropped alert left no trace in the logs. Each drop now produces a `warn` log, rate-limited to one every 10 seconds; each warning carries the number of results dropped since the previous one and the last delivery error. A failed write to a configured DLQ sink is now counted on the new `rsigma_dlq_write_failures_total` counter, and the delivery layer and the daemon's DLQ bridge log a failure they could not queue instead of discarding it silently. The docs now say that a sink commits a failed result by handing it to the DLQ writer, so the source ack does not wait for the DLQ sink's write. + ### Match detail costs one evaluation pass (#613) With `MatchDetailLevel` above `Off`, a rule that fired was evaluated again to collect its selection names and then walked a third time to collect its field matches, re-running every matcher, including regexes. Field matches are now recorded while the selections are evaluated, so a firing rule costs one detail pass, and array members are matched once per quantifier instead of again for the report. The output is unchanged. On the `eval_matched_detail` benchmark (200 rules that all fire), `Full` drops from 314 to 307 µs and `Off` from 252 to 244 µs. diff --git a/crates/rsigma-cli/README.md b/crates/rsigma-cli/README.md index 686edfae0..bdfcd07ae 100644 --- a/crates/rsigma-cli/README.md +++ b/crates/rsigma-cli/README.md @@ -493,7 +493,8 @@ The first three triggers all funnel through one debounced reload task that re-re | `rsigma_output_queue_depth` | gauge | | Current results buffered in engine-to-sink channel | | `rsigma_back_pressure_events_total` | counter | | Times a source was blocked on a full event channel | | `rsigma_uptime_seconds` | gauge | | Daemon uptime in seconds | -| `rsigma_dlq_events_total` | counter | | Events routed to the dead-letter queue | +| `rsigma_dlq_events_total` | counter | | Events routed to the dead-letter queue, with or without a `--dlq` sink | +| `rsigma_dlq_write_failures_total` | counter | | Dead-letter queue entries the `--dlq` sink failed to write | | `rsigma_source_resolves_total` | counter | `source_id` | Total dynamic source resolution attempts | | `rsigma_source_resolve_errors_total` | counter | `source_id` | Total dynamic source resolution failures | | `rsigma_source_resolve_latency_seconds` | histogram | | Source resolution latency | diff --git a/crates/rsigma-cli/src/daemon/metrics.rs b/crates/rsigma-cli/src/daemon/metrics.rs index eba33044b..910272fcb 100644 --- a/crates/rsigma-cli/src/daemon/metrics.rs +++ b/crates/rsigma-cli/src/daemon/metrics.rs @@ -26,6 +26,7 @@ pub struct Metrics { pub pipeline_latency: Histogram, pub batch_size_histogram: Histogram, pub dlq_events: IntCounter, + pub dlq_write_failures: IntCounter, pub sink_queue_depth: IntGaugeVec, pub sink_retries_total: IntCounterVec, pub sink_dropped_total: IntCounterVec, @@ -240,6 +241,11 @@ impl Metrics { "Events routed to dead-letter queue", )) .unwrap(); + let dlq_write_failures = IntCounter::with_opts(Opts::new( + "rsigma_dlq_write_failures_total", + "Dead-letter queue entries the DLQ sink failed to write", + )) + .unwrap(); let sink_queue_depth = IntGaugeVec::new( Opts::new( "rsigma_sink_queue_depth", @@ -357,6 +363,9 @@ impl Metrics { .register(Box::new(batch_size_histogram.clone())) .unwrap(); registry.register(Box::new(dlq_events.clone())).unwrap(); + registry + .register(Box::new(dlq_write_failures.clone())) + .unwrap(); registry .register(Box::new(sink_queue_depth.clone())) .unwrap(); @@ -1156,6 +1165,7 @@ impl Metrics { pipeline_latency, batch_size_histogram, dlq_events, + dlq_write_failures, sink_queue_depth, sink_retries_total, sink_dropped_total, diff --git a/crates/rsigma-cli/src/daemon/server.rs b/crates/rsigma-cli/src/daemon/server.rs index 357d3011d..3d0080bf7 100644 --- a/crates/rsigma-cli/src/daemon/server.rs +++ b/crates/rsigma-cli/src/daemon/server.rs @@ -2,7 +2,7 @@ use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; -use std::time::Instant; +use std::time::{Duration, Instant}; use arc_swap::ArcSwap; use axum::extract::{Query, State}; @@ -39,6 +39,54 @@ struct DlqEntry { timestamp: String, } +/// Minimum gap between two warnings about delivery failures dropped for lack +/// of a DLQ sink. +const DROPPED_FAILURE_WARN_INTERVAL: Duration = Duration::from_secs(10); + +/// Warns about delivery failures dropped because no `--dlq` sink is set. The +/// first drop warns at once; later drops fold into at most one warning per +/// [`DROPPED_FAILURE_WARN_INTERVAL`] that carries their count. +#[derive(Default)] +struct DroppedFailureLog { + last_warn: Option, + pending: u64, +} + +impl DroppedFailureLog { + /// Record one dropped failure. Returns the count it warned about, or + /// `None` when the warning is deferred to a later drop or [`Self::flush`]. + fn record(&mut self, error: &str, now: Instant) -> Option { + self.pending += 1; + if self + .last_warn + .is_some_and(|last| now.duration_since(last) < DROPPED_FAILURE_WARN_INTERVAL) + { + return None; + } + let dropped = std::mem::take(&mut self.pending); + self.last_warn = Some(now); + tracing::warn!( + dropped, + error, + "Delivery failures dropped: no DLQ sink is configured (use --dlq to persist them)" + ); + Some(dropped) + } + + /// Warn about any drops still waiting for the next interval. + fn flush(&mut self) -> Option { + let dropped = std::mem::take(&mut self.pending); + if dropped == 0 { + return None; + } + tracing::warn!( + dropped, + "Delivery failures dropped: no DLQ sink is configured (use --dlq to persist them)" + ); + Some(dropped) + } +} + /// One evaluated batch waiting for ordered sink/ack dispatch. struct EvaluatedBatch { outcome: BatchProcessOutcome, @@ -1999,13 +2047,17 @@ pub async fn run_daemon(config: DaemonConfig) { let (df_tx, mut df_rx) = mpsc::channel::(buffer_size); let df_handle = tokio::spawn(async move { while let Some(failure) = df_rx.recv().await { - let _ = dlq_tx - .send(DlqEntry { - original_event: failure.serialized, - error: failure.error, - timestamp: chrono::Utc::now().to_rfc3339(), - }) - .await; + let entry = DlqEntry { + original_event: failure.serialized, + error: failure.error, + timestamp: chrono::Utc::now().to_rfc3339(), + }; + if let Err(mpsc::error::SendError(entry)) = dlq_tx.send(entry).await { + tracing::warn!( + error = %entry.error, + "DLQ writer closed, delivery failure dropped" + ); + } } }); @@ -2270,24 +2322,27 @@ pub async fn run_daemon(config: DaemonConfig) { dispatcher.shutdown().await; }); - // DLQ writer task: writes DLQ entries to the configured DLQ sink. + // DLQ writer task: writes DLQ entries to the configured DLQ sink. Without + // one, only delivery failures arrive here (parse errors are not routed), + // and they are dropped with a rate-limited warning. let dlq_metrics = metrics.clone(); let dlq_handle = tokio::spawn(async move { tracing::debug!("DLQ task started"); let mut dlq_sink = dlq_sink; - let mut no_sink_logged = false; + let mut dropped = DroppedFailureLog::default(); while let Some(entry) = dlq_rx.recv().await { dlq_metrics.dlq_events.inc(); if let Some(ref mut sink) = dlq_sink { let json = serde_json::to_string(&entry).unwrap_or_default(); if let Err(e) = sink.send_raw(&json).await { - tracing::warn!(error = %e, "Failed to write to DLQ sink"); + dlq_metrics.dlq_write_failures.inc(); + tracing::warn!(error = %e, "Failed to write to DLQ sink, entry dropped"); } - } else if !no_sink_logged { - tracing::debug!("DLQ entry counted but no sink configured (use --dlq to persist)"); - no_sink_logged = true; + } else { + dropped.record(&entry.error, Instant::now()); } } + dropped.flush(); tracing::debug!("DLQ task finished"); }); @@ -4516,6 +4571,23 @@ impl rsigma_runtime::LogsService for OtlpLogsGrpcService { mod tests { use super::*; + #[test] + fn dropped_failure_warnings_are_rate_limited() { + let mut log = DroppedFailureLog::default(); + let t0 = Instant::now(); + assert_eq!(log.record("boom", t0), Some(1)); + assert_eq!(log.record("boom", t0 + Duration::from_secs(1)), None); + assert_eq!(log.record("boom", t0 + Duration::from_secs(9)), None); + assert_eq!( + log.record("boom", t0 + DROPPED_FAILURE_WARN_INTERVAL), + Some(3) + ); + assert_eq!(log.flush(), None); + assert_eq!(log.record("boom", t0 + Duration::from_secs(11)), None); + assert_eq!(log.flush(), Some(1)); + assert_eq!(log.flush(), None); + } + #[test] fn detection_inflight_default_scales_to_five() { assert_eq!(default_detect_inflight(1), 1); diff --git a/crates/rsigma-cli/tests/cli_daemon_webhook.rs b/crates/rsigma-cli/tests/cli_daemon_webhook.rs index da08933f9..a83857844 100644 --- a/crates/rsigma-cli/tests/cli_daemon_webhook.rs +++ b/crates/rsigma-cli/tests/cli_daemon_webhook.rs @@ -12,7 +12,7 @@ mod common; use std::time::Duration; -use common::{DaemonProcess, SIMPLE_RULE, http_post, poll_until, temp_file}; +use common::{DaemonProcess, SIMPLE_RULE, http_get, http_post, poll_until, temp_file}; use wiremock::matchers::method; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -421,3 +421,57 @@ fn webhook_does_not_block_file_sink_under_fanout() { "an unreachable webhook must not block the file sink behind it", ); } + +#[test] +fn delivery_failures_without_dlq_log_one_rate_limited_warning() { + let cfg = WEBHOOK_RETRY_CFG + .replace("__URL__", "http://127.0.0.1:1") + .replace("attempts: 3", "attempts: 1") + .replace("backoff: 100ms", "backoff: 10ms"); + let webhook = temp_file(".yml", &cfg); + let rule = temp_file(".yml", SIMPLE_RULE); + let daemon = DaemonProcess::spawn_http_with_args( + rule.path().to_str().unwrap(), + &["--webhook", webhook.path().to_str().unwrap()], + ); + + for _ in 0..3 { + let (status, _) = http_post(&daemon.url("/api/v1/events"), MATCHING_EVENT); + assert_eq!(status, 200); + } + + let metrics = poll_until(Duration::from_secs(10), || { + let (_, body) = http_get(&daemon.url("/metrics")); + body.contains(r#"rsigma_sink_delivery_failures_total{sink="slack"} 3"#) + .then_some(body) + }) + .expect("all three deliveries should fail"); + assert!( + metrics.contains("rsigma_dlq_write_failures_total 0"), + "nothing reaches a DLQ sink when none is configured: {metrics}" + ); + + let warnings: Vec = poll_until(Duration::from_secs(5), || { + let lines: Vec = daemon + .stderr_lines() + .into_iter() + .filter(|l| l.contains("no DLQ sink is configured")) + .collect(); + (!lines.is_empty()).then_some(lines) + }) + .expect("a dropped delivery failure must be logged without --dlq"); + assert_eq!( + warnings.len(), + 1, + "later drops within the interval fold into the next warning: {warnings:?}" + ); + let warning: serde_json::Value = serde_json::from_str(&warnings[0]).unwrap(); + assert_eq!(warning["level"], "WARN"); + assert_eq!(warning["fields"]["dropped"], 1); + assert!( + warning["fields"]["error"] + .as_str() + .is_some_and(|e| e.contains("sink delivery failure")), + "{warning}" + ); +} diff --git a/crates/rsigma-cli/tests/common/mod.rs b/crates/rsigma-cli/tests/common/mod.rs index 569cf09a0..9294fe015 100644 --- a/crates/rsigma-cli/tests/common/mod.rs +++ b/crates/rsigma-cli/tests/common/mod.rs @@ -4,6 +4,7 @@ use std::io::{BufRead, BufReader, Write}; use std::net::TcpStream; use std::process::{Command as StdCommand, Stdio}; +use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use assert_cmd::Command; @@ -101,6 +102,7 @@ impl Drop for ChildGuard { pub struct DaemonProcess { child: std::process::Child, api_addr: String, + stderr: Arc>>, } impl DaemonProcess { @@ -147,7 +149,9 @@ impl DaemonProcess { } let stderr = guard.as_child_mut().stderr.take().unwrap(); + let stderr_lines = Arc::new(Mutex::new(Vec::new())); let (tx, rx) = std::sync::mpsc::channel::(); + let collected = stderr_lines.clone(); std::thread::spawn(move || { for line in BufReader::new(stderr).lines() { let Ok(line) = line else { return }; @@ -159,6 +163,7 @@ impl DaemonProcess { if line.contains("Sink started") { let _ = tx.send(StartupEvent::SinkStarted); } + collected.lock().unwrap().push(line); } }); @@ -196,6 +201,7 @@ impl DaemonProcess { return Self { child: guard.disarm(), api_addr, + stderr: stderr_lines, }; } if Instant::now() >= deadline { @@ -272,6 +278,11 @@ impl DaemonProcess { &self.api_addr } + /// The daemon's stderr log lines read so far. + pub fn stderr_lines(&self) -> Vec { + self.stderr.lock().unwrap().clone() + } + fn kill(&mut self) { terminate_child(&mut self.child, Duration::from_secs(5)); } diff --git a/crates/rsigma-runtime/src/io/delivery.rs b/crates/rsigma-runtime/src/io/delivery.rs index f7786f1b1..4575e5797 100644 --- a/crates/rsigma-runtime/src/io/delivery.rs +++ b/crates/rsigma-runtime/src/io/delivery.rs @@ -3,10 +3,12 @@ //! The daemon fans every result into a [`Dispatcher`], which owns one //! [`SinkWorker`] per leaf sink. Each worker drains its queue sequentially, //! batches opportunistically, retries with bounded exponential backoff, and -//! routes terminal failures to the DLQ. Acknowledgments use the lifetime of an -//! [`AckGuard`]: each event's ack tokens fire only once every worker has -//! durably committed (delivered or DLQ-parked) the event, so the at-least-once -//! contract survives fan-out. +//! routes terminal failures to the DLQ channel. Acknowledgments use the +//! lifetime of an [`AckGuard`]: each event's ack tokens fire only once every +//! worker has finished with the event, either delivering it or handing the +//! failure to the DLQ channel, so the at-least-once contract survives fan-out. +//! The ack does not wait for whoever drains the DLQ channel to persist the +//! failure. use std::future::Future; use std::pin::Pin; @@ -191,7 +193,9 @@ impl Default for DeliveryContext { /// A result that could not be delivered after exhausting retries. /// /// Emitted to the DLQ channel; the CLI wraps it into its own DLQ record so the -/// runtime stays free of CLI types. +/// runtime stays free of CLI types. The event is acked once the failure is +/// queued on the channel, so a receiver that drops or fails to persist it +/// loses the result. pub struct DeliveryFailure { /// The serialized `ProcessResult` that failed to deliver. pub serialized: String, @@ -307,6 +311,9 @@ impl Dispatcher { /// [`crate::io::Sink::into_leaves`]); each carries its own /// [`DeliveryConfig`] so a sink (e.g. a webhook) can override the global /// retry/backoff/queue defaults. + /// + /// Terminal failures go to `dlq_tx`. With `None`, or once its receiver is + /// gone, each one is logged at `warn` and dropped. pub fn spawn( sinks: Vec<(S, OnFull, DeliveryConfig)>, dlq_tx: Option>, @@ -494,12 +501,17 @@ async fn deliver_one( DeliverTarget::Incident(e) => e.json.clone(), DeliverTarget::FormattedIncident(e) => e.native().json.clone(), }; - let _ = dlq - .send(DeliveryFailure { - serialized, - error: format!("sink delivery failure: {e}"), - }) - .await; + let failure = DeliveryFailure { + serialized, + error: format!("sink delivery failure: {e}"), + }; + if dlq.send(failure).await.is_err() { + tracing::warn!( + sink = label, + error = %e, + "Sink delivery failed and the DLQ channel is closed; dropping result", + ); + } } None => { tracing::warn!( @@ -784,6 +796,50 @@ mod tests { ); } + #[tokio::test] + async fn terminal_failure_with_closed_dlq_counts_and_acks() { + #[derive(Default)] + struct FailureCounter(AtomicUsize); + impl MetricsHook for FailureCounter { + fn on_sink_delivery_failed(&self, _sink: &str) { + self.0.fetch_add(1, Ordering::SeqCst); + } + fn on_parse_error(&self) {} + fn on_events_processed(&self, _: u64) {} + fn on_detection_matches(&self, _: u64) {} + fn on_correlation_matches(&self, _: u64) {} + fn observe_processing_latency(&self, _: f64) {} + fn on_input_queue_depth_change(&self, _: i64) {} + fn on_back_pressure(&self) {} + fn observe_batch_size(&self, _: u64) {} + fn on_output_queue_depth_change(&self, _: i64) {} + fn observe_pipeline_latency(&self, _: f64) {} + fn set_correlation_state_entries(&self, _: u64) {} + } + + let (ack_tx, mut ack_rx) = mpsc::unbounded_channel(); + let (dlq_tx, dlq_rx) = mpsc::channel(8); + drop(dlq_rx); + let counter = Arc::new(FailureCounter::default()); + let mut sink = MockSink::new("mock"); + sink.always_fail = true; + let dispatcher = Dispatcher::spawn( + vec![(sink, OnFull::Block, fast_cfg())], + Some(dlq_tx), + ack_tx, + counter.clone(), + ); + + dispatcher.dispatch(result(), vec![AckToken::Noop]).await; + dispatcher.shutdown().await; + + assert_eq!(counter.0.load(Ordering::SeqCst), 1, "failure still counted"); + assert!( + ack_rx.try_recv().is_ok(), + "token acked even though the DLQ channel is closed", + ); + } + #[tokio::test] async fn formatted_incident_failure_routes_native_line_to_dlq() { let (ack_tx, _ack_rx) = mpsc::unbounded_channel(); diff --git a/docs/_data/vars.yml b/docs/_data/vars.yml index 597c7fff0..48362210b 100644 --- a/docs/_data/vars.yml +++ b/docs/_data/vars.yml @@ -18,8 +18,8 @@ rsigma: # Referenced as {{ rsigma.metrics.names }} etc. Keep in sync with # crates/rsigma-cli/src/daemon/metrics.rs and docs/content/reference/metrics.md. metrics: - names: 95 # total metric names in the definition catalogue - always: 90 # registered without feature gates (names - otlp - tls) + names: 96 # total metric names in the definition catalogue + always: 91 # registered without feature gates (names - otlp - tls) otlp: 3 # daemon-otlp tls: 2 # daemon-tls crates: diff --git a/docs/content/cli/engine/daemon.md b/docs/content/cli/engine/daemon.md index f697ee5de..667c4e16e 100644 --- a/docs/content/cli/engine/daemon.md +++ b/docs/content/cli/engine/daemon.md @@ -49,7 +49,7 @@ For narrative coverage see [Streaming Detection](../../guide/streaming-detection | Flag | Default | Description | |------|---------|-------------| | `--output ` | `stdout` | Detection sink. Schemes: `stdout`, `file://`, `nats://:/`, `otlp(s)://:` (OTLP/gRPC), `otlphttp(s)://:` (OTLP/HTTP, posts to `/v1/logs`); the `s` variants use TLS. On Unix, `unix:///path/to.sock` writes NDJSON to a local collector socket. Repeatable for fan-out. Query params: `?on_full=drop` (best-effort), `?format=ndjson\|ocsf` (line sinks only; see [OCSF Findings](../../guide/ocsf-findings.md)), `?compression=gzip` (OTLP), `?timeout=` {{ added "unreleased" }} (OTLP per-attempt export deadline, default `10s`), and for TLS `?ca=`, `?client_cert=`, `?client_key=` (PEM paths, the last two for mutual TLS) and `?tls_domain=` (SNI override). OTLP schemes require the `daemon-otlp` build. | -| `--dlq ` | unset | Dead-letter queue for events that fail parsing or sink delivery. Same schemes as `--output`. When unset, failed events are logged and discarded. | +| `--dlq ` | unset | Dead-letter queue for events that fail parsing or sink delivery. Same schemes as `--output`. When unset, parse failures are counted on `rsigma_events_parse_errors_total` and discarded, and delivery failures are discarded with a rate-limited `warn` log that carries how many were dropped. {{ added "unreleased" }} | | `--include-event` | off | Embed the full event JSON in every detection match. | | `--match-detail ` | `off` | Match-detail verbosity: `off` (field + value only), `summary` (adds matcher kind, selection, case sensitivity, and reports keyword/absence matches), or `full` (also records the matched pattern). Also settable via `daemon.engine.match_detail`. See [Evaluating Rules](../../guide/evaluating-rules.md#match-detail). | | `--pretty` | off | Pretty-print JSON output. | @@ -61,7 +61,7 @@ For narrative coverage see [Streaming Detection](../../guide/streaming-detection Each sink serializes in its own wire format. The default is RSigma's native NDJSON; `?format=ocsf` emits [OCSF Detection Finding](../../guide/ocsf-findings.md) (class 2004) JSON instead, one finding per line. The parameter is accepted on `stdout`, `file://`, `nats://`, and `unix://` sinks only, and `format=` on an OTLP spec, the `--dlq`, or the `daemon.api.audit.sink` fails startup, as does an unknown value. -Each `--output` sink runs its own bounded queue and worker: results are delivered with bounded exponential-backoff retry, and a result is sent to the DLQ only after retries are exhausted. Fan-out is isolated up to each sink's queue depth, so a slow sink does not immediately stall the others; a slow durable sink eventually applies backpressure, the cost of preserving at-least-once delivery. An acknowledgment is released to the source only once every sink has committed the result, and `?on_full=drop` opts a sink out of that contract in favor of never stalling. +Each `--output` sink runs its own bounded queue and worker: results are delivered with bounded exponential-backoff retry, and a result is sent to the DLQ only after retries are exhausted. Fan-out is isolated up to each sink's queue depth, so a slow sink does not immediately stall the others; a slow durable sink eventually applies backpressure, the cost of preserving at-least-once delivery. An acknowledgment is released to the source only once every sink has committed the result, and `?on_full=drop` opts a sink out of that contract in favor of never stalling. A sink commits a failed result by handing it to the DLQ writer, so the ack does not wait for the DLQ sink's write; a failed DLQ write is logged and counted on `rsigma_dlq_write_failures_total`. {{ added "unreleased" }} Acknowledgment guarantees differ by transport. NATS is durable at-least-once: results un-acked at shutdown are redelivered on restart. The HTTP and OTLP push receivers are at-most-once on ingest, since the request is accepted as soon as the event is queued. diff --git a/docs/content/guide/nats-streaming.md b/docs/content/guide/nats-streaming.md index d501bda36..5a3b76894 100644 --- a/docs/content/guide/nats-streaming.md +++ b/docs/content/guide/nats-streaming.md @@ -72,13 +72,13 @@ rsigma engine daemon -r rules/ \ ## At-least-once delivery -When `--input nats://...` is used, the daemon switches from at-most-once to at-least-once semantics. Each message is wrapped in an `AckToken` that is held until every output sink has committed the corresponding result (delivered or DLQ-parked). A dedicated ack task resolves tokens after that confirmation. +When `--input nats://...` is used, the daemon switches from at-most-once to at-least-once semantics. Each message is wrapped in an `AckToken` that is held until every output sink has committed the corresponding result: delivered it, or handed the failure to the DLQ writer. A dedicated ack task resolves tokens after that confirmation. What this guarantees: - If the daemon panics mid-processing, NATS redelivers after `ack_wait` (configured on the JetStream consumer). - While a sink is retrying (file write error, NATS publish-ack timeout), the source message stays unacked so a crash still redelivers. -- After retries are exhausted (or a permanent sink error), the result is parked in the [DLQ](#dead-letter-queue) when `--dlq` is set, or dropped with a warning when it is not. Either way the source message is then acked so the same failure does not loop forever. +- After retries are exhausted (or a permanent sink error), the result is handed to the [DLQ](#dead-letter-queue) writer and the source message is acked, so the same failure does not loop forever. With `--dlq` set, the writer then persists it; a failed DLQ write is logged and counted on `rsigma_dlq_write_failures_total`, and the result is lost. Without `--dlq`, the result is dropped with a rate-limited warning. {{ added "unreleased" }} - If a parse error happens before the engine, the failed event is routed to the DLQ (when configured) and the source message is acked, preventing infinite redelivery of unparseable data. What this does NOT guarantee: @@ -177,7 +177,7 @@ Each DLQ entry is a JSON object: } ``` -The `rsigma_dlq_events_total` Prometheus counter tracks DLQ volume. Alert on rate-of-change to catch upstream encoding regressions early. See [Prometheus metrics reference](../reference/metrics.md). +The `rsigma_dlq_events_total` Prometheus counter tracks DLQ volume, including delivery failures dropped for lack of a `--dlq` sink. Alert on rate-of-change to catch upstream encoding regressions early, and on `rsigma_dlq_write_failures_total` to catch a DLQ sink that cannot write. {{ added "unreleased" }} See [Prometheus metrics reference](../reference/metrics.md). ## Connection lifecycle and reconnects @@ -201,7 +201,7 @@ The `rsigma_dlq_events_total` Prometheus counter tracks DLQ volume. Alert on rat | `--batch-size` | Events per engine mutex acquisition. Default 128, capped at `--buffer-size`; lower it when burst tail latency matters more than sustained throughput. | | `--drain-timeout` | Seconds to wait for in-flight events on shutdown. Default 5. Raise to 30 in production so SIGTERM does not lose work. | | `--state-save-interval` | Periodic SQLite snapshot interval. Default 30 s. Lower means less work to redo after a crash; higher means less disk I/O. | -| `--dlq` | Set in production. Never set means parse errors are silently dropped. | +| `--dlq` | Set in production. Without it, parse errors are dropped with only a metric, and delivery failures are dropped with a rate-limited warning. | ## Example: production NATS deployment diff --git a/docs/content/reference/metrics.md b/docs/content/reference/metrics.md index 1c44ef34c..e4bcb0641 100644 --- a/docs/content/reference/metrics.md +++ b/docs/content/reference/metrics.md @@ -30,7 +30,8 @@ These always show up. They cover ingest, matches, queue depth, back-pressure, re | `rsigma_batch_phase_duration_seconds` | histogram | `phase` (`parse`, `decode_merge`, `observe`, `evaluate`, `result_merge`, `dispatch`) | Wall-clock duration of each batch-processing phase. `parse` is the parallel format parse; `decode_merge` is the ordered post-parse merge and optional event filter; `observe` covers field/schema observers and decoded taps; `evaluate` is engine detection/correlation; `result_merge` is per-line result assembly and counters; `dispatch` is the daemon's post-batch ack/sink enqueue loop. Compare `_sum` deltas over the same interval to rank serial share without inferring it from throughput. | | `rsigma_pipeline_latency_seconds` | histogram | none | End-to-end latency from event dequeue to sink send. | | `rsigma_batch_size` | histogram | none | Number of events processed per batch. | -| `rsigma_dlq_events_total` | counter | none | Events routed to the dead-letter queue. | +| `rsigma_dlq_events_total` | counter | none | Events routed to the dead-letter queue, counted whether or not a `--dlq` sink is configured. | +| `rsigma_dlq_write_failures_total` | counter | none | Dead-letter queue entries the `--dlq` sink failed to write. Each one is a lost event that was already acked. {{ added "unreleased" }} | | `rsigma_sink_queue_depth` | gauge | `sink` | Results buffered in each sink's delivery queue. | | `rsigma_sink_retries_total` | counter | `sink` | Sink delivery retries after a retryable failure. | | `rsigma_sink_dropped_total` | counter | `sink` | Results dropped because a lossy sink's queue was full (`?on_full=drop`). |