feat(metrique-util): prototype request-scoped metrics pools

metrique-util/Cargo.toml

@@ -14,6 +14,7 @@ unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)', 'cfg(shu
 [features]
 default = []
 state = ["dep:arc-swap"]
+metrics-pool = ["dep:Inflector", "dep:metrique", "dep:tracing"]
 tokio-metrics-bridge = [
     "state",
     "dep:metrique",
@@ -31,6 +32,7 @@ _shuttle = ["dep:shuttle", "pending-sink", "metrique-writer-core/_shuttle"]
 [dependencies]
 metrique-core = { workspace = true }
 metrique = { workspace = true, optional = true }
+Inflector = { workspace = true, optional = true }
 arc-swap = { version = "1", optional = true }
 tokio = { workspace = true, optional = true, features = ["time", "rt"] }
 tokio-metrics = { version = "0.5.0", optional = true, features = ["rt", "metrique-integration"] }
@@ -47,14 +49,21 @@ shuttle = { workspace = true, optional = true }
 
 [dev-dependencies]
 assert2 = { workspace = true }
+aws-smithy-runtime-api = { version = "1", features = ["client"] }
+aws-smithy-types = "1"
+divan = "0.1"
 metrique = { workspace = true, features = ["emf", "test-util", "service-metrics"] }
 metrique-writer = { workspace = true, features = ["test-util"] }
 metrique-writer-core = { workspace = true, features = ["test-util"] }
 tokio = { workspace = true, features = ["full", "test-util"] }
 rstest = { workspace = true }
-metrique-util = { path = ".", features = ["state", "pending-sink", "sysinfo-bridge", "tokio-metrics-bridge"] }
+metrique-util = { path = ".", features = ["metrics-pool", "state", "pending-sink", "sysinfo-bridge", "tokio-metrics-bridge"] }
 tracing-subscriber = { workspace = true }
 
+[[example]]
+name = "sdk-interceptor-metrics-pool"
+required-features = ["metrics-pool"]
+
 [[example]]
 name = "global-state"
 required-features = ["state", "sysinfo-bridge", "tokio-metrics-bridge"]
@@ -83,6 +92,11 @@ name = "sysinfo-folded-static"
 required-features = ["sysinfo-bridge"]
 doc-scrape-examples = false
 
+[[bench]]
+name = "metrics_pool"
+harness = false
+required-features = ["metrics-pool"]
+
 [package.metadata.docs.rs]
 all-features = true
 targets = ["x86_64-unknown-linux-gnu"]

metrique-util/README.md

@@ -5,6 +5,7 @@ Additional utilities for [metrique].
 ## Features
 
 - `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.
+- `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.
 - `tokio-metrics-bridge`: Subscribes [tokio-metrics] runtime snapshots to a global entry sink. The reporter task is automatically aborted when the `AttachHandle` is dropped.
 - `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.
 - `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.
 [metrique]: https://crates.io/crates/metrique
 [metrique documentation]: https://docs.rs/metrique
 [`State<T>`]: https://docs.rs/metrique-util/latest/metrique_util/state/struct.State.html
+[`MetricsPool`]: https://docs.rs/metrique-util/latest/metrique_util/struct.MetricsPool.html
+[`with_metrics_pool`]: https://docs.rs/metrique-util/latest/metrique_util/fn.with_metrics_pool.html
 [`pending_sink::new()`]: https://docs.rs/metrique-util/latest/metrique_util/pending_sink/fn.new.html
 [`PendingSinkResolver::resolve`]: https://docs.rs/metrique-util/latest/metrique_util/pending_sink/struct.PendingSinkResolver.html#method.resolve

metrique-util/benches/metrics_pool.rs

@@ -0,0 +1,139 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Benchmarks for `MetricsPool` type erasure and its two-pass write path.
+//!
+//! Run: `cargo bench -p metrique-util --bench metrics_pool --features metrics-pool`
+
+use std::borrow::Cow;
+use std::time::SystemTime;
+
+use divan::{Bencher, black_box};
+use metrique::unit_of_work::metrics;
+use metrique::writer::{EntryConfig, EntryWriter, Value};
+use metrique::{CloseValue, InflectableEntry, PascalCase};
+use metrique_util::MetricsPool;
+
+#[global_allocator]
+static ALLOC: divan::AllocProfiler = divan::AllocProfiler::system();
+
+fn main() {
+    divan::main();
+}
+
+const SIZES: &[usize] = &[1, 4, 16];
+const PREFIXES: &[&str] = &[
+    "child_00", "child_01", "child_02", "child_03", "child_04", "child_05", "child_06", "child_07",
+    "child_08", "child_09", "child_10", "child_11", "child_12", "child_13", "child_14", "child_15",
+];
+
+#[metrics]
+struct ChildMetrics {
+    count: u64,
+    operation: &'static str,
+}
+
+#[derive(Default)]
+struct CountingWriter {
+    values: usize,
+}
+
+impl<'a> EntryWriter<'a> for CountingWriter {
+    fn timestamp(&mut self, _timestamp: SystemTime) {}
+
+    fn value(&mut self, _name: impl Into<Cow<'a, str>>, _value: &(impl Value + ?Sized)) {
+        self.values += 1;
+    }
+
+    fn config(&mut self, _config: &'a dyn EntryConfig) {}
+}
+
+fn populated_pool(entries: usize, collide: bool) -> MetricsPool {
+    let pool = MetricsPool::new();
+    let base = pool.handle();
+    for index in 0..entries {
+        let handle = if collide {
+            base.clone()
+        } else {
+            base.with_prefix([PREFIXES[index]])
+        };
+        handle.append(ChildMetrics {
+            count: index as u64,
+            operation: "PutObject",
+        });
+    }
+    pool
+}
+
+#[divan::bench(args = SIZES)]
+fn append_unique(bencher: Bencher, entries: usize) {
+    bencher
+        .counter(entries)
+        .with_inputs(|| {
+            let pool = MetricsPool::new();
+            let base = pool.handle();
+            let handles = (0..entries)
+                .map(|index| base.with_prefix([PREFIXES[index]]))
+                .collect::<Vec<_>>();
+            (pool, handles)
+        })
+        .bench_values(|(pool, handles)| {
+            for (index, handle) in handles.into_iter().enumerate() {
+                handle.append(ChildMetrics {
+                    count: black_box(index as u64),
+                    operation: black_box("PutObject"),
+                });
+            }
+            black_box(pool);
+        });
+}
+
+#[divan::bench(args = SIZES)]
+fn close_and_write_unique(bencher: Bencher, entries: usize) {
+    bencher
+        .counter(entries)
+        .with_inputs(|| populated_pool(entries, false))
+        .bench_values(|pool| {
+            let closed = pool.close();
+            let mut writer = CountingWriter::default();
+            InflectableEntry::<PascalCase>::write(&closed, &mut writer);
+            black_box(writer.values);
+        });
+}
+
+#[divan::bench(args = SIZES)]
+fn close_and_write_collisions(bencher: Bencher, entries: usize) {
+    bencher
+        .counter(entries)
+        .with_inputs(|| populated_pool(entries, true))
+        .bench_values(|pool| {
+            let closed = pool.close();
+            let mut writer = CountingWriter::default();
+            InflectableEntry::<PascalCase>::write(&closed, &mut writer);
+            black_box(writer.values);
+        });
+}
+
+#[divan::bench(args = SIZES)]
+fn direct_write_baseline(bencher: Bencher, entries: usize) {
+    bencher
+        .counter(entries)
+        .with_inputs(|| {
+            (0..entries)
+                .map(|index| {
+                    ChildMetrics {
+                        count: index as u64,
+                        operation: "PutObject",
+                    }
+                    .close()
+                })
+                .collect::<Vec<_>>()
+        })
+        .bench_values(|children| {
+            let mut writer = CountingWriter::default();
+            for child in &children {
+                InflectableEntry::<PascalCase>::write(child, &mut writer);
+            }
+            black_box(writer.values);
+        });
+}