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 105 : pub fn enabled() -> bool {
35 105 : std::env::var("ANTARES_TELEMETRY").is_ok_and(|v| !crate::is_off(&v))
36 105 : }
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 57 : 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 57 : let env_filter =
87 57 : tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into());
88 57 : 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 57 : 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 57 : _ => (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 57 : let console: Option<tracing_subscriber::layer::Identity> = None;
151 :
152 57 : tracing_subscriber::registry()
153 57 : .with(env_filter)
154 57 : .with(fmt)
155 57 : .with(otlp)
156 57 : .with(logs)
157 57 : .with(console)
158 57 : .init();
159 :
160 57 : 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 57 : 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 57 : }
171 57 : 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 57 : }
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_policy_failures_total",
215 : Unit::Count,
216 : "policy decisions the seam had to make itself because the engine did not, by reason"
217 : );
218 0 : describe_counter!(
219 : "antares_notifications_sent_total",
220 : Unit::Count,
221 : "notifications delivered successfully, by sink scheme"
222 : );
223 0 : describe_counter!(
224 : "antares_notifications_failed_total",
225 : Unit::Count,
226 : "notification deliveries that failed, by sink scheme"
227 : );
228 0 : describe_histogram!(
229 : "antares_change_lag_seconds",
230 : Unit::Seconds,
231 : "bus=nats: change-event age (stream publish -> matcher processing)"
232 : );
233 0 : describe_gauge!(
234 : "antares_draining",
235 : Unit::Count,
236 : "1 while this instance drains — a roll is visible on a dashboard"
237 : );
238 0 : describe_gauge!(
239 : "antares_uptime_seconds",
240 : Unit::Seconds,
241 : "seconds since process start"
242 : );
243 0 : describe_gauge!(
244 : "antares_memory_allocated_bytes",
245 : Unit::Bytes,
246 : "jemalloc allocated (live) bytes"
247 : );
248 0 : describe_gauge!(
249 : "antares_memory_resident_bytes",
250 : Unit::Bytes,
251 : "jemalloc resident bytes (RSS ~ live x1.2 is the section-2.1 target)"
252 : );
253 0 : describe_gauge!(
254 : "antares_commit_queue_depth",
255 : Unit::Count,
256 : "file mode: writers queued behind the single redb committer"
257 : );
258 0 : describe_histogram!(
259 : "antares_pg_transaction_begin_seconds",
260 : Unit::Seconds,
261 : "postgres: time to obtain a pooled connection and open a transaction \
262 : — the pool wait plus one BEGIN round trip, so this is where pool \
263 : pressure shows"
264 : );
265 0 : describe_counter!(
266 : "antares_pg_pool_timeouts_total",
267 : Unit::Count,
268 : "postgres: acquire timeouts — the request was answered 503 with Retry-After"
269 : );
270 0 : describe_gauge!(
271 : "antares_limit_rejections_total",
272 : Unit::Count,
273 : "bounds-wall rejections, by limit"
274 : );
275 0 : }
276 :
277 : /// The 5 s gauge sampler: process-level state that has no natural
278 : /// increment site. Spawned once per process from `run`. With the switch
279 : /// off there is no recorder to feed — no task is spawned at all.
280 12 : pub fn spawn_sampler(state: antares_api::AppState) {
281 12 : if !enabled() {
282 12 : return;
283 0 : }
284 0 : tokio::spawn(async move {
285 0 : let mut tick = tokio::time::interval(std::time::Duration::from_secs(5));
286 : loop {
287 0 : tick.tick().await;
288 0 : metrics::gauge!("antares_uptime_seconds").set(state.started.elapsed().as_secs_f64());
289 0 : metrics::gauge!("antares_draining").set(f64::from(u8::from(
290 0 : state.draining.load(std::sync::atomic::Ordering::Relaxed),
291 : )));
292 0 : if let Some(mem) = &state.mem_stats {
293 0 : let m = mem();
294 0 : if let Some(a) = m.get("allocatedBytes").and_then(serde_json::Value::as_u64) {
295 0 : metrics::gauge!("antares_memory_allocated_bytes").set(a as f64);
296 0 : }
297 0 : if let Some(r) = m.get("residentBytes").and_then(serde_json::Value::as_u64) {
298 0 : metrics::gauge!("antares_memory_resident_bytes").set(r as f64);
299 0 : }
300 0 : }
301 0 : if let Some((depth, _peak)) = state.store.commit_queue() {
302 0 : metrics::gauge!("antares_commit_queue_depth").set(depth as f64);
303 0 : }
304 : // Limit counters live in LimitStats (incremented at rejection
305 : // sites); exported here so the wall is observable BEFORE users
306 : // hit it.
307 0 : if let Some(map) = state.limits.snapshot().as_object() {
308 0 : for (key, n) in map {
309 0 : if let Some(limit) = key.strip_prefix("rejected") {
310 0 : if let Some(n) = n.as_u64() {
311 0 : metrics::gauge!(
312 0 : "antares_limit_rejections_total",
313 0 : "limit" => limit.to_owned()
314 0 : )
315 0 : .set(n as f64);
316 0 : }
317 0 : }
318 : }
319 0 : }
320 : }
321 : });
322 12 : }
323 : #[cfg(test)]
324 : mod logs_endpoint_tests {
325 : #[test]
326 2 : fn traces_suffix_becomes_logs_anything_else_is_kept() {
327 2 : assert_eq!(
328 2 : super::logs_endpoint("http://c:4318/v1/traces"),
329 : "http://c:4318/v1/logs"
330 : );
331 2 : assert_eq!(
332 2 : super::logs_endpoint("http://c:4318/otlp"),
333 : "http://c:4318/otlp"
334 : );
335 2 : }
336 : }
337 :
338 : #[cfg(test)]
339 : mod tests {
340 : use super::*;
341 :
342 : /// The switch is the whole stack's gate, and the environment is
343 : /// process-global — every spelling is asserted in ONE test. The DEFAULT
344 : /// is off, so only an explicit off value may keep it off: a knob that
345 : /// recognized `1|true|on` alone disabled the whole observability stack on
346 : /// `TRUE` without a word.
347 : #[test]
348 2 : fn telemetry_switch_is_off_by_default_and_tolerant_of_spelling() {
349 2 : std::env::remove_var("ANTARES_TELEMETRY");
350 2 : assert!(!enabled(), "the default must allocate nothing");
351 16 : for on in ["1", "true", "on", "TRUE", "On", "yes", "1 ", " 1"] {
352 16 : std::env::set_var("ANTARES_TELEMETRY", on);
353 16 : assert!(enabled(), "ANTARES_TELEMETRY={on:?} must arm the stack");
354 : }
355 18 : for off in ["0", "false", "off", "", " ", "FALSE", "Off", "no", " 0 "] {
356 18 : std::env::set_var("ANTARES_TELEMETRY", off);
357 18 : assert!(
358 18 : !enabled(),
359 : "ANTARES_TELEMETRY={off:?} must NOT arm the stack"
360 : );
361 : }
362 2 : std::env::remove_var("ANTARES_TELEMETRY");
363 2 : }
364 :
365 : /// A collector endpoint is a URL and may carry `user:password@` userinfo
366 : /// (RFC 3986 clause 3.2.1). It is logged at startup, so the credential must be
367 : /// stripped before it reaches the log.
368 : #[test]
369 2 : fn otlp_endpoint_userinfo_never_reaches_the_log() {
370 2 : let redacted = redact_url("http://otel:s3cr3t@collector.internal:4318/v1/traces");
371 2 : assert!(
372 2 : !redacted.contains("s3cr3t") && !redacted.contains("otel:"),
373 : "userinfo leaked into the log line: {redacted}"
374 : );
375 2 : assert!(
376 2 : redacted.contains("collector.internal:4318/v1/traces"),
377 : "the useful part of the endpoint must survive: {redacted}"
378 : );
379 : // No userinfo: byte-identical, including an '@' that belongs to the
380 : // path rather than the authority.
381 8 : for plain in [
382 2 : "http://collector:4318/v1/traces",
383 2 : "https://collector:4318/v1/@traces",
384 2 : "collector:4318",
385 2 : "",
386 2 : ] {
387 8 : assert_eq!(redact_url(plain), plain, "rewrote a credential-free URL");
388 : }
389 2 : }
390 :
391 : /// An exporter that refuses its endpoint prints the endpoint back
392 : /// verbatim, so the credential survives the build failure and lands in
393 : /// the startup error `main` prints — the leak `redact_url` closes on the
394 : /// two paths that succeed. Both exporters are built from the same
395 : /// operator-supplied URL, so both have to lose it.
396 : #[test]
397 2 : fn a_rejected_collector_endpoint_loses_its_password() {
398 : use opentelemetry_otlp::WithExportConfig as _;
399 : // A space is not a legal URI character, so the build fails with no
400 : // collector to reach — and the rejected endpoint rides the error.
401 2 : let endpoint = "http://otel:s3cr3t@collector .internal:4318/v1/traces";
402 2 : let Err(spans) = opentelemetry_otlp::SpanExporter::builder()
403 2 : .with_http()
404 2 : .with_endpoint(endpoint)
405 2 : .build()
406 : else {
407 0 : panic!("a URI carrying a space is rejected")
408 : };
409 2 : let logs_endpoint = logs_endpoint(endpoint);
410 2 : let Err(logs) = opentelemetry_otlp::LogExporter::builder()
411 2 : .with_http()
412 2 : .with_endpoint(logs_endpoint.clone())
413 2 : .build()
414 : else {
415 0 : panic!("a URI carrying a space is rejected")
416 : };
417 4 : for (raw, needle) in [
418 2 : (spans.to_string(), endpoint),
419 2 : (logs.to_string(), logs_endpoint.as_str()),
420 2 : ] {
421 4 : let safe = no_userinfo(raw, needle);
422 4 : assert!(!safe.contains("s3cr3t"), "the password reached it: {safe}");
423 4 : assert!(!safe.contains("otel:"), "the userinfo reached it: {safe}");
424 4 : assert!(
425 4 : safe.contains("collector"),
426 : "the destination stays readable: {safe}"
427 : );
428 : }
429 2 : }
430 :
431 : /// Metric label cardinality: the only labelled instrument this module
432 : /// feeds is the limit-rejection gauge, whose label comes from a fixed key
433 : /// set in the bounds snapshot — never from a client-controlled string
434 : /// (a tenant, a URI or a header would blow up the time-series count).
435 : #[test]
436 2 : fn limit_rejection_labels_are_a_closed_identifier_set() {
437 2 : let snapshot = antares_api::bounds::LimitStats::default().snapshot();
438 2 : let labels: Vec<String> = snapshot
439 2 : .as_object()
440 2 : .expect("snapshot is an object")
441 2 : .keys()
442 50 : .filter_map(|k| k.strip_prefix("rejected").map(str::to_owned))
443 2 : .collect();
444 2 : assert_eq!(
445 2 : labels.len(),
446 : 3,
447 : "the rejection label set changed — keep it closed and bounded: {labels:?}"
448 : );
449 6 : for l in &labels {
450 6 : assert!(
451 66 : l.chars().all(|c| c.is_ascii_alphanumeric()),
452 : "label value {l:?} is not a fixed identifier"
453 : );
454 : }
455 2 : }
456 : }
|
