hubcio commented on code in PR #4480:
URL: https://github.com/apache/iggy/pull/4480#discussion_r4247744439
##########
core/server/src/http/handlers.rs:
##########
@@ -607,22 +607,44 @@ pub(in crate::http) async fn get_stats(
}
/// `GET <http.metrics.endpoint>`: the metric set in prometheus text
-/// exposition. Auth-only, like `/stats`: the `Identity` extractor rejects a
-/// missing or invalid bearer with 401, and any authenticated user may scrape
-/// (no RBAC rule guards it). Scrapers present a JWT or a raw PAT the same way
-/// every read route accepts them.
+/// exposition. The `Identity` extractor rejects a missing or invalid bearer
+/// with 401, and any authenticated user may scrape the aggregate gauges.
+/// The per-topic series name every stream and topic, so they are exported only
+/// to a user the `/stats` rule ([`Permissioner::get_stats`]) admits; anyone
+/// else gets the aggregates with no per-topic series. Scrapers present a JWT
+/// or a raw PAT the same way every read route accepts them.
///
/// The entity gauges sample the same reads `/stats` serves: the metadata STM
/// stream and user maps plus the stats-registry rollups, whose partition-plane
/// increments are relaxed, so scraped values are approximate while writes are
-/// in flight. The clients count scatter-gathers the per-shard session managers
-/// exactly like `GET /clients` and turns partial when a shard misses the reply
-/// deadline.
+/// in flight. The per-topic series come from the same walk and the same
+/// rollups `GetTopic` reads, and encode synchronously on shard 0 in time
linear
+/// in the topic count. The clients count scatter-gathers the per-shard session
+/// managers exactly like `GET /clients` and turns partial when a shard misses
+/// the reply deadline.
+///
+/// Per-topic series are this replica's local view, so a lagging backup may
+/// report smaller values than the primary.
pub(in crate::http) async fn get_metrics(
State(state): State<HttpState>,
- _identity: Identity,
+ identity: Identity,
) -> String {
- let (streams_count, topics_count, partitions_count, segments_count,
messages_count) = state
+ // Awaited before the permission check and the walk so check, walk, replace
+ // and encode run without a yield, and a scrape that finishes later never
+ // publishes an older per-topic snapshot.
+ let clients_count =
SendWrapper::new(state.shard.count_all_clients()).await as u64;
+ let exports_topics = state
+ .shard
+ .plane
+ .metadata()
+ .mux_stm
+ .users()
+ .authorize(|permissioner| permissioner.get_stats(identity.user_id))
Review Comment:
critical: `/metrics` can expose topic names and counters through revoked
permissions while a restarted replica awaits metadata recovery. await
`await_recovery_barrier` before checking topic permissions, and propagate
recovery failures.
##########
core/server/src/http/metrics.rs:
##########
@@ -73,6 +81,111 @@ impl Collector for StatsRollupUnderflows {
}
}
+/// One topic's usage as the scrape handler sampled it from the streams STM.
+#[derive(Debug)]
+pub(in crate::http) struct TopicUsageSample {
+ pub(in crate::http) stream: Arc<str>,
+ pub(in crate::http) topic: Arc<str>,
+ pub(in crate::http) size_bytes: u64,
+ pub(in crate::http) messages: u64,
+ /// `None` for an uncapped topic, so its series is absent rather than 0 and
+ /// `topic_size_bytes / topic_max_size_bytes` never yields `+Inf`.
+ pub(in crate::http) max_size_bytes: Option<u64>,
+}
+
+/// Per-topic `{stream, topic}` series, replaced wholesale on every scrape so a
+/// deleted or renamed topic drops out on the next one.
+///
+/// The collector cannot read the streams STM itself: it lives on shard 0 and
+/// is `!Sync`, while a registered collector must be `Send + Sync`, so the
+/// scrape handler hands it a snapshot instead.
+#[derive(Debug, Clone, Default)]
+pub(in crate::http) struct TopicUsage {
+ samples: Arc<Mutex<Vec<TopicUsageSample>>>,
+}
+
+impl TopicUsage {
+ pub(in crate::http) fn replace(&self, samples: Vec<TopicUsageSample>) {
+ *self.samples.lock().unwrap_or_else(PoisonError::into_inner) = samples;
+ }
+
+ fn take(&self) -> Vec<TopicUsageSample> {
+ mem::take(&mut
*self.samples.lock().unwrap_or_else(PoisonError::into_inner))
+ }
+}
+
+impl Collector for TopicUsage {
+ fn encode(&self, mut encoder: DescriptorEncoder) -> Result<(),
std::fmt::Error> {
+ let samples =
self.samples.lock().unwrap_or_else(PoisonError::into_inner);
Review Comment:
simplification: `TopicUsage` holds the snapshot lock throughout encoding,
then `HttpMetrics` locks again to clear it. use `mem::take` in
`TopicUsage::encode` and let the local vector drop on return.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
