natx_core/
observability.rs

1//! 连接等待、事务连接槽和拒绝门禁的有界原子观测；连接池状态只在指标出口读取。
2
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::sync::{Arc, Mutex, OnceLock};
5use std::time::{Duration, Instant};
6
7use nametrics_core::atomic::{add, AtomicHistogram, LATENCY_SECONDS};
8use nametrics_core::{
9    LegacyMetricsSource, MetricDescriptor, MetricKind, MetricSample, MetricValue,
10};
11use nanotify_core::{AlertRoute, EventKind, Notification, NotificationFields, Severity};
12
13use crate::{DatabaseDriver, MAX_MANAGED_DATASOURCES};
14
15/// 受管连接的语句日志策略；关闭时 SQLx 不产生逐条语句事件。
16#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
17pub enum StatementLogging {
18    #[default]
19    /// 关闭逐条语句日志，不关闭基础指标。
20    Disabled,
21    /// 按 Debug 级别输出语句。
22    Debug,
23    /// 按 Trace 级别输出语句。
24    Trace,
25}
26
27/// 调用方显式声明的连接用途，不从任务、SQL 或错误文本推断。
28#[derive(Clone, Copy, Debug, PartialEq, Eq)]
29pub enum ConnectionPurpose {
30    /// Mapper 业务 SQL 使用的连接。
31    Mapper,
32    /// 数据库迁移使用的连接。
33    Migration,
34    /// 连接或服务健康探测使用的连接。
35    Probe,
36    /// 业务直接访问连接池。
37    Direct,
38}
39
40impl ConnectionPurpose {
41    /// 业务作用：返回连接用途对应的固定指标值。
42    ///
43    /// 参数说明：无。
44    ///
45    /// 返回：不会包含业务输入的静态标识。
46    pub const fn as_str(self) -> &'static str {
47        match self {
48            Self::Mapper => "mapper",
49            Self::Migration => "migration",
50            Self::Probe => "probe",
51            Self::Direct => "direct",
52        }
53    }
54}
55
56/// 连接获取的稳定完成分类。
57#[derive(Clone, Copy, Debug, PartialEq, Eq)]
58pub enum AcquireOutcome {
59    /// 连接获取成功。
60    Ok,
61    /// 获取连接超过期限，不能据此推断是池饱和还是建连失败。
62    Timeout,
63    /// 连接池或事务连接槽已关闭。
64    Closed,
65    /// 数据库连接工作任务退出导致获取失败。
66    Worker,
67    /// 等待方取消，未取得连接。
68    Cancelled,
69    /// 其它连接获取错误。
70    Other,
71}
72
73impl AcquireOutcome {
74    /// 业务作用：把详细完成分类投影到延迟分布的固定状态。
75    ///
76    /// 参数说明：无。
77    ///
78    /// 返回：成功、失败或调用方取消的数组位置。
79    const fn status(self) -> usize {
80        match self {
81            Self::Ok => 0,
82            Self::Cancelled => 2,
83            _ => 1,
84        }
85    }
86}
87
88/// SQL 执行前拒绝原因，不与数据库调用失败混合。
89#[derive(Clone, Copy, Debug, PartialEq, Eq)]
90pub enum ConnectionRejection {
91    /// 当前操作要求事务，但没有可用的 ambient transaction。
92    TxRequired,
93    /// 当前操作禁止加入已有事务。
94    TxForbidden,
95    /// 目标数据源与当前事务绑定的数据源不同。
96    CrossDatasource,
97    /// 目标驱动与当前事务绑定的驱动不同。
98    CrossDriver,
99    /// 目标名称不在数据源目录中。
100    UnknownDatasource,
101    /// 数据源注册表尚未可用或已经关闭。
102    RegistryUnavailable,
103    /// 事务连接已被其它操作独占，当前调用无法取得访问权。
104    TxConnectionBusy,
105}
106
107/// 等待日志的固定级别。
108#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
109pub enum WaitLogLevel {
110    /// 以普通运行信息记录等待超阈值。
111    Info,
112    #[default]
113    /// 以告警级别记录等待超阈值。
114    Warn,
115}
116
117/// 启动期冻结的成功等待日志策略，不关闭基础指标。
118#[derive(Clone, Copy, Debug, PartialEq, Eq)]
119pub struct WaitObservationPolicy {
120    /// 成功等待达到该时长时命中日志规则。
121    pub threshold: Duration,
122    /// 是否输出成功等待超阈值日志。
123    pub log_enabled: bool,
124    /// 等待超阈值日志的级别。
125    pub log_level: WaitLogLevel,
126    /// 同一等待观测单元两次日志之间的最小间隔。
127    pub log_cooldown: Duration,
128}
129
130/// 启动期冻结的连接超时通知路由；只持有具体有界队列，不接受任意回调。
131pub struct AcquireAlertPolicy {
132    /// 启动期绑定的有界通知队列路由。
133    pub route: AlertRoute,
134    /// 连接超时通知的业务严重度。
135    pub severity: Severity,
136    /// 同一连接观测槽的通知最小间隔。
137    pub cooldown: Duration,
138    /// 允许触发超时通知的连接用途集合。
139    pub purposes: Vec<ConnectionPurpose>,
140    /// 产生通知的稳定服务名。
141    pub service: String,
142    /// 产生通知的进程实例身份。
143    pub instance: String,
144    /// 可选部署环境身份。
145    pub environment: Option<String>,
146    /// 可选部署集群身份。
147    pub cluster: Option<String>,
148}
149
150impl Default for WaitObservationPolicy {
151    /// 业务作用：为连接等待提供默认关闭日志的稳定策略。
152    ///
153    /// 参数说明：无。
154    ///
155    /// 返回：250 毫秒阈值与 60 秒冷却，基础统计仍全量记录。
156    fn default() -> Self {
157        Self {
158            threshold: Duration::from_millis(250),
159            log_enabled: false,
160            log_level: WaitLogLevel::Warn,
161            log_cooldown: Duration::from_secs(60),
162        }
163    }
164}
165
166struct WaitCells {
167    outcomes: [AtomicU64; 6],
168    durations: [AtomicHistogram; 3],
169    in_flight: AtomicU64,
170}
171
172impl WaitCells {
173    /// 业务作用：预分配等待路径的全部计数与分布。
174    ///
175    /// 参数说明：无。
176    ///
177    /// 返回：零值单元；后续完成路径无需动态构造指标。
178    const fn new() -> Self {
179        Self {
180            outcomes: [const { AtomicU64::new(0) }; 6],
181            durations: [const { AtomicHistogram::new(LATENCY_SECONDS) }; 3],
182            in_flight: AtomicU64::new(0),
183        }
184    }
185}
186
187/// 进程生命周期连接统计槽；未知名称共用固定回退槽。
188pub struct ConnectionSlot {
189    driver: DatabaseDriver,
190    datasource: Box<str>,
191    acquire: [WaitCells; 4],
192    transaction: WaitCells,
193    rejections: [AtomicU64; 7],
194    policies: OnceLock<(WaitObservationPolicy, WaitObservationPolicy)>,
195    log_next: [AtomicU64; 2],
196    acquire_alert: OnceLock<AcquireAlertPolicy>,
197    alert_next: AtomicU64,
198    alert_sequence: AtomicU64,
199}
200
201impl ConnectionSlot {
202    /// 业务作用：在启动期建立一个 datasource 的固定容量观测槽。
203    ///
204    /// 参数说明：
205    /// - `driver`：已验证连接配置对应的数据库 driver。
206    /// - `datasource`：用于固定观测身份的具名数据源名称。
207    ///
208    /// 返回：返回调用、等待、拒绝与通知计数均为零且策略尚未安装的观测槽，不登记全局目录或访问数据库。
209    fn new(driver: DatabaseDriver, datasource: &str) -> Self {
210        Self {
211            driver,
212            datasource: datasource.into(),
213            acquire: [const { WaitCells::new() }; 4],
214            transaction: WaitCells::new(),
215            rejections: [const { AtomicU64::new(0) }; 7],
216            policies: OnceLock::new(),
217            log_next: [const { AtomicU64::new(0) }; 2],
218            acquire_alert: OnceLock::new(),
219            alert_next: AtomicU64::new(0),
220            alert_sequence: AtomicU64::new(0),
221        }
222    }
223
224    /// 业务作用：记录连接发放之前的安全门禁拒绝。
225    ///
226    /// 参数说明：
227    /// - `reason`：是由连接权威判定的固定分类。
228    ///
229    /// 返回：无；不产生数据库 operation 或连接等待样本。
230    pub fn reject(&self, reason: ConnectionRejection) {
231        add(&self.rejections[reason as usize], 1);
232    }
233
234    /// 业务作用：开始观察普通 Pool 获取，保证取消时回收 in-flight。
235    ///
236    /// 参数说明：
237    /// - `purpose`：是调用方明确声明的连接用途。
238    ///
239    /// 返回：持有进程静态统计槽的等待守卫。
240    pub fn acquire(&'static self, purpose: ConnectionPurpose) -> WaitGuard {
241        WaitGuard::new(self, purpose, false)
242    }
243
244    /// 业务作用：开始观察 ambient transaction 的连接槽等待。
245    ///
246    /// 参数说明：无。
247    ///
248    /// 返回：与 Pool acquire 独立计时的取消安全守卫。
249    pub fn transaction_slot(&'static self) -> WaitGuard {
250        WaitGuard::new(self, ConnectionPurpose::Direct, true)
251    }
252}
253
254static SLOTS: [OnceLock<ConnectionSlot>; MAX_MANAGED_DATASOURCES * 2] =
255    [const { OnceLock::new() }; MAX_MANAGED_DATASOURCES * 2];
256static REGISTRATION: Mutex<()> = Mutex::new(());
257static UNKNOWN_MYSQL: OnceLock<ConnectionSlot> = OnceLock::new();
258static UNKNOWN_PGSQL: OnceLock<ConnectionSlot> = OnceLock::new();
259static UNKNOWN_SQLITE: OnceLock<ConnectionSlot> = OnceLock::new();
260static CLOCK: OnceLock<Instant> = OnceLock::new();
261
262/// 业务作用：在启动期为已知 datasource 预分配槽，不允许运行时名称扩大指标基数。
263///
264/// 参数说明：
265/// - `driver`：已验证连接配置对应的数据库 driver。
266/// - `datasource`：用于固定观测身份的具名数据源名称。
267///
268/// 返回：名称合法且容量可用时返回进程静态槽，同 driver 同名称重复调用复用原槽；非法名称或总槽位耗尽返回错误，不允许动态指标名无限增长。
269pub fn register_datasource(
270    driver: DatabaseDriver,
271    datasource: &str,
272) -> anyhow::Result<&'static ConnectionSlot> {
273    crate::validate_datasource_name(datasource)?;
274    let _registration = REGISTRATION
275        .lock()
276        .unwrap_or_else(|error| error.into_inner());
277    CLOCK.get_or_init(Instant::now);
278    fallback(driver);
279    if let Some(slot) = known_slot(driver, datasource) {
280        return Ok(slot);
281    }
282    let cell = SLOTS
283        .iter()
284        .find(|cell| cell.get().is_none())
285        .ok_or_else(|| anyhow::anyhow!("connection observation datasource capacity exceeded"))?;
286    Ok(cell.get_or_init(|| ConnectionSlot::new(driver, datasource)))
287}
288
289/// 业务作用：冻结 datasource 的等待日志策略，拒绝运行期静默替换。
290///
291/// 参数说明：
292/// - `driver`：已验证连接配置对应的数据库 driver。
293/// - `datasource`：用于固定观测身份的具名数据源名称。
294/// - `acquire`：调用方已验证的池获取等待观测策略。
295/// - `slot`：调用方已验证的事务连接槽等待观测策略。
296///
297/// 返回：首次冻结或传入值与已冻结策略完全相同则成功，策略变化返回错误。不重新验证策略内部数值，失败前可能已经登记该数据源的观测槽。
298pub fn configure_datasource(
299    driver: DatabaseDriver,
300    datasource: &str,
301    acquire: WaitObservationPolicy,
302    slot: WaitObservationPolicy,
303) -> anyhow::Result<()> {
304    let cells = register_datasource(driver, datasource)?;
305    let policy = (acquire, slot);
306    anyhow::ensure!(
307        *cells.policies.get_or_init(|| policy) == policy,
308        "connection observation policy is frozen; restart is required"
309    );
310    Ok(())
311}
312
313/// 业务作用：将已验证的通知引用绑定到固定 datasource 槽。
314///
315/// 参数说明：
316/// - `driver`：已验证连接配置对应的数据库 driver。
317/// - `datasource`：用于固定观测身份的具名数据源名称。
318/// - `policy`：包含用途、队列、冷却和实例身份的通知策略。
319///
320/// 返回：用途集合非空、至多四项、无重复且此前未安装通知时绑定策略；否则返回错误，不发网络请求。失败前可能已登记观测槽，具体通知队列等前置校验由调用方负责。
321pub fn configure_acquire_alert(
322    driver: DatabaseDriver,
323    datasource: &str,
324    policy: AcquireAlertPolicy,
325) -> anyhow::Result<()> {
326    anyhow::ensure!(
327        !policy.purposes.is_empty() && policy.purposes.len() <= 4,
328        "acquire alert purposes must be nonempty and bounded"
329    );
330    let mut seen = [false; 4];
331    for purpose in &policy.purposes {
332        anyhow::ensure!(!seen[*purpose as usize], "duplicate acquire alert purpose");
333        seen[*purpose as usize] = true;
334    }
335    register_datasource(driver, datasource)?
336        .acquire_alert
337        .set(policy)
338        .map_err(|_| anyhow::anyhow!("acquire alert policy is frozen; restart is required"))
339}
340
341/// 业务作用：查询已预分配槽，未知名称只归入固定回退身份。
342///
343/// 参数说明：
344/// - `driver`：已验证连接配置对应的数据库 driver。
345/// - `datasource`：用于固定观测身份的具名数据源名称。
346///
347/// 返回：返回已注册静态槽，未知身份归入该 driver 的固定回退槽；不登记动态 datasource 标签。
348pub fn connection_slot(driver: DatabaseDriver, datasource: &str) -> &'static ConnectionSlot {
349    known_slot(driver, datasource).unwrap_or_else(|| fallback(driver))
350}
351
352/// 业务作用：在固定目录中查找连接槽，不获取全局锁。
353///
354/// 参数说明：
355/// - `driver`：已验证连接配置对应的数据库 driver。
356/// - `datasource`：用于固定观测身份的具名数据源名称。
357///
358/// 返回：找到同 driver 同名称的已登记槽时返回静态借用，否则返回 None，不分配槽或取得全局注册锁。
359fn known_slot(driver: DatabaseDriver, datasource: &str) -> Option<&'static ConnectionSlot> {
360    SLOTS
361        .iter()
362        .filter_map(OnceLock::get)
363        .find(|slot| slot.driver == driver && slot.datasource.as_ref() == datasource)
364}
365
366/// 业务作用：把未知 datasource 的拒绝归入固定低基数槽。
367///
368/// 参数说明：
369/// - `driver`：指定后端。
370///
371/// 返回：不会包含用户输入名称的静态槽。
372fn fallback(driver: DatabaseDriver) -> &'static ConnectionSlot {
373    match driver {
374        DatabaseDriver::MySql => &UNKNOWN_MYSQL,
375        DatabaseDriver::PostgreSql => &UNKNOWN_PGSQL,
376        DatabaseDriver::Sqlite => &UNKNOWN_SQLITE,
377    }
378    .get_or_init(|| ConnectionSlot::new(driver, "__unknown__"))
379}
380
381/// 连接等待生命周期守卫；显式完成与取消只裁决一次。
382pub struct WaitGuard {
383    slot: &'static ConnectionSlot,
384    purpose: ConnectionPurpose,
385    transaction: bool,
386    started: Instant,
387    finished: bool,
388}
389
390impl WaitGuard {
391    /// 业务作用：进入等待状态并增加对应的在途数量。
392    ///
393    /// 参数说明：
394    /// - `slot`：为固定身份
395    /// - `purpose`：为用途
396    /// - `transaction`：区分事务槽。
397    ///
398    /// 返回：尚未裁决的等待守卫。
399    fn new(slot: &'static ConnectionSlot, purpose: ConnectionPurpose, transaction: bool) -> Self {
400        let guard = Self {
401            slot,
402            purpose,
403            transaction,
404            started: Instant::now(),
405            finished: false,
406        };
407        guard.cells().in_flight.fetch_add(1, Ordering::Relaxed);
408        guard
409    }
410
411    /// 业务作用：选取当前等待语义的固定原子单元。
412    ///
413    /// 参数说明：无。
414    ///
415    /// 返回：Pool 用途单元或事务连接槽单元。
416    fn cells(&self) -> &WaitCells {
417        if self.transaction {
418            &self.slot.transaction
419        } else {
420            &self.slot.acquire[self.purpose as usize]
421        }
422    }
423
424    /// 业务作用：记录连接等待的唯一终态，日志不会改写连接结果。
425    ///
426    /// 参数说明：
427    /// - `outcome`：必须在驱动错误类型被擦除前确定。
428    ///
429    /// 返回：无；重复完成不会重复计数。
430    pub fn finish(&mut self, outcome: AcquireOutcome) {
431        if self.finished {
432            return;
433        }
434        self.finished = true;
435        let elapsed = self.started.elapsed();
436        let cells = self.cells();
437        add(&cells.outcomes[outcome as usize], 1);
438        cells.durations[outcome.status()].observe(elapsed);
439        cells.in_flight.fetch_sub(1, Ordering::Relaxed);
440        if outcome == AcquireOutcome::Ok {
441            self.log_wait(elapsed);
442        }
443        if outcome == AcquireOutcome::Timeout && !self.transaction {
444            self.notify_timeout(elapsed);
445        }
446    }
447
448    /// 业务作用：仅在超时规则与原子冷却均命中后构造一个有界通知。
449    ///
450    /// 参数说明：
451    /// - `elapsed`：是本次连接获取时长。
452    ///
453    /// 返回：无；入队拥塞和停机只影响通知自己的指标，不能改变连接错误。
454    fn notify_timeout(&self, elapsed: Duration) {
455        let Some(policy) = self.slot.acquire_alert.get() else {
456            return;
457        };
458        if !policy.purposes.contains(&self.purpose) {
459            return;
460        }
461        let now = CLOCK
462            .get_or_init(Instant::now)
463            .elapsed()
464            .as_millis()
465            .min(u64::MAX as u128) as u64;
466        if !policy.cooldown.is_zero()
467            && self
468                .slot
469                .alert_next
470                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
471                    (value <= now).then(|| {
472                        now.saturating_add(policy.cooldown.as_millis().min(u64::MAX as u128) as u64)
473                    })
474                })
475                .is_err()
476        {
477            return;
478        }
479        let sequence = self.slot.alert_sequence.fetch_add(1, Ordering::Relaxed);
480        let occurred_at_unix_ms = std::time::SystemTime::now()
481            .duration_since(std::time::UNIX_EPOCH)
482            .unwrap_or_default()
483            .as_millis()
484            .min(u64::MAX as u128) as u64;
485        use std::hash::{Hash, Hasher};
486        let mut identity = std::collections::hash_map::DefaultHasher::new();
487        (
488            policy.instance.as_str(),
489            driver_name(self.slot.driver),
490            self.slot.datasource.as_ref(),
491        )
492            .hash(&mut identity);
493        let notification = Notification::new(NotificationFields {
494            // 来源文本可能很长；固定长度指纹保留末尾序列，避免消息截断抹去同槽通知的区分信息。
495            id: format!(
496                "acquire-{:x}-{occurred_at_unix_ms}-{sequence}",
497                identity.finish()
498            ),
499            event: EventKind::AcquireTimeout,
500            severity: policy.severity,
501            occurred_at_unix_ms,
502            service: policy.service.clone(),
503            instance: policy.instance.clone(),
504            environment: policy.environment.clone(),
505            cluster: policy.cluster.clone(),
506            driver: driver_name(self.slot.driver).to_owned(),
507            datasource: self.slot.datasource.to_string(),
508            operation: "acquire".to_owned(),
509            purpose: Some(self.purpose.as_str().to_owned()),
510            duration: elapsed,
511            outcome: "timeout".to_owned(),
512            error_kind: Some("timeout".to_owned()),
513            ..NotificationFields::default()
514        });
515        let _ = policy.route.try_send(notification);
516    }
517
518    /// 业务作用：只为成功且超阈值的等待输出有冷却的结构化日志。
519    ///
520    /// 参数说明：
521    /// - `elapsed`：是单调时钟测得的完整等待时长。
522    ///
523    /// 返回：无；默认不输出，且不携带 SQL、参数或错误原文。
524    fn log_wait(&self, elapsed: Duration) {
525        let Some(policies) = self.slot.policies.get() else {
526            return;
527        };
528        let policy = if self.transaction {
529            policies.1
530        } else {
531            policies.0
532        };
533        if !policy.log_enabled || elapsed < policy.threshold {
534            return;
535        }
536        let next = &self.slot.log_next[usize::from(self.transaction)];
537        let now = CLOCK
538            .get_or_init(Instant::now)
539            .elapsed()
540            .as_millis()
541            .min(u64::MAX as u128) as u64;
542        if policy.log_cooldown != Duration::ZERO
543            && next
544                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
545                    (value <= now).then(|| {
546                        now.saturating_add(
547                            policy.log_cooldown.as_millis().min(u64::MAX as u128) as u64
548                        )
549                    })
550                })
551                .is_err()
552        {
553            return;
554        }
555        let event = if self.transaction {
556            "transaction_slot_wait"
557        } else {
558            "connection_acquire_wait"
559        };
560        // 连接已取得且指标终态已提交，日志 subscriber 的展开不能撤销该业务结果。
561        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
562            match policy.log_level {
563                WaitLogLevel::Info => {
564                    tracing::info!(target: "natx::connection", event, datasource = self.slot.datasource.as_ref(), driver = driver_name(self.slot.driver), purpose = self.purpose.as_str(), duration_seconds = elapsed.as_secs_f64(), "数据库连接等待超过阈值")
565                }
566                WaitLogLevel::Warn => {
567                    tracing::warn!(target: "natx::connection", event, datasource = self.slot.datasource.as_ref(), driver = driver_name(self.slot.driver), purpose = self.purpose.as_str(), duration_seconds = elapsed.as_secs_f64(), "数据库连接等待超过阈值")
568                }
569            }
570        }));
571        if let Err(payload) = result {
572            // 异常载荷的自定义析构同样属于外部代码，二次展开也必须在诊断边界内终止。
573            if let Err(nested) =
574                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(payload)))
575            {
576                std::mem::forget(nested);
577            }
578        }
579    }
580}
581
582impl Drop for WaitGuard {
583    /// 业务作用：在 Future 取消或展开时回收在途计数并记录取消。
584    ///
585    /// 参数说明：无。
586    ///
587    /// 返回：无；已完成守卫不再产生样本。
588    fn drop(&mut self) {
589        self.finish(AcquireOutcome::Cancelled);
590    }
591}
592
593/// 业务作用：返回数据库后端的固定指标身份。
594///
595/// 参数说明：
596/// - `driver`：是已解析的后端枚举。
597///
598/// 返回：与配置 driver 一致的固定字符串。
599fn driver_name(driver: DatabaseDriver) -> &'static str {
600    match driver {
601        DatabaseDriver::MySql => "mysql",
602        DatabaseDriver::PostgreSql => "postgresql",
603        DatabaseDriver::Sqlite => "sqlite",
604    }
605}
606
607const OUTCOMES: [&str; 6] = ["ok", "timeout", "closed", "worker", "cancelled", "other"];
608const STATUSES: [&str; 3] = ["success", "failure", "cancelled"];
609const PURPOSES: [&str; 4] = ["mapper", "migration", "probe", "direct"];
610const REJECTIONS: [&str; 7] = [
611    "tx_required",
612    "tx_forbidden",
613    "cross_datasource",
614    "cross_driver",
615    "unknown_datasource",
616    "registry_unavailable",
617    "tx_connection_busy",
618];
619
620macro_rules! descriptor {
621    ($name:ident, $text:literal, $kind:ident, $labels:expr, $bounds:expr) => {
622        static $name: MetricDescriptor = MetricDescriptor {
623            name: $text,
624            help: "数据库连接与等待状态。",
625            unit: if matches!(MetricKind::$kind, MetricKind::Histogram) {
626                "seconds"
627            } else {
628                ""
629            },
630            kind: MetricKind::$kind,
631            label_names: $labels,
632            histogram_bounds: $bounds,
633        };
634    };
635}
636descriptor!(
637    ACQUIRES,
638    "natx_connection_acquires_total",
639    Counter,
640    &["datasource", "driver", "purpose", "outcome"],
641    &[]
642);
643descriptor!(
644    ACQUIRE_DURATION,
645    "natx_connection_acquire_duration_seconds",
646    Histogram,
647    &["datasource", "driver", "purpose", "status"],
648    LATENCY_SECONDS
649);
650descriptor!(
651    ACQUIRE_IN_FLIGHT,
652    "natx_connection_acquire_in_flight",
653    Gauge,
654    &["datasource", "driver", "purpose"],
655    &[]
656);
657descriptor!(
658    SLOT_WAITS,
659    "natx_transaction_slot_waits_total",
660    Counter,
661    &["datasource", "driver", "outcome"],
662    &[]
663);
664descriptor!(
665    SLOT_DURATION,
666    "natx_transaction_slot_wait_duration_seconds",
667    Histogram,
668    &["datasource", "driver", "status"],
669    LATENCY_SECONDS
670);
671descriptor!(
672    SLOT_IN_FLIGHT,
673    "natx_transaction_slot_wait_in_flight",
674    Gauge,
675    &["datasource", "driver"],
676    &[]
677);
678descriptor!(
679    REJECTION,
680    "natx_connection_rejections_total",
681    Counter,
682    &["datasource", "driver", "reason"],
683    &[]
684);
685descriptor!(
686    POOL_CONNECTIONS,
687    "natx_pool_connections",
688    Gauge,
689    &["datasource", "driver", "state"],
690    &[]
691);
692descriptor!(
693    POOL_MAX,
694    "natx_pool_max_connections",
695    Gauge,
696    &["datasource", "driver"],
697    &[]
698);
699static CONNECTION_DESCRIPTORS: &[&MetricDescriptor] = &[
700    &ACQUIRES,
701    &ACQUIRE_DURATION,
702    &ACQUIRE_IN_FLIGHT,
703    &SLOT_WAITS,
704    &SLOT_DURATION,
705    &SLOT_IN_FLIGHT,
706    &REJECTION,
707];
708static POOL_DESCRIPTORS: &[&MetricDescriptor] = &[&POOL_CONNECTIONS, &POOL_MAX];
709
710/// 按后端冻结槽目录的原子等待指标源，Prometheus 与 OTLP 共享同一快照与系列边界。
711pub struct ConnectionMetricsSource {
712    slots: Box<[&'static ConnectionSlot]>,
713}
714
715impl ConnectionMetricsSource {
716    /// 业务作用：在指标源注册前冻结指定后端的槽目录，后续注册不扩大已预留的系列预算。
717    ///
718    /// 参数说明：
719    /// - `driver`：限定该源拥有的已登记槽及固定未知身份槽。
720    ///
721    /// 返回：不创建后台 worker 的指标源；原子值持续更新，目录不再变化。
722    pub fn new(driver: DatabaseDriver) -> Self {
723        // 与槽登记共用门禁形成一致的目录边界，预算与所有后续快照必须消费同一组静态引用。
724        let _registration = REGISTRATION
725            .lock()
726            .unwrap_or_else(|error| error.into_inner());
727        let slots = SLOTS
728            .iter()
729            .filter_map(OnceLock::get)
730            .filter(|slot| slot.driver == driver)
731            .chain(std::iter::once(fallback(driver)))
732            .collect::<Vec<_>>()
733            .into_boxed_slice();
734        Self {
735            slots,
736        }
737    }
738
739    /// 业务作用：计算当前冻结 datasource 目录的最坏 Prometheus 系列预算。
740    ///
741    /// 参数说明：无。
742    ///
743    /// 返回：包含固定未知身份和全部 counter、gauge、桶、sum、count 的系列数。
744    pub fn series_budget(&self) -> usize {
745        self.slots.len() * (5 * (6 + 3 * (LATENCY_SECONDS.len() + 3) + 1) + 7)
746    }
747}
748
749impl LegacyMetricsSource for ConnectionMetricsSource {
750    /// 业务作用：发布等待与门禁族的统一描述符。
751    ///
752    /// 参数说明：无。
753    ///
754    /// 返回：固定且与结构化快照一致的目录。
755    fn descriptors(&self) -> &'static [&'static MetricDescriptor] {
756        CONNECTION_DESCRIPTORS
757    }
758
759    /// 业务作用：在出口侧读取原子单元并构造低基数样本。
760    ///
761    /// 参数说明：无。
762    ///
763    /// 返回：冻结目录内的实时 counter、gauge 和非累积 Histogram 快照，不包含源创建后登记的槽。
764    fn snapshot(&self) -> Option<Vec<MetricSample>> {
765        let mut samples = Vec::new();
766        for slot in &self.slots {
767            let base = vec![
768                ("datasource", slot.datasource.to_string()),
769                ("driver", driver_name(slot.driver).to_owned()),
770            ];
771            for (purpose, cells) in PURPOSES.iter().zip(&slot.acquire) {
772                let mut labels = base.clone();
773                labels.push(("purpose", (*purpose).to_owned()));
774                snapshot_wait(
775                    &mut samples,
776                    cells,
777                    labels,
778                    &ACQUIRES,
779                    &ACQUIRE_DURATION,
780                    &ACQUIRE_IN_FLIGHT,
781                );
782            }
783            snapshot_wait(
784                &mut samples,
785                &slot.transaction,
786                base.clone(),
787                &SLOT_WAITS,
788                &SLOT_DURATION,
789                &SLOT_IN_FLIGHT,
790            );
791            for (reason, cell) in REJECTIONS.iter().zip(&slot.rejections) {
792                let mut labels = base.clone();
793                labels.push(("reason", (*reason).to_owned()));
794                samples.push(MetricSample {
795                    name: REJECTION.name,
796                    labels,
797                    value: MetricValue::Counter(cell.load(Ordering::Relaxed)),
798                });
799            }
800        }
801        Some(samples)
802    }
803
804    /// 业务作用：由 MetricHub 统一渲染结构化快照，避免重复文本系列。
805    ///
806    /// 参数说明：
807    /// - `_output`：为兼容文本缓冲区。
808    ///
809    /// 返回：无；该源不使用旧文本回退路径。
810    fn render_prometheus(&self, _output: &mut String) {}
811}
812
813/// 业务作用：把一种等待路径的全部固定单元投影为出口样本。
814///
815/// 参数说明：
816/// - `samples`：接收观测出口样本的追加序列。
817/// - `cells`：一种等待路径已经固定的计数、分布与在途原子单元。
818/// - `labels`：driver、datasource、用途等由调用方冻结的基础标签。
819/// - `counter`：各等待结局累计计数的指标描述符。
820/// - `histogram`：各状态等待时长分布的指标描述符。
821/// - `gauge`：当前在途等待数的指标描述符。
822///
823/// 返回：向 samples 追加各固定结局计数、状态时长分布与当前在途 gauge；分别读取原子值，不保证全部样本属于同一时刻，不修改原计数。
824fn snapshot_wait(
825    samples: &mut Vec<MetricSample>,
826    cells: &WaitCells,
827    labels: Vec<(&'static str, String)>,
828    counter: &'static MetricDescriptor,
829    histogram: &'static MetricDescriptor,
830    gauge: &'static MetricDescriptor,
831) {
832    for (outcome, cell) in OUTCOMES.iter().zip(&cells.outcomes) {
833        let mut labels = labels.clone();
834        labels.push(("outcome", (*outcome).to_owned()));
835        samples.push(MetricSample {
836            name: counter.name,
837            labels,
838            value: MetricValue::Counter(cell.load(Ordering::Relaxed)),
839        });
840    }
841    for (status, cell) in STATUSES.iter().zip(&cells.durations) {
842        let mut labels = labels.clone();
843        labels.push(("status", (*status).to_owned()));
844        samples.push(MetricSample {
845            name: histogram.name,
846            labels,
847            value: cell.snapshot(),
848        });
849    }
850    samples.push(MetricSample {
851        name: gauge.name,
852        labels,
853        value: MetricValue::Gauge(cells.in_flight.load(Ordering::Relaxed) as f64),
854    });
855}
856
857/// SQLx Pool 的近似瞬时状态；各字段不是事务式同时读取。
858pub struct PoolState {
859    /// 采样时连接池内的连接总数。
860    pub total: u32,
861    /// 采样时可供借用的空闲连接数。
862    pub idle: u32,
863    /// 连接池配置允许的最大连接数。
864    pub max: u32,
865}
866
867/// 后端只在抓取时提供连接池状态，不参与业务热路径。
868pub trait PoolSnapshot: Send + Sync {
869    /// 业务作用：读取连接池当前规模与配置上限。
870    ///
871    /// 参数说明：无。
872    ///
873    /// 返回：允许并发变化的近似瞬时状态。
874    fn snapshot(&self) -> PoolState;
875}
876
877/// 持有只读 Pool 句柄的结构化指标源。
878pub struct PoolMetricsSource {
879    driver: DatabaseDriver,
880    pools: Vec<(String, Arc<dyn PoolSnapshot>)>,
881}
882
883impl PoolMetricsSource {
884    /// 业务作用：从启动期连接目录建立 Pool 状态出口。
885    ///
886    /// 参数说明：
887    /// - `driver`：指定后端
888    /// - `pools`：为固定名称与只读句柄。
889    ///
890    /// 返回：名称合法且不重复时返回源，未创建任何采样线程。
891    pub fn new(
892        driver: DatabaseDriver,
893        pools: Vec<(String, Arc<dyn PoolSnapshot>)>,
894    ) -> anyhow::Result<Self> {
895        anyhow::ensure!(
896            pools.len() <= MAX_MANAGED_DATASOURCES,
897            "pool observation capacity exceeded"
898        );
899        let mut names = std::collections::BTreeSet::new();
900        for (name, _) in &pools {
901            crate::validate_datasource_name(name)?;
902            anyhow::ensure!(names.insert(name), "duplicate pool observation datasource");
903        }
904        Ok(Self {
905            driver,
906            pools,
907        })
908    }
909
910    /// 业务作用：预留每个 Pool 的固定 gauge 系列预算。
911    ///
912    /// 参数说明：无。
913    ///
914    /// 返回：total、idle、in_use 和上限的系列总数。
915    pub fn series_budget(&self) -> usize {
916        self.pools.len() * 4
917    }
918}
919
920impl LegacyMetricsSource for PoolMetricsSource {
921    /// 业务作用：发布后端中立的 Pool gauge 描述符。
922    ///
923    /// 参数说明：无。
924    ///
925    /// 返回：状态规模和配置上限两个固定族。
926    fn descriptors(&self) -> &'static [&'static MetricDescriptor] {
927        POOL_DESCRIPTORS
928    }
929
930    /// 业务作用：抓取时读取 Pool 并用饱和差值计算使用数。
931    ///
932    /// 参数说明：无。
933    ///
934    /// 返回：不因 total 与 idle 并发变化而下溢的近似快照。
935    fn snapshot(&self) -> Option<Vec<MetricSample>> {
936        let mut samples = Vec::with_capacity(self.series_budget());
937        for (name, pool) in &self.pools {
938            let state = pool.snapshot();
939            let base = vec![
940                ("datasource", name.clone()),
941                ("driver", driver_name(self.driver).to_owned()),
942            ];
943            for (label, value) in [
944                ("total", state.total),
945                ("idle", state.idle),
946                ("in_use", state.total.saturating_sub(state.idle)),
947            ] {
948                let mut labels = base.clone();
949                labels.push(("state", label.to_owned()));
950                samples.push(MetricSample {
951                    name: POOL_CONNECTIONS.name,
952                    labels,
953                    value: MetricValue::Gauge(value as f64),
954                });
955            }
956            samples.push(MetricSample {
957                name: POOL_MAX.name,
958                labels: base,
959                value: MetricValue::Gauge(state.max as f64),
960            });
961        }
962        Some(samples)
963    }
964
965    /// 业务作用：保持所有出口统一消费结构化 Pool 样本。
966    ///
967    /// 参数说明：
968    /// - `_output`：为旧文本回退缓冲区。
969    ///
970    /// 返回：无；由 MetricHub 完成文本编码。
971    fn render_prometheus(&self, _output: &mut String) {}
972}
natx_core/observability.rs

natx_core/
observability.rs
