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
//! Backpressure handles (`spate_backpressure_*`).
use super::MetricsError;
use super::labels::{ComponentLabels, OwnedGauge};
use super::names;
use super::ownership::{SeriesClaim, series_key};
use metrics::Counter;
use std::time::Duration;
/// Backpressure handles (`spate_backpressure_*`).
#[derive(Debug)]
pub struct BackpressureMetrics {
paused: OwnedGauge,
paused_seconds: OwnedGauge,
pause_events: Counter,
inflight_bytes: OwnedGauge,
_claim: Option<SeriesClaim>,
}
impl BackpressureMetrics {
/// Resolve all backpressure handles.
///
/// Claims the `spate_backpressure_*` series for these labels; if another
/// live handle set already owns them this one logs and becomes a shadow,
/// counting but publishing no gauge (see "Series ownership" in
/// `docs/METRICS.md`). Each pipeline thread's controller gets its own
/// `component` label, so they own separate series rather than sharing one.
pub fn new(labels: &ComponentLabels) -> Self {
Self::build(labels, SeriesClaim::claim_or_shadow(Self::key(labels)))
}
/// Resolve all backpressure handles, failing when another live handle set
/// already owns the series. The pipeline runtime's path.
///
/// # Errors
///
/// [`MetricsError::DuplicateSeries`] on a collision.
pub fn try_new(labels: &ComponentLabels) -> Result<Self, MetricsError> {
let claim = SeriesClaim::try_claim(Self::key(labels))?;
Ok(Self::build(labels, Some(claim)))
}
fn key(labels: &ComponentLabels) -> String {
series_key("backpressure", labels, "")
}
fn build(labels: &ComponentLabels, claim: Option<SeriesClaim>) -> Self {
let owned = claim.is_some();
BackpressureMetrics {
paused: OwnedGauge::new(labels.gauge(names::BACKPRESSURE_PAUSED), owned),
paused_seconds: OwnedGauge::new(
labels.gauge(names::BACKPRESSURE_PAUSED_SECONDS_TOTAL),
owned,
),
pause_events: labels.counter(names::BACKPRESSURE_PAUSE_EVENTS_TOTAL),
inflight_bytes: OwnedGauge::new(
labels.gauge(names::BACKPRESSURE_INFLIGHT_BYTES),
owned,
),
_claim: claim,
}
}
/// Record a pause transition.
pub fn pause_started(&self) {
self.paused.set(1.0);
self.pause_events.increment(1);
}
/// Record a resume transition and the time spent paused.
pub fn pause_ended(&self, paused_for: Duration) {
self.paused.set(0.0);
// Monotonic accumulator; gauge because the facade counter is
// integer-only (see names.rs).
self.paused_seconds.increment(paused_for.as_secs_f64());
}
/// Set the current in-flight byte budget usage.
#[inline]
pub fn set_inflight_bytes(&self, bytes: usize) {
self.inflight_bytes.set(bytes as f64);
}
}