mcp-usage-tower 0.4.0

Tower hot path for MCP authentication, caching, and usage attribution.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

//! Pull-based OpenTelemetry instruments for edge counters.

use std::sync::{Arc, Weak};

use opentelemetry::metrics::{Meter, ObservableCounter};

use crate::{EdgeMetrics, MetricsSnapshot};

type MetricDefinition = (&'static str, &'static str, fn(MetricsSnapshot) -> u64);

/// Registration guard for the OpenTelemetry observable instruments.
///
/// Keep this value alive for as long as metrics should be exported. Dropping it
/// unregisters the callbacks according to the active OpenTelemetry provider.
#[derive(Debug)]
pub struct OpenTelemetryMetrics {
    instruments: Vec<ObservableCounter<u64>>,
}

impl OpenTelemetryMetrics {
    /// Number of registered observable counters.
    #[must_use]
    pub fn len(&self) -> usize {
        self.instruments.len()
    }

    /// Whether no observable counters are registered.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.instruments.is_empty()
    }
}

/// Register low-cardinality observable counters with an application-owned meter.
#[must_use]
pub fn install_opentelemetry(metrics: &Arc<EdgeMetrics>, meter: &Meter) -> OpenTelemetryMetrics {
    let definitions: [MetricDefinition; 13] = [
        ("mcp.usage.classified", "Classified MCP requests", |s| {
            s.classified
        }),
        ("mcp.usage.rejected", "Rejected MCP requests", |s| {
            s.rejected
        }),
        (
            "mcp.usage.unauthenticated",
            "MCP requests refused for an invalid API key",
            |s| s.unauthenticated,
        ),
        (
            "mcp.usage.throttled",
            "MCP requests refused for repeated credential failures",
            |s| s.throttled,
        ),
        ("mcp.usage.cache_hits", "MCP response cache hits", |s| {
            s.cache_hits
        }),
        ("mcp.usage.cache_misses", "MCP response cache misses", |s| {
            s.cache_misses
        }),
        (
            "mcp.usage.billable_deliveries",
            "Billable MCP deliveries",
            |s| s.billed,
        ),
        (
            "mcp.usage.recorded_units",
            "Recorded MCP usage units",
            |s| s.billed_units,
        ),
        ("mcp.usage.free_deliveries", "Free MCP deliveries", |s| {
            s.free
        }),
        (
            "mcp.usage.duplicates",
            "Suppressed MCP usage duplicates",
            |s| s.duplicates,
        ),
        (
            "mcp.usage.record_failures",
            "MCP usage record failures",
            |s| s.record_failures,
        ),
        (
            "mcp.usage.deferred_completions",
            "Terminal accounting parked for later completion",
            |s| s.deferred,
        ),
        (
            "mcp.usage.unrecognized_responses",
            "Unrecognized terminal MCP responses",
            |s| s.unrecognized_responses,
        ),
    ];
    let instruments = definitions
        .into_iter()
        .map(|(name, description, value)| {
            observable_counter(meter, Arc::downgrade(metrics), name, description, value)
        })
        .collect();
    OpenTelemetryMetrics { instruments }
}

fn observable_counter(
    meter: &Meter,
    metrics: Weak<EdgeMetrics>,
    name: &'static str,
    description: &'static str,
    value: fn(MetricsSnapshot) -> u64,
) -> ObservableCounter<u64> {
    meter
        .u64_observable_counter(name)
        .with_description(description)
        .with_callback(move |observer| {
            if let Some(metrics) = metrics.upgrade() {
                observer.observe(value(metrics.snapshot()), &[]);
            }
        })
        .build()
}