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 64b0bae71 feat(python): expose get_stats on IggyClient (#4018)
64b0bae71 is described below
commit 64b0bae7175d42df120bb9e93c66ceb8a2e2e07f
Author: BingEdward <[email protected]>
AuthorDate: Mon Sep 7 15:27:13 2026 +0800
feat(python): expose get_stats on IggyClient (#4018)
---
.../actions/python-maturin/pre-merge/action.yml | 9 +
core/bench/report/src/types/server_stats.rs | 2 +-
core/common/src/types/stats/mod.rs | 2 +-
foreign/csharp/Iggy_SDK/Contracts/StatsResponse.cs | 2 +-
foreign/python/apache_iggy.pyi | 245 +++++++++++
foreign/python/src/client.rs | 25 ++
foreign/python/src/lib.rs | 5 +
foreign/python/src/stats.rs | 449 +++++++++++++++++++++
foreign/python/tests/test_stats.py | 250 ++++++++++++
9 files changed, 986 insertions(+), 3 deletions(-)
diff --git a/.github/actions/python-maturin/pre-merge/action.yml
b/.github/actions/python-maturin/pre-merge/action.yml
index 6a28d0e5b..234c36efe 100644
--- a/.github/actions/python-maturin/pre-merge/action.yml
+++ b/.github/actions/python-maturin/pre-merge/action.yml
@@ -90,6 +90,15 @@ runs:
echo "pyrefly version: $(uv run pyrefly --version)"
shell: bash
+ # The crate is outside the root workspace, so the Rust test jobs never
+ # reach it; without this its unit tests would only ever be compiled.
+ - name: Rust unit tests for Python extension
+ if: inputs.task == 'lint'
+ run: |
+ cd foreign/python
+ cargo test --manifest-path Cargo.toml
+ shell: bash
+
- name: Build Python wheel
if: inputs.task == 'test'
run: |
diff --git a/core/bench/report/src/types/server_stats.rs
b/core/bench/report/src/types/server_stats.rs
index 0eca6a939..e2b24f37b 100644
--- a/core/bench/report/src/types/server_stats.rs
+++ b/core/bench/report/src/types/server_stats.rs
@@ -72,7 +72,7 @@ pub struct BenchmarkServerStats {
pub kernel_version: String,
/// The version of the Iggy server.
pub iggy_server_version: String,
- /// The semantic version of the Iggy server in the numeric format e.g.
1.2.3 -> 100200300 (major * 1000000 + minor * 1000 + patch).
+ /// The semantic version of the Iggy server in the numeric format e.g.
1.2.3 -> 1002003 (major * 1000000 + minor * 1000 + patch).
pub iggy_server_semver: Option<u32>,
/// Cache metrics per partition
#[serde(with = "cache_metrics_serializer")]
diff --git a/core/common/src/types/stats/mod.rs
b/core/common/src/types/stats/mod.rs
index 5de7f03b4..fcec5cf4c 100644
--- a/core/common/src/types/stats/mod.rs
+++ b/core/common/src/types/stats/mod.rs
@@ -70,7 +70,7 @@ pub struct Stats {
pub kernel_version: String,
/// The version of the Iggy server.
pub iggy_server_version: String,
- /// The semantic version of the Iggy server in the numeric format e.g.
1.2.3 -> 100200300 (major * 1000000 + minor * 1000 + patch).
+ /// The semantic version of the Iggy server in the numeric format e.g.
1.2.3 -> 1002003 (major * 1000000 + minor * 1000 + patch).
pub iggy_server_semver: Option<u32>,
/// Cache metrics per partition
#[serde(with = "cache_metrics_serializer")]
diff --git a/foreign/csharp/Iggy_SDK/Contracts/StatsResponse.cs
b/foreign/csharp/Iggy_SDK/Contracts/StatsResponse.cs
index 772fbaa57..4b6efa627 100644
--- a/foreign/csharp/Iggy_SDK/Contracts/StatsResponse.cs
+++ b/foreign/csharp/Iggy_SDK/Contracts/StatsResponse.cs
@@ -148,7 +148,7 @@ public sealed class StatsResponse
public required string IggyServerVersion { get; init; }
/// <summary>
- /// Semantic version of the Iggy server in the numeric format e.g.
1.2.3 -> 100200300 (major * 1000000 + minor * 1000 +
+ /// Semantic version of the Iggy server in the numeric format e.g.
1.2.3 -> 1002003 (major * 1000000 + minor * 1000 +
/// patch).
/// </summary>
public uint IggyServerSemver { get; init; }
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index c0f603e58..ed15ed2f5 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -30,6 +30,8 @@ __all__ = [
"AutoCommitAfter",
"AutoCommitWhen",
"AutoLogin",
+ "CacheMetrics",
+ "CacheMetricsKey",
"Consumer",
"ConsumerGroup",
"ConsumerGroupDetails",
@@ -49,6 +51,7 @@ __all__ = [
"SendMessage",
"SendMessagesConfirmation",
"SendMessagesResponse",
+ "Stats",
"StreamDetails",
"StreamPermissions",
"TcpConfig",
@@ -290,6 +293,57 @@ class AutoLogin:
"""
def __repr__(self) -> builtins.str: ...
[email protected]
+class CacheMetrics:
+ r"""
+ Cache metrics for a specific partition.
+ """
+ @property
+ def hits(self) -> builtins.int:
+ r"""
+ Number of cache hits.
+ """
+ @property
+ def misses(self) -> builtins.int:
+ r"""
+ Number of cache misses.
+ """
+ @property
+ def hit_ratio(self) -> builtins.float:
+ r"""
+ Hit ratio (hits / (hits + misses)).
+ """
+ def __repr__(self) -> builtins.str: ...
+
[email protected]
+class CacheMetricsKey:
+ r"""
+ Key identifying the partition a `CacheMetrics` entry belongs to.
+
+ Hashable and comparable, so it can key the `Stats.cache_metrics` dict.
+ """
+ @property
+ def stream_id(self) -> builtins.int:
+ r"""
+ The unique identifier (numeric) of the stream.
+ """
+ @property
+ def topic_id(self) -> builtins.int:
+ r"""
+ The unique identifier (numeric) of the topic within the stream.
+ """
+ @property
+ def partition_id(self) -> builtins.int:
+ r"""
+ The unique identifier (numeric) of the partition within the topic.
+ """
+ def __eq__(self, other: builtins.object, /) -> builtins.bool: ...
+ def __hash__(self) -> builtins.int: ...
+ def __new__(
+ cls, stream_id: builtins.int, topic_id: builtins.int, partition_id:
builtins.int
+ ) -> CacheMetricsKey: ...
+ def __repr__(self) -> builtins.str: ...
+
class Consumer:
r"""
The consumer polling the messages. It selects both the consumer kind and
the
@@ -872,6 +926,21 @@ class IggyClient:
Sends a ping request to the server to check connectivity.
Raises `RuntimeError` if the connection fails.
"""
+ def get_stats(self) -> collections.abc.Awaitable[Stats]:
+ r"""
+ Get the statistics and details of the server and its running process.
+
+ Requires an authenticated session whose user holds the `read_servers`
+ or `manage_servers` global permission.
+
+ Returns:
+ An awaitable that resolves to `Stats`.
+
+ Raises:
+ RuntimeError: If the client is not connected, the session is not
+ authenticated, the user lacks the permission, or the request
+ fails.
+ """
def describe_options(
self, scope: builtins.str
) -> collections.abc.Awaitable[list[OptionSpec]]:
@@ -1836,6 +1905,182 @@ class SendMessagesResponse:
with an offset a client has already recorded.
"""
[email protected]
+class Stats:
+ r"""
+ The statistics and details of the server and its running process.
+
+ The fields are gathered from several sources while the request is served
+ (metadata counters, a process probe, a disk probe), so they are not an
+ atomic snapshot of one instant.
+ """
+ @property
+ def process_id(self) -> builtins.int:
+ r"""
+ The unique identifier of the server process.
+ """
+ @property
+ def cpu_usage(self) -> builtins.float:
+ r"""
+ The CPU usage of the server process, in percent summed over the cores
+ it ran on, so it exceeds 100 whenever the process uses more than one
+ core.
+
+ Measured as a delta since the previous `get_stats` served by the same
+ server shard, so the first sample a shard serves is 0.
+ """
+ @property
+ def total_cpu_usage(self) -> builtins.float:
+ r"""
+ The total CPU usage of the system, in percent averaged over the cores
+ the server may run on when confined by an affinity/cpuset mask (over
+ every host core otherwise), so it stays within 0-100.
+
+ Same per-shard delta sampling as `cpu_usage`: the first sample a shard
+ serves is 0.
+ """
+ @property
+ def memory_usage(self) -> builtins.int:
+ r"""
+ The memory usage of the server process, in bytes.
+ """
+ @property
+ def total_memory(self) -> builtins.int:
+ r"""
+ The total memory of the system, in bytes, or the effective cgroup
memory
+ limit when the server runs inside a memory-capped cgroup (container,
+ systemd slice).
+ """
+ @property
+ def available_memory(self) -> builtins.int:
+ r"""
+ The available memory of the system, in bytes, scoped to the cgroup
+ limit when one applies.
+ """
+ @property
+ def run_time(self) -> datetime.timedelta:
+ r"""
+ The run time of the server process, with whole-second precision.
+ """
+ @property
+ def start_time(self) -> builtins.int:
+ r"""
+ The start time of the server process, in microseconds since the Unix
+ epoch, with whole-second precision.
+ """
+ @property
+ def read_bytes(self) -> builtins.int:
+ r"""
+ The total number of bytes read.
+ """
+ @property
+ def written_bytes(self) -> builtins.int:
+ r"""
+ The total number of bytes written.
+ """
+ @property
+ def messages_size_bytes(self) -> builtins.int:
+ r"""
+ The total size of the messages, in bytes.
+ """
+ @property
+ def streams_count(self) -> builtins.int:
+ r"""
+ The total number of streams.
+ """
+ @property
+ def topics_count(self) -> builtins.int:
+ r"""
+ The total number of topics.
+ """
+ @property
+ def partitions_count(self) -> builtins.int:
+ r"""
+ The total number of partitions.
+ """
+ @property
+ def segments_count(self) -> builtins.int:
+ r"""
+ The total number of segments.
+ """
+ @property
+ def messages_count(self) -> builtins.int:
+ r"""
+ The total number of messages.
+ """
+ @property
+ def clients_count(self) -> builtins.int:
+ r"""
+ The total number of connected clients.
+ """
+ @property
+ def consumer_groups_count(self) -> builtins.int:
+ r"""
+ The total number of consumer groups.
+ """
+ @property
+ def hostname(self) -> builtins.str:
+ r"""
+ The name of the host the server runs on.
+ """
+ @property
+ def os_name(self) -> builtins.str:
+ r"""
+ The name of the operating system.
+ """
+ @property
+ def os_version(self) -> builtins.str:
+ r"""
+ The version of the operating system.
+ """
+ @property
+ def kernel_version(self) -> builtins.str:
+ r"""
+ The version of the kernel.
+ """
+ @property
+ def iggy_server_version(self) -> builtins.str:
+ r"""
+ The version of the Iggy server.
+ """
+ @property
+ def iggy_server_semver(self) -> builtins.int | None:
+ r"""
+ The numeric semantic version of the Iggy server, or `None` when
unknown.
+ E.g. 1.2.3 -> 1002003 (major * 1000000 + minor * 1000 + patch).
+ """
+ @property
+ def cache_metrics(self) -> builtins.dict[CacheMetricsKey, CacheMetrics]:
+ r"""
+ Cache metrics per partition.
+
+ Current servers do not populate this and reply with an empty map. Each
+ access builds a fresh dict, so mutating the returned dict does not
+ change the stats.
+ """
+ @property
+ def threads_count(self) -> builtins.int:
+ r"""
+ The number of threads in the server process.
+ """
+ @property
+ def free_disk_space(self) -> builtins.int:
+ r"""
+ The available (free) disk space for the data directory, in bytes.
+
+ 0 when the server does not know its data directory or the disk probe
+ fails.
+ """
+ @property
+ def total_disk_space(self) -> builtins.int:
+ r"""
+ The total disk space for the data directory, in bytes.
+
+ 0 when the server does not know its data directory or the disk probe
+ fails.
+ """
+ def __repr__(self) -> builtins.str: ...
+
@typing.final
class StreamDetails:
@property
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 10c5c776b..8252156a3 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -42,6 +42,7 @@ use crate::options::OptionSpec as PyOptionSpec;
use crate::permissions::Permissions as PyPermissions;
use crate::receive_message::{PollingStrategy, ReceiveMessage};
use crate::send_message::{SendMessage, SendMessagesResponse as
PySendMessagesResponse};
+use crate::stats::Stats as PyStats;
use crate::stream::StreamDetails;
use crate::topic::{IggyExpiry, MaxTopicSize, Topic, TopicDetails};
use crate::user::{
@@ -157,6 +158,30 @@ impl IggyClient {
})
}
+ /// Get the statistics and details of the server and its running process.
+ ///
+ /// Requires an authenticated session whose user holds the `read_servers`
+ /// or `manage_servers` global permission.
+ ///
+ /// Returns:
+ /// An awaitable that resolves to `Stats`.
+ ///
+ /// Raises:
+ /// RuntimeError: If the client is not connected, the session is not
+ /// authenticated, the user lacks the permission, or the request
+ /// fails.
+
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[Stats]",
imports=("collections.abc")))]
+ fn get_stats<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
+ let inner = self.inner.clone();
+ future_into_py(py, async move {
+ let stats = inner
+ .get_stats()
+ .await
+ .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?;
+ Ok(PyStats::from(stats))
+ })
+ }
+
/// Describe the option catalog for a resource scope.
///
/// This is the discovery surface for the `options` argument on
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 5f2e12826..d4397d5ab 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -24,6 +24,7 @@ mod options;
mod permissions;
mod receive_message;
mod send_message;
+mod stats;
mod stream;
mod topic;
mod user;
@@ -40,6 +41,7 @@ use permissions::{GlobalPermissions, Permissions,
StreamPermissions, TopicPermis
use pyo3::prelude::*;
use receive_message::{PollingStrategy, ReceiveMessage};
use send_message::{SendMessage, SendMessagesConfirmation,
SendMessagesResponse};
+use stats::{CacheMetrics, CacheMetricsKey, Stats};
use stream::StreamDetails;
use topic::{IggyExpiry, MaxTopicSize, Partition, Topic, TopicDetails};
use user::{UserInfo, UserInfoDetails, UserStatus};
@@ -57,6 +59,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) ->
PyResult<()> {
m.add_class::<TcpConfig>()?;
m.add_class::<TcpReconnectionConfig>()?;
m.add_class::<StreamDetails>()?;
+ m.add_class::<Stats>()?;
+ m.add_class::<CacheMetrics>()?;
+ m.add_class::<CacheMetricsKey>()?;
m.add_class::<Topic>()?;
m.add_class::<TopicDetails>()?;
m.add_class::<IggyExpiry>()?;
diff --git a/foreign/python/src/stats.rs b/foreign/python/src/stats.rs
new file mode 100644
index 000000000..af48425a0
--- /dev/null
+++ b/foreign/python/src/stats.rs
@@ -0,0 +1,449 @@
+// 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.
+
+use crate::duration::iggy_duration_to_py_delta;
+use iggy::prelude::{
+ CacheMetrics as RustCacheMetrics, CacheMetricsKey as RustCacheMetricsKey,
Stats as RustStats,
+};
+use pyo3::prelude::*;
+use pyo3::types::{PyDelta, PyDict};
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
+
+/// Key identifying the partition a `CacheMetrics` entry belongs to.
+///
+/// Hashable and comparable, so it can key the `Stats.cache_metrics` dict.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[gen_stub_pyclass]
+#[pyclass(eq, frozen, hash, skip_from_py_object)]
+pub struct CacheMetricsKey {
+ /// The unique identifier (numeric) of the stream.
+ #[pyo3(get)]
+ pub stream_id: u32,
+ /// The unique identifier (numeric) of the topic within the stream.
+ #[pyo3(get)]
+ pub topic_id: u32,
+ /// The unique identifier (numeric) of the partition within the topic.
+ #[pyo3(get)]
+ pub partition_id: u32,
+}
+
+impl From<&RustCacheMetricsKey> for CacheMetricsKey {
+ fn from(key: &RustCacheMetricsKey) -> Self {
+ Self {
+ stream_id: key.stream_id,
+ topic_id: key.topic_id,
+ partition_id: key.partition_id,
+ }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl CacheMetricsKey {
+ #[new]
+ fn new(stream_id: u32, topic_id: u32, partition_id: u32) -> Self {
+ Self {
+ stream_id,
+ topic_id,
+ partition_id,
+ }
+ }
+
+ fn __repr__(&self) -> String {
+ format!(
+ "CacheMetricsKey(stream_id={}, topic_id={}, partition_id={})",
+ self.stream_id, self.topic_id, self.partition_id
+ )
+ }
+}
+
+/// Cache metrics for a specific partition.
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct CacheMetrics {
+ /// Number of cache hits.
+ #[pyo3(get)]
+ pub hits: u64,
+ /// Number of cache misses.
+ #[pyo3(get)]
+ pub misses: u64,
+ /// Hit ratio (hits / (hits + misses)).
+ #[pyo3(get)]
+ pub hit_ratio: f32,
+}
+
+impl From<&RustCacheMetrics> for CacheMetrics {
+ fn from(metrics: &RustCacheMetrics) -> Self {
+ Self {
+ hits: metrics.hits,
+ misses: metrics.misses,
+ hit_ratio: metrics.hit_ratio,
+ }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl CacheMetrics {
+ fn __repr__(&self) -> String {
+ format!(
+ "CacheMetrics(hits={}, misses={}, hit_ratio={})",
+ self.hits, self.misses, self.hit_ratio
+ )
+ }
+}
+
+/// The statistics and details of the server and its running process.
+///
+/// The fields are gathered from several sources while the request is served
+/// (metadata counters, a process probe, a disk probe), so they are not an
+/// atomic snapshot of one instant.
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct Stats {
+ pub(crate) inner: RustStats,
+}
+
+impl From<RustStats> for Stats {
+ fn from(stats: RustStats) -> Self {
+ Self { inner: stats }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Stats {
+ /// The unique identifier of the server process.
+ #[getter]
+ pub fn process_id(&self) -> u32 {
+ self.inner.process_id
+ }
+
+ /// The CPU usage of the server process, in percent summed over the cores
+ /// it ran on, so it exceeds 100 whenever the process uses more than one
+ /// core.
+ ///
+ /// Measured as a delta since the previous `get_stats` served by the same
+ /// server shard, so the first sample a shard serves is 0.
+ #[getter]
+ pub fn cpu_usage(&self) -> f32 {
+ self.inner.cpu_usage
+ }
+
+ /// The total CPU usage of the system, in percent averaged over the cores
+ /// the server may run on when confined by an affinity/cpuset mask (over
+ /// every host core otherwise), so it stays within 0-100.
+ ///
+ /// Same per-shard delta sampling as `cpu_usage`: the first sample a shard
+ /// serves is 0.
+ #[getter]
+ pub fn total_cpu_usage(&self) -> f32 {
+ self.inner.total_cpu_usage
+ }
+
+ /// The memory usage of the server process, in bytes.
+ #[getter]
+ pub fn memory_usage(&self) -> u64 {
+ self.inner.memory_usage.as_bytes_u64()
+ }
+
+ /// The total memory of the system, in bytes, or the effective cgroup
memory
+ /// limit when the server runs inside a memory-capped cgroup (container,
+ /// systemd slice).
+ #[getter]
+ pub fn total_memory(&self) -> u64 {
+ self.inner.total_memory.as_bytes_u64()
+ }
+
+ /// The available memory of the system, in bytes, scoped to the cgroup
+ /// limit when one applies.
+ #[getter]
+ pub fn available_memory(&self) -> u64 {
+ self.inner.available_memory.as_bytes_u64()
+ }
+
+ /// The run time of the server process, with whole-second precision.
+ #[getter]
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta",
imports=("datetime")))]
+ pub fn run_time<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>>
{
+ iggy_duration_to_py_delta(py, self.inner.run_time)
+ }
+
+ /// The start time of the server process, in microseconds since the Unix
+ /// epoch, with whole-second precision.
+ #[getter]
+ pub fn start_time(&self) -> u64 {
+ self.inner.start_time.as_micros()
+ }
+
+ /// The total number of bytes read.
+ #[getter]
+ pub fn read_bytes(&self) -> u64 {
+ self.inner.read_bytes.as_bytes_u64()
+ }
+
+ /// The total number of bytes written.
+ #[getter]
+ pub fn written_bytes(&self) -> u64 {
+ self.inner.written_bytes.as_bytes_u64()
+ }
+
+ /// The total size of the messages, in bytes.
+ #[getter]
+ pub fn messages_size_bytes(&self) -> u64 {
+ self.inner.messages_size_bytes.as_bytes_u64()
+ }
+
+ /// The total number of streams.
+ #[getter]
+ pub fn streams_count(&self) -> u32 {
+ self.inner.streams_count
+ }
+
+ /// The total number of topics.
+ #[getter]
+ pub fn topics_count(&self) -> u32 {
+ self.inner.topics_count
+ }
+
+ /// The total number of partitions.
+ #[getter]
+ pub fn partitions_count(&self) -> u32 {
+ self.inner.partitions_count
+ }
+
+ /// The total number of segments.
+ #[getter]
+ pub fn segments_count(&self) -> u32 {
+ self.inner.segments_count
+ }
+
+ /// The total number of messages.
+ #[getter]
+ pub fn messages_count(&self) -> u64 {
+ self.inner.messages_count
+ }
+
+ /// The total number of connected clients.
+ #[getter]
+ pub fn clients_count(&self) -> u32 {
+ self.inner.clients_count
+ }
+
+ /// The total number of consumer groups.
+ #[getter]
+ pub fn consumer_groups_count(&self) -> u32 {
+ self.inner.consumer_groups_count
+ }
+
+ /// The name of the host the server runs on.
+ #[getter]
+ pub fn hostname(&self) -> String {
+ self.inner.hostname.clone()
+ }
+
+ /// The name of the operating system.
+ #[getter]
+ pub fn os_name(&self) -> String {
+ self.inner.os_name.clone()
+ }
+
+ /// The version of the operating system.
+ #[getter]
+ pub fn os_version(&self) -> String {
+ self.inner.os_version.clone()
+ }
+
+ /// The version of the kernel.
+ #[getter]
+ pub fn kernel_version(&self) -> String {
+ self.inner.kernel_version.clone()
+ }
+
+ /// The version of the Iggy server.
+ #[getter]
+ pub fn iggy_server_version(&self) -> String {
+ self.inner.iggy_server_version.clone()
+ }
+
+ /// The numeric semantic version of the Iggy server, or `None` when
unknown.
+ /// E.g. 1.2.3 -> 1002003 (major * 1000000 + minor * 1000 + patch).
+ #[getter]
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ pub fn iggy_server_semver(&self) -> Option<u32> {
+ self.inner.iggy_server_semver
+ }
+
+ /// Cache metrics per partition.
+ ///
+ /// Current servers do not populate this and reply with an empty map. Each
+ /// access builds a fresh dict, so mutating the returned dict does not
+ /// change the stats.
+ #[getter]
+ #[gen_stub(override_return_type(type_repr =
"builtins.dict[CacheMetricsKey, CacheMetrics]"))]
+ pub fn cache_metrics<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a,
PyDict>> {
+ let dict = PyDict::new(py);
+ for (key, metrics) in &self.inner.cache_metrics {
+ dict.set_item(CacheMetricsKey::from(key),
CacheMetrics::from(metrics))?;
+ }
+ Ok(dict)
+ }
+
+ /// The number of threads in the server process.
+ #[getter]
+ pub fn threads_count(&self) -> u32 {
+ self.inner.threads_count
+ }
+
+ /// The available (free) disk space for the data directory, in bytes.
+ ///
+ /// 0 when the server does not know its data directory or the disk probe
+ /// fails.
+ #[getter]
+ pub fn free_disk_space(&self) -> u64 {
+ self.inner.free_disk_space.as_bytes_u64()
+ }
+
+ /// The total disk space for the data directory, in bytes.
+ ///
+ /// 0 when the server does not know its data directory or the disk probe
+ /// fails.
+ #[getter]
+ pub fn total_disk_space(&self) -> u64 {
+ self.inner.total_disk_space.as_bytes_u64()
+ }
+
+ fn __repr__(&self) -> String {
+ format!(
+ "Stats(hostname='{}', iggy_server_version='{}', streams_count={}, \
+ topics_count={}, partitions_count={}, messages_count={},
clients_count={})",
+ self.inner.hostname,
+ self.inner.iggy_server_version,
+ self.inner.streams_count,
+ self.inner.topics_count,
+ self.inner.partitions_count,
+ self.inner.messages_count,
+ self.inner.clients_count
+ )
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy::prelude::{IggyByteSize, IggyDuration, IggyTimestamp};
+ use std::collections::HashMap;
+
+ /// Two entries sharing a stream and topic, so a key collision in the
+ /// `HashMap` -> `PyDict` conversion drops one of them.
+ fn cache_metrics_entries() -> HashMap<RustCacheMetricsKey,
RustCacheMetrics> {
+ HashMap::from([
+ (
+ RustCacheMetricsKey {
+ stream_id: 1,
+ topic_id: 2,
+ partition_id: 3,
+ },
+ RustCacheMetrics {
+ hits: 7,
+ misses: 3,
+ hit_ratio: 0.7,
+ },
+ ),
+ (
+ RustCacheMetricsKey {
+ stream_id: 1,
+ topic_id: 2,
+ partition_id: 4,
+ },
+ RustCacheMetrics {
+ hits: 0,
+ misses: 5,
+ hit_ratio: 0.0,
+ },
+ ),
+ ])
+ }
+
+ fn rust_stats(iggy_server_semver: Option<u32>) -> RustStats {
+ RustStats {
+ process_id: 1,
+ cpu_usage: 0.0,
+ total_cpu_usage: 0.0,
+ memory_usage: IggyByteSize::default(),
+ total_memory: IggyByteSize::default(),
+ available_memory: IggyByteSize::default(),
+ run_time: IggyDuration::default(),
+ start_time: IggyTimestamp::default(),
+ read_bytes: IggyByteSize::default(),
+ written_bytes: IggyByteSize::default(),
+ messages_size_bytes: IggyByteSize::default(),
+ streams_count: 0,
+ topics_count: 0,
+ partitions_count: 0,
+ segments_count: 0,
+ messages_count: 0,
+ clients_count: 0,
+ consumer_groups_count: 0,
+ hostname: String::new(),
+ os_name: String::new(),
+ os_version: String::new(),
+ kernel_version: String::new(),
+ iggy_server_version: String::new(),
+ iggy_server_semver,
+ cache_metrics: cache_metrics_entries(),
+ threads_count: 0,
+ free_disk_space: IggyByteSize::default(),
+ total_disk_space: IggyByteSize::default(),
+ }
+ }
+
+ #[test]
+ fn given_stats_when_converting_should_preserve_semver_option() {
+ Python::initialize();
+
+ assert_eq!(Stats::from(rust_stats(None)).iggy_server_semver(), None);
+ assert_eq!(
+ Stats::from(rust_stats(Some(1_002_003))).iggy_server_semver(),
+ Some(1_002_003)
+ );
+ }
+
+ #[test]
+ fn given_populated_cache_metrics_when_reading_should_key_every_entry() {
+ Python::initialize();
+
+ let stats = Stats::from(rust_stats(None));
+
+ Python::attach(|py| {
+ let dict = stats.cache_metrics(py).expect("build cache metrics
dict");
+ assert_eq!(dict.len(), cache_metrics_entries().len());
+
+ let entry = dict
+ .get_item(CacheMetricsKey::new(1, 2, 4))
+ .expect("look up cache metrics entry")
+ .expect("entry for an independently built key");
+ assert_eq!(
+ entry
+ .getattr("misses")
+ .and_then(|misses| misses.extract::<u64>())
+ .expect("read misses"),
+ 5
+ );
+ });
+ }
+}
diff --git a/foreign/python/tests/test_stats.py
b/foreign/python/tests/test_stats.py
new file mode 100644
index 000000000..2b4423cb8
--- /dev/null
+++ b/foreign/python/tests/test_stats.py
@@ -0,0 +1,250 @@
+# 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.
+
+import datetime
+
+import pytest
+
+from apache_iggy import CacheMetricsKey, GlobalPermissions, IggyClient,
Permissions
+from apache_iggy import SendMessage as Message
+
+from .utils import (
+ get_server_config,
+ login_fresh_client,
+ unique_credentials,
+ wait_for_server,
+)
+
+# Fields that describe the server process itself and must not change between
+# two calls within one server run.
+PROCESS_IDENTITY_FIELDS = (
+ "process_id",
+ "start_time",
+ "hostname",
+ "os_name",
+ "os_version",
+ "kernel_version",
+ "iggy_server_version",
+ "iggy_server_semver",
+)
+
+
+class TestStats:
+ """Test server statistics retrieval."""
+
+ @pytest.mark.asyncio
+ async def test_get_stats(self, iggy_client: IggyClient, unique_name):
+ """Sending messages moves the server counts reported by get_stats."""
+ stats_before = await iggy_client.get_stats()
+
+ stream_name = unique_name()
+ topic_name = unique_name()
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=0,
+ messages=[Message(f"stats message {i}") for i in range(3)],
+ )
+
+ stats = await iggy_client.get_stats()
+
+ # `>=` rather than exact equality: the counters are server-global, so
+ # concurrently running tests (e.g. under pytest-xdist) may bump them
too.
+ assert stats.streams_count >= stats_before.streams_count + 1
+ assert stats.topics_count >= stats_before.topics_count + 1
+ assert stats.partitions_count >= stats_before.partitions_count + 1
+ assert stats.messages_count >= stats_before.messages_count + 3
+ assert stats.messages_size_bytes > stats_before.messages_size_bytes
+ assert stats.clients_count >= 1
+
+ assert stats.iggy_server_version
+ assert stats.hostname
+ assert stats.os_name
+ assert stats.os_version
+ assert stats.kernel_version
+ assert stats.process_id > 0
+ # sysinfo cannot enumerate a process's threads on macOS, so a server
+ # running there reports 0.
+ if stats.os_name != "Darwin":
+ assert stats.threads_count > 0
+ assert stats.start_time > 0
+ assert stats.total_memory > 0
+ assert stats.available_memory <= stats.total_memory
+ assert stats.total_disk_space > 0
+ assert stats.free_disk_space <= stats.total_disk_space
+
+ assert isinstance(stats.run_time, datetime.timedelta)
+ assert stats.run_time >= stats_before.run_time
+ for field in PROCESS_IDENTITY_FIELDS:
+ assert getattr(stats, field) == getattr(stats_before, field)
+
+ assert f"streams_count={stats.streams_count}" in repr(stats)
+ assert stats.hostname in repr(stats)
+
+ @pytest.mark.asyncio
+ async def test_get_stats_reflects_topology(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Streams, topics, partitions, consumer groups and a second client
show
+ up in the counters, and deleting the streams brings them back down."""
+ stats_before = await iggy_client.get_stats()
+
+ streams = [unique_name() for _ in range(2)]
+ topics_per_stream = 2
+ partitions_per_topic = 3
+ topics = []
+ for stream_name in streams:
+ await iggy_client.create_stream(stream_name)
+ for _ in range(topics_per_stream):
+ topic_name = unique_name()
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=partitions_per_topic,
+ )
+ await iggy_client.create_consumer_group(
+ stream_name, topic_name, unique_name()
+ )
+ topics.append((stream_name, topic_name))
+ # Bound only to keep a second connection open until the test ends.
+ _second_client = await login_fresh_client("iggy", "iggy")
+
+ topics_created = len(topics)
+ partitions_created = topics_created * partitions_per_topic
+
+ stats = await iggy_client.get_stats()
+
+ assert stats.streams_count >= stats_before.streams_count + len(streams)
+ assert stats.topics_count >= stats_before.topics_count + topics_created
+ assert (
+ stats.partitions_count >= stats_before.partitions_count +
partitions_created
+ )
+ # Every new partition opens with one segment.
+ assert stats.segments_count >= stats_before.segments_count +
partitions_created
+ assert (
+ stats.consumer_groups_count
+ >= stats_before.consumer_groups_count + topics_created
+ )
+ # Both `iggy_client` and `_second_client` are connected at this point,
so
+ # the cross-shard total covers them regardless of what else the server
+ # reaped in between.
+ assert stats.clients_count >= 2
+
+ # The SDK has no delete_stream, so the streams stay behind and only the
+ # topic-level counters are expected to drop. The baseline is re-read
+ # right before the deletes to keep the window in which a concurrent
+ # test can bump the server-global counters as small as possible.
+ stats_before_delete = await iggy_client.get_stats()
+ for stream_name, topic_name in topics:
+ await iggy_client.delete_topic(stream_name, topic_name)
+
+ stats_after = await iggy_client.get_stats()
+
+ assert (
+ stats_after.topics_count
+ <= stats_before_delete.topics_count - topics_created
+ )
+ assert (
+ stats_after.partitions_count
+ <= stats_before_delete.partitions_count - partitions_created
+ )
+ assert (
+ stats_after.segments_count
+ <= stats_before_delete.segments_count - partitions_created
+ )
+ assert (
+ stats_after.consumer_groups_count
+ <= stats_before_delete.consumer_groups_count - topics_created
+ )
+
+ @pytest.mark.asyncio
+ async def test_get_stats_cache_metrics_dict(self, iggy_client: IggyClient):
+ """cache_metrics is a dict the server currently leaves empty."""
+ stats = await iggy_client.get_stats()
+
+ assert stats.cache_metrics == {}
+
+ @pytest.mark.unit
+ def test_cache_metrics_key_is_constructible_and_hashable(self):
+ """A key built in Python can address a cache_metrics dict entry."""
+ key = CacheMetricsKey(stream_id=1, topic_id=2, partition_id=3)
+
+ assert key.stream_id == 1
+ assert key.topic_id == 2
+ assert key.partition_id == 3
+ assert repr(key) == "CacheMetricsKey(stream_id=1, topic_id=2,
partition_id=3)"
+
+ equal_key = CacheMetricsKey(stream_id=1, topic_id=2, partition_id=3)
+ other_key = CacheMetricsKey(stream_id=1, topic_id=2, partition_id=4)
+ assert key == equal_key
+ assert key != other_key
+ assert hash(key) == hash(equal_key)
+
+ # An equal key constructed independently hits the same dict slot.
+ metrics_by_key = {key: "metrics"}
+ assert metrics_by_key[equal_key] == "metrics"
+ assert other_key not in metrics_by_key
+
+ @pytest.mark.asyncio
+ async def test_get_stats_requires_connection_and_auth(self):
+ """get_stats fails before connecting, before login, and after
logout."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(f"{host}:{port}")
+ with pytest.raises(RuntimeError, match="Not connected"):
+ await client.get_stats()
+
+ await client.connect()
+ with pytest.raises(RuntimeError, match="Unauthenticated"):
+ await client.get_stats()
+
+ await client.login_user("iggy", "iggy")
+ await client.get_stats()
+
+ await client.logout_user()
+ with pytest.raises(RuntimeError, match="Unauthenticated"):
+ await client.get_stats()
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("flag", ["read_servers", "manage_servers"])
+ async def test_get_stats_requires_server_permission(
+ self, iggy_client: IggyClient, unique_name, flag
+ ):
+ """A user without read_servers or manage_servers is denied; either
grants."""
+ username, password = unique_credentials(unique_name)
+ created = await iggy_client.create_user(username, password)
+
+ try:
+ denied_client = await login_fresh_client(username, password)
+ with pytest.raises(RuntimeError, match="Unauthorized"):
+ await denied_client.get_stats()
+
+ await iggy_client.update_permissions(
+ created.id,
+ Permissions(global_permissions=GlobalPermissions(**{flag:
True})),
+ )
+
+ granted_client = await login_fresh_client(username, password)
+ stats = await granted_client.get_stats()
+ assert stats.process_id > 0
+ finally:
+ await iggy_client.delete_user(created.id)
