This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 789dfe6b5 feat(server): expose per-topic size and message count on
/metrics (#4480)
789dfe6b5 is described below
commit 789dfe6b5b1fd620d64d19b5ba056e02cb7773df
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Tue Oct 13 09:12:58 2026 +0200
feat(server): expose per-topic size and message count on /metrics (#4480)
---
core/common/src/utils/topic_size.rs | 32 ++++
core/integration/tests/server/mod.rs | 2 +-
core/integration/tests/server/topic_metrics_vsr.rs | 178 ++++++++++++++++-
core/server/config.toml | 4 +-
core/server/src/http/handlers.rs | 107 ++++++++---
core/server/src/http/metrics.rs | 210 ++++++++++++++++++++-
core/server/src/segment_cleaner.rs | 41 ++--
7 files changed, 506 insertions(+), 68 deletions(-)
diff --git a/core/common/src/utils/topic_size.rs
b/core/common/src/utils/topic_size.rs
index 1cfe4c13b..5ac6e8ad2 100644
--- a/core/common/src/utils/topic_size.rs
+++ b/core/common/src/utils/topic_size.rs
@@ -59,6 +59,24 @@ impl MaxTopicSize {
MaxTopicSize::Custom(iggy_byte_size) =>
iggy_byte_size.as_bytes_u64(),
}
}
+
+ /// The cap in bytes with `ServerDefault` resolved against `default`, or
+ /// `None` for "no cap".
+ ///
+ /// For `ServerDefault`, a `default` of 0 or `u64::MAX` yields `None`: 0
+ /// means "no cap", never a budget that trims everything. `ServerDefault`
+ /// therefore never comes back as `Some(0)`, so a caller can size against
+ /// any returned value directly.
+ pub fn resolve(self, default: u64) -> Option<u64> {
+ let resolved = match self {
+ MaxTopicSize::ServerDefault => MaxTopicSize::from(default),
+ sized => sized,
+ };
+ match resolved {
+ MaxTopicSize::Custom(size) => Some(size.as_bytes_u64()),
+ MaxTopicSize::Unlimited | MaxTopicSize::ServerDefault => None,
+ }
+ }
}
impl From<IggyByteSize> for MaxTopicSize {
@@ -187,3 +205,17 @@ impl fmt::Display for MaxTopicSize {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::MaxTopicSize;
+
+ #[test]
+ fn
given_max_topic_size_variants_when_resolving_should_yield_only_a_finite_cap() {
+ assert_eq!(MaxTopicSize::from(1024).resolve(u64::MAX), Some(1024));
+ assert_eq!(MaxTopicSize::Unlimited.resolve(4096), None);
+ assert_eq!(MaxTopicSize::ServerDefault.resolve(4096), Some(4096));
+ assert_eq!(MaxTopicSize::ServerDefault.resolve(u64::MAX), None);
+ assert_eq!(MaxTopicSize::ServerDefault.resolve(0), None);
+ }
+}
diff --git a/core/integration/tests/server/mod.rs
b/core/integration/tests/server/mod.rs
index ade46c258..e747305cd 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -39,7 +39,7 @@ mod poll_semantics_vsr;
mod topic_admission_vsr;
// Stats aggregates the cross-shard connected-client count, not a hardcoded 0.
mod stats_vsr;
-// /metrics carries per-topic size and message count (#4473).
+// /metrics carries per-topic size and message count.
mod topic_metrics_vsr;
// Shared HTTP transport plumbing (session + verb helpers) for the raw-HTTP
// server suites below.
diff --git a/core/integration/tests/server/topic_metrics_vsr.rs
b/core/integration/tests/server/topic_metrics_vsr.rs
index 9cc18ee80..c94d6a6a5 100644
--- a/core/integration/tests/server/topic_metrics_vsr.rs
+++ b/core/integration/tests/server/topic_metrics_vsr.rs
@@ -21,27 +21,40 @@
//! totals (`messages`, `segments`) can't say which topic is growing.
use iggy::prelude::*;
+use integration::harness::TestHarness;
use integration::iggy_harness;
use super::http_client::HttpClient;
const STREAM: &str = "metrics-stream";
const TOPIC: &str = "metrics-topic";
+const METRICS_USER: &str = "metrics-reader";
+const METRICS_PASSWORD: &str = "metrics-reader-pass";
+const TOPIC_SERIES: [&str; 3] = ["topic_size_bytes", "topic_messages",
"topic_max_size_bytes"];
+
+/// The value of the unlabelled `name` gauge.
+fn gauge(exposition: &str, name: &str) -> Option<u64> {
+ let series_prefix = format!("{name} ");
+ exposition
+ .lines()
+ .find_map(|line| line.strip_prefix(&series_prefix))
+ .and_then(|value| value.trim().parse().ok())
+}
/// The value of the `name` series labelled with this test's stream and topic.
fn topic_series(exposition: &str, name: &str) -> Option<u64> {
let stream_label = format!("stream=\"{STREAM}\"");
let topic_label = format!("topic=\"{TOPIC}\"");
+ let series_prefix = format!("{name}{{");
exposition
.lines()
- .filter(|line| line.starts_with(&format!("{name}{{")))
+ .filter(|line| line.starts_with(&series_prefix))
.find(|line| line.contains(&stream_label) &&
line.contains(&topic_label))
.and_then(|line| line.rsplit(' ').next())
.and_then(|value| value.parse().ok())
}
#[iggy_harness(cluster_nodes = 1)]
-#[ignore = "#4473: /metrics has no per-topic series yet"]
async fn
given_topic_with_messages_when_scraping_metrics_should_report_its_size_and_count(
harness: &TestHarness,
) {
@@ -93,13 +106,7 @@ async fn
given_topic_with_messages_when_scraping_metrics_should_report_its_size_
);
assert!(topic.size.as_bytes_u64() > 0);
- let http = HttpClient::login_root(harness).await;
- let exposition = http
- .get("/metrics")
- .await
- .text()
- .await
- .expect("metrics text");
+ let exposition = scrape(harness).await;
assert_eq!(
topic_series(&exposition, "topic_size_bytes"),
@@ -111,4 +118,157 @@ async fn
given_topic_with_messages_when_scraping_metrics_should_report_its_size_
Some(topic.messages_count),
"/metrics must report the topic's message count as GetTopic
does:\n{exposition}"
);
+ assert_eq!(
+ topic_series(&exposition, "topic_max_size_bytes"),
+ None,
+ "an uncapped topic must not export a cap series:\n{exposition}"
+ );
+}
+
+#[iggy_harness(cluster_nodes = 1)]
+async fn given_topic_with_max_size_when_scraping_metrics_should_report_its_cap(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ let max_topic_size = MaxTopicSize::from(2 * 1024 * 1024 * 1024u64);
+ create_topic(&client, max_topic_size).await;
+
+ let exposition = scrape(harness).await;
+
+ assert_eq!(
+ topic_series(&exposition, "topic_max_size_bytes"),
+ Some(max_topic_size.as_bytes_u64()),
+ "/metrics must report the topic's configured cap:\n{exposition}"
+ );
+}
+
+#[iggy_harness(cluster_nodes = 1)]
+async fn
given_scraped_topic_when_deleted_should_drop_its_series_on_next_scrape(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ create_topic(&client, MaxTopicSize::from(2 * 1024 * 1024 * 1024u64)).await;
+
+ let before = scrape(harness).await;
+ for name in TOPIC_SERIES {
+ assert!(
+ topic_series(&before, name).is_some(),
+ "the capped topic must export its {name} series before the
delete:\n{before}"
+ );
+ }
+
+ let stream_id = Identifier::from_str_value(STREAM).expect("stream
identifier");
+ let topic_id = Identifier::from_str_value(TOPIC).expect("topic
identifier");
+ client
+ .delete_topic(&stream_id, &topic_id)
+ .await
+ .expect("delete topic");
+
+ let after = scrape(harness).await;
+ for name in TOPIC_SERIES {
+ assert_eq!(
+ topic_series(&after, name),
+ None,
+ "a deleted topic must not keep its {name} series:\n{after}"
+ );
+ }
+}
+
+#[iggy_harness(cluster_nodes = 1)]
+async fn
given_user_without_server_read_when_scraping_metrics_should_omit_topic_series(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ create_topic(&client, MaxTopicSize::from(2 * 1024 * 1024 * 1024u64)).await;
+ // Topic reads alone must not unlock the per-topic series: the gate is the
+ // server-read rule `/stats` uses, not visibility of any one topic.
+ let permissions = Permissions {
+ global: GlobalPermissions {
+ read_streams: true,
+ read_topics: true,
+ ..GlobalPermissions::default()
+ },
+ streams: None,
+ };
+ client
+ .create_user(
+ METRICS_USER,
+ METRICS_PASSWORD,
+ UserStatus::Active,
+ Some(permissions),
+ )
+ .await
+ .expect("create metrics user");
+ let root = HttpClient::login_root(harness).await;
+ let user = root.login(METRICS_USER, METRICS_PASSWORD).await;
+
+ // A privileged scrape first, so a denied scrape that re-encoded the
+ // previous snapshot would leak these series.
+ let privileged_before = scrape_as(&root).await;
+ for name in TOPIC_SERIES {
+ assert!(
+ topic_series(&privileged_before, name).is_some(),
+ "root must see the {name} series:\n{privileged_before}"
+ );
+ }
+
+ let restricted = scrape_as(&user).await;
+ assert_eq!(
+ gauge(&restricted, "streams"),
+ Some(1),
+ "aggregate gauges stay login-only:\n{restricted}"
+ );
+ assert_eq!(
+ gauge(&restricted, "topics"),
+ Some(1),
+ "aggregate gauges stay login-only:\n{restricted}"
+ );
+ for name in TOPIC_SERIES {
+ assert!(
+ !restricted
+ .lines()
+ .any(|line| line.starts_with(&format!("{name}{{"))),
+ "a user without server read must not see any {name}
series:\n{restricted}"
+ );
+ }
+
+ let privileged_after = scrape_as(&root).await;
+ for name in TOPIC_SERIES {
+ assert!(
+ topic_series(&privileged_after, name).is_some(),
+ "a denied scrape must not empty the next privileged one
({name}):\n{privileged_after}"
+ );
+ }
+}
+
+async fn create_topic(client: &IggyClient, max_topic_size: MaxTopicSize) {
+ client.create_stream(STREAM).await.expect("create stream");
+ let stream_id = Identifier::from_str_value(STREAM).expect("stream
identifier");
+ client
+ .create_topic(
+ &stream_id,
+ TOPIC,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ max_topic_size: Some(max_topic_size),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+}
+
+async fn scrape(harness: &TestHarness) -> String {
+ scrape_as(&HttpClient::login_root(harness).await).await
+}
+
+async fn scrape_as(session: &HttpClient) -> String {
+ let response = session.get("/metrics").await;
+ assert!(
+ response.status().is_success(),
+ "/metrics scrape failed: {}",
+ response.status()
+ );
+ response.text().await.expect("metrics text")
}
diff --git a/core/server/config.toml b/core/server/config.toml
index b8ef18ae5..f488c2752 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -191,7 +191,9 @@ enabled = true
# Specifies the endpoint for accessing metrics, e.g., "/metrics".
# This route requires authentication like every other read: scrapers must
-# present a bearer credential (JWT or personal access token).
+# present a bearer credential (JWT or personal access token). Any
+# authenticated user gets the aggregate gauges; the per-topic series are
+# exported only to users with the read_servers or manage_servers permission.
endpoint = "/metrics"
# TLS (Transport Layer Security) configuration for HTTP.
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index 15adb538f..0f5e2f473 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -103,13 +103,13 @@ use iggy_common::wire_conversions::{
};
use iggy_common::{
ClientInfo, ClientInfoDetails, ClusterMetadata, CompressionAlgorithm,
Consumer, ConsumerGroup,
- ConsumerGroupDetails, ConsumerOffsetInfo, Identifier, IdentityInfo,
IggyError, IggyExpiry,
- MaxTopicSize, OptionSpec, OptionsScope, PersonalAccessTokenInfo,
PollMessages, PolledMessages,
- RawPersonalAccessToken, SendMessages, SendMessagesConfirmations, Stats,
Stream, StreamDetails,
- StreamUpdateOptions, TokenInfo, Topic, TopicCreateOptions, TopicDetails,
TopicUpdateOptions,
- UPDATABLE_STREAM_OPTION_KEYS, UPDATABLE_TOPIC_OPTION_KEYS,
UPDATABLE_USER_OPTION_KEYS,
- UserInfo, UserInfoDetails, UserUpdateOptions, Validatable,
validate_preallocated_topic_bytes,
- validate_topic_segment_size,
+ ConsumerGroupDetails, ConsumerOffsetInfo, DEFAULT_MAX_TOPIC_SIZE,
Identifier, IdentityInfo,
+ IggyError, IggyExpiry, MaxTopicSize, OptionSpec, OptionsScope,
PersonalAccessTokenInfo,
+ PollMessages, PolledMessages, RawPersonalAccessToken, SendMessages,
SendMessagesConfirmations,
+ Stats, Stream, StreamDetails, StreamUpdateOptions, TokenInfo, Topic,
TopicCreateOptions,
+ TopicDetails, TopicUpdateOptions, UPDATABLE_STREAM_OPTION_KEYS,
UPDATABLE_TOPIC_OPTION_KEYS,
+ UPDATABLE_USER_OPTION_KEYS, UserInfo, UserInfoDetails, UserUpdateOptions,
Validatable,
+ validate_preallocated_topic_bytes, validate_topic_segment_size,
};
use metadata::impls::metadata::StreamsFrontend;
use metadata::permissioner::Permissioner;
@@ -127,10 +127,10 @@ use crate::http::error::{
ReadError, WriteError,
};
use crate::http::extractor::{Authenticated, Identity};
-use crate::http::metrics::gauge_value;
+use crate::http::metrics::{TopicUsageSample, gauge_value};
use crate::http::reads::{
- authorize_data_plane, gate_local_read, read_local, resolve_gate_stream,
resolve_gate_topic,
- resolve_gate_topic_ids, resolve_gate_user, topic_durability,
+ authorize_data_plane, await_recovery_barrier, gate_local_read, read_local,
resolve_gate_stream,
+ resolve_gate_topic, resolve_gate_topic_ids, resolve_gate_user,
topic_durability,
};
use crate::http::reply::{
committed_payload, decode_consumer_group_details, decode_raw_pat_token,
decode_stream_details,
@@ -607,22 +607,51 @@ 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.
+///
+/// Held behind the post-restart recovery barrier like every other local read:
+/// until the recovered WAL suffix re-commits, the users map may still grant a
+/// permission a committed op revoked, and the walk may name a topic a
+/// committed op deleted. An expired barrier answers the retryable 503 rather
+/// than a partial exposition.
pub(in crate::http) async fn get_metrics(
State(state): State<HttpState>,
- _identity: Identity,
-) -> String {
- let (streams_count, topics_count, partitions_count, segments_count,
messages_count) = state
+ identity: Identity,
+) -> Result<String, ReadError> {
+ SendWrapper::new(await_recovery_barrier(&state.shard)).await?;
+ // 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))
+ .is_ok();
+ let (
+ (streams_count, topics_count, partitions_count, segments_count,
messages_count),
+ topic_usage,
+ ) = state
.shard
.plane
.metadata()
@@ -633,6 +662,17 @@ pub(in crate::http) async fn get_metrics(
let mut partitions_count = 0u64;
let mut segments_count = 0u64;
let mut messages_count = 0u64;
+ let mut topic_usage = if exports_topics {
+ Vec::with_capacity(
+ streams
+ .items
+ .iter()
+ .map(|(_, stream)| stream.topics.len())
+ .sum(),
+ )
+ } else {
+ Vec::new()
+ };
for (_, stream) in &streams.items {
topics_count = topics_count.saturating_add(stream.topics.len()
as u64);
segments_count = segments_count
@@ -642,14 +682,27 @@ pub(in crate::http) async fn get_metrics(
for (_, topic) in &stream.topics {
partitions_count =
partitions_count.saturating_add(topic.partitions.len()
as u64);
+ if !exports_topics {
+ continue;
+ }
+ topic_usage.push(TopicUsageSample {
+ stream: Arc::clone(&stream.name),
+ topic: Arc::clone(&topic.name),
+ size_bytes: topic.stats.size_bytes_inconsistent(),
+ messages: topic.stats.messages_count_inconsistent(),
+ max_size_bytes:
topic.max_topic_size.resolve(DEFAULT_MAX_TOPIC_SIZE),
+ });
}
}
(
- streams.items.len() as u64,
- topics_count,
- partitions_count,
- segments_count,
- messages_count,
+ (
+ streams.items.len() as u64,
+ topics_count,
+ partitions_count,
+ segments_count,
+ messages_count,
+ ),
+ topic_usage,
)
});
let users_count = state
@@ -659,7 +712,6 @@ pub(in crate::http) async fn get_metrics(
.mux_stm
.users()
.read(|users| users.items.len() as u64);
- let clients_count =
SendWrapper::new(state.shard.count_all_clients()).await as u64;
let metrics = &state.metrics;
metrics.streams.set(gauge_value(streams_count));
@@ -669,7 +721,8 @@ pub(in crate::http) async fn get_metrics(
metrics.messages.set(gauge_value(messages_count));
metrics.users.set(gauge_value(users_count));
metrics.clients.set(gauge_value(clients_count));
- metrics.formatted_output()
+ metrics.topic_usage.replace(topic_usage);
+ Ok(metrics.formatted_output())
}
/// `POST /snapshot`: collect a diagnostic archive and return it as a ZIP
diff --git a/core/server/src/http/metrics.rs b/core/server/src/http/metrics.rs
index dfc1d4a69..8bc278b09 100644
--- a/core/server/src/http/metrics.rs
+++ b/core/server/src/http/metrics.rs
@@ -20,13 +20,21 @@
//! whether the route is mounted. The scrape handler itself lives with the
//! other route handlers so this leaf never imports the state hub.
+use std::fmt::Write;
+use std::mem;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex, PoisonError};
+
use configs::http::HttpMetricsConfig;
use iggy_common::{IggyError, stats_rollup_underflows};
use prometheus_client::collector::Collector;
use prometheus_client::encoding::text::encode;
-use prometheus_client::encoding::{DescriptorEncoder, EncodeMetric};
+use prometheus_client::encoding::{
+ DescriptorEncoder, EncodeLabelValue, EncodeMetric, LabelValueEncoder,
+};
+use prometheus_client::metrics::MetricType;
use prometheus_client::metrics::counter::{ConstCounter, Counter};
-use prometheus_client::metrics::gauge::Gauge;
+use prometheus_client::metrics::gauge::{ConstGauge, Gauge};
use prometheus_client::registry::Registry;
use tracing::error;
@@ -73,6 +81,110 @@ 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;
+ }
+}
+
+impl Collector for TopicUsage {
+ fn encode(&self, mut encoder: DescriptorEncoder) -> Result<(),
std::fmt::Error> {
+ // The scrape handler refills the snapshot before every encode, so
+ // keeping it past this encode only pins every topic name until the
+ // next scrape.
+ let samples = mem::take(&mut
*self.samples.lock().unwrap_or_else(PoisonError::into_inner));
+ encode_topic_gauge(
+ &mut encoder,
+ "topic_size_bytes",
+ "size of the topic's stored messages in bytes",
+ &samples,
+ |sample| Some(sample.size_bytes),
+ )?;
+ encode_topic_gauge(
+ &mut encoder,
+ "topic_messages",
+ "count of the topic's stored messages",
+ &samples,
+ |sample| Some(sample.messages),
+ )?;
+ encode_topic_gauge(
+ &mut encoder,
+ "topic_max_size_bytes",
+ concat!(
+ "configured size cap of the topic in bytes, absent when
unlimited; ",
+ "retention may keep more",
+ ),
+ &samples,
+ |sample| sample.max_size_bytes,
+ )
+ }
+}
+
+fn encode_topic_gauge(
+ encoder: &mut DescriptorEncoder,
+ name: &str,
+ help: &str,
+ samples: &[TopicUsageSample],
+ value: impl Fn(&TopicUsageSample) -> Option<u64>,
+) -> Result<(), std::fmt::Error> {
+ let mut metric_encoder = encoder.encode_descriptor(name, help, None,
MetricType::Gauge)?;
+ for sample in samples {
+ let Some(value) = value(sample) else {
+ continue;
+ };
+ let labels = [
+ ("stream", EscapedLabelValue(&sample.stream)),
+ ("topic", EscapedLabelValue(&sample.topic)),
+ ];
+ ConstGauge::new(value).encode(metric_encoder.encode_family(&labels)?)?;
+ }
+ Ok(())
+}
+
+/// prometheus-client writes label values verbatim, and stream and topic names
+/// may contain any character the exposition format treats as syntax.
+struct EscapedLabelValue<'a>(&'a str);
+
+impl EncodeLabelValue for EscapedLabelValue<'_> {
+ fn encode(&self, encoder: &mut LabelValueEncoder) -> Result<(),
std::fmt::Error> {
+ if !self.0.contains(['\\', '"', '\n']) {
+ return encoder.write_str(self.0);
+ }
+ for character in self.0.chars() {
+ match character {
+ '\\' => encoder.write_str("\\\\")?,
+ '"' => encoder.write_str("\\\"")?,
+ '\n' => encoder.write_str("\\n")?,
+ other => encoder.write_char(other)?,
+ }
+ }
+ Ok(())
+ }
+}
+
/// The legacy server's metric set, registered under the same names and help
/// texts so existing dashboards and alerts keep working unchanged.
///
@@ -90,6 +202,8 @@ pub(in crate::http) struct HttpMetrics {
pub(in crate::http) messages: Gauge,
pub(in crate::http) users: Gauge,
pub(in crate::http) clients: Gauge,
+ pub(in crate::http) topic_usage: TopicUsage,
+ last_output_len: AtomicUsize,
}
impl HttpMetrics {
@@ -122,6 +236,8 @@ impl HttpMetrics {
// Not a legacy-parity metric, and not a mirrored one: the source is a
// process-wide static the scrape reads directly.
registry.register_collector(Box::new(StatsRollupUnderflows));
+ let topic_usage = TopicUsage::default();
+ registry.register_collector(Box::new(topic_usage.clone()));
// Every shard's drop / reconcile / partition counters, one
// `shard`-labelled sub-registry per shard so series stay per-shard
// without a `shard_id` label in the counter label sets (see
@@ -145,6 +261,8 @@ impl HttpMetrics {
messages,
users,
clients,
+ topic_usage,
+ last_output_len: AtomicUsize::new(0),
}
}
@@ -155,10 +273,12 @@ impl HttpMetrics {
}
pub(in crate::http) fn formatted_output(&self) -> String {
- let mut buffer = String::new();
+ let last = self.last_output_len.load(Ordering::Relaxed);
+ let mut buffer = String::with_capacity(last + last / 8);
if let Err(error) = encode(&mut buffer, &self.registry) {
error!(%error, "failed to encode metrics");
}
+ self.last_output_len.store(buffer.len(), Ordering::Relaxed);
buffer
}
}
@@ -319,6 +439,90 @@ mod tests {
assert_eq!(endpoint.as_deref(), Some("/metrics"));
}
+ fn topic_sample(stream: &str, topic: &str, max_size_bytes: Option<u64>) ->
TopicUsageSample {
+ TopicUsageSample {
+ stream: Arc::from(stream),
+ topic: Arc::from(topic),
+ size_bytes: 4096,
+ messages: 20,
+ max_size_bytes,
+ }
+ }
+
+ #[test]
+ fn
given_topic_samples_when_encoding_should_label_each_series_by_stream_and_topic()
{
+ let metrics = HttpMetrics::init(&[]);
+ metrics.topic_usage.replace(vec![
+ topic_sample("orders", "eu", Some(1_000_000)),
+ topic_sample("orders", "us", None),
+ ]);
+ let output = metrics.formatted_output();
+ for line in [
+ "# TYPE topic_size_bytes gauge",
+ "# TYPE topic_messages gauge",
+ "# TYPE topic_max_size_bytes gauge",
+ r#"topic_size_bytes{stream="orders",topic="eu"} 4096"#,
+ r#"topic_size_bytes{stream="orders",topic="us"} 4096"#,
+ r#"topic_messages{stream="orders",topic="eu"} 20"#,
+ r#"topic_messages{stream="orders",topic="us"} 20"#,
+ r#"topic_max_size_bytes{stream="orders",topic="eu"} 1000000"#,
+ ] {
+ assert!(
+ output.contains(&format!("\n{line}\n")),
+ "expected `{line}` in exposition:\n{output}"
+ );
+ }
+ assert!(
+
!output.contains(r#"topic_max_size_bytes{stream="orders",topic="us"}"#),
+ "an uncapped topic must not export a max size:\n{output}"
+ );
+ }
+
+ #[test]
+ fn
given_replaced_snapshot_when_encoding_should_drop_topics_no_longer_present() {
+ let metrics = HttpMetrics::init(&[]);
+ metrics
+ .topic_usage
+ .replace(vec![topic_sample("orders", "eu", None)]);
+ metrics.topic_usage.replace(Vec::new());
+ let output = metrics.formatted_output();
+ assert!(
+ !output.contains(r#"topic="eu""#),
+ "a deleted topic must not outlive the next scrape:\n{output}"
+ );
+ }
+
+ #[test]
+ fn
given_encoded_snapshot_when_scraping_again_without_refill_should_not_retain_it()
{
+ let metrics = HttpMetrics::init(&[]);
+ metrics
+ .topic_usage
+ .replace(vec![topic_sample("orders", "eu", None)]);
+ let first = metrics.formatted_output();
+ assert!(
+ first.contains(r#"topic="eu""#),
+ "the snapshot must be encoded before it is released:\n{first}"
+ );
+ let second = metrics.formatted_output();
+ assert!(
+ !second.contains(r#"topic="eu""#),
+ "the snapshot must not outlive the scrape that encoded
it:\n{second}"
+ );
+ }
+
+ #[test]
+ fn given_name_with_exposition_syntax_when_encoding_should_escape_it() {
+ let metrics = HttpMetrics::init(&[]);
+ metrics
+ .topic_usage
+ .replace(vec![topic_sample("a\"b", "c\\d\ne", None)]);
+ let output = metrics.formatted_output();
+ assert!(
+ output.contains(r#"topic_messages{stream="a\"b",topic="c\\d\ne"}
20"#),
+ "label values must be escaped:\n{output}"
+ );
+ }
+
#[test]
fn validated_endpoint_rejects_missing_leading_slash() {
assert!(matches!(
diff --git a/core/server/src/segment_cleaner.rs
b/core/server/src/segment_cleaner.rs
index 724666b24..b70f45932 100644
--- a/core/server/src/segment_cleaner.rs
+++ b/core/server/src/segment_cleaner.rs
@@ -178,39 +178,26 @@ struct PartitionSizeBudget {
/// one maximum bus frame past `segment_size`. An operator who raises that knob
/// grows real sealed segments with it.
///
-/// `ServerDefault` is resolved against the node default HERE, at enforcement
-/// time. Create admission rewrites the sentinel before replication, but an
-/// UPDATE to `ServerDefault` leaves it in committed state, and reading that as
-/// "no cap" made an updated topic behave differently from an identically
-/// configured created one. A node default of unlimited (the shipped config)
-/// still yields `None`.
-///
-/// `ServerDefault` must never reach a sized branch: its `as_bytes_u64()` is 0,
-/// which would trim every sealed segment.
+/// `ServerDefault` is resolved against `DEFAULT_MAX_TOPIC_SIZE` HERE, at
+/// enforcement time. Create admission rewrites the sentinel before
+/// replication, but an UPDATE to `ServerDefault` leaves it in committed state,
+/// and reading that as "no cap" made an updated topic behave differently from
+/// an identically configured created one. `DEFAULT_MAX_TOPIC_SIZE` is a
+/// compile-time const set to unlimited, so this still yields `None` today.
fn per_partition_size_budget(
max_topic_size: MaxTopicSize,
default_max_topic_size: u64,
partition_count: usize,
sealed_segment_ceiling: u64,
) -> Option<PartitionSizeBudget> {
- let resolved = match max_topic_size {
- MaxTopicSize::ServerDefault =>
MaxTopicSize::from(default_max_topic_size),
- sized => sized,
- };
- match resolved {
- MaxTopicSize::Custom(size) => {
- let divisor = u64::try_from(partition_count).unwrap_or(1).max(1);
- let configured_share = size.as_bytes_u64() / divisor;
- Some(PartitionSizeBudget {
- max_bytes: configured_share.max(sealed_segment_ceiling),
- configured_share,
- resolved_cap: size.as_bytes_u64(),
- })
- }
- // `From<u64>` maps 0 back to `ServerDefault`, so a node default of 0
- // lands here as "no cap" rather than as a trim-everything budget.
- MaxTopicSize::Unlimited | MaxTopicSize::ServerDefault => None,
- }
+ let resolved_cap = max_topic_size.resolve(default_max_topic_size)?;
+ let divisor = u64::try_from(partition_count).unwrap_or(1).max(1);
+ let configured_share = resolved_cap / divisor;
+ Some(PartitionSizeBudget {
+ max_bytes: configured_share.max(sealed_segment_ceiling),
+ configured_share,
+ resolved_cap,
+ })
}
#[cfg(test)]
