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 2eac34938 test(server): expect per-topic size and message count on
/metrics (#4475)
2eac34938 is described below
commit 2eac3493847040c2e0f436110565f363360fe3cb
Author: Mark Aron Szulyovszky <[email protected]>
AuthorDate: Fri Oct 9 11:25:20 2026 +0100
test(server): expect per-topic size and message count on /metrics (#4475)
---
core/integration/tests/server/mod.rs | 2 +
core/integration/tests/server/topic_metrics_vsr.rs | 114 +++++++++++++++++++++
2 files changed, 116 insertions(+)
diff --git a/core/integration/tests/server/mod.rs
b/core/integration/tests/server/mod.rs
index 935d99f61..6b4f8b3a6 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -39,6 +39,8 @@ 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).
+mod topic_metrics_vsr;
// Purge durability: applied generation survives restart; journal-resident
// purged batches stay fenced behind the purge floor.
mod purge_vsr;
diff --git a/core/integration/tests/server/topic_metrics_vsr.rs
b/core/integration/tests/server/topic_metrics_vsr.rs
new file mode 100644
index 000000000..9cc18ee80
--- /dev/null
+++ b/core/integration/tests/server/topic_metrics_vsr.rs
@@ -0,0 +1,114 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `/metrics` per topic: an operator alerting on a topic's growth (against its
+//! `max_topic_size`, or behind a stopped consumer that holds retention back)
+//! needs the topic's size and message count as labelled series. The cluster
+//! totals (`messages`, `segments`) can't say which topic is growing.
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+
+use super::http_client::HttpClient;
+
+const STREAM: &str = "metrics-stream";
+const TOPIC: &str = "metrics-topic";
+
+/// 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}\"");
+ exposition
+ .lines()
+ .filter(|line| line.starts_with(&format!("{name}{{")))
+ .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,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ client.create_stream(STREAM).await.expect("create stream");
+ let stream_id = Identifier::from_str_value(STREAM).expect("stream
identifier");
+ let topic_id = Identifier::from_str_value(TOPIC).expect("topic
identifier");
+ client
+ .create_topic(
+ &stream_id,
+ TOPIC,
+ &TopicCreateOptions {
+ partitions_count: Some(2),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+
+ for partition in 0..2 {
+ let mut messages: Vec<IggyMessage> = (0..10)
+ .map(|index| {
+ IggyMessage::builder()
+
.payload(format!("partition-{partition}-message-{index}").into())
+ .build()
+ .expect("build message")
+ })
+ .collect();
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(partition),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+ }
+
+ let topic = client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("get topic")
+ .expect("topic exists");
+ assert_eq!(
+ topic.messages_count, 20,
+ "GetTopic must count both partitions' sends, or the comparison below
proves nothing"
+ );
+ 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");
+
+ assert_eq!(
+ topic_series(&exposition, "topic_size_bytes"),
+ Some(topic.size.as_bytes_u64()),
+ "/metrics must report the topic's size as GetTopic does:\n{exposition}"
+ );
+ assert_eq!(
+ topic_series(&exposition, "topic_messages"),
+ Some(topic.messages_count),
+ "/metrics must report the topic's message count as GetTopic
does:\n{exposition}"
+ );
+}
