You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2025/09/02 19:59:16 UTC
(iggy) branch get-cluster-metadata created (now a0df4ecb)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch get-cluster-metadata
in repository https://gitbox.apache.org/repos/asf/iggy.git
at a0df4ecb feat(server,sdk): implement GetClusterMetadata command
This branch includes the following new commits:
new a0df4ecb feat(server,sdk): implement GetClusterMetadata command
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(server,sdk): implement GetClusterMetadata command
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch get-cluster-metadata
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit a0df4ecbd2c4f387cd74d2c3d74c18bcf6d95033
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Sep 2 21:55:55 2025 +0200
feat(server,sdk): implement GetClusterMetadata command
---
.../src/commands/system/get_cluster_metadata.rs | 59 ++++++++
core/common/src/commands/system/mod.rs | 1 +
core/common/src/lib.rs | 2 +
core/common/src/types/cluster/metadata.rs | 154 +++++++++++++++++++++
.../src/{commands/system => types/cluster}/mod.rs | 10 +-
core/common/src/types/cluster/node.rs | 140 +++++++++++++++++++
core/common/src/types/cluster/role.rs | 72 ++++++++++
core/common/src/types/cluster/status.rs | 63 +++++++++
core/common/src/types/command/mod.rs | 3 +
core/common/src/types/mod.rs | 1 +
core/configs/server.toml | 39 ++++++
core/integration/tests/streaming/segment.rs | 2 +
core/integration/tests/streaming/snapshot.rs | 2 +
core/integration/tests/streaming/system.rs | 5 +
core/server/src/binary/command.rs | 2 +
.../cluster/get_cluster_metadata_handler.rs | 63 +++++++++
.../src/binary/handlers/cluster}/mod.rs | 7 +-
core/server/src/binary/handlers/mod.rs | 1 +
.../systems/mod.rs => configs/cluster.rs} | 36 ++---
core/server/src/configs/defaults.rs | 34 ++++-
core/server/src/configs/mod.rs | 1 +
core/server/src/configs/server.rs | 2 +
core/server/src/main.rs | 1 +
core/server/src/streaming/systems/cluster.rs | 68 +++++++++
core/server/src/streaming/systems/mod.rs | 1 +
core/server/src/streaming/systems/streams.rs | 1 +
core/server/src/streaming/systems/system.rs | 6 +
27 files changed, 747 insertions(+), 29 deletions(-)
diff --git a/core/common/src/commands/system/get_cluster_metadata.rs b/core/common/src/commands/system/get_cluster_metadata.rs
new file mode 100644
index 00000000..ad9693ab
--- /dev/null
+++ b/core/common/src/commands/system/get_cluster_metadata.rs
@@ -0,0 +1,59 @@
+/* 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::error::IggyError;
+use crate::{BytesSerializable, Command, GET_CLUSTER_METADATA_CODE, Validatable};
+use bytes::Bytes;
+use serde::{Deserialize, Serialize};
+use std::fmt::Display;
+
+/// `GetClusterMetadata` command is used to retrieve cluster metadata including
+/// available nodes, their roles, and connection information.
+#[derive(Debug, Default, Serialize, Deserialize, PartialEq)]
+pub struct GetClusterMetadata {}
+
+impl Command for GetClusterMetadata {
+ fn code(&self) -> u32 {
+ GET_CLUSTER_METADATA_CODE
+ }
+}
+
+impl Validatable<IggyError> for GetClusterMetadata {
+ fn validate(&self) -> Result<(), IggyError> {
+ Ok(())
+ }
+}
+
+impl BytesSerializable for GetClusterMetadata {
+ fn to_bytes(&self) -> Bytes {
+ Bytes::new()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<GetClusterMetadata, IggyError> {
+ if !bytes.is_empty() {
+ return Err(IggyError::InvalidCommand);
+ }
+ Ok(GetClusterMetadata {})
+ }
+}
+
+impl Display for GetClusterMetadata {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(f, "GetClusterMetadata")
+ }
+}
diff --git a/core/common/src/commands/system/mod.rs b/core/common/src/commands/system/mod.rs
index e6aa4ac3..862f8222 100644
--- a/core/common/src/commands/system/mod.rs
+++ b/core/common/src/commands/system/mod.rs
@@ -18,6 +18,7 @@
pub mod get_client;
pub mod get_clients;
+pub mod get_cluster_metadata;
pub mod get_me;
pub mod get_snapshot;
pub mod get_stats;
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index 6603a7d5..2ae8a027 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -35,6 +35,7 @@ pub use commands::partitions::*;
pub use commands::personal_access_tokens::*;
pub use commands::segments::*;
pub use commands::streams::*;
+pub use commands::system::get_cluster_metadata::*;
pub use commands::system::*;
pub use commands::topics::*;
pub use commands::users::*;
@@ -47,6 +48,7 @@ pub use traits::validatable::Validatable;
pub use types::args::*;
pub use types::client::client_info::*;
pub use types::client_state::ClientState;
+pub use types::cluster::*;
pub use types::command::*;
pub use types::compression::compression_algorithm::*;
pub use types::configuration::auth_config::auto_login::*;
diff --git a/core/common/src/types/cluster/metadata.rs b/core/common/src/types/cluster/metadata.rs
new file mode 100644
index 00000000..636a5039
--- /dev/null
+++ b/core/common/src/types/cluster/metadata.rs
@@ -0,0 +1,154 @@
+/* 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::{BytesSerializable, IggyError, types::cluster::node::ClusterNode};
+use bytes::{BufMut, Bytes, BytesMut};
+use serde::{Deserialize, Serialize};
+use std::fmt::Display;
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ClusterMetadata {
+ pub name: String,
+ pub node_id: u32,
+ pub nodes: Vec<ClusterNode>,
+}
+
+impl BytesSerializable for ClusterMetadata {
+ fn to_bytes(&self) -> Bytes {
+ let name_bytes = self.name.as_bytes();
+
+ // Calculate size for each node
+ let nodes_size: usize = self.nodes.iter().map(|node| node.get_buffer_size()).sum();
+ let size = 4 + name_bytes.len() + 4 + 4 + nodes_size; // name_len + name + node_id + nodes_len + nodes
+
+ let mut bytes = BytesMut::with_capacity(size);
+
+ // Write name length and name
+ bytes.put_u32_le(name_bytes.len() as u32);
+ bytes.put_slice(name_bytes);
+
+ // Write node_id
+ bytes.put_u32_le(self.node_id);
+
+ // Write nodes count
+ bytes.put_u32_le(self.nodes.len() as u32);
+
+ // Write each node
+ for node in &self.nodes {
+ node.write_to_buffer(&mut bytes);
+ }
+
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<ClusterMetadata, IggyError> {
+ if bytes.len() < 12 {
+ return Err(IggyError::InvalidCommand);
+ }
+
+ let mut position = 0;
+
+ // Read name length
+ let name_len = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ) as usize;
+ position += 4;
+
+ // Read name
+ if bytes.len() < position + name_len {
+ return Err(IggyError::InvalidCommand);
+ }
+ let name = String::from_utf8(bytes[position..position + name_len].to_vec())
+ .map_err(|_| IggyError::InvalidCommand)?;
+ position += name_len;
+
+ // Read node_id
+ if bytes.len() < position + 4 {
+ return Err(IggyError::InvalidCommand);
+ }
+ let node_id = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ position += 4;
+
+ // Read nodes count
+ if bytes.len() < position + 4 {
+ return Err(IggyError::InvalidCommand);
+ }
+ let nodes_count = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ) as usize;
+ position += 4;
+
+ // Read nodes
+ let mut nodes = Vec::with_capacity(nodes_count);
+ for _ in 0..nodes_count {
+ let node = ClusterNode::from_bytes(bytes.slice(position..))?;
+ position += node.get_buffer_size();
+ nodes.push(node);
+ }
+
+ Ok(ClusterMetadata {
+ name,
+ node_id,
+ nodes,
+ })
+ }
+
+ fn write_to_buffer(&self, buf: &mut BytesMut) {
+ let name_bytes = self.name.as_bytes();
+
+ // Write name length and name
+ buf.put_u32_le(name_bytes.len() as u32);
+ buf.put_slice(name_bytes);
+
+ // Write node_id
+ buf.put_u32_le(self.node_id);
+
+ // Write nodes count
+ buf.put_u32_le(self.nodes.len() as u32);
+
+ // Write each node
+ for node in &self.nodes {
+ node.write_to_buffer(buf);
+ }
+ }
+
+ fn get_buffer_size(&self) -> usize {
+ let nodes_size: usize = self.nodes.iter().map(|node| node.get_buffer_size()).sum();
+ 4 + self.name.len() + 4 + 4 + nodes_size // name_len + name + node_id + nodes_len + nodes
+ }
+}
+
+impl Display for ClusterMetadata {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "ClusterMetadata {{ id: {}, node_id: {}, nodes: {} }}",
+ self.name,
+ self.node_id,
+ self.nodes.len()
+ )
+ }
+}
diff --git a/core/common/src/commands/system/mod.rs b/core/common/src/types/cluster/mod.rs
similarity index 87%
copy from core/common/src/commands/system/mod.rs
copy to core/common/src/types/cluster/mod.rs
index e6aa4ac3..06c5ed89 100644
--- a/core/common/src/commands/system/mod.rs
+++ b/core/common/src/types/cluster/mod.rs
@@ -16,9 +16,7 @@
* under the License.
*/
-pub mod get_client;
-pub mod get_clients;
-pub mod get_me;
-pub mod get_snapshot;
-pub mod get_stats;
-pub mod ping;
+pub mod metadata;
+pub mod node;
+pub mod role;
+pub mod status;
diff --git a/core/common/src/types/cluster/node.rs b/core/common/src/types/cluster/node.rs
new file mode 100644
index 00000000..aa230804
--- /dev/null
+++ b/core/common/src/types/cluster/node.rs
@@ -0,0 +1,140 @@
+/* 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::{
+ BytesSerializable, IggyError,
+ types::cluster::{role::ClusterNodeRole, status::ClusterNodeStatus},
+};
+use bytes::{BufMut, Bytes, BytesMut};
+use serde::{Deserialize, Serialize};
+use std::fmt::Display;
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ClusterNode {
+ pub id: u32,
+ pub name: String,
+ pub address: String,
+ pub role: ClusterNodeRole,
+ pub status: ClusterNodeStatus,
+}
+
+impl BytesSerializable for ClusterNode {
+ fn to_bytes(&self) -> Bytes {
+ let size = self.get_buffer_size();
+ let mut bytes = BytesMut::with_capacity(size);
+ self.write_to_buffer(&mut bytes);
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<Self, IggyError> {
+ if bytes.len() < 10 {
+ // Minimum: 4 (id) + 4 (name_len) + 4 (address_len) + 1 (role) + 1 (status)
+ return Err(IggyError::InvalidCommand);
+ }
+
+ let mut position = 0;
+
+ // Read id
+ let id = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ position += 4;
+
+ // Read name length
+ let name_len = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ) as usize;
+ position += 4;
+
+ // Read name
+ if bytes.len() < position + name_len {
+ return Err(IggyError::InvalidCommand);
+ }
+ let name = String::from_utf8(bytes[position..position + name_len].to_vec())
+ .map_err(|_| IggyError::InvalidCommand)?;
+ position += name_len;
+
+ // Read address length
+ if bytes.len() < position + 4 {
+ return Err(IggyError::InvalidCommand);
+ }
+ let address_len = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ) as usize;
+ position += 4;
+
+ // Read address
+ if bytes.len() < position + address_len {
+ return Err(IggyError::InvalidCommand);
+ }
+ let address = String::from_utf8(bytes[position..position + address_len].to_vec())
+ .map_err(|_| IggyError::InvalidCommand)?;
+ position += address_len;
+
+ // Read role
+ if bytes.len() < position + 1 {
+ return Err(IggyError::InvalidCommand);
+ }
+ let role = ClusterNodeRole::try_from(bytes[position])?;
+ position += 1;
+
+ // Read status
+ if bytes.len() < position + 1 {
+ return Err(IggyError::InvalidCommand);
+ }
+ let status = ClusterNodeStatus::try_from(bytes[position])?;
+
+ Ok(ClusterNode {
+ id,
+ name,
+ address,
+ role,
+ status,
+ })
+ }
+
+ fn write_to_buffer(&self, buf: &mut BytesMut) {
+ buf.put_u32_le(self.id);
+ buf.put_u32_le(self.name.len() as u32);
+ buf.put_slice(self.name.as_bytes());
+ buf.put_u32_le(self.address.len() as u32);
+ buf.put_slice(self.address.as_bytes());
+ self.role.write_to_buffer(buf);
+ self.status.write_to_buffer(buf);
+ }
+
+ fn get_buffer_size(&self) -> usize {
+ 4 + 4 + self.name.len() + 4 + self.address.len() + 1 + 1 // id + name_len + name + address_len + address + role + status
+ }
+}
+
+impl Display for ClusterNode {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "ClusterNode {{ id: {}, name: {}, address: {}, role: {}, status: {} }}",
+ self.id, self.name, self.address, self.role, self.status
+ )
+ }
+}
diff --git a/core/common/src/types/cluster/role.rs b/core/common/src/types/cluster/role.rs
new file mode 100644
index 00000000..944c48cd
--- /dev/null
+++ b/core/common/src/types/cluster/role.rs
@@ -0,0 +1,72 @@
+/* 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::{BytesSerializable, IggyError};
+use bytes::{BufMut, Bytes, BytesMut};
+use serde::{Deserialize, Serialize};
+use strum::{AsRefStr, Display as StrumDisplay, EnumString};
+
+/// Node role within the cluster
+#[derive(
+ Debug, Serialize, Deserialize, PartialEq, Clone, Copy, AsRefStr, StrumDisplay, EnumString,
+)]
+#[serde(rename_all = "lowercase")]
+#[strum(serialize_all = "lowercase")]
+#[repr(u8)]
+pub enum ClusterNodeRole {
+ /// Primary/Leader node - handles all writes
+ Leader,
+ /// Follower/Secondary node - read replica
+ Follower,
+}
+
+impl BytesSerializable for ClusterNodeRole {
+ fn to_bytes(&self) -> Bytes {
+ let mut bytes = BytesMut::with_capacity(1);
+ self.write_to_buffer(&mut bytes);
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<ClusterNodeRole, IggyError> {
+ if bytes.is_empty() {
+ return Err(IggyError::InvalidCommand);
+ }
+
+ ClusterNodeRole::try_from(bytes[0])
+ }
+
+ fn write_to_buffer(&self, buf: &mut BytesMut) {
+ buf.put_u8(*self as u8);
+ }
+
+ fn get_buffer_size(&self) -> usize {
+ 1
+ }
+}
+
+impl TryFrom<u8> for ClusterNodeRole {
+ type Error = IggyError;
+
+ fn try_from(value: u8) -> Result<Self, Self::Error> {
+ match value {
+ v if v == ClusterNodeRole::Leader as u8 => Ok(ClusterNodeRole::Leader),
+ v if v == ClusterNodeRole::Follower as u8 => Ok(ClusterNodeRole::Follower),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
diff --git a/core/common/src/types/cluster/status.rs b/core/common/src/types/cluster/status.rs
new file mode 100644
index 00000000..ef0f14d8
--- /dev/null
+++ b/core/common/src/types/cluster/status.rs
@@ -0,0 +1,63 @@
+use crate::{BytesSerializable, IggyError};
+use bytes::{BufMut, Bytes, BytesMut};
+use serde::{Deserialize, Serialize};
+use strum::{AsRefStr, Display as StrumDisplay, EnumString};
+
+/// Current status of a node
+#[derive(
+ Debug, Serialize, Deserialize, PartialEq, Clone, Copy, AsRefStr, StrumDisplay, EnumString,
+)]
+#[serde(rename_all = "lowercase")]
+#[strum(serialize_all = "lowercase")]
+#[repr(u8)]
+pub enum ClusterNodeStatus {
+ /// Node is healthy and responsive
+ Healthy,
+ /// Node is starting up
+ Starting,
+ /// Node is shutting down
+ Stopping,
+ /// Node is unreachable
+ Unreachable,
+ /// Node is in maintenance mode
+ Maintenance,
+}
+
+impl TryFrom<u8> for ClusterNodeStatus {
+ type Error = IggyError;
+
+ fn try_from(value: u8) -> Result<Self, Self::Error> {
+ match value {
+ v if v == ClusterNodeStatus::Healthy as u8 => Ok(ClusterNodeStatus::Healthy),
+ v if v == ClusterNodeStatus::Starting as u8 => Ok(ClusterNodeStatus::Starting),
+ v if v == ClusterNodeStatus::Stopping as u8 => Ok(ClusterNodeStatus::Stopping),
+ v if v == ClusterNodeStatus::Unreachable as u8 => Ok(ClusterNodeStatus::Unreachable),
+ v if v == ClusterNodeStatus::Maintenance as u8 => Ok(ClusterNodeStatus::Maintenance),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+impl BytesSerializable for ClusterNodeStatus {
+ fn to_bytes(&self) -> Bytes {
+ let mut bytes = BytesMut::with_capacity(1);
+ self.write_to_buffer(&mut bytes);
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<ClusterNodeStatus, IggyError> {
+ if bytes.is_empty() {
+ return Err(IggyError::InvalidCommand);
+ }
+
+ ClusterNodeStatus::try_from(bytes[0])
+ }
+
+ fn write_to_buffer(&self, buf: &mut BytesMut) {
+ buf.put_u8(*self as u8);
+ }
+
+ fn get_buffer_size(&self) -> usize {
+ 1
+ }
+}
diff --git a/core/common/src/types/command/mod.rs b/core/common/src/types/command/mod.rs
index 4c576381..2ba735b8 100644
--- a/core/common/src/types/command/mod.rs
+++ b/core/common/src/types/command/mod.rs
@@ -31,6 +31,8 @@ pub const GET_STATS: &str = "stats";
pub const GET_STATS_CODE: u32 = 10;
pub const GET_SNAPSHOT_FILE: &str = "snapshot";
pub const GET_SNAPSHOT_FILE_CODE: u32 = 11;
+pub const GET_CLUSTER_METADATA: &str = "cluster.metadata";
+pub const GET_CLUSTER_METADATA_CODE: u32 = 12;
pub const GET_ME: &str = "me";
pub const GET_ME_CODE: u32 = 20;
pub const GET_CLIENT: &str = "client.get";
@@ -165,6 +167,7 @@ pub fn get_name_from_code(code: u32) -> Result<&'static str, IggyError> {
JOIN_CONSUMER_GROUP_CODE => Ok(JOIN_CONSUMER_GROUP),
LEAVE_CONSUMER_GROUP_CODE => Ok(LEAVE_CONSUMER_GROUP),
GET_SNAPSHOT_FILE_CODE => Ok(GET_SNAPSHOT_FILE),
+ GET_CLUSTER_METADATA_CODE => Ok(GET_CLUSTER_METADATA),
_ => Err(IggyError::InvalidCommand),
}
}
diff --git a/core/common/src/types/mod.rs b/core/common/src/types/mod.rs
index 2376abd6..dd8a2dea 100644
--- a/core/common/src/types/mod.rs
+++ b/core/common/src/types/mod.rs
@@ -18,6 +18,7 @@
pub(crate) mod args;
pub(crate) mod client;
pub(crate) mod client_state;
+pub(crate) mod cluster;
pub(crate) mod command;
pub(crate) mod compression;
pub(crate) mod configuration;
diff --git a/core/configs/server.toml b/core/configs/server.toml
index 5445ff18..2713d396 100644
--- a/core/configs/server.toml
+++ b/core/configs/server.toml
@@ -541,3 +541,42 @@ size = "4 GiB"
# and holds different buffer sizes, from 256 B to 512 MiB.
# Note: This number has to be a power of 2. Minimum value is 128 due to internal implementation details.
bucket_capacity = 8192
+
+# Cluster configuration
+[cluster]
+# Enables or disables cluster mode (boolean).
+# When enabled, this node will participate in the cluster and coordinate with other nodes.
+enabled = false
+
+# Unique cluster identifier (string).
+# All nodes in the same cluster must share the same cluster_id.
+# This prevents accidental cross-cluster communication.
+id = "iggy-cluster"
+
+# Current node configuration
+[cluster.node]
+# This node unique identifier within the cluster (u32).
+# Must be unique across all nodes and match one of the ids in the nodes list.
+id = 1
+
+# List of all nodes in the cluster (array of ClusterNodeConfig).
+# This is the cluster topology - all nodes should have the same list.
+# Currently:
+# - only static configuration is supported
+# - configuration changes require restart to take effect
+# - only TCP protocol is supported
+# - only 2 nodes are supported
+#
+# Environment variables for nodes can be set using array index:
+# IGGY_CLUSTER_NODES_0_ID=1
+# IGGY_CLUSTER_NODES_0_NAME=iggy-node-1
+# IGGY_CLUSTER_NODES_0_ADDRESS=127.0.0.1:8090
+[[cluster.nodes]]
+id = 1
+name = "iggy-node-1"
+address = "127.0.0.1:8090"
+
+[[cluster.nodes]]
+id = 2
+name = "iggy-node-2"
+address = "127.0.0.1:8091"
diff --git a/core/integration/tests/streaming/segment.rs b/core/integration/tests/streaming/segment.rs
index 403418d8..6a6d434a 100644
--- a/core/integration/tests/streaming/segment.rs
+++ b/core/integration/tests/streaming/segment.rs
@@ -21,6 +21,7 @@ use crate::streaming::create_message;
use bytes::Bytes;
use iggy::prelude::locking::IggySharedMutFn;
use iggy::prelude::*;
+use server::configs::cluster::ClusterConfig;
use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
use server::configs::system::{PartitionConfig, SegmentConfig, SystemConfig};
use server::streaming::segments::*;
@@ -472,6 +473,7 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
let setup = TestSetup::init_with_config(config).await;
let mut system = System::new(
setup.config.clone(),
+ ClusterConfig::default(),
DataMaintenanceConfig::default(),
PersonalAccessTokenConfig::default(),
);
diff --git a/core/integration/tests/streaming/snapshot.rs b/core/integration/tests/streaming/snapshot.rs
index 52b9419a..4279ce18 100644
--- a/core/integration/tests/streaming/snapshot.rs
+++ b/core/integration/tests/streaming/snapshot.rs
@@ -18,6 +18,7 @@
use crate::streaming::common::test_setup::TestSetup;
use iggy::prelude::{SnapshotCompression, SystemSnapshotType};
+use server::configs::cluster::ClusterConfig;
use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
use server::streaming::session::Session;
use server::streaming::systems::system::System;
@@ -30,6 +31,7 @@ async fn should_create_snapshot_file() {
let setup = TestSetup::init().await;
let mut system = System::new(
setup.config.clone(),
+ ClusterConfig::default(),
DataMaintenanceConfig::default(),
PersonalAccessTokenConfig::default(),
);
diff --git a/core/integration/tests/streaming/system.rs b/core/integration/tests/streaming/system.rs
index ce950e84..83a83d4d 100644
--- a/core/integration/tests/streaming/system.rs
+++ b/core/integration/tests/streaming/system.rs
@@ -18,6 +18,7 @@
use crate::streaming::common::test_setup::TestSetup;
use iggy::prelude::Identifier;
+use server::configs::cluster::ClusterConfig;
use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
use server::streaming::session::Session;
use server::streaming::systems::system::System;
@@ -29,6 +30,7 @@ async fn should_initialize_system_and_base_directories() {
let setup = TestSetup::init().await;
let mut system = System::new(
setup.config.clone(),
+ ClusterConfig::default(),
DataMaintenanceConfig::default(),
PersonalAccessTokenConfig::default(),
);
@@ -52,6 +54,7 @@ async fn should_create_and_persist_stream() {
let setup = TestSetup::init().await;
let mut system = System::new(
setup.config.clone(),
+ ClusterConfig::default(),
DataMaintenanceConfig::default(),
PersonalAccessTokenConfig::default(),
);
@@ -73,6 +76,7 @@ async fn should_create_and_persist_stream_with_automatically_generated_id() {
let setup = TestSetup::init().await;
let mut system = System::new(
setup.config.clone(),
+ ClusterConfig::default(),
DataMaintenanceConfig::default(),
PersonalAccessTokenConfig::default(),
);
@@ -94,6 +98,7 @@ async fn should_delete_persisted_stream() {
let setup = TestSetup::init().await;
let mut system = System::new(
setup.config.clone(),
+ ClusterConfig::default(),
DataMaintenanceConfig::default(),
PersonalAccessTokenConfig::default(),
);
diff --git a/core/server/src/binary/command.rs b/core/server/src/binary/command.rs
index c7802be9..d12d6c05 100644
--- a/core/server/src/binary/command.rs
+++ b/core/server/src/binary/command.rs
@@ -39,6 +39,7 @@ use iggy_common::delete_topic::DeleteTopic;
use iggy_common::delete_user::DeleteUser;
use iggy_common::get_client::GetClient;
use iggy_common::get_clients::GetClients;
+use iggy_common::get_cluster_metadata::GetClusterMetadata;
use iggy_common::get_consumer_group::GetConsumerGroup;
use iggy_common::get_consumer_groups::GetConsumerGroups;
use iggy_common::get_consumer_offset::GetConsumerOffset;
@@ -76,6 +77,7 @@ define_server_command_enum! {
GetClient(GetClient), GET_CLIENT_CODE, GET_CLIENT, true;
GetClients(GetClients), GET_CLIENTS_CODE, GET_CLIENTS, false;
GetSnapshot(GetSnapshot), GET_SNAPSHOT_FILE_CODE, GET_SNAPSHOT_FILE, false;
+ GetClusterMetadata(GetClusterMetadata), GET_CLUSTER_METADATA_CODE, GET_CLUSTER_METADATA, false;
PollMessages(PollMessages), POLL_MESSAGES_CODE, POLL_MESSAGES, true;
FlushUnsavedBuffer(FlushUnsavedBuffer), FLUSH_UNSAVED_BUFFER_CODE, FLUSH_UNSAVED_BUFFER, true;
GetUser(GetUser), GET_USER_CODE, GET_USER, true;
diff --git a/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs b/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
new file mode 100644
index 00000000..7d9d863c
--- /dev/null
+++ b/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
@@ -0,0 +1,63 @@
+/* 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::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
+use crate::binary::handlers::utils::receive_and_validate;
+use crate::binary::sender::SenderKind;
+use crate::streaming::session::Session;
+use crate::streaming::systems::system::SharedSystem;
+use anyhow::Result;
+use iggy_common::get_cluster_metadata::GetClusterMetadata;
+use iggy_common::{BytesSerializable, IggyError};
+use tracing::{debug, instrument};
+
+impl ServerCommandHandler for GetClusterMetadata {
+ fn code(&self) -> u32 {
+ iggy_common::GET_CLUSTER_METADATA_CODE
+ }
+
+ #[instrument(skip_all, name = "trace_get_cluster_metadata", fields(iggy_user_id = session.get_user_id(), iggy_client_id = session.client_id))]
+ async fn handle(
+ self,
+ sender: &mut SenderKind,
+ _length: u32,
+ session: &Session,
+ system: &SharedSystem,
+ ) -> Result<(), IggyError> {
+ debug!("session: {session}, command: {self}");
+
+ let system = system.read().await;
+ let cluster_metadata = system.get_cluster_metadata(session)?;
+
+ let response = cluster_metadata.to_bytes();
+ sender.send_ok_response(&response).await?;
+ Ok(())
+ }
+}
+
+impl BinaryServerCommand for GetClusterMetadata {
+ async fn from_sender(sender: &mut SenderKind, code: u32, length: u32) -> Result<Self, IggyError>
+ where
+ Self: Sized,
+ {
+ match receive_and_validate(sender, code, length).await? {
+ ServerCommand::GetClusterMetadata(get_cluster_metadata) => Ok(get_cluster_metadata),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
diff --git a/core/common/src/commands/system/mod.rs b/core/server/src/binary/handlers/cluster/mod.rs
similarity index 87%
copy from core/common/src/commands/system/mod.rs
copy to core/server/src/binary/handlers/cluster/mod.rs
index e6aa4ac3..4b3a71c6 100644
--- a/core/common/src/commands/system/mod.rs
+++ b/core/server/src/binary/handlers/cluster/mod.rs
@@ -16,9 +16,4 @@
* under the License.
*/
-pub mod get_client;
-pub mod get_clients;
-pub mod get_me;
-pub mod get_snapshot;
-pub mod get_stats;
-pub mod ping;
+pub mod get_cluster_metadata_handler;
diff --git a/core/server/src/binary/handlers/mod.rs b/core/server/src/binary/handlers/mod.rs
index 2e659416..22807ba8 100644
--- a/core/server/src/binary/handlers/mod.rs
+++ b/core/server/src/binary/handlers/mod.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+pub mod cluster;
pub mod consumer_groups;
pub mod consumer_offsets;
pub mod messages;
diff --git a/core/server/src/streaming/systems/mod.rs b/core/server/src/configs/cluster.rs
similarity index 63%
copy from core/server/src/streaming/systems/mod.rs
copy to core/server/src/configs/cluster.rs
index 7039e919..cd0a6bd9 100644
--- a/core/server/src/streaming/systems/mod.rs
+++ b/core/server/src/configs/cluster.rs
@@ -16,20 +16,24 @@
* under the License.
*/
-pub mod clients;
-pub mod consumer_groups;
-pub mod consumer_offsets;
-pub mod info;
-pub mod messages;
-pub mod partitions;
-pub mod personal_access_tokens;
-pub mod segments;
-pub mod snapshot;
-pub mod stats;
-pub mod storage;
-pub mod streams;
-pub mod system;
-pub mod topics;
-pub mod users;
+use serde::{Deserialize, Serialize};
-pub const COMPONENT: &str = "STREAMING_SYSTEMS";
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ClusterConfig {
+ pub enabled: bool,
+ pub id: String,
+ pub node: NodeConfig,
+ pub nodes: Vec<ClusterNodeConfig>,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct NodeConfig {
+ pub id: u32,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ClusterNodeConfig {
+ pub id: u32,
+ pub name: String,
+ pub address: String,
+}
diff --git a/core/server/src/configs/defaults.rs b/core/server/src/configs/defaults.rs
index 82b560cf..45600d17 100644
--- a/core/server/src/configs/defaults.rs
+++ b/core/server/src/configs/defaults.rs
@@ -18,6 +18,7 @@
use super::system::MemoryPoolConfig;
use super::tcp::TcpSocketConfig;
+use crate::configs::cluster::{ClusterConfig, ClusterNodeConfig, NodeConfig};
use crate::configs::http::{
HttpConfig, HttpCorsConfig, HttpJwtConfig, HttpMetricsConfig, HttpTlsConfig,
};
@@ -41,7 +42,7 @@ use std::time::Duration;
static_toml::static_toml! {
// static_toml crate always starts from CARGO_MANIFEST_DIR (in this case iggy-server root directory)
- static SERVER_CONFIG = include_toml!("../configs/server.toml");
+ pub static SERVER_CONFIG = include_toml!("../configs/server.toml");
}
impl Default for ServerConfig {
@@ -56,6 +57,7 @@ impl Default for ServerConfig {
tcp: TcpConfig::default(),
http: HttpConfig::default(),
telemetry: TelemetryConfig::default(),
+ cluster: ClusterConfig::default(),
}
}
}
@@ -533,3 +535,33 @@ impl Default for TelemetryTracesConfig {
}
}
}
+
+impl Default for ClusterConfig {
+ fn default() -> ClusterConfig {
+ ClusterConfig {
+ enabled: SERVER_CONFIG.cluster.enabled,
+ id: SERVER_CONFIG.cluster.id.parse().unwrap(),
+ node: NodeConfig::default(),
+ nodes: vec![
+ ClusterNodeConfig {
+ id: SERVER_CONFIG.cluster.nodes[0].id as u32,
+ name: SERVER_CONFIG.cluster.nodes[0].name.parse().unwrap(),
+ address: SERVER_CONFIG.cluster.nodes[0].address.parse().unwrap(),
+ },
+ ClusterNodeConfig {
+ id: SERVER_CONFIG.cluster.nodes[1].id as u32,
+ name: SERVER_CONFIG.cluster.nodes[1].name.parse().unwrap(),
+ address: SERVER_CONFIG.cluster.nodes[1].address.parse().unwrap(),
+ },
+ ],
+ }
+ }
+}
+
+impl Default for NodeConfig {
+ fn default() -> NodeConfig {
+ NodeConfig {
+ id: SERVER_CONFIG.cluster.node.id as u32,
+ }
+ }
+}
diff --git a/core/server/src/configs/mod.rs b/core/server/src/configs/mod.rs
index 1a45ecf9..936994ac 100644
--- a/core/server/src/configs/mod.rs
+++ b/core/server/src/configs/mod.rs
@@ -17,6 +17,7 @@
*/
pub mod cache_indexes;
+pub mod cluster;
pub mod config_provider;
pub mod defaults;
pub mod displays;
diff --git a/core/server/src/configs/server.rs b/core/server/src/configs/server.rs
index 252df9a7..2e897922 100644
--- a/core/server/src/configs/server.rs
+++ b/core/server/src/configs/server.rs
@@ -18,6 +18,7 @@
use crate::archiver::ArchiverKindType;
use crate::configs::COMPONENT;
+use crate::configs::cluster::ClusterConfig;
use crate::configs::config_provider::ConfigProviderKind;
use crate::configs::http::HttpConfig;
use crate::configs::quic::QuicConfig;
@@ -45,6 +46,7 @@ pub struct ServerConfig {
pub tcp: TcpConfig,
pub http: HttpConfig,
pub telemetry: TelemetryConfig,
+ pub cluster: ClusterConfig,
}
#[serde_as]
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index d7351b5f..c6375f4b 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -96,6 +96,7 @@ async fn main() -> Result<(), ServerError> {
let system = SharedSystem::new(System::new(
config.system.clone(),
+ config.cluster.clone(),
config.data_maintenance.clone(),
config.personal_access_token.clone(),
));
diff --git a/core/server/src/streaming/systems/cluster.rs b/core/server/src/streaming/systems/cluster.rs
new file mode 100644
index 00000000..34c04205
--- /dev/null
+++ b/core/server/src/streaming/systems/cluster.rs
@@ -0,0 +1,68 @@
+/* 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::streaming::session::Session;
+use crate::streaming::systems::system::System;
+use iggy_common::IggyError;
+use iggy_common::metadata::ClusterMetadata;
+use iggy_common::node::ClusterNode;
+use iggy_common::role::ClusterNodeRole;
+use iggy_common::status::ClusterNodeStatus;
+use tracing::trace;
+
+impl System {
+ pub fn get_cluster_metadata(&self, session: &Session) -> Result<ClusterMetadata, IggyError> {
+ trace!("Getting cluster metadata for session: {session}");
+
+ // Map the cluster configuration to metadata
+ let nodes: Vec<ClusterNode> = self
+ .cluster_config
+ .nodes
+ .iter()
+ .map(|node_config| {
+ // Determine role based on node id (first node is leader for now)
+ let role = if node_config.id == 1 {
+ ClusterNodeRole::Leader
+ } else {
+ ClusterNodeRole::Follower
+ };
+
+ // For now, all nodes are healthy if cluster is enabled
+ let status = if self.cluster_config.enabled {
+ ClusterNodeStatus::Healthy
+ } else {
+ ClusterNodeStatus::Stopping
+ };
+
+ ClusterNode {
+ id: node_config.id,
+ name: node_config.name.clone(),
+ address: node_config.address.clone(),
+ role,
+ status,
+ }
+ })
+ .collect();
+
+ Ok(ClusterMetadata {
+ name: self.cluster_config.id.clone(),
+ node_id: self.cluster_config.node.id,
+ nodes,
+ })
+ }
+}
diff --git a/core/server/src/streaming/systems/mod.rs b/core/server/src/streaming/systems/mod.rs
index 7039e919..a0961c8b 100644
--- a/core/server/src/streaming/systems/mod.rs
+++ b/core/server/src/streaming/systems/mod.rs
@@ -17,6 +17,7 @@
*/
pub mod clients;
+pub mod cluster;
pub mod consumer_groups;
pub mod consumer_offsets;
pub mod info;
diff --git a/core/server/src/streaming/systems/streams.rs b/core/server/src/streaming/systems/streams.rs
index 43d24fde..067412f6 100644
--- a/core/server/src/streaming/systems/streams.rs
+++ b/core/server/src/streaming/systems/streams.rs
@@ -484,6 +484,7 @@ mod tests {
let stream_name = "test";
let mut system = System::create(
config,
+ crate::configs::cluster::ClusterConfig::default(),
storage,
Arc::new(StateKind::Mock(MockState::new())),
None,
diff --git a/core/server/src/streaming/systems/system.rs b/core/server/src/streaming/systems/system.rs
index 0e69b79c..f4d83de6 100644
--- a/core/server/src/streaming/systems/system.rs
+++ b/core/server/src/streaming/systems/system.rs
@@ -17,6 +17,7 @@
*/
use crate::archiver::{ArchiverKind, ArchiverKindType};
+use crate::configs::cluster::ClusterConfig;
use crate::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
use crate::configs::system::SystemConfig;
use crate::map_toggle_str;
@@ -82,6 +83,7 @@ pub struct System {
pub(crate) streams_ids: AHashMap<String, u32>,
pub(crate) users: AHashMap<UserId, User>,
pub(crate) config: Arc<SystemConfig>,
+ pub(crate) cluster_config: ClusterConfig,
pub(crate) client_manager: IggySharedMut<ClientManager>,
pub(crate) encryptor: Option<Arc<EncryptorKind>>,
pub(crate) metrics: Metrics,
@@ -93,6 +95,7 @@ pub struct System {
impl System {
pub fn new(
config: Arc<SystemConfig>,
+ cluster_config_param: ClusterConfig,
data_maintenance_config: DataMaintenanceConfig,
pat_config: PersonalAccessTokenConfig,
) -> System {
@@ -120,6 +123,7 @@ impl System {
)));
Self::create(
config.clone(),
+ cluster_config_param,
SystemStorage::new(config, partition_persister),
state,
encryptor,
@@ -137,6 +141,7 @@ impl System {
pub fn create(
system_config: Arc<SystemConfig>,
+ cluster_config: ClusterConfig,
storage: SystemStorage,
state: Arc<StateKind>,
encryptor: Option<Arc<EncryptorKind>>,
@@ -170,6 +175,7 @@ impl System {
System {
config: system_config,
+ cluster_config,
streams: AHashMap::new(),
streams_ids: AHashMap::new(),
storage: Arc::new(storage),