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
//! Connector-owned metric families, registered through the source's
//! [`Meter`] (namespace `s3`, role `source` → `spate_s3_source_*`).
//!
//! All handles are resolved once at `open` and cloned into the lanes and
//! fetchers; nothing resolves names on the record path. Counters are
//! incremented at batch/object/chunk boundaries per the taxonomy's
//! hot-path discipline.
use crate::lane_ctx::PoisonKind;
use spate_core::metrics::{Counter, Gauge, Meter};
/// Pre-registered `spate_s3_source_*` handles.
#[derive(Clone)]
pub(crate) struct S3Metrics {
/// Objects, or byte ranges of one, fully framed and handed to the
/// pipeline.
pub(crate) objects_completed: Counter,
/// Objects not yet completed across this worker's currently-held
/// units, a cut object counting once per range (rises by the objects a
/// lane reads when it opens, falls per completed object, settles when the
/// lane closes).
pub(crate) objects_remaining: Gauge,
/// Bytes read from the store (as stored, pre-decompression).
pub(crate) bytes_read: Counter,
/// Bytes after decompression, as the framer receives them. For a plain
/// object read whole this equals `bytes_read`; the bytes a byte-range
/// read trims count in `bytes_read` alone.
pub(crate) bytes_decoded: Counter,
/// Object GET and HEAD attempts beyond the first (transient failures).
pub(crate) get_retries: Counter,
/// Keys whose read an overwrite cut short in a notification unit, once
/// per key and delivery, whether or not the key was re-enqueued.
pub(crate) objects_overwritten: Counter,
/// Of those, the keys acknowledged at the re-enqueue cap without a
/// re-enqueue.
pub(crate) overwrites_dropped: Counter,
/// Objects, or byte ranges of one, that failed their unit, by bounded
/// reason, with one pre-registered handle per [`PoisonKind`].
objects_failed_not_found: Counter,
objects_failed_etag_drift: Counter,
objects_failed_undecodable: Counter,
objects_failed_retries_exhausted: Counter,
}
impl std::fmt::Debug for S3Metrics {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("S3Metrics")
}
}
impl S3Metrics {
/// Resolve every family under the runtime-minted scope. Build-time
/// only.
pub(crate) fn new(meter: &Meter) -> S3Metrics {
let failed = |reason: &'static str| {
meter.counter("objects_failed_total", &[("reason", reason.into())])
};
S3Metrics {
objects_completed: meter.counter("objects_completed_total", &[]),
objects_remaining: meter.gauge("objects_remaining", &[]),
bytes_read: meter.counter("bytes_read_total", &[]),
bytes_decoded: meter.counter("bytes_decoded_total", &[]),
get_retries: meter.counter("get_retries_total", &[]),
objects_overwritten: meter.counter("objects_overwritten_total", &[]),
overwrites_dropped: meter.counter("overwrites_dropped_total", &[]),
objects_failed_not_found: failed(PoisonKind::NotFound.reason_label()),
objects_failed_etag_drift: failed(PoisonKind::EtagDrift.reason_label()),
objects_failed_undecodable: failed(PoisonKind::Undecodable.reason_label()),
objects_failed_retries_exhausted: failed(PoisonKind::RetriesExhausted.reason_label()),
}
}
/// The pre-registered failure counter for `kind` (never resolves a
/// name at increment time).
pub(crate) fn objects_failed(&self, kind: PoisonKind) -> &Counter {
match kind {
PoisonKind::NotFound => &self.objects_failed_not_found,
PoisonKind::EtagDrift => &self.objects_failed_etag_drift,
PoisonKind::Undecodable => &self.objects_failed_undecodable,
PoisonKind::RetriesExhausted => &self.objects_failed_retries_exhausted,
}
}
}