951a522
use std::sync::LazyLock;
use std::time::Duration;
use opentelemetry::global;
use opentelemetry::metrics::{Counter, Gauge, UpDownCounter};
use opentelemetry::{KeyValue, metrics::Meter};
use crate::redis::AppRedis;
fn meter() -> Meter {
global::meter("gitdata")
}
static AUTH_LOGINS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("auth.logins")
.with_description("Login attempts by outcome")
.build()
});
static GIT_OPS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("git.ops")
.with_description("Git protocol operations by operation and outcome")
.build()
});
static PUSH_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("push.deliveries")
.with_description("Web push deliveries by outcome")
.build()
});
static IM_CONNECTIONS: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {
meter()
.i64_up_down_counter("im.connections")
.with_description("Open chat WebSocket connections")
.build()
});
static CHAT_WEBHOOK_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("chat.webhook.deliveries")
.with_description("Outbound chat webhook deliveries by outcome")
.build()
});
static QUEUE_DEPTH: LazyLock<Gauge<i64>> = LazyLock::new(|| {
meter()
.i64_gauge("queue.depth")
.with_description("Pending queue entries by topic")
.build()
});
pub fn auth_login(outcome: &'static str) {
AUTH_LOGINS.add(1, &[KeyValue::new("outcome", outcome)]);
}
pub fn git_op(op: &'static str, outcome: &'static str) {
GIT_OPS.add(
1,
&[KeyValue::new("op", op), KeyValue::new("outcome", outcome)],
);
}
pub fn push_delivery(outcome: &'static str) {
PUSH_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);
}
pub fn im_connection(delta: i64) {
IM_CONNECTIONS.add(delta, &[]);
}
/// `Drop`-based guard so every socket exit path (incl. early returns) balances
/// the connection gauge.
pub struct ImConnectionGuard;
impl ImConnectionGuard {
#[must_use]
pub fn new() -> Self {
im_connection(1);
Self
}
}
impl Default for ImConnectionGuard {
fn default() -> Self {
Self::new()
}
}
impl Drop for ImConnectionGuard {
fn drop(&mut self) {
im_connection(-1);
}
}
pub fn chat_webhook_delivery(outcome: &'static str) {
CHAT_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);
}
const QUEUE_DEPTH_INTERVAL: Duration = Duration::from_secs(30);
/// 周期性上报队列与死信队列的积压深度；进程退出时随运行时一起结束。
pub fn spawn_queue_depth_reporter(redis: AppRedis, queues: &[(&'static str, String)]) {
let queues: Vec<(&'static str, String)> = queues.to_vec();
if queues.is_empty() {
return;
}
tokio::spawn(async move {
let mut ticker = tokio::time::interval(QUEUE_DEPTH_INTERVAL);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
ticker.tick().await;
for (topic, key) in &queues {
match redis.llen(key).await {
Ok(depth) => QUEUE_DEPTH.record(depth, &[KeyValue::new("topic", *topic)]),
Err(error) => {
tracing::debug!(%error, %key, "queue depth sample failed");
}
}
}
}
});
}