kerniq_observability_prometheus/
lib.rs

1//! Prometheus scraping for Kerniq Server.
2//!
3//! This crate is configuration over two existing libraries: the
4//! [`opentelemetry-prometheus`](https://docs.rs/opentelemetry-prometheus)
5//! exporter (an OpenTelemetry `MetricReader` that collects on demand) and the
6//! [`prometheus`](https://docs.rs/prometheus) crate's text encoder. Kerniq
7//! adds the HTTP endpoint and, optionally, a dedicated listener.
8//!
9//! ```
10//! use kerniq_observability_prometheus::{PrometheusConfig, PrometheusExporter};
11//! use opentelemetry::metrics::MeterProvider;
12//! use opentelemetry_sdk::metrics::SdkMeterProvider;
13//!
14//! let (reader, exporter) = PrometheusExporter::build(PrometheusConfig::default()).unwrap();
15//! let provider = SdkMeterProvider::builder().with_reader(reader).build();
16//! provider.meter("demo").u64_counter("jobs").build().add(1, &[]);
17//!
18//! let text = exporter.render().unwrap();
19//! assert!(text.contains("jobs_total"));
20//!
21//! // Serve it: merge into the application, or bind a separate listener.
22//! let router: axum::Router = exporter.router();
23//! ```
24//!
25//! # Cost model
26//!
27//! Nothing runs between scrapes: there is no background task and no copy of
28//! the metrics. Each scrape asks the OpenTelemetry SDK to aggregate its
29//! current state and encodes it on a blocking-pool thread, so encoding never
30//! occupies an async worker that is serving requests.
31
32#![warn(missing_docs)]
33#![cfg_attr(docsrs, feature(doc_cfg))]
34
35use std::future::Future;
36use std::net::SocketAddr;
37
38use axum::Router;
39use axum::http::{StatusCode, header};
40use axum::response::{IntoResponse, Response};
41use axum::routing::get;
42use prometheus::{Encoder, TextEncoder};
43
44pub use opentelemetry_prometheus::PrometheusExporter as PrometheusReader;
45
46/// Errors from building or rendering the exporter.
47#[derive(Debug, thiserror::Error)]
48pub enum Error {
49    /// The OpenTelemetry Prometheus exporter could not be built.
50    #[error("failed to build the Prometheus exporter: {0}")]
51    Build(String),
52    /// Encoding the scrape failed.
53    #[error("failed to encode metrics: {0}")]
54    Encode(#[from] prometheus::Error),
55    /// The metrics listener could not be bound or served.
56    #[error("metrics listener error: {0}")]
57    Io(#[from] std::io::Error),
58}
59
60/// Where and how metrics are exposed.
61#[derive(Debug, Clone)]
62#[must_use]
63pub struct PrometheusConfig {
64    path: String,
65    listen: Option<SocketAddr>,
66    scope_labels: bool,
67    target_info: bool,
68    namespace: Option<String>,
69    resource_labels: Vec<String>,
70}
71
72impl Default for PrometheusConfig {
73    fn default() -> Self {
74        Self {
75            path: "/metrics".to_owned(),
76            listen: None,
77            scope_labels: false,
78            target_info: true,
79            namespace: None,
80            resource_labels: vec!["service.name".to_owned()],
81        }
82    }
83}
84
85impl PrometheusConfig {
86    /// `/metrics` on the application listener.
87    pub fn new() -> Self {
88        Self::default()
89    }
90
91    /// The scrape path. Default `/metrics`.
92    ///
93    /// # Panics
94    ///
95    /// Panics if `path` does not start with `/`.
96    pub fn path(mut self, path: impl Into<String>) -> Self {
97        let path = path.into();
98        assert!(path.starts_with('/'), "metrics path must start with '/'");
99        self.path = path;
100        self
101    }
102
103    /// Serves metrics on a separate listener instead of the application's.
104    /// Recommended in production: the port can be firewalled to the
105    /// Prometheus servers, and scrapes never share a connection limit or
106    /// middleware with user traffic.
107    pub fn listen(mut self, address: SocketAddr) -> Self {
108        self.listen = Some(address);
109        self
110    }
111
112    /// Adds `otel_scope_name`/`otel_scope_version` labels to every series.
113    /// Off by default: they add a label to every series without helping
114    /// most dashboards.
115    pub fn scope_labels(mut self, enabled: bool) -> Self {
116        self.scope_labels = enabled;
117        self
118    }
119
120    /// Emits the `target_info` series carrying resource attributes. Default
121    /// on; join it in PromQL to filter by service or environment.
122    pub fn target_info(mut self, enabled: bool) -> Self {
123        self.target_info = enabled;
124        self
125    }
126
127    /// Resource attributes copied onto every series as labels (dots become
128    /// underscores). Default: `service.name`, so dashboards can filter by
129    /// `service_name` both here and behind an OpenTelemetry Collector with
130    /// `resource_to_telemetry_conversion`. Each entry is one constant label
131    /// per process and does not increase cardinality.
132    pub fn resource_labels<I, K>(mut self, keys: I) -> Self
133    where
134        I: IntoIterator<Item = K>,
135        K: Into<String>,
136    {
137        self.resource_labels = keys.into_iter().map(Into::into).collect();
138        self
139    }
140
141    /// Prefixes every metric name with `namespace_`.
142    pub fn namespace(mut self, namespace: impl Into<String>) -> Self {
143        self.namespace = Some(namespace.into());
144        self
145    }
146
147    /// The configured scrape path.
148    pub fn scrape_path(&self) -> &str {
149        &self.path
150    }
151
152    /// The separate listener address, if any.
153    pub fn listen_address(&self) -> Option<SocketAddr> {
154        self.listen
155    }
156}
157
158/// A handle that renders and serves the metrics collected by the paired
159/// [`PrometheusReader`]. Cheap to clone.
160#[derive(Debug, Clone)]
161pub struct PrometheusExporter {
162    registry: prometheus::Registry,
163    config: PrometheusConfig,
164}
165
166impl PrometheusExporter {
167    /// Builds the reader (register it with an `SdkMeterProvider`) and the
168    /// handle that renders it.
169    pub fn build(config: PrometheusConfig) -> Result<(PrometheusReader, Self), Error> {
170        let registry = prometheus::Registry::new();
171        let mut builder = opentelemetry_prometheus::exporter()
172            .with_registry(registry.clone())
173            .scope_info_enabled(config.scope_labels);
174        if !config.target_info {
175            builder = builder.without_target_info();
176        }
177        if !config.resource_labels.is_empty() {
178            let keys: std::collections::HashSet<opentelemetry::Key> = config
179                .resource_labels
180                .iter()
181                .map(|k| opentelemetry::Key::new(k.clone()))
182                .collect();
183            builder = builder.with_resource_selector(keys);
184        }
185        if let Some(ns) = &config.namespace {
186            builder = builder.with_namespace(ns.clone());
187        }
188        let reader = builder.build().map_err(|e| Error::Build(e.to_string()))?;
189        Ok((reader, Self { registry, config }))
190    }
191
192    /// The configuration this exporter was built with.
193    pub fn config(&self) -> &PrometheusConfig {
194        &self.config
195    }
196
197    /// Collects and encodes the current metrics in the Prometheus text
198    /// format (0.0.4). Synchronous and CPU-bound; prefer [`render_async`]
199    /// from async code.
200    ///
201    /// [`render_async`]: Self::render_async
202    pub fn render(&self) -> Result<String, Error> {
203        let families = self.registry.gather();
204        let mut buffer = Vec::with_capacity(16 * 1024);
205        TextEncoder::new().encode(&families, &mut buffer)?;
206        Ok(String::from_utf8_lossy(&buffer).into_owned())
207    }
208
209    /// [`render`](Self::render) on Tokio's blocking pool.
210    pub async fn render_async(&self) -> Result<String, Error> {
211        let this = self.clone();
212        match tokio::task::spawn_blocking(move || this.render()).await {
213            Ok(result) => result,
214            Err(join) => Err(Error::Io(std::io::Error::other(join))),
215        }
216    }
217
218    /// A router answering `GET <path>`. It carries no Kerniq middleware, so
219    /// merging it after the application's middleware has been applied keeps
220    /// scrapes out of the HTTP metrics.
221    pub fn router<S>(&self) -> Router<S>
222    where
223        S: Clone + Send + Sync + 'static,
224    {
225        let exporter = self.clone();
226        Router::new().route(
227            &self.config.path,
228            get(move || {
229                let exporter = exporter.clone();
230                async move { exporter.scrape().await }
231            }),
232        )
233    }
234
235    async fn scrape(&self) -> Response {
236        match self.render_async().await {
237            Ok(body) => (
238                [(
239                    header::CONTENT_TYPE,
240                    TextEncoder::new().format_type().to_owned(),
241                )],
242                body,
243            )
244                .into_response(),
245            Err(err) => {
246                tracing::error!(error = %err, "failed to render Prometheus metrics");
247                StatusCode::INTERNAL_SERVER_ERROR.into_response()
248            }
249        }
250    }
251
252    /// Serves the metrics endpoint on `listener` until `shutdown` completes.
253    /// Used for [`PrometheusConfig::listen`]; usable directly for custom
254    /// setups.
255    pub async fn serve(
256        &self,
257        listener: tokio::net::TcpListener,
258        shutdown: impl Future<Output = ()> + Send + 'static,
259    ) -> Result<(), Error> {
260        let address = listener.local_addr()?;
261        tracing::info!(%address, path = %self.config.path, "Prometheus metrics listener started");
262        axum::serve(listener, self.router::<()>())
263            .with_graceful_shutdown(shutdown)
264            .await?;
265        Ok(())
266    }
267}
kerniq_observability_prometheus/lib.rs

kerniq_observability_prometheus/
lib.rs
