mcp_usage_tower/
opentelemetry_metrics.rs

1//! Pull-based OpenTelemetry instruments for edge counters.
2
3use std::sync::{Arc, Weak};
4
5use opentelemetry::metrics::{Meter, ObservableCounter};
6
7use crate::{EdgeMetrics, MetricsSnapshot};
8
9type MetricDefinition = (&'static str, &'static str, fn(MetricsSnapshot) -> u64);
10
11/// Registration guard for the OpenTelemetry observable instruments.
12///
13/// Keep this value alive for as long as metrics should be exported. Dropping it
14/// unregisters the callbacks according to the active OpenTelemetry provider.
15#[derive(Debug)]
16pub struct OpenTelemetryMetrics {
17    instruments: Vec<ObservableCounter<u64>>,
18}
19
20impl OpenTelemetryMetrics {
21    /// Number of registered observable counters.
22    #[must_use]
23    pub fn len(&self) -> usize {
24        self.instruments.len()
25    }
26
27    /// Whether no observable counters are registered.
28    #[must_use]
29    pub fn is_empty(&self) -> bool {
30        self.instruments.is_empty()
31    }
32}
33
34/// Register low-cardinality observable counters with an application-owned meter.
35#[must_use]
36pub fn install_opentelemetry(metrics: &Arc<EdgeMetrics>, meter: &Meter) -> OpenTelemetryMetrics {
37    let definitions: [MetricDefinition; 13] = [
38        ("mcp.usage.classified", "Classified MCP requests", |s| {
39            s.classified
40        }),
41        ("mcp.usage.rejected", "Rejected MCP requests", |s| {
42            s.rejected
43        }),
44        (
45            "mcp.usage.unauthenticated",
46            "MCP requests refused for an invalid API key",
47            |s| s.unauthenticated,
48        ),
49        (
50            "mcp.usage.throttled",
51            "MCP requests refused for repeated credential failures",
52            |s| s.throttled,
53        ),
54        ("mcp.usage.cache_hits", "MCP response cache hits", |s| {
55            s.cache_hits
56        }),
57        ("mcp.usage.cache_misses", "MCP response cache misses", |s| {
58            s.cache_misses
59        }),
60        (
61            "mcp.usage.billable_deliveries",
62            "Billable MCP deliveries",
63            |s| s.billed,
64        ),
65        (
66            "mcp.usage.recorded_units",
67            "Recorded MCP usage units",
68            |s| s.billed_units,
69        ),
70        ("mcp.usage.free_deliveries", "Free MCP deliveries", |s| {
71            s.free
72        }),
73        (
74            "mcp.usage.duplicates",
75            "Suppressed MCP usage duplicates",
76            |s| s.duplicates,
77        ),
78        (
79            "mcp.usage.record_failures",
80            "MCP usage record failures",
81            |s| s.record_failures,
82        ),
83        (
84            "mcp.usage.deferred_completions",
85            "Terminal accounting parked for later completion",
86            |s| s.deferred,
87        ),
88        (
89            "mcp.usage.unrecognized_responses",
90            "Unrecognized terminal MCP responses",
91            |s| s.unrecognized_responses,
92        ),
93    ];
94    let instruments = definitions
95        .into_iter()
96        .map(|(name, description, value)| {
97            observable_counter(meter, Arc::downgrade(metrics), name, description, value)
98        })
99        .collect();
100    OpenTelemetryMetrics { instruments }
101}
102
103fn observable_counter(
104    meter: &Meter,
105    metrics: Weak<EdgeMetrics>,
106    name: &'static str,
107    description: &'static str,
108    value: fn(MetricsSnapshot) -> u64,
109) -> ObservableCounter<u64> {
110    meter
111        .u64_observable_counter(name)
112        .with_description(description)
113        .with_callback(move |observer| {
114            if let Some(metrics) = metrics.upgrade() {
115                observer.observe(value(metrics.snapshot()), &[]);
116            }
117        })
118        .build()
119}
mcp_usage_tower/opentelemetry_metrics.rs

mcp_usage_tower/
opentelemetry_metrics.rs
