You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/08/16 18:05:34 UTC
(iggy) branch io_uring_tpc updated: feat(io_uring): Make metrics shared between different shards (#2104)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new a7d09f06 feat(io_uring): Make metrics shared between different shards (#2104)
a7d09f06 is described below
commit a7d09f06cdb016e1f52da015518ada017cca5194
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 16 23:35:31 2025 +0530
feat(io_uring): Make metrics shared between different shards (#2104)
Refactor shard local metrics to be an `Arc` so they can be shared and
accessed across shards.
Closes: https://github.com/apache/iggy/issues/2054
---
core/server/src/main.rs | 5 ++
core/server/src/shard/builder.rs | 17 ++++-
core/server/src/shard/mod.rs | 2 -
.../src/shard/tasks/auxilary/maintain_messages.rs | 8 +--
core/server/src/streaming/diagnostics/metrics.rs | 80 ++++++++++++----------
5 files changed, 67 insertions(+), 45 deletions(-)
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 02a2ae95..6d72d048 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -52,6 +52,7 @@ use server::state::command::EntryCommand;
use server::state::file::FileState;
use server::state::models::CreateUserWithId;
use server::state::system::SystemState;
+use server::streaming::diagnostics::metrics::Metrics;
use server::streaming::storage::SystemStorage;
use server::streaming::utils::MemoryPool;
use server::versioning::SemanticVersion;
@@ -228,6 +229,8 @@ async fn main() -> Result<(), ServerError> {
}
}
+ let shared_metrics = Metrics::init();
+
// TWELFTH DISCRETE LOADING STEP.
info!("Starting {} shard(s)", shards_set.len());
let (connections, shutdown_handles) = create_shard_connections(&shards_set);
@@ -241,6 +244,7 @@ async fn main() -> Result<(), ServerError> {
let encryptor = encryptor.clone();
let archiver = archiver.clone();
let init_state = init_state.clone();
+ let shared_metrics = shared_metrics.clone();
let state_persister = resolve_persister(config.system.state.enforce_fsync);
let state = StateKind::File(FileState::new(
&config.system.get_state_messages_file_path(),
@@ -266,6 +270,7 @@ async fn main() -> Result<(), ServerError> {
.version(current_version)
.state(state)
.init_state(init_state)
+ .metrics(shared_metrics)
.build();
let shard = Rc::new(shard);
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index 67a17043..153a474d 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -16,7 +16,11 @@
* under the License.
*/
-use std::{cell::Cell, rc::Rc, sync::atomic::AtomicBool};
+use std::{
+ cell::Cell,
+ rc::Rc,
+ sync::{Arc, atomic::AtomicBool},
+};
use iggy_common::{Aes256GcmEncryptor, EncryptorKind};
use tracing::info;
@@ -45,6 +49,7 @@ pub struct IggyShardBuilder {
archiver: Option<ArchiverKind>,
state: Option<StateKind>,
init_state: Option<SystemState>,
+ metrics: Option<Metrics>,
}
impl IggyShardBuilder {
@@ -92,6 +97,11 @@ impl IggyShardBuilder {
self
}
+ pub fn metrics(mut self, metrics: Metrics) -> Self {
+ self.metrics = Some(metrics);
+ self
+ }
+
// TODO: Too much happens in there, some of those bootstrapping logic should be moved outside.
pub fn build(self) -> IggyShard {
let id = self.id.unwrap();
@@ -117,6 +127,9 @@ impl IggyShardBuilder {
let shards = connections.into_iter().map(Shard::new).collect();
let archiver = self.archiver.map(Rc::new);
let storage = Rc::new(storage);
+
+ // Initialize metrics
+ let metrics = self.metrics.unwrap_or_else(|| Metrics::init());
IggyShard {
id: id,
shards: shards,
@@ -131,7 +144,7 @@ impl IggyShardBuilder {
stop_receiver: stop_receiver,
stop_sender: stop_sender,
messages_receiver: Cell::new(Some(frame_receiver)),
- metrics: Metrics::init(),
+ metrics: metrics,
task_registry: TaskRegistry::new(),
is_shutting_down: AtomicBool::new(false),
tcp_bound_address: Cell::new(None),
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index e7cb032a..14dc9101 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -160,7 +160,6 @@ pub struct IggyShard {
pub(crate) active_sessions: RefCell<Vec<Rc<Session>>>,
pub(crate) permissioner: RefCell<Permissioner>,
pub(crate) users: RefCell<HashMap<UserId, User>>,
-
pub(crate) metrics: Metrics,
pub messages_receiver: Cell<Option<Receiver<ShardFrame>>>,
pub(crate) stop_receiver: StopReceiver,
@@ -178,7 +177,6 @@ impl IggyShard {
pub fn default_from_config(server_config: ServerConfig) -> Self {
use crate::bootstrap::resolve_persister;
use crate::state::file::FileState;
- use crate::streaming::diagnostics::metrics::Metrics;
use crate::streaming::storage::SystemStorage;
use crate::versioning::SemanticVersion;
diff --git a/core/server/src/shard/tasks/auxilary/maintain_messages.rs b/core/server/src/shard/tasks/auxilary/maintain_messages.rs
index 5ae8779f..99a9a995 100644
--- a/core/server/src/shard/tasks/auxilary/maintain_messages.rs
+++ b/core/server/src/shard/tasks/auxilary/maintain_messages.rs
@@ -98,12 +98,8 @@ pub async fn spawn_message_maintainance_task(shard: Rc<IggyShard>) -> Result<(),
topic.topic_id
);
- shard
- .metrics
- .decrement_segments(deleted_segments.segments_count);
- shard
- .metrics
- .decrement_messages(deleted_segments.messages_count);
+ shard.metrics.decrement_segments(deleted_segments.segments_count);
+ shard.metrics.decrement_messages(deleted_segments.messages_count);
}
}
}
diff --git a/core/server/src/streaming/diagnostics/metrics.rs b/core/server/src/streaming/diagnostics/metrics.rs
index b5a602c5..aa24c050 100644
--- a/core/server/src/streaming/diagnostics/metrics.rs
+++ b/core/server/src/streaming/diagnostics/metrics.rs
@@ -20,11 +20,12 @@ use prometheus_client::encoding::text::encode;
use prometheus_client::metrics::counter::Counter;
use prometheus_client::metrics::gauge::Gauge;
use prometheus_client::registry::Registry;
+use std::sync::Arc;
use tracing::error;
-#[derive(Debug)]
-pub(crate) struct Metrics {
- registry: Registry,
+#[derive(Debug, Clone)]
+pub struct Metrics {
+ registry: Arc<Registry>,
http_requests: Counter,
streams: Gauge,
topics: Gauge,
@@ -37,38 +38,47 @@ pub(crate) struct Metrics {
impl Metrics {
pub fn init() -> Self {
- let mut metrics = Metrics {
- registry: <Registry>::default(),
- http_requests: Counter::default(),
- streams: Gauge::default(),
- topics: Gauge::default(),
- partitions: Gauge::default(),
- segments: Gauge::default(),
- messages: Gauge::default(),
- users: Gauge::default(),
- clients: Gauge::default(),
- };
-
- metrics.register_counter("http_requests", metrics.http_requests.clone());
- metrics.register_gauge("streams", metrics.streams.clone());
- metrics.register_gauge("topics", metrics.topics.clone());
- metrics.register_gauge("partitions", metrics.partitions.clone());
- metrics.register_gauge("segments", metrics.segments.clone());
- metrics.register_gauge("messages", metrics.messages.clone());
- metrics.register_gauge("users", metrics.users.clone());
- metrics.register_gauge("clients", metrics.clients.clone());
-
- metrics
- }
-
- fn register_counter(&mut self, name: &str, counter: Counter) {
- self.registry
- .register(name, format!("total count of {name}"), counter)
- }
-
- fn register_gauge(&mut self, name: &str, gauge: Gauge) {
- self.registry
- .register(name, format!("total count of {name}"), gauge)
+ let mut registry = Registry::default();
+
+ let http_requests = Counter::default();
+ let streams = Gauge::default();
+ let topics = Gauge::default();
+ let partitions = Gauge::default();
+ let segments = Gauge::default();
+ let messages = Gauge::default();
+ let users = Gauge::default();
+ let clients = Gauge::default();
+
+ registry.register(
+ "http_requests",
+ "total count of http_requests",
+ http_requests.clone(),
+ );
+ registry.register("streams", "total count of streams", streams.clone());
+ registry.register("topics", "total count of topics", topics.clone());
+ registry.register(
+ "partitions",
+ "total count of partitions",
+ partitions.clone(),
+ );
+ registry.register("segments", "total count of segments", segments.clone());
+ registry.register("messages", "total count of messages", messages.clone());
+ registry.register("users", "total count of users", users.clone());
+ registry.register("clients", "total count of clients", clients.clone());
+
+ let registry = Arc::new(registry);
+
+ Metrics {
+ registry,
+ http_requests,
+ streams,
+ topics,
+ partitions,
+ segments,
+ messages,
+ users,
+ clients,
+ }
}
pub fn get_formatted_output(&self) -> String {