futu_server/metrics/

prometheus.rs

use std::sync::Arc;
use std::sync::atomic::Ordering;

use super::{GatewayMetrics, HourBreakdown};

/// v1.4.90 P1-B: 把 24 小时 hour breakdown 渲染成 Prometheus 行 (with `hour` label).
///
/// 输出形如 `<metric_name>{hour="00"} 7\n<metric_name>{hour="01"} 0\n...`
/// 每个 metric 24 行 (UTC hour 0..23).
fn render_hour_breakdown_prom(metric_name: &str, hb: &HourBreakdown) -> String {
    let snap = hb.snapshot();
    let mut out = String::with_capacity(24 * 60);
    for (h, v) in snap.iter().enumerate() {
        out.push_str(&format!("{}{{hour=\"{:02}\"}} {}\n", metric_name, h, v));
    }
    out
}

impl GatewayMetrics {
    /// v1.4.90 P1-B: 输出 Prometheus text exposition 格式, 涵盖:
    ///
    /// - 连接 / 请求 / 响应 / 后端连接基础 counter
    /// - per-cmd_id push counter (cmd_6212_quote / cmd_4716 / cmd_14716 /
    ///   cmd_5300 / cmd_other) — v1.4.83/84 telnet 已暴露, 此处补 Prometheus
    /// - per-cmd_id × UTC-hour 24-bucket breakdown (v1.4.84 §14)
    /// - 订阅 / KeepAlive 计数
    /// - 延迟 p50/p95/p99 gauge (取最近 1000 样本)
    ///
    /// 注册到 [`futu_auth::metrics::Registry`] extension renderer 后, 在
    /// `/metrics` HTTP 输出末尾自动追加. 见
    /// [`install_prometheus_extension`].
    #[must_use]
    pub fn render_prometheus(&self) -> String {
        let mut s = String::with_capacity(8192);

        // ===== 连接 =====
        s.push_str("# HELP futu_gateway_connections_total Total accepted client connections\n");
        s.push_str("# TYPE futu_gateway_connections_total counter\n");
        s.push_str(&format!(
            "futu_gateway_connections_total {}\n",
            self.total_connections.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_disconnections_total Total client disconnections\n# TYPE futu_gateway_disconnections_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_disconnections_total {}\n",
            self.total_disconnections.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_rejected_connections_total Connections rejected (limit hit)\n# TYPE futu_gateway_rejected_connections_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_rejected_connections_total {}\n",
            self.rejected_connections.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_keepalive_timeouts_total KeepAlive timeout disconnects\n# TYPE futu_gateway_keepalive_timeouts_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_keepalive_timeouts_total {}\n",
            self.keepalive_timeouts.load(Ordering::Relaxed)
        ));

        // ===== 请求 =====
        s.push_str(
            "# HELP futu_gateway_requests_total Total handled client requests\n# TYPE futu_gateway_requests_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_requests_total {}\n",
            self.total_requests.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_request_errors_total Handler-returned-None or decryption errors\n# TYPE futu_gateway_request_errors_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_request_errors_total {}\n",
            self.total_request_errors.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_response_bytes_total Cumulative response payload bytes\n# TYPE futu_gateway_response_bytes_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_response_bytes_total {}\n",
            self.total_response_bytes.load(Ordering::Relaxed)
        ));

        // ===== 后端 =====
        s.push_str(
            "# HELP futu_gateway_backend_online Backend connection state (1=online,0=offline)\n# TYPE futu_gateway_backend_online gauge\n",
        );
        s.push_str(&format!(
            "futu_gateway_backend_online {}\n",
            self.backend_online.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_backend_reconnects_total Backend reconnect attempts\n# TYPE futu_gateway_backend_reconnects_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_backend_reconnects_total {}\n",
            self.backend_reconnects.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_backend_reconnect_failures_total Backend reconnect failures\n# TYPE futu_gateway_backend_reconnect_failures_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_backend_reconnect_failures_total {}\n",
            self.backend_reconnect_failures.load(Ordering::Relaxed)
        ));

        // ===== 推送总数 =====
        s.push_str(
            "# HELP futu_gateway_backend_pushes_received_total Pushes received from backend\n# TYPE futu_gateway_backend_pushes_received_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_backend_pushes_received_total {}\n",
            self.backend_pushes_received.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_backend_control_event_drops_total Backend lifecycle control events dropped before dispatcher processing\n# TYPE futu_gateway_backend_control_event_drops_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_backend_control_event_drops_total {}\n",
            self.backend_control_event_drops.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_message_center_async_dropped_total CMD5300 option-event work dropped before async dispatcher processing\n# TYPE futu_gateway_message_center_async_dropped_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_message_center_async_dropped_total {}\n",
            self.message_center_async_dropped_total
                .load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_client_pushes_sent_total Pushes forwarded to clients\n# TYPE futu_gateway_client_pushes_sent_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_client_pushes_sent_total {}\n",
            self.client_pushes_sent.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_client_push_send_failures_total Client push send failures because the downstream channel was closed\n# TYPE futu_gateway_client_push_send_failures_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_client_push_send_failures_total {}\n",
            self.client_push_send_failures.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_ordinary_client_push_backpressure_disconnects_total Slow ordinary-push clients closed after their bounded queue filled\n# TYPE futu_gateway_ordinary_client_push_backpressure_disconnects_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_ordinary_client_push_backpressure_disconnects_total {}\n",
            self.ordinary_client_push_backpressure_disconnects
                .load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_qot_client_push_backpressure_drops_total Quote push frames dropped for clients whose downstream channel is full\n# TYPE futu_gateway_qot_client_push_backpressure_drops_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_qot_client_push_backpressure_drops_total {}\n",
            self.qot_client_push_backpressure_drops
                .load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_qot_client_push_backpressure_drops_by_sub_type_total Quote push frames dropped by full client channel, grouped by Qot_Common.SubType\n# TYPE futu_gateway_qot_client_push_backpressure_drops_by_sub_type_total counter\n",
        );
        for (sub_type, count) in self
            .qot_client_push_backpressure_drops_per_sub_type()
            .iter()
            .enumerate()
        {
            s.push_str(&format!(
                "futu_gateway_qot_client_push_backpressure_drops_by_sub_type_total{{sub_type=\"{}\"}} {}\n",
                sub_type, count
            ));
        }

        // ===== per-cmd push counters (v1.4.83 §14) =====
        // 这是 P1-B 修复的核心: telnet 已有但 /metrics 之前没暴露
        s.push_str(
            "# HELP futu_gateway_backend_pushes_cmd_total Backend pushes by cmd_id (v1.4.83 §14)\n# TYPE futu_gateway_backend_pushes_cmd_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_backend_pushes_cmd_total{{cmd=\"6212_quote\"}} {}\n",
            self.backend_pushes_cmd_quote.load(Ordering::Relaxed)
        ));
        s.push_str(&format!(
            "futu_gateway_backend_pushes_cmd_total{{cmd=\"4716_trade_legacy\"}} {}\n",
            self.backend_pushes_cmd_trade_legacy.load(Ordering::Relaxed)
        ));
        s.push_str(&format!(
            "futu_gateway_backend_pushes_cmd_total{{cmd=\"14716_trade_new\"}} {}\n",
            self.backend_pushes_cmd_trade_new.load(Ordering::Relaxed)
        ));
        s.push_str(&format!(
            "futu_gateway_backend_pushes_cmd_total{{cmd=\"5300_msg_center\"}} {}\n",
            self.backend_pushes_cmd_msg_center.load(Ordering::Relaxed)
        ));
        s.push_str(&format!(
            "futu_gateway_backend_pushes_cmd_total{{cmd=\"other\"}} {}\n",
            self.backend_pushes_cmd_other.load(Ordering::Relaxed)
        ));

        // ===== per-cmd × UTC hour breakdown (v1.4.84 §14) =====
        s.push_str(
            "# HELP futu_gateway_backend_pushes_cmd_quote_by_hour Cmd 6212 quote pushes per UTC hour\n# TYPE futu_gateway_backend_pushes_cmd_quote_by_hour counter\n",
        );
        s.push_str(&render_hour_breakdown_prom(
            "futu_gateway_backend_pushes_cmd_quote_by_hour",
            &self.backend_pushes_cmd_quote_by_hour,
        ));
        s.push_str(
            "# HELP futu_gateway_backend_pushes_cmd_trade_legacy_by_hour Cmd 4716 trade-legacy pushes per UTC hour\n# TYPE futu_gateway_backend_pushes_cmd_trade_legacy_by_hour counter\n",
        );
        s.push_str(&render_hour_breakdown_prom(
            "futu_gateway_backend_pushes_cmd_trade_legacy_by_hour",
            &self.backend_pushes_cmd_trade_legacy_by_hour,
        ));
        s.push_str(
            "# HELP futu_gateway_backend_pushes_cmd_trade_new_by_hour Cmd 14716 trade-new pushes per UTC hour (v1.4.84 §14 tester subject)\n# TYPE futu_gateway_backend_pushes_cmd_trade_new_by_hour counter\n",
        );
        s.push_str(&render_hour_breakdown_prom(
            "futu_gateway_backend_pushes_cmd_trade_new_by_hour",
            &self.backend_pushes_cmd_trade_new_by_hour,
        ));
        s.push_str(
            "# HELP futu_gateway_backend_pushes_cmd_msg_center_by_hour Cmd 5300 msg-center pushes per UTC hour\n# TYPE futu_gateway_backend_pushes_cmd_msg_center_by_hour counter\n",
        );
        s.push_str(&render_hour_breakdown_prom(
            "futu_gateway_backend_pushes_cmd_msg_center_by_hour",
            &self.backend_pushes_cmd_msg_center_by_hour,
        ));

        // ===== 订阅 =====
        s.push_str(
            "# HELP futu_gateway_qot_subscribe_ops_total Quote subscribe operations\n# TYPE futu_gateway_qot_subscribe_ops_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_qot_subscribe_ops_total {}\n",
            self.qot_subscribe_ops.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_qot_unsubscribe_ops_total Quote unsubscribe operations\n# TYPE futu_gateway_qot_unsubscribe_ops_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_qot_unsubscribe_ops_total {}\n",
            self.qot_unsubscribe_ops.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_resubscribe_ops_total Re-subscribe ops after reconnect (legacy, == resubscribe_applied_keys)\n# TYPE futu_gateway_resubscribe_ops_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_resubscribe_ops_total {}\n",
            self.resubscribe_ops.load(Ordering::Relaxed)
        ));
        // v1.4.106 codex 0631 F5: dual counter — attempts 触发 vs applied_keys 真生效.
        s.push_str(
            "# HELP futu_gateway_resubscribe_attempts_total Re-subscribe trigger count (each reconnect/staleness loop +=1)\n# TYPE futu_gateway_resubscribe_attempts_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_resubscribe_attempts_total {}\n",
            self.resubscribe_attempts.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_resubscribe_applied_keys_total Re-subscribe applied keys total (cache resolve OK + backend ack OK)\n# TYPE futu_gateway_resubscribe_applied_keys_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_resubscribe_applied_keys_total {}\n",
            self.resubscribe_applied_keys.load(Ordering::Relaxed)
        ));

        // v1.4.110 codex audit Round2 P3 #19: cold-cache wait 监控.
        // hit/total 比看 backend push 延迟健康度, timeout/total 比看常超时.
        s.push_str(
            "# HELP futu_gateway_cold_cache_wait_total Cold-cache wait entries (cache miss + IsSub)\n# TYPE futu_gateway_cold_cache_wait_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_cold_cache_wait_total {}\n",
            self.cold_cache_wait_total.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_cold_cache_wait_hit_total Cold-cache wait hits (push filled cache within timeout)\n# TYPE futu_gateway_cold_cache_wait_hit_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_cold_cache_wait_hit_total {}\n",
            self.cold_cache_wait_hit.load(Ordering::Relaxed)
        ));
        s.push_str(
            "# HELP futu_gateway_cold_cache_wait_timeout_total Cold-cache wait timeouts (3s elapsed, cache still miss)\n# TYPE futu_gateway_cold_cache_wait_timeout_total counter\n",
        );
        s.push_str(&format!(
            "futu_gateway_cold_cache_wait_timeout_total {}\n",
            self.cold_cache_wait_timeout.load(Ordering::Relaxed)
        ));

        // ===== 延迟 (gauge, 取 ring 当前 stats) =====
        let lat = self.latency_stats();
        s.push_str(
            "# HELP futu_gateway_request_latency_us Request latency percentiles (microseconds, recent ring)\n# TYPE futu_gateway_request_latency_us gauge\n",
        );
        s.push_str(&format!(
            "futu_gateway_request_latency_us{{quantile=\"p50\"}} {}\n",
            lat.p50_us
        ));
        s.push_str(&format!(
            "futu_gateway_request_latency_us{{quantile=\"p95\"}} {}\n",
            lat.p95_us
        ));
        s.push_str(&format!(
            "futu_gateway_request_latency_us{{quantile=\"p99\"}} {}\n",
            lat.p99_us
        ));
        s.push_str(&format!(
            "futu_gateway_request_latency_us{{quantile=\"max\"}} {}\n",
            lat.max_us
        ));

        s
    }
}

/// v1.4.90 P1-B: 把 [`GatewayMetrics`] 注册为 [`futu_auth::metrics::Registry`]
/// 的 extension renderer, 让 `/metrics` HTTP 端点自动包含 per-cmd / per-hour
/// counter.
///
/// 调用方 (futu-opend `main.rs`) 在创建 `GatewayMetrics` Arc 之后调一次:
///
/// ```ignore
/// futu_auth::metrics::install(Arc::new(MetricsRegistry::default()));
/// // ... bridge / server 初始化, 都共享同一份 Arc<GatewayMetrics> ...
/// futu_server::metrics::install_prometheus_extension(server.metrics().clone());
/// ```
///
/// 多次调用会注册多个 renderer (无害但重复输出); 实际上只该调用一次.
pub fn install_prometheus_extension(metrics: Arc<GatewayMetrics>) {
    futu_auth::metrics::register_global_renderer(move || metrics.render_prometheus());
}