d3d11f4
use std::sync::LazyLock;
use std::time::Duration;
use opentelemetry::global;
use opentelemetry::metrics::{Counter, Gauge, Histogram, 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 CI_EVENTS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("ci.events")
.with_description("CI events consumed by kind")
.build()
});
static CI_RUNS_CREATED: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("ci.runs.created")
.with_description("Workflow runs created from events")
.build()
});
static CI_JOBS_FINISHED: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("ci.jobs.finished")
.with_description("Runner jobs finished by conclusion")
.build()
});
static CI_JOB_DURATION: LazyLock<Histogram<f64>> = LazyLock::new(|| {
meter()
.f64_histogram("ci.job.duration")
.with_unit("s")
.with_description("Runner job wall-clock duration")
.build()
});
static CI_WEBHOOK_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("ci.webhook.deliveries")
.with_description("Outbound CI webhook deliveries by outcome")
.build()
});
static AI_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("ai.jobs")
.with_description("AI jobs processed by kind and outcome")
.build()
});
static AI_JOB_DURATION: LazyLock<Histogram<f64>> = LazyLock::new(|| {
meter()
.f64_histogram("ai.job.duration")
.with_unit("s")
.with_description("AI job wall-clock duration")
.build()
});
static AI_TOKENS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("ai.tokens")
.with_description("Model tokens consumed by type")
.build()
});
static EMAIL_SENT: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("email.sent")
.with_description("Email sends by 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 SYNC_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("git.sync.jobs")
.with_description("Repository sync jobs by outcome")
.build()
});
static IMPORT_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("import.jobs")
.with_description("Repository import jobs by outcome")
.build()
});
static SEARCH_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("search.index.jobs")
.with_description("Search index jobs 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 ci_event(kind: &'static str) {
CI_EVENTS.add(1, &[KeyValue::new("kind", kind)]);
}
pub fn ci_runs_created(event: &'static str, count: usize) {
if count > 0 {
CI_RUNS_CREATED.add(count as u64, &[KeyValue::new("event", event)]);
}
}
pub fn ci_job_finished(conclusion: &'static str, duration: Option<Duration>) {
CI_JOBS_FINISHED.add(1, &[KeyValue::new("conclusion", conclusion)]);
if let Some(duration) = duration {
CI_JOB_DURATION.record(
duration.as_secs_f64(),
&[KeyValue::new("conclusion", conclusion)],
);
}
}
pub fn ci_webhook_delivery(outcome: &'static str) {
CI_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);
}
pub fn ai_job(kind: &str, outcome: &'static str) {
AI_JOBS.add(
1,
&[
KeyValue::new("kind", kind.to_string()),
KeyValue::new("outcome", outcome),
],
);
}
pub fn ai_job_duration(kind: &str, duration: Duration) {
AI_JOB_DURATION.record(
duration.as_secs_f64(),
&[KeyValue::new("kind", kind.to_string())],
);
}
pub fn ai_tokens(prompt: Option<i32>, completion: Option<i32>) {
if let Some(prompt) = prompt {
AI_TOKENS.add(
u64::try_from(prompt).unwrap_or(0),
&[KeyValue::new("type", "prompt")],
);
}
if let Some(completion) = completion {
AI_TOKENS.add(
u64::try_from(completion).unwrap_or(0),
&[KeyValue::new("type", "completion")],
);
}
}
pub fn email_sent(outcome: &'static str) {
EMAIL_SENT.add(1, &[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)]);
}
pub fn sync_job(outcome: &'static str) {
SYNC_JOBS.add(1, &[KeyValue::new("outcome", outcome)]);
}
pub fn import_job(outcome: &'static str) {
IMPORT_JOBS.add(1, &[KeyValue::new("outcome", outcome)]);
}
pub fn search_index_job(outcome: &'static str) {
SEARCH_JOBS.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");
}
}
}
}
});
}