From e9f855689247d942a85cbac2cbc6fd5c960a1aae Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 08:17:10 +0800 Subject: [PATCH] feat(cdr): export push and queue pipeline metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror the RWI webhook pipeline observability for CDR: producers now count enqueues and drops at the bounded-channel boundary (reporter try_send), the manager counts pushed vs push-failed records per saver batch, and queue depth/capacity gauges are sampled every 5s. An opt-in [callrecord] track_queue_latency flag records the queueing-wait histogram cdr_queue_latency_seconds (enqueued → manager dequeued) via a transient RecordEnqueuedAt instant stashed in the record extensions — save/push time is excluded by design. --- docs/config/06-media-recording.md | 13 ++++++++++ docs/observability.md | 25 +++++++++++++++++++ src/callrecord/mod.rs | 40 ++++++++++++++++++++++++++++++- src/callrecord/tests.rs | 7 ++++++ src/config.rs | 6 +++++ src/metrics.rs | 37 ++++++++++++++++++++++++++++ src/proxy/proxy_call/reporter.rs | 17 ++++++++++--- 7 files changed, 141 insertions(+), 4 deletions(-) diff --git a/docs/config/06-media-recording.md b/docs/config/06-media-recording.md index 547e0598d..24feb6a39 100644 --- a/docs/config/06-media-recording.md +++ b/docs/config/06-media-recording.md @@ -227,6 +227,19 @@ The `[callrecord]` section is **optional**. It adds a secondary raw-CDR sink on `max_concurrent` controls how many post-call CDR save/upload/hook tasks may run at once. The default is `64`; values below `1` are clamped to `1`. +Common pipeline options (independent of the storage type): + +| Key | Default | Description | +|---|---|---| +| `channel_capacity` | 2048 | Bounded queue length between call producers and the CDR manager. Producers drop (and count `cdr_records_dropped_total`) when full. | +| `batch_size` | 64 | Max records per manager batch. | +| `track_queue_latency` | false | Record the `cdr_queue_latency_seconds` histogram (queueing wait only — save/push time excluded). | + +The pipeline exports Prometheus metrics (`cdr_records_enqueued_total`, +`cdr_records_pushed_total`, `cdr_records_push_failed_total`, +`cdr_records_dropped_total`, `cdr_queue_size`, `cdr_queue_current`, +`cdr_queue_latency_seconds`) — see [observability.md](../observability.md#call-record-cdr-pipeline). + ### Database Writes CDR JSON to a separate database table (default: `call_records`). diff --git a/docs/observability.md b/docs/observability.md index 4173dd267..8b7c05f1e 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -109,6 +109,31 @@ All metrics emitted by RustPBX, organized by category: | `rustpbx_call_duration_seconds` | Histogram | `direction` | Wall-clock time from INVITE to BYE | | `rustpbx_call_talk_time_seconds` | Histogram | `direction` | Talk time (only for answered calls) | +#### Call Record (CDR) Pipeline + +Metrics for the call-record pipeline: producers enqueue finished CDRs into a +bounded queue (`[callrecord] channel_capacity`), and a single manager task +drains the queue in batches and hands them to the configured saver +(`http`, `database`, `local`, `s3`). + +| Metric | Type | Labels | Description | +|---|---|---|---| +| `cdr_records_enqueued_total` | Counter | - | CDRs accepted into the queue | +| `cdr_records_pushed_total` | Counter | - | CDRs persisted by the saver (batch success) | +| `cdr_records_push_failed_total` | Counter | - | CDRs in batches the saver failed to persist | +| `cdr_records_dropped_total` | Counter | - | CDRs lost because the queue was full or the manager was gone | +| `cdr_queue_size` | Gauge | - | Configured queue capacity | +| `cdr_queue_current` | Gauge | - | CDRs currently queued (sampled every 5 s) | +| `cdr_queue_latency_seconds` | Histogram | - | Queueing wait (record enqueued → manager dequeued); opt-in via `[callrecord] track_queue_latency` | + +A healthy pipeline keeps `cdr_records_enqueued_total == cdr_records_pushed_total`, +`cdr_records_dropped_total == 0` and `cdr_queue_current` near 0. Rising +`cdr_queue_current` or any `cdr_records_dropped_total` increments mean the +saver endpoint is slower than the call completion rate — scale the endpoint or +raise `channel_capacity` (memory-bounded). `cdr_queue_latency_seconds` +measures how long a record waits in the queue only; a slow endpoint does not +inflate it, backlog does. + #### Trunk Metrics | Metric | Type | Labels | Description | diff --git a/src/callrecord/mod.rs b/src/callrecord/mod.rs index ef6db91ef..8bf7cb9d5 100644 --- a/src/callrecord/mod.rs +++ b/src/callrecord/mod.rs @@ -186,6 +186,13 @@ pub struct CallRecord { pub extensions: http::Extensions, } +/// Extension key stashing the enqueue instant on a `CallRecord` so the +/// manager can measure queueing wait (enqueued → dequeued) for the +/// opt-in `cdr_queue_latency_seconds` histogram without changing the +/// channel item type. Never serialized. +#[derive(Debug, Clone, Copy)] +pub struct RecordEnqueuedAt(pub Instant); + impl Clone for CallRecord { fn clone(&self) -> Self { Self { @@ -538,6 +545,8 @@ pub struct CallRecordManager { pub batch_size: usize, pub sender: CallRecordSender, pub stats: Arc, + channel_capacity: usize, + track_queue_latency: bool, cancel_token: CancellationToken, receiver: CallRecordReceiver, saver: Box, @@ -628,6 +637,7 @@ impl CallRecordManagerBuilder { .unwrap_or(DEFAULT_CALL_RECORD_BATCH_SIZE) .max(1); let (sender, receiver) = tokio::sync::mpsc::channel(channel_capacity); + let track_queue_latency = config.as_ref().is_some_and(|c| c.track_queue_latency); let saver: Box = match config.map(|c| c.storage) { // No [callrecord] section → default: database → rustpbx_call_records None => { @@ -763,6 +773,8 @@ impl CallRecordManagerBuilder { Ok(CallRecordManager { batch_size, stats: Arc::new(CallRecordStats::new()), + channel_capacity, + track_queue_latency, cancel_token, sender, receiver, @@ -1328,7 +1340,17 @@ impl CallRecordManager { pub async fn serve(&mut self) { let token = self.cancel_token.clone(); let batch_size = self.batch_size.max(1); - info!(batch_size, "CallRecordManager serving"); + let track_queue_latency = self.track_queue_latency; + info!( + batch_size, + channel_capacity = self.channel_capacity, + track_queue_latency, + "CallRecordManager serving" + ); + crate::metrics::cdr::set_queue_size(self.channel_capacity); + crate::metrics::cdr::set_queue_current(0); + let mut gauge_interval = tokio::time::interval(Duration::from_secs(5)); + gauge_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let receiver = &mut self.receiver; let saver = self.saver.as_ref(); let hooks = &self.hooks; @@ -1350,6 +1372,17 @@ impl CallRecordManager { continue; } + if track_queue_latency { + let now = Instant::now(); + for record in records.iter_mut() { + if let Some(enqueued_at) = record.extensions.remove::() { + crate::metrics::cdr::queue_latency_seconds( + now.duration_since(enqueued_at.0).as_secs_f64(), + ); + } + } + } + let started_at = Instant::now(); for hook in hooks { if let Err(e) = hook.on_record_enrich(&mut records).await { @@ -1359,6 +1392,7 @@ impl CallRecordManager { match saver.save(&records).await { Ok(file_names) => { + crate::metrics::cdr::pushed(records.len() as u64); if file_names.len() != records.len() { warn!( batch_size = records.len(), @@ -1376,6 +1410,7 @@ impl CallRecordManager { ); } Err(err) => { + crate::metrics::cdr::push_failed(records.len() as u64); warn!(batch_size = records.len(), "Failed to save call record batch: {}", err); } } @@ -1386,6 +1421,9 @@ impl CallRecordManager { } } } + _ = gauge_interval.tick(), if !shutting_down => { + crate::metrics::cdr::set_queue_current(receiver.len()); + } _ = token.cancelled(), if !shutting_down => { shutting_down = true; info!(pending = receiver.len(), "CallRecordManager received shutdown"); diff --git a/src/callrecord/tests.rs b/src/callrecord/tests.rs index 059628c17..08905f033 100644 --- a/src/callrecord/tests.rs +++ b/src/callrecord/tests.rs @@ -377,6 +377,7 @@ async fn test_database_saver_without_url_needs_main_db() { .with_config(CallRecordConfig { channel_capacity: 2048, batch_size: 64, + track_queue_latency: false, storage: CallRecordStorageConfig::Database { database_url: None, table_name: "custom_table".to_string(), @@ -424,6 +425,7 @@ async fn test_local_config_without_main_db_ok() { .with_config(CallRecordConfig { channel_capacity: 2048, batch_size: 4, + track_queue_latency: false, storage: CallRecordStorageConfig::Local { root: root.clone() }, }) .build() @@ -450,6 +452,7 @@ async fn test_builder_uses_configured_channel_and_batch_settings() { .with_config(CallRecordConfig { channel_capacity: 17, batch_size: 3, + track_queue_latency: false, storage: CallRecordStorageConfig::Local { root: tmp.path().to_string_lossy().into_owned(), }, @@ -470,6 +473,7 @@ async fn test_http_config_without_main_db_ok() { .with_config(CallRecordConfig { channel_capacity: 2048, batch_size: 4, + track_queue_latency: false, storage: CallRecordStorageConfig::Http { url: "http://127.0.0.1:1/cdr".to_string(), headers: None, @@ -494,6 +498,7 @@ async fn test_s3_config_without_main_db_ok() { .with_config(CallRecordConfig { channel_capacity: 2048, batch_size: 4, + track_queue_latency: false, storage: CallRecordStorageConfig::S3 { vendor: crate::config::S3Vendor::Minio, bucket: "test-bucket".to_string(), @@ -523,6 +528,7 @@ async fn test_database_with_url_without_main_db_ok() { .with_config(CallRecordConfig { channel_capacity: 2048, batch_size: 4, + track_queue_latency: false, storage: CallRecordStorageConfig::Database { database_url: Some("sqlite::memory:".to_string()), table_name: "custom_cdr".to_string(), @@ -556,6 +562,7 @@ async fn test_local_saver_does_not_write_to_db() { .with_config(CallRecordConfig { channel_capacity: 2048, batch_size: 4, + track_queue_latency: false, storage: CallRecordStorageConfig::Local { root: root.clone() }, }) .build() diff --git a/src/config.rs b/src/config.rs index bf7e89546..6fcbbab30 100644 --- a/src/config.rs +++ b/src/config.rs @@ -641,6 +641,11 @@ pub struct CallRecordConfig { /// Maximum number of records passed to hooks and the saver at once. #[serde(default = "default_call_record_batch_size")] pub batch_size: usize, + /// Record the queueing-wait histogram `cdr_queue_latency_seconds` + /// (record enqueued → manager dequeued). The save/push time itself is + /// excluded; disabled by default. + #[serde(default)] + pub track_queue_latency: bool, #[serde(flatten)] pub storage: CallRecordStorageConfig, } @@ -1769,6 +1774,7 @@ impl Default for CallRecordConfig { Self { channel_capacity: default_call_record_channel_capacity(), batch_size: default_call_record_batch_size(), + track_queue_latency: false, storage: CallRecordStorageConfig::Local { #[cfg(target_os = "windows")] root: "./config/cdr".to_string(), diff --git a/src/metrics.rs b/src/metrics.rs index 80be0f691..aa729d98a 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -768,6 +768,43 @@ pub mod cc { } } +pub mod cdr { + /// Record accepted into the bounded queue (producer → manager). + pub fn enqueued() { + metrics::counter!("cdr_records_enqueued_total").increment(1); + } + + /// Record lost because the queue was full or the manager was gone. + pub fn dropped() { + metrics::counter!("cdr_records_dropped_total").increment(1); + } + + /// Records handed to the saver and persisted successfully. + pub fn pushed(n: u64) { + metrics::counter!("cdr_records_pushed_total").increment(n); + } + + /// Records in a batch the saver failed to persist. + pub fn push_failed(n: u64) { + metrics::counter!("cdr_records_push_failed_total").increment(n); + } + + pub fn set_queue_size(capacity: usize) { + metrics::gauge!("cdr_queue_size").set(capacity as f64); + } + + pub fn set_queue_current(n: usize) { + metrics::gauge!("cdr_queue_current").set(n as f64); + } + + /// Queueing wait (record enqueued → manager dequeued). Excludes the + /// save/push time itself; a slow endpoint does NOT inflate this — + /// queue backlog does. Opt-in via `[callrecord] track_queue_latency`. + pub fn queue_latency_seconds(duration_secs: f64) { + metrics::histogram!("cdr_queue_latency_seconds").record(duration_secs); + } +} + pub fn init_static_gauges() { let version = crate::version::get_short_version(); metrics::gauge!("rustpbx_info", "version" => version).set(1.0); diff --git a/src/proxy/proxy_call/reporter.rs b/src/proxy/proxy_call/reporter.rs index b18ed59b6..063c6f72a 100644 --- a/src/proxy/proxy_call/reporter.rs +++ b/src/proxy/proxy_call/reporter.rs @@ -323,9 +323,20 @@ impl CallReporter { // Bounded channel: drop new records (with a warn log) if the // saver has fallen behind, instead of buffering indefinitely. // `try_send` is sync, so the existing synchronous emit path is - // preserved. - if let Err(tokio::sync::mpsc::error::TrySendError::Full(_)) = sender.try_send(record) { - tracing::warn!("call record channel full; dropping record to bound memory"); + // preserved. The enqueue instant rides in `extensions` for the + // manager's opt-in queueing-latency histogram. + record + .extensions + .insert(crate::callrecord::RecordEnqueuedAt(std::time::Instant::now())); + match sender.try_send(record) { + Ok(()) => crate::metrics::cdr::enqueued(), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + crate::metrics::cdr::dropped(); + tracing::warn!("call record channel full; dropping record to bound memory"); + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + crate::metrics::cdr::dropped(); + } } } }