Skip to content

Commit be0955c

Browse files
committed
feat(metrique-util): prototype request-scoped metrics pools
1 parent b986459 commit be0955c

9 files changed

Lines changed: 1614 additions & 2 deletions

File tree

Cargo.lock

Lines changed: 4 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

metrique-util/Cargo.toml

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)', 'cfg(shu
1414
[features]
1515
default = []
1616
state = ["dep:arc-swap"]
17+
metrics-pool = ["dep:Inflector", "dep:metrique", "dep:tracing"]
1718
tokio-metrics-bridge = [
1819
"state",
1920
"dep:metrique",
@@ -31,6 +32,7 @@ _shuttle = ["dep:shuttle", "pending-sink", "metrique-writer-core/_shuttle"]
3132
[dependencies]
3233
metrique-core = { workspace = true }
3334
metrique = { workspace = true, optional = true }
35+
Inflector = { workspace = true, optional = true }
3436
arc-swap = { version = "1", optional = true }
3537
tokio = { workspace = true, optional = true, features = ["time", "rt"] }
3638
tokio-metrics = { version = "0.5.0", optional = true, features = ["rt", "metrique-integration"] }
@@ -47,14 +49,21 @@ shuttle = { workspace = true, optional = true }
4749

4850
[dev-dependencies]
4951
assert2 = { workspace = true }
52+
aws-smithy-runtime-api = { version = "1", features = ["client"] }
53+
aws-smithy-types = "1"
54+
divan = "0.1"
5055
metrique = { workspace = true, features = ["emf", "test-util", "service-metrics"] }
5156
metrique-writer = { workspace = true, features = ["test-util"] }
5257
metrique-writer-core = { workspace = true, features = ["test-util"] }
5358
tokio = { workspace = true, features = ["full", "test-util"] }
5459
rstest = { workspace = true }
55-
metrique-util = { path = ".", features = ["state", "pending-sink", "sysinfo-bridge", "tokio-metrics-bridge"] }
60+
metrique-util = { path = ".", features = ["metrics-pool", "state", "pending-sink", "sysinfo-bridge", "tokio-metrics-bridge"] }
5661
tracing-subscriber = { workspace = true }
5762

63+
[[example]]
64+
name = "sdk-interceptor-metrics-pool"
65+
required-features = ["metrics-pool"]
66+
5867
[[example]]
5968
name = "global-state"
6069
required-features = ["state", "sysinfo-bridge", "tokio-metrics-bridge"]
@@ -83,6 +92,11 @@ name = "sysinfo-folded-static"
8392
required-features = ["sysinfo-bridge"]
8493
doc-scrape-examples = false
8594

95+
[[bench]]
96+
name = "metrics_pool"
97+
harness = false
98+
required-features = ["metrics-pool"]
99+
86100
[package.metadata.docs.rs]
87101
all-features = true
88102
targets = ["x86_64-unknown-linux-gnu"]

metrique-util/README.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ Additional utilities for [metrique].
55
## Features
66

77
- `state`: Provides [`State<T>`], an atomically swappable shared value with snapshot-on-first-read semantics. Useful for shared runtime state (feature flags, config reloads, routing tables) that should appear on every metric record.
8+
- `metrics-pool`: Provides [`MetricsPool`], which collects independently-created metrics and flattens them into a parent metric entry. [`with_metrics_pool`] installs a pool while a future is polled.
89
- `tokio-metrics-bridge`: Subscribes [tokio-metrics] runtime snapshots to a global entry sink. The reporter task is automatically aborted when the `AttachHandle` is dropped.
910
- `sysinfo-bridge`: Subscribes [sysinfo] system and current-process snapshots to a global entry sink, capturing metrics like CPU usage, disk space, and network rx/tx. The reporter task is automatically aborted when the `AttachHandle` is dropped.
1011
- `pending-sink`: Provides [`pending_sink::new()`], which creates a `(BoxEntrySink, PendingSinkResolver)` pair for deferred sink attachment with bounded buffering. Entries are buffered in a ring buffer until [`PendingSinkResolver::resolve`] drains them into the real sink and switches to direct forwarding. If the resolver is dropped without calling `resolve`, buffered entries are discarded and the sink becomes a no-op.
@@ -24,5 +25,7 @@ See the [metrique documentation] for the full framework.
2425
[metrique]: https://crates.io/crates/metrique
2526
[metrique documentation]: https://docs.rs/metrique
2627
[`State<T>`]: https://docs.rs/metrique-util/latest/metrique_util/state/struct.State.html
28+
[`MetricsPool`]: https://docs.rs/metrique-util/latest/metrique_util/struct.MetricsPool.html
29+
[`with_metrics_pool`]: https://docs.rs/metrique-util/latest/metrique_util/fn.with_metrics_pool.html
2730
[`pending_sink::new()`]: https://docs.rs/metrique-util/latest/metrique_util/pending_sink/fn.new.html
2831
[`PendingSinkResolver::resolve`]: https://docs.rs/metrique-util/latest/metrique_util/pending_sink/struct.PendingSinkResolver.html#method.resolve
Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
//! Benchmarks for `MetricsPool` type erasure and its two-pass write path.
5+
//!
6+
//! Run: `cargo bench -p metrique-util --bench metrics_pool --features metrics-pool`
7+
8+
use std::borrow::Cow;
9+
use std::time::SystemTime;
10+
11+
use divan::{Bencher, black_box};
12+
use metrique::unit_of_work::metrics;
13+
use metrique::writer::{EntryConfig, EntryWriter, Value};
14+
use metrique::{CloseValue, InflectableEntry, PascalCase};
15+
use metrique_util::MetricsPool;
16+
17+
#[global_allocator]
18+
static ALLOC: divan::AllocProfiler = divan::AllocProfiler::system();
19+
20+
fn main() {
21+
divan::main();
22+
}
23+
24+
const SIZES: &[usize] = &[1, 4, 16];
25+
const PREFIXES: &[&str] = &[
26+
"child_00", "child_01", "child_02", "child_03", "child_04", "child_05", "child_06", "child_07",
27+
"child_08", "child_09", "child_10", "child_11", "child_12", "child_13", "child_14", "child_15",
28+
];
29+
30+
#[metrics]
31+
struct ChildMetrics {
32+
count: u64,
33+
operation: &'static str,
34+
}
35+
36+
#[derive(Default)]
37+
struct CountingWriter {
38+
values: usize,
39+
}
40+
41+
impl<'a> EntryWriter<'a> for CountingWriter {
42+
fn timestamp(&mut self, _timestamp: SystemTime) {}
43+
44+
fn value(&mut self, _name: impl Into<Cow<'a, str>>, _value: &(impl Value + ?Sized)) {
45+
self.values += 1;
46+
}
47+
48+
fn config(&mut self, _config: &'a dyn EntryConfig) {}
49+
}
50+
51+
fn populated_pool(entries: usize, collide: bool) -> MetricsPool {
52+
let pool = MetricsPool::new();
53+
let base = pool.handle();
54+
for index in 0..entries {
55+
let handle = if collide {
56+
base.clone()
57+
} else {
58+
base.with_prefix([PREFIXES[index]])
59+
};
60+
handle.append(ChildMetrics {
61+
count: index as u64,
62+
operation: "PutObject",
63+
});
64+
}
65+
pool
66+
}
67+
68+
#[divan::bench(args = SIZES)]
69+
fn append_unique(bencher: Bencher, entries: usize) {
70+
bencher
71+
.counter(entries)
72+
.with_inputs(|| {
73+
let pool = MetricsPool::new();
74+
let base = pool.handle();
75+
let handles = (0..entries)
76+
.map(|index| base.with_prefix([PREFIXES[index]]))
77+
.collect::<Vec<_>>();
78+
(pool, handles)
79+
})
80+
.bench_values(|(pool, handles)| {
81+
for (index, handle) in handles.into_iter().enumerate() {
82+
handle.append(ChildMetrics {
83+
count: black_box(index as u64),
84+
operation: black_box("PutObject"),
85+
});
86+
}
87+
black_box(pool);
88+
});
89+
}
90+
91+
#[divan::bench(args = SIZES)]
92+
fn close_and_write_unique(bencher: Bencher, entries: usize) {
93+
bencher
94+
.counter(entries)
95+
.with_inputs(|| populated_pool(entries, false))
96+
.bench_values(|pool| {
97+
let closed = pool.close();
98+
let mut writer = CountingWriter::default();
99+
InflectableEntry::<PascalCase>::write(&closed, &mut writer);
100+
black_box(writer.values);
101+
});
102+
}
103+
104+
#[divan::bench(args = SIZES)]
105+
fn close_and_write_collisions(bencher: Bencher, entries: usize) {
106+
bencher
107+
.counter(entries)
108+
.with_inputs(|| populated_pool(entries, true))
109+
.bench_values(|pool| {
110+
let closed = pool.close();
111+
let mut writer = CountingWriter::default();
112+
InflectableEntry::<PascalCase>::write(&closed, &mut writer);
113+
black_box(writer.values);
114+
});
115+
}
116+
117+
#[divan::bench(args = SIZES)]
118+
fn direct_write_baseline(bencher: Bencher, entries: usize) {
119+
bencher
120+
.counter(entries)
121+
.with_inputs(|| {
122+
(0..entries)
123+
.map(|index| {
124+
ChildMetrics {
125+
count: index as u64,
126+
operation: "PutObject",
127+
}
128+
.close()
129+
})
130+
.collect::<Vec<_>>()
131+
})
132+
.bench_values(|children| {
133+
let mut writer = CountingWriter::default();
134+
for child in &children {
135+
InflectableEntry::<PascalCase>::write(child, &mut writer);
136+
}
137+
black_box(writer.values);
138+
});
139+
}

0 commit comments

Comments
 (0)