Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Tracing + Prometheus metrics + env-gated OTLP span export.
3 : //!
4 : //! Naming follows Prometheus conventions with the `antares_` prefix and
5 : //! unit suffixes. The `metrics` facade is what core crates speak; THIS
6 : //! module is the only place an exporter exists (only the
7 : //! composition root knows). `/q/metrics` renders the Prometheus text
8 : //! format via the closure wired onto AppState.
9 : //!
10 : //! ALL of it sits behind a RUNTIME switch: ANTARES_TELEMETRY=1 installs
11 : //! the recorder, the sampler and (with ANTARES_OTLP_ENDPOINT) the OTLP
12 : //! pipeline at startup; the default constructs NONE of it — `metrics::`
13 : //! macro calls no-op without a recorder (zero allocations), and
14 : //! /q/metrics answers 404. One build, lean by default; flip the
15 : //! env and restart where a dashboard actually scrapes.
16 : //!
17 : //! OTLP: set ANTARES_OTLP_ENDPOINT (e.g. http://collector:4318/v1/traces)
18 : //! and spans flow out over OTLP/HTTP, log records to its `v1/logs` twin
19 : //! with the same resource; unset (the default) costs nothing.
20 : //! tokio-console: cargo feature `console` + RUSTFLAGS="--cfg tokio_unstable"
21 : //! (the layer only arms when BOTH are present — an --all-features build
22 : //! without the RUSTFLAGS must not panic at startup).
23 :
24 : use std::sync::Arc;
25 :
26 : /// What /q/metrics renders through — exporter type erased so `main` builds
27 : /// identically with and without the `telemetry` feature.
28 : pub type MetricsRender = Arc<dyn Fn() -> String + Send + Sync>;
29 :
30 : /// Is the observability stack switched on for this process? The default is
31 : /// off, so ANY value that is not an explicit off spelling arms it — a knob
32 : /// that recognized only `1|true|on` disabled the whole stack on `TRUE`
33 : /// without a word.
34 107 : pub fn enabled() -> bool {
35 107 : std::env::var("ANTARES_TELEMETRY").is_ok_and(|v| !crate::is_off(&v))
36 107 : }
37 :
38 : /// Strip `user:password@` userinfo (RFC 3986 clause 3.2.1) out of a URL before
39 : /// it is logged. A string without an authority component — no scheme, or an
40 : /// `@` that belongs to the path — comes back byte-identical.
41 14 : fn redact_url(url: &str) -> String {
42 14 : let Some((scheme, rest)) = url.split_once("://") else {
43 4 : return url.to_owned();
44 : };
45 10 : let (authority, path) = match rest.find('/') {
46 10 : Some(i) => rest.split_at(i),
47 0 : None => (rest, ""),
48 : };
49 10 : match authority.rsplit_once('@') {
50 6 : Some((_, host)) => format!("{scheme}://{host}{path}"),
51 4 : None => url.to_owned(),
52 : }
53 14 : }
54 :
55 : /// An exporter that refuses its endpoint prints the endpoint back verbatim
56 : /// (`invalid URI {0}`), so a rejected URL carrying `user:password@` userinfo
57 : /// (RFC 3986 clause 3.2.1) reaches the startup error the way it would reach a
58 : /// log line. The endpoint is known at the call site, so the failure carries
59 : /// its redacted form instead.
60 4 : fn no_userinfo(message: String, endpoint: &str) -> String {
61 4 : let safe = redact_url(endpoint);
62 4 : if safe == endpoint {
63 0 : message
64 : } else {
65 4 : message.replace(endpoint, &safe)
66 : }
67 4 : }
68 :
69 : /// The OTLP/HTTP logs endpoint paired with a traces endpoint: the standard
70 : /// `v1/traces` suffix becomes `v1/logs`; any other URL is used as given.
71 6 : fn logs_endpoint(traces: &str) -> String {
72 6 : match traces.strip_suffix("/v1/traces") {
73 4 : Some(base) => format!("{base}/v1/logs"),
74 2 : None => traces.to_owned(),
75 : }
76 6 : }
77 :
78 : /// Install the tracing subscriber stack and (ANTARES_TELEMETRY=1) the
79 : /// Prometheus recorder. Call once, before the runtime spins up anything
80 : /// measurable. Returns the /q/metrics render closure, or None when the
81 : /// switch is off — in which case nothing telemetry-shaped is allocated.
82 59 : pub fn init() -> Result<Option<MetricsRender>, Box<dyn std::error::Error>> {
83 : use tracing_subscriber::layer::SubscriberExt;
84 : use tracing_subscriber::util::SubscriberInitExt;
85 :
86 59 : let env_filter =
87 59 : tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into());
88 59 : let fmt = tracing_subscriber::fmt::layer();
89 :
90 : // The collector endpoint is a URL and may carry `user:password@` userinfo
91 : // (RFC 3986 clause 3.2.1); it is logged at startup, so the credential is
92 : // stripped first.
93 : // Env-gated OTLP pipeline — needs the switch AND an endpoint. Spans and
94 : // log records share one resource so a collector joins them.
95 59 : let (otlp, logs) = match std::env::var("ANTARES_OTLP_ENDPOINT") {
96 0 : Ok(endpoint) if enabled() => {
97 : use opentelemetry::trace::TracerProvider as _;
98 : use opentelemetry_otlp::WithExportConfig as _;
99 : use tracing_subscriber::Layer as _;
100 0 : let resource = opentelemetry_sdk::Resource::builder()
101 0 : .with_service_name("antares")
102 0 : .build();
103 0 : let exporter = opentelemetry_otlp::SpanExporter::builder()
104 0 : .with_http()
105 0 : .with_endpoint(endpoint.clone())
106 0 : .build()
107 0 : .map_err(|e| no_userinfo(e.to_string(), &endpoint))?;
108 0 : let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
109 0 : .with_batch_exporter(exporter)
110 0 : .with_resource(resource.clone())
111 0 : .build();
112 0 : let tracer = provider.tracer("antares");
113 : // Log records: batch exporter on its own thread with a bounded
114 : // queue, so a dead collector drops records instead of stalling
115 : // a request. The exporter's own HTTP stack is filtered out of
116 : // the bridge, else every export would log another export.
117 0 : let logs_endpoint = logs_endpoint(&endpoint);
118 0 : let log_exporter = opentelemetry_otlp::LogExporter::builder()
119 0 : .with_http()
120 0 : .with_endpoint(logs_endpoint.clone())
121 0 : .build()
122 0 : .map_err(|e| no_userinfo(e.to_string(), &logs_endpoint))?;
123 0 : let logger_provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder()
124 0 : .with_batch_exporter(log_exporter)
125 0 : .with_resource(resource)
126 0 : .build();
127 0 : let bridge = opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge::new(
128 0 : &logger_provider,
129 : )
130 0 : .with_filter(tracing_subscriber::filter::filter_fn(|m| {
131 0 : !["opentelemetry", "hyper", "reqwest", "h2", "tonic"]
132 0 : .iter()
133 0 : .any(|t| m.target().starts_with(t))
134 0 : }));
135 0 : tracing::info!(
136 0 : endpoint = redact_url(&endpoint),
137 : "OTLP span and log export enabled"
138 : );
139 0 : (
140 0 : Some(tracing_opentelemetry::layer().with_tracer(tracer)),
141 0 : Some(bridge),
142 0 : )
143 : }
144 59 : _ => (None, None),
145 : };
146 :
147 : #[cfg(all(feature = "console", tokio_unstable))]
148 : let console = Some(console_subscriber::spawn());
149 : #[cfg(not(all(feature = "console", tokio_unstable)))]
150 59 : let console: Option<tracing_subscriber::layer::Identity> = None;
151 :
152 59 : tracing_subscriber::registry()
153 59 : .with(env_filter)
154 59 : .with(fmt)
155 59 : .with(otlp)
156 59 : .with(logs)
157 59 : .with(console)
158 59 : .init();
159 :
160 59 : if !enabled() {
161 : // An endpoint without the switch is a configuration the operator
162 : // believes is exporting: say so rather than drop it silently. Logged
163 : // here because the subscriber above is what makes a log visible.
164 59 : if let Ok(endpoint) = std::env::var("ANTARES_OTLP_ENDPOINT") {
165 0 : tracing::warn!(
166 0 : endpoint = redact_url(&endpoint),
167 : "ANTARES_OTLP_ENDPOINT is set but ANTARES_TELEMETRY is off — \
168 : no spans or logs are exported"
169 : );
170 59 : }
171 59 : return Ok(None); // the recorder, registry and sampler are never built
172 0 : }
173 0 : let handle = metrics_exporter_prometheus::PrometheusBuilder::new()
174 0 : .set_buckets(LATENCY_BUCKETS_SECONDS)?
175 0 : .install_recorder()?;
176 0 : describe();
177 0 : Ok(Some(Arc::new(move || handle.render())))
178 59 : }
179 :
180 : /// Bucket bounds, in seconds, for every histogram this binary registers —
181 : /// both are request-shaped latencies.
182 : ///
183 : /// Without them the exporter renders a histogram as a rolling summary, whose
184 : /// quantiles are computed over a sliding window it owns rather than over the
185 : /// scrape: an idle window reports `0` for every quantile while `_count` and
186 : /// `_sum` keep climbing, a busy one reports only the last window, and neither
187 : /// can be aggregated across instances or read back over a rollout. Buckets
188 : /// are cumulative and belong to the scrape, so `histogram_quantile()` works
189 : /// on them.
190 : ///
191 : /// The range is set from measured service time: single-digit milliseconds
192 : /// when the broker is healthy, tens of seconds once the accept path is the
193 : /// bottleneck. The exporter's own default stops at 10 s, which files every
194 : /// slow request under `+Inf` — precisely the requests a latency dashboard
195 : /// exists to show.
196 : const LATENCY_BUCKETS_SECONDS: &[f64] = &[
197 : 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0,
198 : ];
199 :
200 : /// Metric metadata — Prometheus-convention names (antares_ prefix, unit suffixes).
201 0 : fn describe() {
202 : use metrics::{describe_counter, describe_gauge, describe_histogram, Unit};
203 0 : describe_counter!(
204 : "antares_http_requests_total",
205 : Unit::Count,
206 : "HTTP requests served, by method and status class"
207 : );
208 0 : describe_histogram!(
209 : "antares_http_request_duration_seconds",
210 : Unit::Seconds,
211 : "HTTP request service time"
212 : );
213 0 : describe_counter!(
214 : "antares_change_final_delivery_total",
215 : Unit::Count,
216 : "change messages handed to the matcher on their last allowed delivery; one that then fails is dropped by the server"
217 : );
218 0 : describe_counter!(
219 : "antares_policy_failures_total",
220 : Unit::Count,
221 : "policy decisions the seam had to make itself because the engine did not, by reason"
222 : );
223 0 : describe_counter!(
224 : "antares_notifications_sent_total",
225 : Unit::Count,
226 : "notifications delivered successfully, by sink scheme"
227 : );
228 0 : describe_counter!(
229 : "antares_notifications_failed_total",
230 : Unit::Count,
231 : "notification deliveries that failed, by sink scheme"
232 : );
233 0 : describe_histogram!(
234 : "antares_change_lag_seconds",
235 : Unit::Seconds,
236 : "bus=nats: change-event age (stream publish -> matcher processing)"
237 : );
238 0 : describe_gauge!(
239 : "antares_draining",
240 : Unit::Count,
241 : "1 while this instance drains — a roll is visible on a dashboard"
242 : );
243 0 : describe_gauge!(
244 : "antares_uptime_seconds",
245 : Unit::Seconds,
246 : "seconds since process start"
247 : );
248 0 : describe_gauge!(
249 : "antares_memory_allocated_bytes",
250 : Unit::Bytes,
251 : "jemalloc allocated (live) bytes"
252 : );
253 0 : describe_gauge!(
254 : "antares_memory_resident_bytes",
255 : Unit::Bytes,
256 : "jemalloc resident bytes (RSS ~ live x1.2 is the section-2.1 target)"
257 : );
258 0 : describe_gauge!(
259 : "antares_commit_queue_depth",
260 : Unit::Count,
261 : "file mode: writers queued behind the single redb committer"
262 : );
263 0 : describe_histogram!(
264 : "antares_pg_transaction_begin_seconds",
265 : Unit::Seconds,
266 : "postgres: time to obtain a pooled connection and open a transaction \
267 : — the pool wait plus one BEGIN round trip, so this is where pool \
268 : pressure shows"
269 : );
270 0 : describe_counter!(
271 : "antares_pg_pool_timeouts_total",
272 : Unit::Count,
273 : "postgres: acquire timeouts — the request was answered 503 with Retry-After"
274 : );
275 0 : describe_gauge!(
276 : "antares_limit_rejections_total",
277 : Unit::Count,
278 : "bounds-wall rejections, by limit"
279 : );
280 0 : }
281 :
282 : /// The 5 s gauge sampler: process-level state that has no natural
283 : /// increment site. Spawned once per process from `run`. With the switch
284 : /// off there is no recorder to feed — no task is spawned at all.
285 12 : pub fn spawn_sampler(state: antares_api::AppState) {
286 12 : if !enabled() {
287 12 : return;
288 0 : }
289 0 : tokio::spawn(async move {
290 0 : let mut tick = tokio::time::interval(std::time::Duration::from_secs(5));
291 : loop {
292 0 : tick.tick().await;
293 0 : metrics::gauge!("antares_uptime_seconds").set(state.started.elapsed().as_secs_f64());
294 0 : metrics::gauge!("antares_draining").set(f64::from(u8::from(
295 0 : state.draining.load(std::sync::atomic::Ordering::Relaxed),
296 : )));
297 0 : if let Some(mem) = &state.mem_stats {
298 0 : let m = mem();
299 0 : if let Some(a) = m.get("allocatedBytes").and_then(serde_json::Value::as_u64) {
300 0 : metrics::gauge!("antares_memory_allocated_bytes").set(a as f64);
301 0 : }
302 0 : if let Some(r) = m.get("residentBytes").and_then(serde_json::Value::as_u64) {
303 0 : metrics::gauge!("antares_memory_resident_bytes").set(r as f64);
304 0 : }
305 0 : }
306 0 : if let Some((depth, _peak)) = state.store.commit_queue() {
307 0 : metrics::gauge!("antares_commit_queue_depth").set(depth as f64);
308 0 : }
309 : // Limit counters live in LimitStats (incremented at rejection
310 : // sites); exported here so the wall is observable BEFORE users
311 : // hit it.
312 0 : if let Some(map) = state.limits.snapshot().as_object() {
313 0 : for (key, n) in map {
314 0 : if let Some(limit) = key.strip_prefix("rejected") {
315 0 : if let Some(n) = n.as_u64() {
316 0 : metrics::gauge!(
317 0 : "antares_limit_rejections_total",
318 0 : "limit" => limit.to_owned()
319 0 : )
320 0 : .set(n as f64);
321 0 : }
322 0 : }
323 : }
324 0 : }
325 : }
326 : });
327 12 : }
328 : #[cfg(test)]
329 : mod logs_endpoint_tests {
330 : #[test]
331 2 : fn traces_suffix_becomes_logs_anything_else_is_kept() {
332 2 : assert_eq!(
333 2 : super::logs_endpoint("http://c:4318/v1/traces"),
334 : "http://c:4318/v1/logs"
335 : );
336 2 : assert_eq!(
337 2 : super::logs_endpoint("http://c:4318/otlp"),
338 : "http://c:4318/otlp"
339 : );
340 2 : }
341 : }
342 :
343 : #[cfg(test)]
344 : mod tests {
345 : use super::*;
346 :
347 : /// The switch is the whole stack's gate, and the environment is
348 : /// process-global — every spelling is asserted in ONE test. The DEFAULT
349 : /// is off, so only an explicit off value may keep it off: a knob that
350 : /// recognized `1|true|on` alone disabled the whole observability stack on
351 : /// `TRUE` without a word.
352 : #[test]
353 2 : fn telemetry_switch_is_off_by_default_and_tolerant_of_spelling() {
354 2 : std::env::remove_var("ANTARES_TELEMETRY");
355 2 : assert!(!enabled(), "the default must allocate nothing");
356 16 : for on in ["1", "true", "on", "TRUE", "On", "yes", "1 ", " 1"] {
357 16 : std::env::set_var("ANTARES_TELEMETRY", on);
358 16 : assert!(enabled(), "ANTARES_TELEMETRY={on:?} must arm the stack");
359 : }
360 18 : for off in ["0", "false", "off", "", " ", "FALSE", "Off", "no", " 0 "] {
361 18 : std::env::set_var("ANTARES_TELEMETRY", off);
362 18 : assert!(
363 18 : !enabled(),
364 : "ANTARES_TELEMETRY={off:?} must NOT arm the stack"
365 : );
366 : }
367 2 : std::env::remove_var("ANTARES_TELEMETRY");
368 2 : }
369 :
370 : /// A collector endpoint is a URL and may carry `user:password@` userinfo
371 : /// (RFC 3986 clause 3.2.1). It is logged at startup, so the credential must be
372 : /// stripped before it reaches the log.
373 : #[test]
374 2 : fn otlp_endpoint_userinfo_never_reaches_the_log() {
375 2 : let redacted = redact_url("http://otel:s3cr3t@collector.internal:4318/v1/traces");
376 2 : assert!(
377 2 : !redacted.contains("s3cr3t") && !redacted.contains("otel:"),
378 : "userinfo leaked into the log line: {redacted}"
379 : );
380 2 : assert!(
381 2 : redacted.contains("collector.internal:4318/v1/traces"),
382 : "the useful part of the endpoint must survive: {redacted}"
383 : );
384 : // No userinfo: byte-identical, including an '@' that belongs to the
385 : // path rather than the authority.
386 8 : for plain in [
387 2 : "http://collector:4318/v1/traces",
388 2 : "https://collector:4318/v1/@traces",
389 2 : "collector:4318",
390 2 : "",
391 2 : ] {
392 8 : assert_eq!(redact_url(plain), plain, "rewrote a credential-free URL");
393 : }
394 2 : }
395 :
396 : /// An exporter that refuses its endpoint prints the endpoint back
397 : /// verbatim, so the credential survives the build failure and lands in
398 : /// the startup error `main` prints — the leak `redact_url` closes on the
399 : /// two paths that succeed. Both exporters are built from the same
400 : /// operator-supplied URL, so both have to lose it.
401 : #[test]
402 2 : fn a_rejected_collector_endpoint_loses_its_password() {
403 : use opentelemetry_otlp::WithExportConfig as _;
404 : // A space is not a legal URI character, so the build fails with no
405 : // collector to reach — and the rejected endpoint rides the error.
406 2 : let endpoint = "http://otel:s3cr3t@collector .internal:4318/v1/traces";
407 2 : let Err(spans) = opentelemetry_otlp::SpanExporter::builder()
408 2 : .with_http()
409 2 : .with_endpoint(endpoint)
410 2 : .build()
411 : else {
412 0 : panic!("a URI carrying a space is rejected")
413 : };
414 2 : let logs_endpoint = logs_endpoint(endpoint);
415 2 : let Err(logs) = opentelemetry_otlp::LogExporter::builder()
416 2 : .with_http()
417 2 : .with_endpoint(logs_endpoint.clone())
418 2 : .build()
419 : else {
420 0 : panic!("a URI carrying a space is rejected")
421 : };
422 4 : for (raw, needle) in [
423 2 : (spans.to_string(), endpoint),
424 2 : (logs.to_string(), logs_endpoint.as_str()),
425 2 : ] {
426 4 : let safe = no_userinfo(raw, needle);
427 4 : assert!(!safe.contains("s3cr3t"), "the password reached it: {safe}");
428 4 : assert!(!safe.contains("otel:"), "the userinfo reached it: {safe}");
429 4 : assert!(
430 4 : safe.contains("collector"),
431 : "the destination stays readable: {safe}"
432 : );
433 : }
434 2 : }
435 :
436 : /// Metric label cardinality: the only labelled instrument this module
437 : /// feeds is the limit-rejection gauge, whose label comes from a fixed key
438 : /// set in the bounds snapshot — never from a client-controlled string
439 : /// (a tenant, a URI or a header would blow up the time-series count).
440 : #[test]
441 2 : fn limit_rejection_labels_are_a_closed_identifier_set() {
442 2 : let snapshot = antares_api::bounds::LimitStats::default().snapshot();
443 2 : let labels: Vec<String> = snapshot
444 2 : .as_object()
445 2 : .expect("snapshot is an object")
446 2 : .keys()
447 68 : .filter_map(|k| k.strip_prefix("rejected").map(str::to_owned))
448 2 : .collect();
449 2 : assert_eq!(
450 2 : labels.len(),
451 : 3,
452 : "the rejection label set changed — keep it closed and bounded: {labels:?}"
453 : );
454 6 : for l in &labels {
455 6 : assert!(
456 66 : l.chars().all(|c| c.is_ascii_alphanumeric()),
457 : "label value {l:?} is not a fixed identifier"
458 : );
459 : }
460 2 : }
461 : }
|
