nasaga_runtime_core/
observability.rs

1//! Saga 运行指标快照与 Prometheus 文本导出。
2//!
3//! 业务状态指标来自当前后端的已提交事实；Kafka 处理耗时和 transport 动作是
4//! 进程内低基数计数。高基数 saga_id、tenant、workflow 只进结构化日志和审计 API，
5//! 禁止进入指标 label。
6
7use std::fmt::Write as _;
8
9use nasaga_backend::SagaStoreMetrics;
10
11/// 业务作用：聚合 Saga 已提交状态与当前进程 Kafka transport 指标。
12#[derive(Debug, Clone, PartialEq, Eq, Default)]
13pub struct SagaOperationalMetrics {
14    /// 历史创建实例数。
15    pub started_total: u64,
16    /// 进入 `COMPLETED` 的历史迁移数。
17    pub completed_total: u64,
18    /// 进入 `COMPENSATED` 的历史迁移数。
19    pub compensated_total: u64,
20    /// 进入 `MANUAL_INTERVENTION` 的历史迁移数。
21    pub manual_intervention_total: u64,
22    /// 进入 `MANUALLY_CLOSED` 的历史迁移数（系统外处置后人工关闭自动化）。
23    pub manually_closed_total: u64,
24    /// 当前进程按租户配额拒绝的创建请求数（不携带租户标签,精确用量走受鉴权管理查询）。
25    pub quota_rejections_total: u64,
26    /// 当前进程按租户速率拒绝的变更类管理动作数（不携带租户标签,当前窗口用量走
27    /// 受鉴权管理查询）。
28    pub action_rate_rejections_total: u64,
29    /// 历史 Unknown attempt 数。
30    pub unknown_result_total: u64,
31    /// 历史业务 attempt 重试数。
32    pub retry_attempt_total: u64,
33    /// 历史互斥事实数。
34    pub conflict_total: u64,
35    /// 当前正向运行实例数。
36    pub running_current: u64,
37    /// 当前等待 Unknown 裁决实例数。
38    pub waiting_resolution_current: u64,
39    /// 当前补偿中实例数。
40    pub compensating_current: u64,
41    /// 当前人工介入实例数。
42    pub manual_intervention_current: u64,
43    /// 当前已到期且可领取 timer 数。
44    pub due_timer_current: u64,
45    /// 终态生命周期样本数。
46    pub lifecycle_duration_count: u64,
47    /// 终态生命周期累计微秒数。
48    pub lifecycle_duration_micros_sum: u64,
49    /// 按冻结 workflow/definition version 分组的端到端终态时延分位数。
50    pub lifecycle_quantiles: Vec<nasaga_backend::SagaLifecycleQuantiles>,
51    /// 当前进程处理 Saga result 次数。
52    pub kafka_result_processing_total: u64,
53    /// 当前进程处理 Saga result 累计微秒数。
54    pub kafka_result_processing_micros_sum: u64,
55    /// 当前进程请求 Kafka 保留 offset 重投的次数。
56    pub kafka_result_retry_total: u64,
57    /// 当前进程请求 `nafka` 进入 durability-first DLT 的次数。
58    pub kafka_result_dlt_requested_total: u64,
59    /// 当前进程在本地事务提交后完成手动 ACK 的次数。
60    pub kafka_result_ack_total: u64,
61    /// 当前进程由 Inbox 幂等吸收后 ACK 的重复结果数。
62    pub kafka_result_duplicate_total: u64,
63    /// 当前进程处理 Saga command 次数。
64    pub kafka_command_processing_total: u64,
65    /// 当前进程处理 Saga command 累计微秒数。
66    pub kafka_command_processing_micros_sum: u64,
67    /// 当前进程因 Participant 事务未提交而保留 command offset 的次数。
68    pub kafka_command_retry_total: u64,
69    /// 当前进程请求 command 进入 durability-first DLT 的次数。
70    pub kafka_command_dlt_requested_total: u64,
71    /// 当前进程在 Participant COMMIT 后完成手动 ACK 的 command 数。
72    pub kafka_command_ack_total: u64,
73    /// 当前进程由 Participant Inbox 幂等吸收后 ACK 的重复 command 数。
74    pub kafka_command_duplicate_total: u64,
75}
76
77impl SagaOperationalMetrics {
78    /// 业务作用：把低基数快照渲染为 Prometheus text exposition 格式。
79    ///
80    /// 参数说明: 无。
81    ///
82    /// 返回：可直接作为 `/metrics` 响应体的 UTF-8 文本，不包含高基数标签。
83    pub fn render_prometheus(&self) -> String {
84        let mut output = String::with_capacity(2_048);
85        counter(&mut output, "nasaga_started_total", self.started_total);
86        counter(&mut output, "nasaga_completed_total", self.completed_total);
87        counter(
88            &mut output,
89            "nasaga_compensated_total",
90            self.compensated_total,
91        );
92        counter(
93            &mut output,
94            "nasaga_manual_intervention_total",
95            self.manual_intervention_total,
96        );
97        counter(
98            &mut output,
99            "nasaga_manually_closed_total",
100            self.manually_closed_total,
101        );
102        counter(
103            &mut output,
104            "nasaga_quota_rejections_total",
105            self.quota_rejections_total,
106        );
107        counter(
108            &mut output,
109            "nasaga_action_rate_rejections_total",
110            self.action_rate_rejections_total,
111        );
112        counter(
113            &mut output,
114            "nasaga_unknown_result_total",
115            self.unknown_result_total,
116        );
117        counter(
118            &mut output,
119            "nasaga_retry_attempt_total",
120            self.retry_attempt_total,
121        );
122        counter(&mut output, "nasaga_conflict_total", self.conflict_total);
123        gauge(&mut output, "nasaga_running", self.running_current);
124        gauge(
125            &mut output,
126            "nasaga_waiting_resolution",
127            self.waiting_resolution_current,
128        );
129        gauge(
130            &mut output,
131            "nasaga_compensating",
132            self.compensating_current,
133        );
134        gauge(
135            &mut output,
136            "nasaga_manual_intervention",
137            self.manual_intervention_current,
138        );
139        gauge(&mut output, "nasaga_due_timer", self.due_timer_current);
140        summary(
141            &mut output,
142            "nasaga_lifecycle_duration_seconds",
143            self.lifecycle_duration_count,
144            self.lifecycle_duration_micros_sum,
145        );
146        for quantiles in &self.lifecycle_quantiles {
147            labeled_gauge(
148                &mut output,
149                "nasaga_lifecycle_duration_seconds",
150                &quantiles.workflow,
151                quantiles.definition_version,
152                "0.5",
153                quantiles.p50_micros,
154            );
155            labeled_gauge(
156                &mut output,
157                "nasaga_lifecycle_duration_seconds",
158                &quantiles.workflow,
159                quantiles.definition_version,
160                "0.95",
161                quantiles.p95_micros,
162            );
163            labeled_gauge(
164                &mut output,
165                "nasaga_lifecycle_duration_seconds",
166                &quantiles.workflow,
167                quantiles.definition_version,
168                "0.99",
169                quantiles.p99_micros,
170            );
171        }
172        summary(
173            &mut output,
174            "nasaga_kafka_result_processing_duration_seconds",
175            self.kafka_result_processing_total,
176            self.kafka_result_processing_micros_sum,
177        );
178        counter(
179            &mut output,
180            "nasaga_kafka_result_retry_total",
181            self.kafka_result_retry_total,
182        );
183        counter(
184            &mut output,
185            "nasaga_kafka_result_dlt_requested_total",
186            self.kafka_result_dlt_requested_total,
187        );
188        counter(
189            &mut output,
190            "nasaga_kafka_result_ack_total",
191            self.kafka_result_ack_total,
192        );
193        counter(
194            &mut output,
195            "nasaga_kafka_result_duplicate_total",
196            self.kafka_result_duplicate_total,
197        );
198        summary(
199            &mut output,
200            "nasaga_kafka_command_processing_duration_seconds",
201            self.kafka_command_processing_total,
202            self.kafka_command_processing_micros_sum,
203        );
204        counter(
205            &mut output,
206            "nasaga_kafka_command_retry_total",
207            self.kafka_command_retry_total,
208        );
209        counter(
210            &mut output,
211            "nasaga_kafka_command_dlt_requested_total",
212            self.kafka_command_dlt_requested_total,
213        );
214        counter(
215            &mut output,
216            "nasaga_kafka_command_ack_total",
217            self.kafka_command_ack_total,
218        );
219        counter(
220            &mut output,
221            "nasaga_kafka_command_duplicate_total",
222            self.kafka_command_duplicate_total,
223        );
224        output.push_str(&crate::render_saga_latency_metrics());
225        output
226    }
227}
228
229impl From<SagaStoreMetrics> for SagaOperationalMetrics {
230    /// 业务作用：将后端已提交快照与当前进程 transport 计数组装为对外指标。
231    ///
232    /// 参数说明：
233    /// - `store`: 当前 Saga 后端聚合的已提交事实。
234    ///
235    /// 返回：包含完整业务与 transport 指标的快照。
236    fn from(store: SagaStoreMetrics) -> Self {
237        let process = crate::process_metrics_snapshot();
238        Self {
239            started_total: store.started_total,
240            completed_total: store.completed_total,
241            compensated_total: store.compensated_total,
242            manual_intervention_total: store.manual_intervention_total,
243            manually_closed_total: store.manually_closed_total,
244            quota_rejections_total: process.quota_rejections_total,
245            action_rate_rejections_total: process.action_rate_rejections_total,
246            unknown_result_total: store.unknown_result_total,
247            retry_attempt_total: store.retry_attempt_total,
248            conflict_total: store.conflict_total,
249            running_current: store.running_current,
250            waiting_resolution_current: store.waiting_resolution_current,
251            compensating_current: store.compensating_current,
252            manual_intervention_current: store.manual_intervention_current,
253            due_timer_current: store.due_timer_current,
254            lifecycle_duration_count: store.lifecycle_duration_count,
255            lifecycle_duration_micros_sum: store.lifecycle_duration_micros_sum,
256            lifecycle_quantiles: store.lifecycle_quantiles,
257            kafka_result_processing_total: process.kafka_result_processing_total,
258            kafka_result_processing_micros_sum: process.kafka_result_processing_micros_sum,
259            kafka_result_retry_total: process.kafka_result_retry_total,
260            kafka_result_dlt_requested_total: process.kafka_result_dlt_requested_total,
261            kafka_result_ack_total: process.kafka_result_ack_total,
262            kafka_result_duplicate_total: process.kafka_result_duplicate_total,
263            kafka_command_processing_total: process.kafka_command_processing_total,
264            kafka_command_processing_micros_sum: process.kafka_command_processing_micros_sum,
265            kafka_command_retry_total: process.kafka_command_retry_total,
266            kafka_command_dlt_requested_total: process.kafka_command_dlt_requested_total,
267            kafka_command_ack_total: process.kafka_command_ack_total,
268            kafka_command_duplicate_total: process.kafka_command_duplicate_total,
269        }
270    }
271}
272
273/// 业务作用：渲染带冻结流程版本和固定 quantile 的生命周期 gauge，保持标签集合低基数。
274///
275/// 参数说明：
276/// - `output`：接收指标文本的现有缓冲。
277/// - `name`：调用方指定的固定指标名。
278/// - `workflow`：来自冻结 definition 的工作流身份。
279/// - `version`：definition 的业务定义版本。
280/// - `quantile`：调用方提供的固定分位标签文本。
281/// - `micros`：持久样本微秒值，转为秒后输出。
282///
283/// 返回：无返回值；向已有文本追加秒单位样本，workflow 中反斜线与双引号被转义，不转义换行或验证指标名及 quantile，调用方须提供受控值。
284fn labeled_gauge(
285    output: &mut String,
286    name: &str,
287    workflow: &str,
288    version: u32,
289    quantile: &str,
290    micros: u64,
291) {
292    let workflow = workflow.replace('\\', "\\\\").replace('"', "\\\"");
293    let _ = writeln!(
294        output,
295        "{name}{{workflow=\"{workflow}\",definition_version=\"{version}\",quantile=\"{quantile}\"}} {}",
296        micros as f64 / 1_000_000.0
297    );
298}
299
300/// 业务作用：累计一次按租户配额拒绝的创建请求,供低基数观测面导出。
301///
302/// 参数说明: 无。
303///
304/// 返回：无返回值。
305pub(crate) fn record_quota_rejection() {
306    crate::record_quota_rejection();
307}
308
309/// 业务作用：累计一次按租户速率拒绝的变更类管理动作,供低基数观测面导出。
310///
311/// 参数说明: 无。
312///
313/// 返回：无返回值。
314pub(crate) fn record_action_rate_rejection() {
315    crate::record_action_rate_rejection();
316}
317
318/// 业务作用：输出一个无 label 的 Prometheus counter。
319///
320/// 参数说明：
321/// - `output`：接收指标文本的现有缓冲。
322/// - `name`：受控的固定指标名。
323/// - `value`：调用方读取的累计计数值。
324///
325/// 返回：无返回值；追加 counter TYPE 声明与整数样本，不验证名称或读取原子计数。
326fn counter(output: &mut String, name: &str, value: u64) {
327    let _ = writeln!(output, "# TYPE {name} counter\n{name} {value}");
328}
329
330/// 业务作用：输出一个无 label 的 Prometheus gauge。
331///
332/// 参数说明：
333/// - `output`：接收指标文本的现有缓冲。
334/// - `name`：受控的固定指标名。
335/// - `value`：调用方读取的当前状态值。
336///
337/// 返回：无返回值；追加 gauge TYPE 声明与整数样本，不验证名称或采集状态。
338fn gauge(output: &mut String, name: &str, value: u64) {
339    let _ = writeln!(output, "# TYPE {name} gauge\n{name} {value}");
340}
341
342/// 业务作用：以 Prometheus summary 的 `_count`/`_sum` 形式输出累计耗时。
343///
344/// 参数说明：
345/// - `output`：接收指标文本的现有缓冲。
346/// - `name`：受控的固定指标名。
347/// - `count`：累计观测次数。
348/// - `micros_sum`：累计耗时微秒和，输出时转换为秒。
349///
350/// 返回：无返回值；追加 summary TYPE、count 与保留六位小数的秒数 sum，不输出 quantile 或验证 count 与 sum 一致。
351fn summary(output: &mut String, name: &str, count: u64, micros_sum: u64) {
352    let seconds_sum = micros_sum as f64 / 1_000_000.0;
353    let _ = writeln!(
354        output,
355        "# TYPE {name} summary\n{name}_count {count}\n{name}_sum {seconds_sum:.6}"
356    );
357}
nasaga_runtime_core/observability.rs

nasaga_runtime_core/
observability.rs
