futu_gateway_core/bridge/push_health/
snapshot.rs

1use std::fmt::Write as _;
2
3/// v1.4.83 §9 F5 结构化 snapshot, 给 `/api/push-subscriber-info` 直接序列化.
4#[derive(Debug, Clone, serde::Serialize)]
5pub struct PushHealthSnapshot {
6    pub push_stream_healthy: bool,
7    pub last_push_received_at_ms: i64,
8    pub last_push_by_sub_type: Vec<PushSubTypeWatermark>,
9    pub consecutive_parse_errors: u32,
10    pub total_parse_errors: u64,
11    pub total_pushes_received: u64,
12    pub resubscribe_triggers: u64,
13    pub circuit_breaker_trips: u64,
14    pub circuit_breaker_tripped_at_ms: i64,
15    pub is_circuit_tripped_now: bool,
16    pub orphan_orders_detected: u64,
17    pub last_orphan_scan_at_ms: i64,
18    pub last_resubscribe_at_ms: i64,
19    /// v1.4.90 P2-B: F2 (TradeReQuery) account-level circuit-break 总 trip 次数.
20    pub f2_circuit_tripped_total: u64,
21    /// v1.4.106 codex 0932 F1: TradeReQuery 队列丢弃总数 (queue full / closed).
22    /// 持续 > 0 表明 dispatcher 跟不上 push 速度, 需要 ops 关注.
23    pub trade_requery_queue_dropped_total: u64,
24    /// v1.4.106 codex 0932 F1: 最近一次 TradeReQuery drop 的 Unix ms.
25    /// 0 = 从未丢弃过. 60s 内非 0 → `is_healthy()` 返 false.
26    pub last_trade_requery_drop_at_ms: i64,
27    /// **v1.4.106 codex 0631 F4 [P2]**: 最近一次 resubscribe 部分失败的 Unix ms.
28    /// 0 = 从未部分失败. 60s 内非 0 → push_stream_healthy=false (degraded).
29    pub last_resubscribe_partial_at_ms: i64,
30    /// **v1.4.106 codex 0631 F4 [P2]**: resubscribe 部分失败累计次数 (monotonic).
31    /// 监控 alarm: rate-of-change 显著 → 后端 / 网络异常.
32    pub resubscribe_partial_total: u64,
33    /// **v1.4.106 codex 0631 F4 [P2]**: 最近一次 resubscribe 缺失的 keys 数
34    /// (cache miss / unresolved). 0 = 全部 OK; >0 = 部分 ghost subscription.
35    pub last_missed_keys_count: u32,
36    /// **v1.4.106 codex 0631 F4 [P2]**: 导致 push_stream_healthy=false 的人类
37    /// 可读理由 (`None` = 健康).
38    pub degraded_reason: Option<String>,
39}
40
41/// Per-SubType QOT push freshness watermark.
42#[derive(Debug, Clone, serde::Serialize)]
43pub struct PushSubTypeWatermark {
44    pub sub_type: i32,
45    pub last_push_received_at_ms: i64,
46}
47
48impl PushHealthSnapshot {
49    /// Render the snapshot as Prometheus text exposition.
50    #[must_use]
51    pub fn render_prometheus(&self) -> String {
52        let mut s = String::with_capacity(4096);
53
54        s.push_str(
55            "# HELP futu_gateway_push_health_stream_healthy Push stream health from PushHealth snapshot (1=healthy,0=degraded)\n# TYPE futu_gateway_push_health_stream_healthy gauge\n",
56        );
57        let healthy = u8::from(self.push_stream_healthy);
58        let degraded = u8::from(self.degraded_reason.is_some());
59        let _ = writeln!(s, "futu_gateway_push_health_stream_healthy {healthy}");
60
61        s.push_str(
62            "# HELP futu_gateway_push_health_last_push_received_at_ms Last successful push timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_push_health_last_push_received_at_ms gauge\n",
63        );
64        let _ = writeln!(
65            s,
66            "futu_gateway_push_health_last_push_received_at_ms {}",
67            self.last_push_received_at_ms
68        );
69
70        s.push_str(
71            "# HELP futu_gateway_push_health_last_push_by_sub_type_at_ms Last regular QOT push dispatch timestamp by FTAPI SubType in Unix milliseconds\n# TYPE futu_gateway_push_health_last_push_by_sub_type_at_ms gauge\n",
72        );
73        for item in &self.last_push_by_sub_type {
74            let _ = writeln!(
75                s,
76                "futu_gateway_push_health_last_push_by_sub_type_at_ms{{sub_type=\"{}\"}} {}",
77                item.sub_type, item.last_push_received_at_ms
78            );
79        }
80
81        s.push_str(
82            "# HELP futu_gateway_push_health_consecutive_parse_errors Consecutive push parse errors\n# TYPE futu_gateway_push_health_consecutive_parse_errors gauge\n",
83        );
84        let _ = writeln!(
85            s,
86            "futu_gateway_push_health_consecutive_parse_errors {}",
87            self.consecutive_parse_errors
88        );
89
90        s.push_str(
91            "# HELP futu_gateway_push_health_total_parse_errors_total Total push parse errors\n# TYPE futu_gateway_push_health_total_parse_errors_total counter\n",
92        );
93        let _ = writeln!(
94            s,
95            "futu_gateway_push_health_total_parse_errors_total {}",
96            self.total_parse_errors
97        );
98
99        s.push_str(
100            "# HELP futu_gateway_push_health_total_pushes_received_total Total successfully parsed pushes\n# TYPE futu_gateway_push_health_total_pushes_received_total counter\n",
101        );
102        let _ = writeln!(
103            s,
104            "futu_gateway_push_health_total_pushes_received_total {}",
105            self.total_pushes_received
106        );
107
108        s.push_str(
109            "# HELP futu_gateway_push_health_resubscribe_triggers_total Push-health re-subscribe triggers\n# TYPE futu_gateway_push_health_resubscribe_triggers_total counter\n",
110        );
111        let _ = writeln!(
112            s,
113            "futu_gateway_push_health_resubscribe_triggers_total {}",
114            self.resubscribe_triggers
115        );
116
117        s.push_str(
118            "# HELP futu_gateway_push_health_circuit_breaker_trips_total Push parse circuit-breaker trips\n# TYPE futu_gateway_push_health_circuit_breaker_trips_total counter\n",
119        );
120        let _ = writeln!(
121            s,
122            "futu_gateway_push_health_circuit_breaker_trips_total {}",
123            self.circuit_breaker_trips
124        );
125
126        s.push_str(
127            "# HELP futu_gateway_push_health_circuit_breaker_tripped_at_ms Push parse circuit-breaker trip timestamp in Unix milliseconds (0=not tripped)\n# TYPE futu_gateway_push_health_circuit_breaker_tripped_at_ms gauge\n",
128        );
129        let _ = writeln!(
130            s,
131            "futu_gateway_push_health_circuit_breaker_tripped_at_ms {}",
132            self.circuit_breaker_tripped_at_ms
133        );
134
135        s.push_str(
136            "# HELP futu_gateway_push_health_circuit_breaker_tripped Push parse circuit-breaker current state (1=tripped,0=closed)\n# TYPE futu_gateway_push_health_circuit_breaker_tripped gauge\n",
137        );
138        let _ = writeln!(
139            s,
140            "futu_gateway_push_health_circuit_breaker_tripped {}",
141            u8::from(self.is_circuit_tripped_now)
142        );
143
144        s.push_str(
145            "# HELP futu_gateway_push_health_orphan_orders_detected_total Orphan in-flight orders detected by push health scan\n# TYPE futu_gateway_push_health_orphan_orders_detected_total counter\n",
146        );
147        let _ = writeln!(
148            s,
149            "futu_gateway_push_health_orphan_orders_detected_total {}",
150            self.orphan_orders_detected
151        );
152
153        s.push_str(
154            "# HELP futu_gateway_push_health_last_orphan_scan_at_ms Last orphan scan timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_push_health_last_orphan_scan_at_ms gauge\n",
155        );
156        let _ = writeln!(
157            s,
158            "futu_gateway_push_health_last_orphan_scan_at_ms {}",
159            self.last_orphan_scan_at_ms
160        );
161
162        s.push_str(
163            "# HELP futu_gateway_push_health_last_resubscribe_at_ms Last re-subscribe timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_push_health_last_resubscribe_at_ms gauge\n",
164        );
165        let _ = writeln!(
166            s,
167            "futu_gateway_push_health_last_resubscribe_at_ms {}",
168            self.last_resubscribe_at_ms
169        );
170
171        s.push_str(
172            "# HELP futu_gateway_push_health_f2_circuit_tripped_total TradeReQuery F2 account-level circuit trips\n# TYPE futu_gateway_push_health_f2_circuit_tripped_total counter\n",
173        );
174        let _ = writeln!(
175            s,
176            "futu_gateway_push_health_f2_circuit_tripped_total {}",
177            self.f2_circuit_tripped_total
178        );
179
180        s.push_str(
181            "# HELP futu_gateway_push_health_trade_requery_queue_dropped_total TradeReQuery notices dropped because worker queue was full or closed\n# TYPE futu_gateway_push_health_trade_requery_queue_dropped_total counter\n",
182        );
183        let _ = writeln!(
184            s,
185            "futu_gateway_push_health_trade_requery_queue_dropped_total {}",
186            self.trade_requery_queue_dropped_total
187        );
188
189        s.push_str(
190            "# HELP futu_gateway_push_health_last_trade_requery_drop_at_ms Last TradeReQuery queue drop timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_push_health_last_trade_requery_drop_at_ms gauge\n",
191        );
192        let _ = writeln!(
193            s,
194            "futu_gateway_push_health_last_trade_requery_drop_at_ms {}",
195            self.last_trade_requery_drop_at_ms
196        );
197
198        s.push_str(
199            "# HELP futu_gateway_push_health_last_resubscribe_partial_at_ms Last partial re-subscribe failure timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_push_health_last_resubscribe_partial_at_ms gauge\n",
200        );
201        let _ = writeln!(
202            s,
203            "futu_gateway_push_health_last_resubscribe_partial_at_ms {}",
204            self.last_resubscribe_partial_at_ms
205        );
206
207        s.push_str(
208            "# HELP futu_gateway_push_health_resubscribe_partial_total Partial re-subscribe failures\n# TYPE futu_gateway_push_health_resubscribe_partial_total counter\n",
209        );
210        let _ = writeln!(
211            s,
212            "futu_gateway_push_health_resubscribe_partial_total {}",
213            self.resubscribe_partial_total
214        );
215
216        s.push_str(
217            "# HELP futu_gateway_push_health_last_missed_keys_count Missing keys from the last partial re-subscribe\n# TYPE futu_gateway_push_health_last_missed_keys_count gauge\n",
218        );
219        let _ = writeln!(
220            s,
221            "futu_gateway_push_health_last_missed_keys_count {}",
222            self.last_missed_keys_count
223        );
224
225        s.push_str(
226            "# HELP futu_gateway_push_health_degraded Push health degraded flag derived from degraded_reason (1=degraded,0=healthy)\n# TYPE futu_gateway_push_health_degraded gauge\n",
227        );
228        let _ = writeln!(s, "futu_gateway_push_health_degraded {degraded}");
229
230        s
231    }
232}
futu_gateway_core/bridge/push_health/snapshot.rs

futu_gateway_core/bridge/push_health/
snapshot.rs
