This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new cfa104c37 feat(server-ng): add cluster.nodes.advertised_address for
clients (#3757)
cfa104c37 is described below
commit cfa104c37c575c843b8f005cf016d6744e18889f
Author: Maciej Modzelewski <[email protected]>
AuthorDate: Tue Jul 28 08:59:51 2026 +0200
feat(server-ng): add cluster.nodes.advertised_address for clients (#3757)
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 value parses into an `AdvertisedAddress`: a literal IP or an
RFC 1123 hostname, rejecting ports, empty labels, and dotted-numeric
strings that are really malformed IPv4, so `10.0.0.256` fails at boot
instead of reaching DNS. Hostnames normalize to lowercase and IPs to
canonical form, so textual variants of one address compare equal and
publish identically; IPv6 hosts are bracketed when rendered with a
port.
Validation now also 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. Only client transports take part,
as replica ports never carry an advertised address. A roster `ip`
that parses as neither IP nor hostname stays legal (Docker service
names with underscores exist in the wild) and falls back to
raw-string comparison instead of failing boot.
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 | 471 +++++++++++++++++++++++-
core/configs/src/server_ng_config/defaults.rs | 1 +
core/configs/src/server_ng_config/validators.rs | 1 +
core/server-ng/config.toml | 22 +-
core/server-ng/src/bootstrap.rs | 1 +
core/server-ng/src/cluster_meta.rs | 81 +++-
core/server-ng/src/http/error.rs | 72 +++-
7 files changed, 628 insertions(+), 21 deletions(-)
diff --git a/core/configs/src/server_ng_config/cluster.rs
b/core/configs/src/server_ng_config/cluster.rs
index 81e4639d1..7f6c20e20 100644
--- a/core/configs/src/server_ng_config/cluster.rs
+++ b/core/configs/src/server_ng_config/cluster.rs
@@ -29,6 +29,9 @@ use configs::ConfigEnv;
use iggy_common::{IggyDuration, Validatable};
use serde::{Deserialize, Serialize};
use serde_with::{DisplayFromStr, serde_as};
+use std::fmt;
+use std::net::{IpAddr, Ipv6Addr, SocketAddr};
+use std::str::FromStr;
use std::time::Duration;
/// The primary heartbeats roughly every second (`PING_TICKS`); a window at
@@ -40,6 +43,13 @@ pub const MIN_CLUSTER_HEARTBEAT_TIMEOUT: Duration =
Duration::from_secs(2);
/// length is accepted.
const MIN_SHARED_SECRET_LEN: usize = 32;
+/// DNS caps a full name at 255 octets on the wire, which leaves 253
+/// characters of presentation text (RFC 1035).
+const MAX_HOSTNAME_LEN: usize = 253;
+
+/// Per-label limit from RFC 1035.
+const MAX_HOSTNAME_LABEL_LEN: usize = 63;
+
/// serde fallback for configs written before the field existed; the value
/// itself lives in `core/server-ng/config.toml` like every other default.
fn default_heartbeat_timeout() -> IggyDuration {
@@ -144,9 +154,15 @@ 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: a literal IP or a DNS hostname,
+ /// validated as [`AdvertisedAddress`] at boot. 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
@@ -172,6 +188,166 @@ pub struct TransportPorts {
pub tcp_replica: Option<u16>,
}
+/// A validated client-facing node address: a literal IP or a DNS hostname.
+///
+/// Hostnames follow RFC 1123: ASCII letters, digits and hyphens in labels of
+/// 1-63 characters that do not start or end with a hyphen, at most
+/// [`MAX_HOSTNAME_LEN`] characters total, no port and no trailing dot. Names
+/// consisting solely of digits and dots are rejected as malformed IPv4 rather
+/// than accepted as hostnames, so `10.0.0.256` fails loudly instead of being
+/// handed to DNS. Hostnames normalize to lowercase and IPs to their canonical
+/// form ([`IpAddr`]), so textual variants of one address
(`Broker.Example.COM`,
+/// `2001:DB8::1`, `[2001:db8::1]`) compare equal.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum AdvertisedAddress {
+ Ip(IpAddr),
+ Hostname(String),
+}
+
+impl AdvertisedAddress {
+ /// Render `host:port` for a URL or endpoint listing, bracketing IPv6
+ /// hosts (`[::1]:8080`) so the port separator stays unambiguous.
+ pub fn authority(&self, port: u16) -> String {
+ match self {
+ Self::Ip(ip) => SocketAddr::new(*ip, port).to_string(),
+ Self::Hostname(hostname) => format!("{hostname}:{port}"),
+ }
+ }
+}
+
+impl FromStr for AdvertisedAddress {
+ type Err = AdvertisedAddressError;
+
+ fn from_str(address: &str) -> Result<Self, Self::Err> {
+ if address.is_empty() {
+ return Err(AdvertisedAddressError::Empty);
+ }
+ if let Ok(ip) = address.parse::<IpAddr>() {
+ return Ok(Self::Ip(ip));
+ }
+ // URL-style bracketed IPv6 (`[2001:db8::1]`) is unambiguous; accept
+ // it and store the inner address.
+ if let Some(inner) = address
+ .strip_prefix('[')
+ .and_then(|rest| rest.strip_suffix(']'))
+ && let Ok(ip) = inner.parse::<Ipv6Addr>()
+ {
+ return Ok(Self::Ip(IpAddr::V6(ip)));
+ }
+ if let Some((host, port)) = address.rsplit_once(':') {
+ // `host:port` and `[v6]:port` are the common misconfigurations;
+ // anything else with a colon can only be a broken IPv6 literal,
+ // since ':' never appears in a hostname.
+ let bracketed_host = host.starts_with('[') && host.ends_with(']');
+ if !port.is_empty()
+ && port.bytes().all(|byte| byte.is_ascii_digit())
+ && (bracketed_host || !host.contains(':'))
+ {
+ return Err(AdvertisedAddressError::PortNotAllowed);
+ }
+ return Err(AdvertisedAddressError::MalformedIpv6);
+ }
+ if address.len() > MAX_HOSTNAME_LEN {
+ return Err(AdvertisedAddressError::HostnameTooLong {
+ length: address.len(),
+ });
+ }
+ let mut all_labels_numeric = true;
+ for label in address.split('.') {
+ if label.is_empty() {
+ return Err(AdvertisedAddressError::EmptyLabel);
+ }
+ if label.len() > MAX_HOSTNAME_LABEL_LEN {
+ return Err(AdvertisedAddressError::LabelTooLong {
+ label: label.to_owned(),
+ });
+ }
+ if label.starts_with('-') || label.ends_with('-') {
+ return Err(AdvertisedAddressError::LabelHyphen {
+ label: label.to_owned(),
+ });
+ }
+ if let Some(character) = label
+ .chars()
+ .find(|character| !character.is_ascii_alphanumeric() &&
*character != '-')
+ {
+ return Err(AdvertisedAddressError::InvalidCharacter {
character });
+ }
+ all_labels_numeric &= label.bytes().all(|byte|
byte.is_ascii_digit());
+ }
+ if all_labels_numeric {
+ return Err(AdvertisedAddressError::MalformedIpv4);
+ }
+ // DNS resolution is case-insensitive; normalizing here makes equality
+ // (and thus endpoint-conflict detection) case-insensitive too.
+ Ok(Self::Hostname(address.to_ascii_lowercase()))
+ }
+}
+
+impl fmt::Display for AdvertisedAddress {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::Ip(ip) => write!(formatter, "{ip}"),
+ Self::Hostname(hostname) => write!(formatter, "{hostname}"),
+ }
+ }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AdvertisedAddressError {
+ Empty,
+ PortNotAllowed,
+ MalformedIpv4,
+ MalformedIpv6,
+ HostnameTooLong { length: usize },
+ EmptyLabel,
+ LabelTooLong { label: String },
+ LabelHyphen { label: String },
+ InvalidCharacter { character: char },
+}
+
+impl fmt::Display for AdvertisedAddressError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::Empty => write!(formatter, "address cannot be empty"),
+ Self::PortNotAllowed => write!(
+ formatter,
+ "address must not include a port; ports are configured in
cluster.nodes.ports"
+ ),
+ Self::MalformedIpv4 => write!(
+ formatter,
+ "address consists only of digits and dots but is not a valid
IPv4 address"
+ ),
+ Self::MalformedIpv6 => write!(
+ formatter,
+ "address contains ':' but is not a valid IPv6 address, and ':'
cannot appear in a hostname"
+ ),
+ Self::HostnameTooLong { length } => write!(
+ formatter,
+ "hostname is {length} characters long; the limit is
{MAX_HOSTNAME_LEN}"
+ ),
+ Self::EmptyLabel => write!(
+ formatter,
+ "hostname contains an empty label (leading, trailing, or
doubled dot)"
+ ),
+ Self::LabelTooLong { label } => write!(
+ formatter,
+ "hostname label '{label}' exceeds {MAX_HOSTNAME_LABEL_LEN}
characters"
+ ),
+ Self::LabelHyphen { label } => write!(
+ formatter,
+ "hostname label '{label}' cannot start or end with a hyphen"
+ ),
+ Self::InvalidCharacter { character } => write!(
+ formatter,
+ "character '{character}' is not allowed in a hostname
(allowed: ASCII letters, digits, '-', '.')"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for AdvertisedAddressError {}
+
/// Whether cluster-wide JWT key material exists: a configured `http.jwt`
/// secret, or the signing key derived from the cluster PSK. When it does, a
/// bearer minted on any node verifies on every node - the invariant
@@ -231,6 +407,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() {
@@ -270,17 +448,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
@@ -298,6 +476,51 @@ impl Validatable<ConfigurationError> for ClusterConfig {
}
}
}
+
+ // An advertised address must parse strictly (IP or RFC 1123
+ // hostname): the value is handed verbatim to every client via
+ // cluster metadata and redirect URLs, so a bad one poisons them
+ // all. The roster `ip` predates this check and is only validated
+ // as non-empty (Docker service names with underscores exist in
+ // the wild), so when it backs the client endpoints an unparsable
+ // value falls back to raw-string comparison instead of failing
+ // boot.
+ let client_address = match node.advertised_address.as_deref() {
+ Some(advertised_address) => match
advertised_address.parse::<AdvertisedAddress>() {
+ Ok(address) => Some(address),
+ Err(error) => {
+ eprintln!(
+ "Invalid cluster configuration: advertised_address
'{advertised_address}' for node '{}': {error}",
+ node.name
+ );
+ return
Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ },
+ None => node.ip.parse::<AdvertisedAddress>().ok(),
+ };
+
+ for (name, port) in &client_ports {
+ if let Some(port) = port {
+ let (endpoint, inserted) = match &client_address {
+ Some(address) => (
+ address.authority(*port),
+ used_advertised_endpoints.insert((address.clone(),
*port)),
+ ),
+ None => {
+ let endpoint = format!("{}:{port}", node.ip);
+ let inserted =
used_raw_advertised_endpoints.insert(endpoint.clone());
+ (endpoint, 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
@@ -385,6 +608,104 @@ 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",
+ "advertise_address": "203.0.113.1",
+ "replica_id": 0,
+ "ports": {}
+ }"#,
+ )
+ .expect_err("misspelled advertised_address must be rejected");
+
+ assert!(
+ error
+ .to_string()
+ .contains("unknown field `advertise_address`"),
+ "unexpected deserialization error: {error}"
+ );
+ }
+}
+
+#[cfg(test)]
+mod advertised_address_tests {
+ use super::*;
+
+ #[test]
+ fn parses_ip_literals_to_canonical_form() {
+ assert_eq!(
+ "203.0.113.1".parse::<AdvertisedAddress>(),
+ Ok(AdvertisedAddress::Ip("203.0.113.1".parse().unwrap()))
+ );
+ for equivalent_address in ["2001:DB8::1", "2001:db8:0:0:0:0:0:1",
"[2001:db8::1]"] {
+ assert_eq!(
+ equivalent_address.parse::<AdvertisedAddress>(),
+ Ok(AdvertisedAddress::Ip("2001:db8::1".parse().unwrap())),
+ "'{equivalent_address}' must parse to canonical 2001:db8::1"
+ );
+ }
+ }
+
+ #[test]
+ fn normalizes_hostname_to_lowercase() {
+ let address = "Broker-1.Example.COM".parse::<AdvertisedAddress>();
+ assert_eq!(
+ address,
+ Ok(AdvertisedAddress::Hostname(
+ "broker-1.example.com".to_owned()
+ ))
+ );
+ }
+
+ #[test]
+ fn authority_brackets_ipv6_hosts_only() {
+ let cases = [
+ ("203.0.113.1", "203.0.113.1:8090"),
+ ("2001:db8::1", "[2001:db8::1]:8090"),
+ ("broker-1.example.com", "broker-1.example.com:8090"),
+ ];
+ for (host, expected_authority) in cases {
+ let address = host.parse::<AdvertisedAddress>().expect("valid
address");
+ assert_eq!(address.authority(8090), expected_authority);
+ }
+ }
+
+ #[test]
+ fn rejects_port_suffixes() {
+ for address_with_port in ["example.com:8090", "10.0.0.1:8090",
"[2001:db8::1]:8090"] {
+ assert_eq!(
+ address_with_port.parse::<AdvertisedAddress>(),
+ Err(AdvertisedAddressError::PortNotAllowed),
+ "'{address_with_port}' must be rejected as host:port"
+ );
+ }
+ }
+
+ #[test]
+ fn rejects_dotted_numeric_strings_as_malformed_ipv4() {
+ for malformed_ip in ["10.0.0.256", "192.168.1", "12345"] {
+ assert_eq!(
+ malformed_ip.parse::<AdvertisedAddress>(),
+ Err(AdvertisedAddressError::MalformedIpv4),
+ "'{malformed_ip}' must not pass as a hostname"
+ );
+ }
+ }
+
+ #[test]
+ fn rejects_broken_ipv6_literals() {
+ for broken_ipv6 in ["2001:db8:::1", "[2001:db8::zz]", "::1::2"] {
+ assert_eq!(
+ broken_ipv6.parse::<AdvertisedAddress>(),
+ Err(AdvertisedAddressError::MalformedIpv6),
+ "'{broken_ipv6}' must be rejected as malformed IPv6"
+ );
+ }
+ }
}
#[cfg(test)]
@@ -395,6 +716,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(),
}
@@ -518,6 +840,147 @@ 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",
"[2001:db8::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_accepts_hostname_advertised_address() {
+ let mut n1 = node("n1", 0);
+ n1.advertised_address = Some("iggy-node-1.example.com".to_owned());
+ n1.ports.tcp = Some(8090);
+
+ assert!(cfg(vec![n1, node("n2", 1)]).validate().is_ok());
+ }
+
+ #[test]
+ fn validate_rejects_malformed_advertised_addresses() {
+ let oversized_label = format!("{}.example.com", "a".repeat(64));
+ let oversized_hostname = format!("{}example.com", "a.".repeat(130));
+ for advertised_address in [
+ "",
+ " 203.0.113.1",
+ "10.0.0.256",
+ "192.168.1",
+ "example.com:8090",
+ "[2001:db8::1]:8090",
+ "2001:db8:::1",
+ "iggy_node.example.com",
+ "-node.example.com",
+ "node-.example.com",
+ ".example.com",
+ "example..com",
+ "example.com.",
+ "ex\u{e4}mple.com",
+ oversized_label.as_str(),
+ oversized_hostname.as_str(),
+ ] {
+ let mut n1 = node("n1", 0);
+ n1.advertised_address = Some(advertised_address.to_owned());
+
+ assert!(
+ cfg(vec![n1, node("n2", 1)]).validate().is_err(),
+ "'{advertised_address}' must be rejected"
+ );
+ }
+ }
+
+ #[test]
+ fn validate_rejects_case_variant_hostname_advertised_endpoints() {
+ let mut n1 = node("n1", 0);
+ n1.ip = "10.0.0.1".to_owned();
+ n1.advertised_address = Some("broker.example.com".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("Broker.Example.COM".to_owned());
+ n2.ports.tcp = Some(8090);
+
+ assert!(cfg(vec![n1, n2]).validate().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_node_ip_hostname_clashing_with_advertised_hostname() {
+ let mut n1 = node("n1", 0);
+ n1.ip = "10.0.0.1".to_owned();
+ n1.advertised_address = Some("broker.example.com".to_owned());
+ n1.ports.tcp = Some(8090);
+ let mut n2 = node("n2", 1);
+ n2.ip = "broker.example.com".to_owned();
+ n2.ports.tcp = Some(8090);
+
+ assert!(cfg(vec![n1, n2]).validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_distinct_hostname_advertised_endpoints() {
+ let mut n1 = node("n1", 0);
+ n1.ip = "10.0.0.1".to_owned();
+ n1.advertised_address = Some("broker-1.example.com".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("broker-2.example.com".to_owned());
+ n2.ports.tcp = Some(8090);
+
+ assert!(cfg(vec![n1, n2]).validate().is_ok());
+ }
+
#[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 8c00ea10e..483fecaed 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -442,6 +442,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 2bed5b4b3..67b995f76 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -642,12 +642,21 @@ ca_file = ""
# flag, which selects the entry in this list that describes the current
# node. All other entries are remote peers.
#
-# 'ip' is the address the cluster advertises for the node (cluster metadata,
-# follower-to-primary HTTP forwarding); it is not the bind interface for
-# quic/http/websocket, which comes from each transport's own 'address'
-# setting above. 'ports' is the single source of listener ports in cluster
-# mode: every enabled transport needs an explicit per-node port, otherwise
-# the server refuses to start.
+# 'ip' is the node's roster address. Replica-to-replica traffic,
+# follower-to-primary HTTP forwarding, and the TCP listener use it. It is not
+# the bind interface for quic/http/websocket, which comes from each transport's
+# own 'address' setting above.
+#
+# 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). Accepts a literal IPv4/IPv6 address or a DNS hostname
+# (RFC 1123: ASCII letters, digits, '-' and '.'; no port, no trailing dot).
+# When unset, clients receive 'ip'.
+#
+# In cluster mode, 'ports' is the single source of listener ports: every
+# enabled transport needs an explicit per-node port, otherwise the server
+# refuses to start.
[[cluster.nodes]]
name = "iggy-node-1"
ip = "127.0.0.1"
@@ -664,6 +673,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 = "iggy-node-3.example.com"
# replica_id = 2
# ports = { tcp = 8092, quic = 8082, http = 3002, websocket = 8094,
tcp_replica = 9092 }
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 1fcd78495..9730f74fd 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -3350,6 +3350,7 @@ mod tests {
configs::ng_cluster::ClusterNodeConfig {
name: "node".to_owned(),
ip: ip.to_owned(),
+ advertised_address: None,
replica_id: 0,
ports: configs::ng_cluster::TransportPorts {
tcp: Some(18070),
diff --git a/core/server-ng/src/cluster_meta.rs
b/core/server-ng/src/cluster_meta.rs
index 9e969f3df..8a1bff535 100644
--- a/core/server-ng/src/cluster_meta.rs
+++ b/core/server-ng/src/cluster_meta.rs
@@ -28,7 +28,7 @@
//! leader, but the full roster is still returned). The self-synthesized single
//! node is the cluster-disabled fallback, shared by both callers.
-use configs::ng_cluster::{ClusterNodeConfig, TransportPorts};
+use configs::ng_cluster::{AdvertisedAddress, ClusterNodeConfig,
TransportPorts};
use iggy_common::{
ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus,
TransportEndpoints,
};
@@ -106,7 +106,7 @@ impl ClusterRoster {
.iter()
.map(|node| ClusterNode {
name: node.name.clone(),
- ip: node.ip.clone(),
+ ip: client_host(node),
endpoints: ports_to_endpoints(&node.ports),
role: role_for(primary_index, node.replica_id),
status: ClusterNodeStatus::Healthy,
@@ -135,6 +135,18 @@ impl ClusterRoster {
}
}
+/// Client-facing host in normalized form (lowercase hostname, canonical IP),
+/// matching what boot validation compared and what redirect URLs render, so
+/// textual config variants of one address publish identical metadata. A
+/// roster `ip` that parses as neither (boot only requires it non-empty)
+/// passes through verbatim; a configured `advertised_address` always parses,
+/// validation rejects it otherwise.
+fn client_host(node: &ClusterNodeConfig) -> String {
+ let host = node.advertised_address.as_deref().unwrap_or(&node.ip);
+ host.parse::<AdvertisedAddress>()
+ .map_or_else(|_| host.to_owned(), |address| address.to_string())
+}
+
const fn role_for(primary_index: Option<u8>, replica_id: u8) ->
ClusterNodeRole {
match primary_index {
Some(primary) if primary == replica_id => ClusterNodeRole::Leader,
@@ -150,3 +162,68 @@ 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");
+ }
+
+ #[test]
+ fn cluster_metadata_normalizes_advertised_hostname_to_lowercase() {
+ let metadata =
roster(Some("Broker.Example.COM".to_owned())).cluster_metadata(Some(0));
+
+ assert_eq!(metadata.nodes[0].ip, "broker.example.com");
+ }
+
+ #[test]
+ fn cluster_metadata_canonicalizes_advertised_ipv6_address() {
+ for equivalent_address in ["2001:DB8::1", "[2001:db8::1]"] {
+ let metadata =
roster(Some(equivalent_address.to_owned())).cluster_metadata(Some(0));
+
+ assert_eq!(
+ metadata.nodes[0].ip, "2001:db8::1",
+ "'{equivalent_address}' must publish canonical form"
+ );
+ }
+ }
+
+ #[test]
+ fn cluster_metadata_passes_unparsable_replica_ip_verbatim() {
+ let mut cluster_roster = roster(None);
+ cluster_roster.nodes[0].ip = "iggy_node".to_owned();
+
+ let metadata = cluster_roster.cluster_metadata(Some(0));
+
+ assert_eq!(metadata.nodes[0].ip, "iggy_node");
+ }
+}
diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs
index 511a1dd22..8ea07561e 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::{AdvertisedAddress, ClusterNodeConfig};
use iggy_binary_protocol::Operation;
use iggy_common::IggyError;
use serde::{Deserialize, Serialize};
@@ -500,39 +501,56 @@ pub(in crate::http) fn primary_redirect_location(
scheme: &str,
path_and_query: &str,
) -> Option<String> {
- let socket = primary_http_socket(roster, primary_index)?;
- Some(format!("{scheme}://{socket}{path_and_query}"))
+ let authority = primary_advertised_http_authority(roster, primary_index)?;
+ Some(format!("{scheme}://{authority}{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 authority (`host:port`) for a redirect. The
+/// advertised address is preferred, with the private roster IP retained as
+/// the compatibility fallback. [`AdvertisedAddress::authority`] brackets IPv6
+/// hosts and passes hostnames through, so the redirect URL stays valid; a
+/// host that is neither a valid IP nor a valid hostname yields `None` so
+/// callers fail closed.
+fn primary_advertised_http_authority(roster: &ClusterRoster, primary_index:
u8) -> Option<String> {
+ let (node, http_port) = primary_node(roster, primary_index)?;
+ let host = node.advertised_address.as_deref().unwrap_or(&node.ip);
+ let address = host.parse::<AdvertisedAddress>().ok()?;
+ Some(address.authority(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 +632,42 @@ 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_redirect_location_uses_advertised_hostname() {
+ let mut primary = node(0, "10.0.0.1", Some(8080));
+ primary.advertised_address = Some("broker-1.example.com".to_owned());
+ let roster = roster(vec![primary]);
+
+ assert_eq!(
+ primary_redirect_location(&roster, 0, "https", READ_PATH),
+
Some("https://broker-1.example.com: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();
