synapse_proxy/
metrics.rs

1//! OpenTelemetry metrics exported in Prometheus text format on /metrics.
2
3use axum::extract::State;
4use axum::http::{header, StatusCode};
5use axum::response::IntoResponse;
6use axum::routing::get;
7use axum::Router;
8use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider as _};
9use opentelemetry::KeyValue;
10use opentelemetry_otlp::WithExportConfig;
11use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
12use opentelemetry_sdk::Resource;
13use prometheus::{Encoder, Registry, TextEncoder};
14
15pub struct Metrics {
16    requests: Counter<u64>,
17    duration: Histogram<f64>,
18    upstream_errors: Counter<u64>,
19    transform_errors: Counter<u64>,
20    provider: SdkMeterProvider,
21}
22
23impl Metrics {
24    /// Prometheus-only (unchanged behaviour). Retained for existing callers.
25    pub fn new() -> anyhow::Result<(Self, Registry)> {
26        Self::with_otlp(None, "synapse-proxy")
27    }
28
29    /// Prometheus reader (always, exported on `:9090`) plus an OTLP/HTTP
30    /// `PeriodicReader` when `otlp_endpoint` is `Some`. `service_name` is
31    /// attached as the `service.name` resource attribute so OTLP series carry
32    /// e.g. `service_name=sandbox-broker`.
33    ///
34    /// `otlp_endpoint` is a base collector URL like `http://host:4318`; the
35    /// `/v1/metrics` signal path is appended here (a programmatically-supplied
36    /// endpoint is used verbatim by opentelemetry-otlp 0.32, so we must append
37    /// the path ourselves).
38    pub fn with_otlp(
39        otlp_endpoint: Option<&str>,
40        service_name: &str,
41    ) -> anyhow::Result<(Self, Registry)> {
42        let registry = Registry::new();
43        let prom = opentelemetry_prometheus::exporter()
44            .with_registry(registry.clone())
45            .build()?;
46        let resource = Resource::builder()
47            .with_service_name(service_name.to_string())
48            .build();
49        let mut builder = SdkMeterProvider::builder()
50            .with_reader(prom)
51            .with_resource(resource);
52        if let Some(endpoint) = otlp_endpoint {
53            let exporter = opentelemetry_otlp::MetricExporter::builder()
54                .with_http()
55                .with_endpoint(format!("{}/v1/metrics", endpoint.trim_end_matches('/')))
56                .build()?;
57            builder = builder.with_reader(PeriodicReader::builder(exporter).build());
58        }
59        let provider = builder.build();
60        let meter = provider.meter("synapse-proxy");
61        let metrics = Self {
62            requests: meter.u64_counter("synapse_proxy_requests_total").build(),
63            duration: meter
64                .f64_histogram("synapse_proxy_request_duration_seconds")
65                .build(),
66            upstream_errors: meter
67                .u64_counter("synapse_proxy_upstream_errors_total")
68                .build(),
69            transform_errors: meter
70                .u64_counter("synapse_proxy_transform_errors_total")
71                .build(),
72            provider,
73        };
74        Ok((metrics, registry))
75    }
76
77    /// A meter on the SAME provider (so downstream crates such as synapse-mcp
78    /// build instruments that share this provider's OTLP + Prometheus readers
79    /// and single `service.name` resource).
80    pub fn meter(&self) -> Meter {
81        self.provider.meter("sandbox-broker")
82    }
83
84    pub fn record(&self, route: &str, method: &str, status: u16, outcome: &str, secs: f64) {
85        let labels = [
86            KeyValue::new("route", route.to_string()),
87            KeyValue::new("method", method.to_string()),
88            KeyValue::new("status", status.to_string()),
89            KeyValue::new("outcome", outcome.to_string()),
90        ];
91        self.requests.add(1, &labels);
92        self.duration.record(
93            secs,
94            &[
95                KeyValue::new("route", route.to_string()),
96                KeyValue::new("method", method.to_string()),
97            ],
98        );
99    }
100
101    pub fn upstream_error(&self, route: &str, reason: &str) {
102        self.upstream_errors.add(
103            1,
104            &[
105                KeyValue::new("route", route.to_string()),
106                KeyValue::new("reason", reason.to_string()),
107            ],
108        );
109    }
110
111    pub fn transform_error(&self, route: &str, transform: &str) {
112        self.transform_errors.add(
113            1,
114            &[
115                KeyValue::new("route", route.to_string()),
116                KeyValue::new("transform", transform.to_string()),
117            ],
118        );
119    }
120}
121
122pub fn metrics_router(registry: Registry) -> Router {
123    Router::new()
124        .route("/metrics", get(serve))
125        .with_state(registry)
126}
127
128async fn serve(State(registry): State<Registry>) -> impl IntoResponse {
129    let mut buf = Vec::new();
130    if TextEncoder::new()
131        .encode(&registry.gather(), &mut buf)
132        .is_err()
133    {
134        return (StatusCode::INTERNAL_SERVER_ERROR, "encode error").into_response();
135    }
136    ([(header::CONTENT_TYPE, "text/plain; version=0.0.4")], buf).into_response()
137}
138
139#[cfg(test)]
140mod tests {
141    use super::*;
142
143    #[test]
144    fn records_and_exports() {
145        let (m, registry) = Metrics::new().unwrap();
146        m.record("cortex", "POST", 200, "forwarded", 0.01);
147        let mut buf = Vec::new();
148        TextEncoder::new()
149            .encode(&registry.gather(), &mut buf)
150            .unwrap();
151        let text = String::from_utf8(buf).unwrap();
152        assert!(text.contains("synapse_proxy_requests_total"));
153        assert!(text.contains("route=\"cortex\""));
154    }
155
156    #[test]
157    fn otlp_and_prometheus_readers_coexist() {
158        // With an OTLP endpoint set, the Prometheus registry still exports the proxy series.
159        let (m, registry) =
160            Metrics::with_otlp(Some("http://127.0.0.1:4318"), "sandbox-broker").unwrap();
161        m.record("cortex", "POST", 200, "forwarded", 0.01);
162        let mut buf = Vec::new();
163        TextEncoder::new()
164            .encode(&registry.gather(), &mut buf)
165            .unwrap();
166        let text = String::from_utf8(buf).unwrap();
167        assert!(text.contains("synapse_proxy_requests_total"));
168        // The exposed meter creates instruments on the same provider.
169        let _c = m.meter().u64_counter("probe_total").build();
170    }
171}
synapse_proxy/metrics.rs

synapse_proxy/
metrics.rs
