You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2026/07/27 13:47:13 UTC
(iggy) branch advertised_ip created (now 5d22cc431)
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a change to branch advertised_ip
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 5d22cc431 feat(server-ng): add cluster.nodes.advertised_address for clients
This branch includes the following new commits:
new 5d22cc431 feat(server-ng): add cluster.nodes.advertised_address for clients
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-ng): add cluster.nodes.advertised_address for clients
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch advertised_ip
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 5d22cc431ba8fad7f43dbc08983b2804d09c27b0
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Mon Jul 27 13:11:10 2026 +0200
feat(server-ng): add cluster.nodes.advertised_address for clients
A node's `ip` is the address it binds for replica traffic, which in
Docker, Kubernetes, or NAT deployments is private. Both client-facing
surfaces reuse it: the cluster roster returned to clients and the
follower's redirect to the metadata primary. Clients outside the
replica network get an address they cannot reach.
Nodes now take an optional `advertised_address` that overrides `ip`
only where an address is handed to a client. Replica traffic still
uses `ip`, so the two planes can live on separate networks. Left
unset, behavior is unchanged. The field is typed as a string but
validated as a literal IP, leaving room to accept hostnames later
without a config-format break.
Validation now checks advertised client endpoints for conflicts
alongside the existing bind-endpoint check, since two nodes on
distinct private IPs can otherwise publish the same client address
and silently shadow each other. Node entries also reject unknown
fields, so a misspelled advertised_address fails at startup
instead of silently leaving the private ip in client responses.
---
core/configs/src/server_ng_config/cluster.rs | 146 +++++++++++++++++++++++-
core/configs/src/server_ng_config/defaults.rs | 1 +
core/configs/src/server_ng_config/validators.rs | 1 +
core/server-ng/config.toml | 7 ++
core/server-ng/src/cluster_meta.rs | 41 ++++++-
core/server-ng/src/http/error.rs | 56 +++++++--
6 files changed, 239 insertions(+), 13 deletions(-)
diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs
index 44b565f58..0e61d6494 100644
--- a/core/configs/src/server_ng_config/cluster.rs
+++ b/core/configs/src/server_ng_config/cluster.rs
@@ -29,6 +29,7 @@ use configs::ConfigEnv;
use iggy_common::{IggyDuration, Validatable};
use serde::{Deserialize, Serialize};
use serde_with::{DisplayFromStr, serde_as};
+use std::net::{IpAddr, SocketAddr};
use std::time::Duration;
/// The primary heartbeats roughly every second (`PING_TICKS`); a window at
@@ -144,9 +145,14 @@ pub struct ClusterTlsConfig {
}
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
pub struct ClusterNodeConfig {
pub name: String,
pub ip: String,
+ /// Optional client-facing address. Replica traffic continues to use
+ /// [`Self::ip`].
+ #[serde(default)]
+ pub advertised_address: Option<String>,
/// Numeric replica ID for VSR consensus (0-based).
///
/// Must be unique across [`ClusterConfig::nodes`] and strictly less than
@@ -226,6 +232,8 @@ impl Validatable<ConfigurationError> for ClusterConfig {
let mut seen_ids = std::collections::HashSet::new();
let mut seen_names = std::collections::HashSet::new();
let mut used_endpoints = std::collections::HashSet::new();
+ let mut used_advertised_endpoints = std::collections::HashSet::new();
+ let mut used_raw_advertised_endpoints = std::collections::HashSet::new();
for node in &self.nodes {
if node.name.trim().is_empty() {
@@ -265,17 +273,17 @@ impl Validatable<ConfigurationError> for ClusterConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
- let port_list = [
+ let client_ports = [
("TCP", node.ports.tcp),
("QUIC", node.ports.quic),
("HTTP", node.ports.http),
("WebSocket", node.ports.websocket),
- ("TCP_REPLICA", node.ports.tcp_replica),
];
+ let replica_port = ("TCP_REPLICA", node.ports.tcp_replica);
- for (name, port_opt) in &port_list {
+ for (name, port_opt) in client_ports.into_iter().chain([replica_port]) {
if let Some(port) = port_opt {
- if *port == 0 {
+ if port == 0 {
eprintln!(
"Invalid cluster configuration: {} port cannot be 0 for node '{}'",
name, node.name
@@ -293,6 +301,47 @@ impl Validatable<ConfigurationError> for ClusterConfig {
}
}
}
+
+ // Strict for now; the String type leaves room to accept hostnames
+ // later without breaking existing configs.
+ let client_ip = match node.advertised_address.as_deref() {
+ Some(advertised_address) => match advertised_address.parse::<IpAddr>() {
+ Ok(ip) => Some(ip),
+ Err(_) => {
+ eprintln!(
+ "Invalid cluster configuration: advertised_address '{advertised_address}' is not a valid IP address for node '{}'",
+ node.name
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ },
+ None => node.ip.parse::<IpAddr>().ok(),
+ };
+
+ let client_address = node.advertised_address.as_deref().unwrap_or(&node.ip);
+ for (name, port) in &client_ports {
+ if let Some(port) = port {
+ let (endpoint, inserted) = client_ip.map_or_else(
+ || {
+ let endpoint = format!("{client_address}:{port}");
+ let inserted = used_raw_advertised_endpoints.insert(endpoint.clone());
+ (endpoint, inserted)
+ },
+ |ip| {
+ let endpoint = SocketAddr::new(ip, *port);
+ let inserted = used_advertised_endpoints.insert(endpoint);
+ (endpoint.to_string(), inserted)
+ },
+ );
+ if !inserted {
+ eprintln!(
+ "Invalid cluster configuration: advertised client endpoint conflict - {endpoint} is already used (node '{}', transport {name})",
+ node.name
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ }
+ }
}
// Replica-auth PSK (only reached when the cluster is enabled; the early
@@ -380,6 +429,27 @@ mod tests {
"shared_secret field present in serialized config: {serialized}"
);
}
+
+ #[test]
+ fn cluster_node_rejects_unknown_fields() {
+ let error = serde_json::from_str::<ClusterNodeConfig>(
+ r#"{
+ "name": "node-0",
+ "ip": "10.0.0.1",
+ "advertised_addres": "203.0.113.1",
+ "replica_id": 0,
+ "ports": {}
+ }"#,
+ )
+ .expect_err("misspelled advertised_address must be rejected");
+
+ assert!(
+ error
+ .to_string()
+ .contains("unknown field `advertised_addres`"),
+ "unexpected deserialization error: {error}"
+ );
+ }
}
#[cfg(test)]
@@ -390,6 +460,7 @@ mod cluster_validate_tests {
ClusterNodeConfig {
name: name.to_string(),
ip: "127.0.0.1".to_string(),
+ advertised_address: None,
replica_id: id,
ports: TransportPorts::default(),
}
@@ -513,6 +584,73 @@ mod cluster_validate_tests {
assert!(c.validate().is_ok());
}
+ #[test]
+ fn validate_rejects_duplicate_advertised_client_endpoint() {
+ let mut n1 = node("n1", 0);
+ n1.ip = "10.0.0.1".to_owned();
+ n1.advertised_address = Some("203.0.113.1".to_owned());
+ n1.ports.tcp = Some(8090);
+ let mut n2 = node("n2", 1);
+ n2.ip = "10.0.0.2".to_owned();
+ n2.advertised_address = n1.advertised_address.clone();
+ n2.ports.tcp = Some(8090);
+
+ assert!(cfg(vec![n1, n2]).validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_equivalent_ipv6_advertised_client_endpoints() {
+ for equivalent_address in ["2001:DB8::1", "2001:db8:0:0:0:0:0:1"] {
+ let mut n1 = node("n1", 0);
+ n1.ip = "10.0.0.1".to_owned();
+ n1.advertised_address = Some("2001:db8::1".to_owned());
+ n1.ports.tcp = Some(8090);
+ let mut n2 = node("n2", 1);
+ n2.ip = "10.0.0.2".to_owned();
+ n2.advertised_address = Some(equivalent_address.to_owned());
+ n2.ports.tcp = Some(8090);
+
+ assert!(
+ cfg(vec![n1, n2]).validate().is_err(),
+ "{equivalent_address} must conflict with 2001:db8::1"
+ );
+ }
+ }
+
+ #[test]
+ fn validate_rejects_equivalent_ipv6_client_endpoints_from_node_ip() {
+ let mut n1 = node("n1", 0);
+ n1.ip = "2001:db8::1".to_owned();
+ n1.ports.tcp = Some(8090);
+ let mut n2 = node("n2", 1);
+ n2.ip = "2001:db8:0:0:0:0:0:1".to_owned();
+ n2.ports.tcp = Some(8090);
+
+ assert!(cfg(vec![n1, n2]).validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_distinct_advertised_client_endpoints() {
+ let mut n1 = node("n1", 0);
+ n1.ip = "10.0.0.1".to_owned();
+ n1.advertised_address = Some("203.0.113.1".to_owned());
+ n1.ports.tcp = Some(8090);
+ let mut n2 = node("n2", 1);
+ n2.ip = "10.0.0.2".to_owned();
+ n2.advertised_address = Some("203.0.113.2".to_owned());
+ n2.ports.tcp = Some(8090);
+
+ assert!(cfg(vec![n1, n2]).validate().is_ok());
+ }
+
+ #[test]
+ fn validate_rejects_non_ip_advertised_address() {
+ let mut n1 = node("n1", 0);
+ n1.advertised_address = Some("iggy-node-1.example.com".to_owned());
+
+ assert!(cfg(vec![n1, node("n2", 1)]).validate().is_err());
+ }
+
#[test]
fn validate_rejects_zero_tcp_replica_port() {
let ports = TransportPorts {
diff --git a/core/configs/src/server_ng_config/defaults.rs b/core/configs/src/server_ng_config/defaults.rs
index 58887a70f..d9c32dc8b 100644
--- a/core/configs/src/server_ng_config/defaults.rs
+++ b/core/configs/src/server_ng_config/defaults.rs
@@ -84,6 +84,7 @@ impl Default for ClusterConfig {
.map(|node| ClusterNodeConfig {
name: node.name.parse().unwrap(),
ip: node.ip.parse().unwrap(),
+ advertised_address: None,
replica_id: u8::try_from(node.replica_id).expect(
"static_toml replica_id must fit in u8 (0..=255); \
fix core/server-ng/config.toml",
diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs
index 4736d9055..e824afe49 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -435,6 +435,7 @@ mod tests {
ClusterNodeConfig {
name: format!("node-{replica_id}"),
ip: "127.0.0.1".to_string(),
+ advertised_address: None,
replica_id,
ports: TransportPorts {
tcp: Some(8090 + u16::from(replica_id)),
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 7a185a80d..34baee0e6 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -639,6 +639,12 @@ ca_file = ""
# node's identity is resolved at launch from the '--replica-id <N>' CLI
# flag, which selects the entry in this list that describes the current
# node. All other entries are remote peers.
+#
+# Each node may also set 'advertised_address': the client-facing address
+# handed out in cluster metadata and leader redirects. Set it when 'ip' is
+# a private replica-network address unreachable by clients (Docker,
+# Kubernetes, NAT). Must be a literal IP address for now. Replica-to-replica
+# traffic always uses 'ip'. Unset = clients get 'ip'.
[[cluster.nodes]]
name = "iggy-node-1"
ip = "127.0.0.1"
@@ -655,6 +661,7 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket = 8093, tcp_replica =
# [[cluster.nodes]]
# name = "iggy-node-3"
# ip = "192.168.1.100"
+# advertised_address = "203.0.113.10"
# replica_id = 2
# ports = { tcp = 8092, http = 3002, tcp_replica = 9092 }
diff --git a/core/server-ng/src/cluster_meta.rs b/core/server-ng/src/cluster_meta.rs
index 9e969f3df..1337b879e 100644
--- a/core/server-ng/src/cluster_meta.rs
+++ b/core/server-ng/src/cluster_meta.rs
@@ -106,7 +106,10 @@ impl ClusterRoster {
.iter()
.map(|node| ClusterNode {
name: node.name.clone(),
- ip: node.ip.clone(),
+ ip: node
+ .advertised_address
+ .clone()
+ .unwrap_or_else(|| node.ip.clone()),
endpoints: ports_to_endpoints(&node.ports),
role: role_for(primary_index, node.replica_id),
status: ClusterNodeStatus::Healthy,
@@ -150,3 +153,39 @@ fn ports_to_endpoints(ports: &TransportPorts) -> TransportEndpoints {
ports.websocket.unwrap_or(0),
)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn roster(advertised_address: Option<String>) -> ClusterRoster {
+ ClusterRoster {
+ enabled: true,
+ name: "test-cluster".to_owned(),
+ nodes: vec![ClusterNodeConfig {
+ name: "node-0".to_owned(),
+ ip: "10.0.0.1".to_owned(),
+ advertised_address,
+ replica_id: 0,
+ ports: TransportPorts::default(),
+ }],
+ self_ip: "127.0.0.1".to_owned(),
+ self_ports: TransportPorts::default(),
+ metadata_view: Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)),
+ }
+ }
+
+ #[test]
+ fn cluster_metadata_uses_advertised_address_when_configured() {
+ let metadata = roster(Some("203.0.113.10".to_owned())).cluster_metadata(Some(0));
+
+ assert_eq!(metadata.nodes[0].ip, "203.0.113.10");
+ }
+
+ #[test]
+ fn cluster_metadata_falls_back_to_replica_ip() {
+ let metadata = roster(None).cluster_metadata(Some(0));
+
+ assert_eq!(metadata.nodes[0].ip, "10.0.0.1");
+ }
+}
diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs
index 511a1dd22..6799d4fa6 100644
--- a/core/server-ng/src/http/error.rs
+++ b/core/server-ng/src/http/error.rs
@@ -25,6 +25,7 @@ use axum::Json;
use axum::http::header::{LOCATION, RETRY_AFTER};
use axum::http::{HeaderValue, StatusCode};
use axum::response::{IntoResponse, Response};
+use configs::ng_cluster::ClusterNodeConfig;
use iggy_binary_protocol::Operation;
use iggy_common::IggyError;
use serde::{Deserialize, Serialize};
@@ -500,39 +501,54 @@ pub(in crate::http) fn primary_redirect_location(
scheme: &str,
path_and_query: &str,
) -> Option<String> {
- let socket = primary_http_socket(roster, primary_index)?;
+ let socket = primary_advertised_http_socket(roster, primary_index)?;
Some(format!("{scheme}://{socket}{path_and_query}"))
}
/// Resolve the VSR primary's HTTP socket from the static roster: the node
-/// whose `replica_id` equals `primary_index`, its `ports.http`, and its `ip`
-/// parsed strictly. `None` on any miss so callers fail closed. Formatting the
-/// returned [`SocketAddr`] brackets an IPv6 host (`[::1]:8080`) rather than
-/// leaving it ambiguous.
+/// whose `replica_id` equals `primary_index`, its `ports.http`, and its private
+/// roster `ip` parsed strictly. Internal replica forwarding uses this address.
pub(in crate::http) fn primary_http_socket(
roster: &ClusterRoster,
primary_index: u8,
) -> Option<SocketAddr> {
+ let (node, http_port) = primary_node(roster, primary_index)?;
+ let ip = node.ip.parse::<IpAddr>().ok()?;
+ Some(SocketAddr::new(ip, http_port))
+}
+
+/// Resolve the client-facing HTTP socket for a redirect. The advertised
+/// address is preferred, with the private roster IP retained as the
+/// compatibility fallback. Constructing a socket address brackets IPv6 hosts,
+/// which keeps the resulting redirect URL valid.
+fn primary_advertised_http_socket(roster: &ClusterRoster, primary_index: u8) -> Option<SocketAddr> {
+ let (node, http_port) = primary_node(roster, primary_index)?;
+ let host = node.advertised_address.as_deref().unwrap_or(&node.ip);
+ let ip = host.parse::<IpAddr>().ok()?;
+ Some(SocketAddr::new(ip, http_port))
+}
+
+fn primary_node(roster: &ClusterRoster, primary_index: u8) -> Option<(&ClusterNodeConfig, u16)> {
let node = roster
.nodes
.iter()
.find(|node| node.replica_id == primary_index)?;
let http_port = node.ports.http?;
- let ip = node.ip.parse::<IpAddr>().ok()?;
- Some(SocketAddr::new(ip, http_port))
+ Some((node, http_port))
}
#[cfg(test)]
mod tests {
use super::*;
- use configs::ng_cluster::{ClusterNodeConfig, TransportPorts};
+ use configs::ng_cluster::TransportPorts;
const READ_PATH: &str = "/streams?consistency=linearizable";
fn node(replica_id: u8, ip: &str, http: Option<u16>) -> ClusterNodeConfig {
ClusterNodeConfig {
name: format!("node-{replica_id}"),
ip: ip.to_owned(),
+ advertised_address: None,
replica_id,
ports: TransportPorts {
tcp: None,
@@ -614,6 +630,30 @@ mod tests {
);
}
+ #[test]
+ fn primary_redirect_location_uses_advertised_address() {
+ let mut primary = node(0, "10.0.0.1", Some(8080));
+ primary.advertised_address = Some("2001:db8::1".to_owned());
+ let roster = roster(vec![primary]);
+
+ assert_eq!(
+ primary_redirect_location(&roster, 0, "https", READ_PATH),
+ Some("https://[2001:db8::1]:8080/streams?consistency=linearizable".to_owned())
+ );
+ }
+
+ #[test]
+ fn primary_http_socket_uses_private_roster_ip() {
+ let mut primary = node(0, "10.0.0.1", Some(8080));
+ primary.advertised_address = Some("203.0.113.1".to_owned());
+ let roster = roster(vec![primary]);
+
+ assert_eq!(
+ primary_http_socket(&roster, 0),
+ Some("10.0.0.1:8080".parse().expect("valid socket address"))
+ );
+ }
+
#[test]
fn transient_not_committed_renders_503_with_retry_after() {
let response = CustomError::from(IggyError::TransientNotCommitted).into_response();