d13fe56
use std::sync::{LazyLock, OnceLock};
use std::time::Instant;
use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter, UpDownCounter};
use opentelemetry::{KeyValue, global};
use opentelemetry_otlp::{MetricExporter, WithExportConfig, WithHttpConfig};
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};
use crate::config::TelemetryConfig;
pub const RPC_DURATION: &str = "rpc.server.duration";
pub const RPC_REQUESTS: &str = "rpc.server.requests";
pub const RPC_ACTIVE: &str = "rpc.server.active_requests";
struct RpcMetrics {
requests: Counter<u64>,
duration: Histogram<f64>,
active: UpDownCounter<i64>,
}
fn rpc() -> &'static RpcMetrics {
static METRICS: OnceLock<RpcMetrics> = OnceLock::new();
METRICS.get_or_init(|| {
let meter = global::meter("identity.grpc");
RpcMetrics {
requests: meter.u64_counter(RPC_REQUESTS).build(),
duration: meter.f64_histogram(RPC_DURATION).with_unit("ms").build(),
active: meter.i64_up_down_counter(RPC_ACTIVE).build(),
}
})
}
/// 在途 RPC 守卫：构造时 +1、Drop 时 -1；`finish` 记录计数、时延与状态。
pub struct ActiveRpcGuard {
service: &'static str,
started: Instant,
}
impl ActiveRpcGuard {
pub fn new(service: &'static str) -> Self {
let metrics = rpc();
metrics
.active
.add(1, &[KeyValue::new("rpc.service", service)]);
Self {
service,
started: Instant::now(),
}
}
/// 记录一次 RPC 结束。`status` 为 `OK` 之外的取值时同时把 span 标成错误。
pub fn finish<T>(&self, method: &'static str, result: &Result<T, tonic::Status>) {
let status_code = status_label(result);
if let Some(error) = result.as_ref().err() {
use tracing_opentelemetry::OpenTelemetrySpanExt;
tracing::Span::current().set_status(opentelemetry::trace::Status::error(
error.message().to_owned(),
));
}
let metrics = rpc();
let attributes = [
KeyValue::new("rpc.system", "grpc"),
KeyValue::new("rpc.service", self.service),
KeyValue::new("rpc.method", method),
KeyValue::new("rpc.grpc.status_code", status_code),
KeyValue::new("error", status_code != "OK"),
];
metrics.requests.add(1, &attributes);
metrics
.duration
.record(self.started.elapsed().as_secs_f64() * 1000.0, &attributes);
}
}
impl Drop for ActiveRpcGuard {
fn drop(&mut self) {
rpc()
.active
.add(-1, &[KeyValue::new("rpc.service", self.service)]);
}
}
pub fn status_label<T>(result: &Result<T, tonic::Status>) -> &'static str {
match result {
Ok(_) => "OK",
Err(status) => match status.code() {
tonic::Code::Ok => "OK",
tonic::Code::Cancelled => "CANCELLED",
tonic::Code::Unknown => "UNKNOWN",
tonic::Code::InvalidArgument => "INVALID_ARGUMENT",
tonic::Code::DeadlineExceeded => "DEADLINE_EXCEEDED",
tonic::Code::NotFound => "NOT_FOUND",
tonic::Code::AlreadyExists => "ALREADY_EXISTS",
tonic::Code::PermissionDenied => "PERMISSION_DENIED",
tonic::Code::ResourceExhausted => "RESOURCE_EXHAUSTED",
tonic::Code::FailedPrecondition => "FAILED_PRECONDITION",
tonic::Code::Aborted => "ABORTED",
tonic::Code::OutOfRange => "OUT_OF_RANGE",
tonic::Code::Unimplemented => "UNIMPLEMENTED",
tonic::Code::Internal => "INTERNAL",
tonic::Code::Unavailable => "UNAVAILABLE",
tonic::Code::DataLoss => "DATA_LOSS",
tonic::Code::Unauthenticated => "UNAUTHENTICATED",
},
}
}
pub fn build(
config: &TelemetryConfig,
resource: Resource,
warnings: &mut Vec<String>,
) -> SdkMeterProvider {
let mut builder = SdkMeterProvider::builder().with_resource(resource);
if let Some(endpoint) = config.metrics_url() {
match exporter(config, &endpoint) {
Ok(exporter) => {
let reader = PeriodicReader::builder(exporter)
.with_interval(config.metrics.interval)
.build();
builder = builder.with_reader(reader);
}
Err(error) => warnings.push(format!(
"otlp metric exporter disabled: {error} (endpoint {endpoint})"
)),
}
}
builder.build()
}
fn exporter(
config: &TelemetryConfig,
endpoint: &str,
) -> Result<MetricExporter, opentelemetry_otlp::ExporterBuildError> {
let mut builder = MetricExporter::builder()
.with_http()
.with_endpoint(endpoint)
.with_temporality(Temporality::default())
.with_timeout(config.timeout);
if !config.metrics.signal.headers.is_empty() {
builder = builder.with_headers(config.metrics.signal.headers.clone());
}
builder.build()
}
fn app_meter() -> Meter {
global::meter("gitdata")
}
static AUTH_LOGINS: LazyLock<Counter<u64>> = LazyLock::new(|| {
app_meter()
.u64_counter("auth.logins")
.with_description("Login attempts by outcome")
.build()
});
static GIT_OPS: LazyLock<Counter<u64>> = LazyLock::new(|| {
app_meter()
.u64_counter("git.ops")
.with_description("Git protocol operations by operation and outcome")
.build()
});
static PUSH_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {
app_meter()
.u64_counter("push.deliveries")
.with_description("Web push deliveries by outcome")
.build()
});
static IM_CONNECTIONS: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {
app_meter()
.i64_up_down_counter("im.connections")
.with_description("Open chat WebSocket connections")
.build()
});
static CHAT_WEBHOOK_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {
app_meter()
.u64_counter("chat.webhook.deliveries")
.with_description("Outbound chat webhook deliveries by outcome")
.build()
});
static QUEUE_DEPTH: LazyLock<Gauge<i64>> = LazyLock::new(|| {
app_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)]);
}
pub fn queue_depth(topic: &'static str, depth: i64) {
QUEUE_DEPTH.record(depth, &[KeyValue::new("topic", topic)]);
}