From fe8d21d771c01f2b978a560dd1b03cdee3c9eaa6 Mon Sep 17 00:00:00 2001 From: 7487 <1042653432@qq.com> Date: Tue, 1 Sep 2026 15:20:14 +0800 Subject: [PATCH 1/2] feat(python): expose get_stats on IggyClient The Python SDK had no way to reach the server's headline diagnostic call, exposed by every other SDK. Wrap Stats, CacheMetrics and CacheMetricsKey in a new stats module following the user.rs pattern. CacheMetricsKey is frozen, hashable and comparable so cache_metrics maps to dict[CacheMetricsKey, CacheMetrics]. Byte sizes are exposed as integer bytes, times as microseconds, matching the existing getters. Closes #4016 --- foreign/python/apache_iggy.pyi | 208 +++++++++++++++++++++ foreign/python/src/client.rs | 20 ++ foreign/python/src/lib.rs | 5 + foreign/python/src/stats.rs | 289 +++++++++++++++++++++++++++++ foreign/python/tests/test_stats.py | 73 ++++++++ 5 files changed, 595 insertions(+) create mode 100644 foreign/python/src/stats.rs create mode 100644 foreign/python/tests/test_stats.py diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 6c3715b54b..28c4a8034e 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,54 @@ class AutoLogin: """ def __repr__(self) -> builtins.str: ... +@typing.final +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: ... + +@typing.final +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 __repr__(self) -> builtins.str: ... + class Consumer: r""" The consumer polling the messages. It selects both the consumer kind and the @@ -872,6 +923,16 @@ 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. + + Returns: + An awaitable that resolves to `Stats`. + + Raises: + RuntimeError: If the request fails. + """ def describe_options( self, scope: builtins.str ) -> collections.abc.Awaitable[list[OptionSpec]]: @@ -1834,6 +1895,153 @@ class SendMessagesResponse: with an offset a client has already recorded. """ +@typing.final +class Stats: + r""" + The statistics and details of the server and its running process. + """ + @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. + """ + @property + def total_cpu_usage(self) -> builtins.float: + r""" + The total CPU usage of the system, in percent. + """ + @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. + """ + @property + def available_memory(self) -> builtins.int: + r""" + The available memory of the system, in bytes. + """ + @property + def run_time(self) -> builtins.int: + r""" + The run time of the server process, in microseconds. + """ + @property + def start_time(self) -> builtins.int: + r""" + The start time of the server process, in microseconds since the Unix epoch. + """ + @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) -> int | None: + r""" + The numeric semantic version of the Iggy server, or `None` when unknown. + E.g. 1.2.3 -> 100200300 (major * 1000000 + minor * 1000 + patch). + """ + @property + def cache_metrics(self) -> builtins.dict[CacheMetricsKey, CacheMetrics]: + r""" + Cache metrics per partition. + """ + @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. + """ + @property + def total_disk_space(self) -> builtins.int: + r""" + The total disk space for the data directory, in bytes. + """ + @typing.final class StreamDetails: @property diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index f669a87484..fdeffedab9 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,25 @@ impl IggyClient { }) } + /// Get the statistics and details of the server and its running process. + /// + /// Returns: + /// An awaitable that resolves to `Stats`. + /// + /// Raises: + /// RuntimeError: If 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> { + let inner = self.inner.clone(); + future_into_py(py, async move { + let stats = inner + .get_stats() + .await + .map_err(|e| PyErr::new::(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 5f2e128264..d4397d5ab0 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::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/foreign/python/src/stats.rs b/foreign/python/src/stats.rs new file mode 100644 index 0000000000..c040552977 --- /dev/null +++ b/foreign/python/src/stats.rs @@ -0,0 +1,289 @@ +// 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 iggy::prelude::{ + CacheMetrics as RustCacheMetrics, CacheMetricsKey as RustCacheMetricsKey, Stats as RustStats, +}; +use pyo3::prelude::*; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use std::collections::HashMap; + +/// 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 { + 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. +#[gen_stub_pyclass] +#[pyclass] +pub struct Stats { + pub(crate) inner: RustStats, +} + +impl From 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. + #[getter] + pub fn cpu_usage(&self) -> f32 { + self.inner.cpu_usage + } + + /// The total CPU usage of the system, in percent. + #[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. + #[getter] + pub fn total_memory(&self) -> u64 { + self.inner.total_memory.as_bytes_u64() + } + + /// The available memory of the system, in bytes. + #[getter] + pub fn available_memory(&self) -> u64 { + self.inner.available_memory.as_bytes_u64() + } + + /// The run time of the server process, in microseconds. + #[getter] + pub fn run_time(&self) -> u64 { + self.inner.run_time.as_micros() + } + + /// The start time of the server process, in microseconds since the Unix epoch. + #[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 -> 100200300 (major * 1000000 + minor * 1000 + patch). + #[getter] + #[gen_stub(override_return_type(type_repr = "int | None"))] + pub fn iggy_server_semver(&self) -> Option { + self.inner.iggy_server_semver + } + + /// Cache metrics per partition. + #[getter] + pub fn cache_metrics(&self) -> HashMap { + self.inner + .cache_metrics + .iter() + .map(|(key, metrics)| (CacheMetricsKey::from(key), CacheMetrics::from(metrics))) + .collect() + } + + /// 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. + #[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. + #[getter] + pub fn total_disk_space(&self) -> u64 { + self.inner.total_disk_space.as_bytes_u64() + } +} diff --git a/foreign/python/tests/test_stats.py b/foreign/python/tests/test_stats.py new file mode 100644 index 0000000000..4a7f231d05 --- /dev/null +++ b/foreign/python/tests/test_stats.py @@ -0,0 +1,73 @@ +# 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 pytest + +from apache_iggy import CacheMetrics, CacheMetricsKey, IggyClient +from apache_iggy import SendMessage as Message + + +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() + + 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.clients_count >= 1 + + assert stats.iggy_server_version + assert stats.hostname + assert stats.process_id > 0 + assert stats.start_time > 0 + assert stats.total_memory > 0 + assert stats.total_disk_space > 0 + + @pytest.mark.asyncio + async def test_get_stats_cache_metrics_dict(self, iggy_client: IggyClient): + """cache_metrics is a dict keyed by hashable CacheMetricsKey.""" + stats = await iggy_client.get_stats() + + assert isinstance(stats.cache_metrics, dict) + for key, metrics in stats.cache_metrics.items(): + assert isinstance(key, CacheMetricsKey) + assert isinstance(metrics, CacheMetrics) + # The key round-trips through a dict lookup. + assert stats.cache_metrics[key] is not None + assert key.stream_id >= 0 + assert metrics.hits >= 0 + assert metrics.misses >= 0 From 7a2823596392529ae16cb581a59036699e0a6ff8 Mon Sep 17 00:00:00 2001 From: 7487 <1042653432@qq.com> Date: Fri, 4 Sep 2026 18:26:29 +0800 Subject: [PATCH 2/2] fix(python): address review feedback on get_stats - CacheMetricsKey gets a #[new] constructor so a key built in Python can address a cache_metrics dict entry directly. - cache_metrics is converted once in From and stored as a Py; every access returns the same dict instead of re-collecting the whole map. - run_time is exposed as datetime.timedelta via the shared duration helper, matching the SDK's other duration surfaces. - The semver stub override uses the builtins.int | None convention, and Stats gains a __repr__ with the headline fields. - The numeric semver docstring example is corrected to 1.2.3 -> 1002003 here and in core/common (get_numeric_version pads minor/patch to three digits). - Tests compare server-global counters with >= (pytest-xdist safe), drop assertions that could not fail, and cover key construction, hashing and dict addressing without a server. The cache metrics map itself stays empty for now: the server replies with a hardcoded empty map. Co-Authored-By: Claude Fable 5 --- core/common/src/types/stats/mod.rs | 2 +- foreign/python/apache_iggy.pyi | 15 +++++-- foreign/python/src/stats.rs | 65 ++++++++++++++++++++++++------ foreign/python/tests/test_stats.py | 48 +++++++++++++++++----- 4 files changed, 103 insertions(+), 27 deletions(-) diff --git a/core/common/src/types/stats/mod.rs b/core/common/src/types/stats/mod.rs index 5de7f03b43..fcec5cf4c6 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, /// Cache metrics per partition #[serde(with = "cache_metrics_serializer")] diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 5ae0c93839..8719844e0e 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -339,6 +339,9 @@ class CacheMetricsKey: """ 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: @@ -1933,9 +1936,9 @@ class Stats: The available memory of the system, in bytes. """ @property - def run_time(self) -> builtins.int: + def run_time(self) -> datetime.timedelta: r""" - The run time of the server process, in microseconds. + The run time of the server process. """ @property def start_time(self) -> builtins.int: @@ -2018,15 +2021,18 @@ class Stats: The version of the Iggy server. """ @property - def iggy_server_semver(self) -> int | None: + 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 -> 100200300 (major * 1000000 + minor * 1000 + patch). + 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. + + Built once when the stats snapshot is created; every access returns the + same dict. """ @property def threads_count(self) -> builtins.int: @@ -2043,6 +2049,7 @@ class Stats: r""" The total disk space for the data directory, in bytes. """ + def __repr__(self) -> builtins.str: ... @typing.final class StreamDetails: diff --git a/foreign/python/src/stats.rs b/foreign/python/src/stats.rs index c040552977..1f3e876f89 100644 --- a/foreign/python/src/stats.rs +++ b/foreign/python/src/stats.rs @@ -15,12 +15,13 @@ // 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}; -use std::collections::HashMap; /// Key identifying the partition a `CacheMetrics` entry belongs to. /// @@ -53,6 +54,15 @@ impl From<&RustCacheMetricsKey> for CacheMetricsKey { #[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={})", @@ -102,11 +112,25 @@ impl CacheMetrics { #[pyclass] pub struct Stats { pub(crate) inner: RustStats, + /// Converted once here so that every `cache_metrics` access returns the + /// same dict instead of re-collecting the whole map. + cache_metrics: Py, } impl From for Stats { fn from(stats: RustStats) -> Self { - Self { inner: stats } + let cache_metrics = Python::attach(|py| { + let dict = PyDict::new(py); + for (key, metrics) in &stats.cache_metrics { + dict.set_item(CacheMetricsKey::from(key), CacheMetrics::from(metrics)) + .expect("insert cache metrics entry"); + } + dict.unbind() + }); + Self { + inner: stats, + cache_metrics, + } } } @@ -149,10 +173,11 @@ impl Stats { self.inner.available_memory.as_bytes_u64() } - /// The run time of the server process, in microseconds. + /// The run time of the server process. #[getter] - pub fn run_time(&self) -> u64 { - self.inner.run_time.as_micros() + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + pub fn run_time<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.run_time) } /// The start time of the server process, in microseconds since the Unix epoch. @@ -252,21 +277,21 @@ impl Stats { } /// The numeric semantic version of the Iggy server, or `None` when unknown. - /// E.g. 1.2.3 -> 100200300 (major * 1000000 + minor * 1000 + patch). + /// E.g. 1.2.3 -> 1002003 (major * 1000000 + minor * 1000 + patch). #[getter] - #[gen_stub(override_return_type(type_repr = "int | None"))] + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] pub fn iggy_server_semver(&self) -> Option { self.inner.iggy_server_semver } /// Cache metrics per partition. + /// + /// Built once when the stats snapshot is created; every access returns the + /// same dict. #[getter] - pub fn cache_metrics(&self) -> HashMap { - self.inner - .cache_metrics - .iter() - .map(|(key, metrics)| (CacheMetricsKey::from(key), CacheMetrics::from(metrics))) - .collect() + #[gen_stub(override_return_type(type_repr = "builtins.dict[CacheMetricsKey, CacheMetrics]"))] + pub fn cache_metrics(&self, py: Python<'_>) -> Py { + self.cache_metrics.clone_ref(py) } /// The number of threads in the server process. @@ -286,4 +311,18 @@ impl Stats { 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 + ) + } } diff --git a/foreign/python/tests/test_stats.py b/foreign/python/tests/test_stats.py index 4a7f231d05..59b2148e90 100644 --- a/foreign/python/tests/test_stats.py +++ b/foreign/python/tests/test_stats.py @@ -15,6 +15,8 @@ # specific language governing permissions and limitations # under the License. +import datetime + import pytest from apache_iggy import CacheMetrics, CacheMetricsKey, IggyClient @@ -44,9 +46,11 @@ async def test_get_stats(self, iggy_client: IggyClient, unique_name): stats = await iggy_client.get_stats() - 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 + # `>=` 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.clients_count >= 1 @@ -57,17 +61,43 @@ async def test_get_stats(self, iggy_client: IggyClient, unique_name): assert stats.total_memory > 0 assert stats.total_disk_space > 0 + assert isinstance(stats.run_time, datetime.timedelta) + assert stats.run_time >= stats_before.run_time + + assert f"streams_count={stats.streams_count}" in repr(stats) + assert stats.hostname in repr(stats) + @pytest.mark.asyncio async def test_get_stats_cache_metrics_dict(self, iggy_client: IggyClient): - """cache_metrics is a dict keyed by hashable CacheMetricsKey.""" + """cache_metrics is a dict keyed by CacheMetricsKey, and repeated + accesses return the same dict rather than rebuilding it.""" stats = await iggy_client.get_stats() assert isinstance(stats.cache_metrics, dict) + # The getter must not re-collect the map on every access. + assert stats.cache_metrics is stats.cache_metrics + # The server does not populate cache metrics yet (`GetStats` replies + # with an empty map), so entries are only checked when present. for key, metrics in stats.cache_metrics.items(): assert isinstance(key, CacheMetricsKey) assert isinstance(metrics, CacheMetrics) - # The key round-trips through a dict lookup. - assert stats.cache_metrics[key] is not None - assert key.stream_id >= 0 - assert metrics.hits >= 0 - assert metrics.misses >= 0 + + 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