diff --git a/README.md b/README.md index 0e1925eeb8..9427205019 100644 --- a/README.md +++ b/README.md @@ -288,7 +288,7 @@ For configuration options and detailed help: You can also use environment variables to override any configuration setting: - Override TCP address - `IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server` + `IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server` - Set custom data path `IGGY_SYSTEM_PATH=/data/iggy cargo run --bin iggy-server` diff --git a/bdd/docker-compose.server.yml b/bdd/docker-compose.server.yml index e22484533c..5563040feb 100644 --- a/bdd/docker-compose.server.yml +++ b/bdd/docker-compose.server.yml @@ -65,6 +65,7 @@ services: - IGGY_ROOT_PASSWORD=iggy - IGGY_SYSTEM_PATH=local_data - IGGY_TCP_ADDRESS=0.0.0.0:8090 + - IGGY_NODE_ADVERTISED_ADDRESS=iggy-server - IGGY_HTTP_ADDRESS=0.0.0.0:3000 - IGGY_QUIC_ADDRESS=0.0.0.0:8080 - IGGY_WEBSOCKET_ADDRESS=0.0.0.0:8070 diff --git a/core/configs/src/server_config/cluster.rs b/core/configs/src/server_config/cluster.rs index a1874a6406..d98b01cd31 100644 --- a/core/configs/src/server_config/cluster.rs +++ b/core/configs/src/server_config/cluster.rs @@ -386,6 +386,7 @@ pub struct ClusterTlsConfig { #[serde(deny_unknown_fields)] pub struct ClusterNodeConfig { pub name: String, + /// Replica-plane address, dialed verbatim by every peer, a literal IP. pub ip: String, /// Optional client-facing address: a literal IP or a DNS hostname, /// validated as [`AdvertisedAddress`] at boot. Replica traffic continues @@ -438,10 +439,7 @@ pub struct AdvertisedAddressSelector { /// once, built wherever a roster is assembled for serving clients /// (listener/shard start). Per-request resolution never re-parses config /// strings: everything is snapshotted here, so mutating the source config -/// after conversion has no effect on what clients are told. Entries that do -/// not parse are dropped at build time; validation already rejects them -/// whenever the cluster is enabled, and a disabled cluster never consults -/// the roster. +/// after conversion has no effect on what clients are told. #[derive(Debug, Clone)] pub struct ResolvedClusterNode { config: ClusterNodeConfig, @@ -449,38 +447,72 @@ pub struct ResolvedClusterNode { /// addresses, in declaration order. selectors: Vec<(IpNet, AdvertisedAddress)>, /// Parsed catch-all: [`ClusterNodeConfig::advertised_address`], else the - /// roster [`ClusterNodeConfig::ip`]. `None` when the configured value - /// does not parse - a set `advertised_address` never falls through to - /// the private roster ip. - catch_all: Option, + /// roster [`ClusterNodeConfig::ip`]. A set `advertised_address` never + /// falls through to the private roster ip. + catch_all: AdvertisedAddress, /// Parsed roster [`ClusterNodeConfig::ip`], the replica-plane dial - /// address. `None` when the roster ip is not a literal IP (boot only - /// requires it non-empty); internal forwarding then has no dial target. - replica_ip: Option, + /// address. + replica_ip: IpAddr, } -impl From for ResolvedClusterNode { - fn from(config: ClusterNodeConfig) -> Self { - let selectors = config - .advertised_addresses - .iter() - .filter_map(|selector| { - let network = selector.client_cidr.parse::().ok()?; - let address = selector.address.parse::().ok()?; - Some((canonical_ip_net(network.trunc()), address)) - }) - .collect(); - let catch_all = match config.advertised_address.as_deref() { - Some(advertised_address) => advertised_address.parse().ok(), - None => config.ip.parse().ok(), - }; - let replica_ip = config.ip.parse().ok(); - Self { +impl TryFrom for ResolvedClusterNode { + type Error = ConfigurationError; + + fn try_from(config: ClusterNodeConfig) -> Result { + let replica_ip = config.ip.parse::().map_err(|error| { + eprintln!( + "Invalid cluster configuration: IP '{}' for node '{}' is not a literal IP \ + address: {error}", + config.ip, config.name + ); + ConfigurationError::InvalidConfigurationValue + })?; + + let catch_all = + match config.advertised_address.as_deref() { + Some(advertised_address) => advertised_address + .parse::() + .map_err(|error| { + eprintln!( + "Invalid cluster configuration: advertised_address \ + '{advertised_address}' for node '{}': {error}", + config.name + ); + ConfigurationError::InvalidConfigurationValue + })?, + None => AdvertisedAddress::Ip(replica_ip), + }; + + let mut selectors = Vec::with_capacity(config.advertised_addresses.len()); + for selector in &config.advertised_addresses { + let network = selector.client_cidr.parse::().map_err(|error| { + eprintln!( + "Invalid cluster configuration: advertised_addresses client_cidr '{}' for \ + node '{}': {error}", + selector.client_cidr, config.name + ); + ConfigurationError::InvalidConfigurationValue + })?; + let address = selector + .address + .parse::() + .map_err(|error| { + eprintln!( + "Invalid cluster configuration: advertised_addresses address '{}' for \ + node '{}': {error}", + selector.address, config.name + ); + ConfigurationError::InvalidConfigurationValue + })?; + selectors.push((canonical_ip_net(network.trunc()), address)); + } + + Ok(Self { config, selectors, catch_all, replica_ip, - } + }) } } @@ -496,33 +528,19 @@ impl ResolvedClusterNode { /// internal request forwarding. Never routed through the advertised /// ladder: this is what servers dial, not what clients are told. #[must_use] - pub fn replica_ip(&self) -> Option { + pub fn replica_ip(&self) -> IpAddr { self.replica_ip } /// The client-facing address for a client connecting from `client_ip`: - /// longest-prefix match over the selector networks, then the parsed - /// catch-all. `None` when no selector matches and the catch-all did not - /// parse; callers choose whether to fail closed (redirect URLs) or to - /// publish [`Self::raw_advertised_fallback`] verbatim (cluster metadata). + /// longest-prefix match over the selector networks, then the catch-all. + /// Always an address, since construction refused a node whose sources did + /// not parse. #[must_use] - pub fn advertised_for(&self, client_ip: Option) -> Option<&AdvertisedAddress> { + pub fn advertised_for(&self, client_ip: Option) -> &AdvertisedAddress { client_ip .and_then(|client_ip| self.selector_address(client_ip)) - .or(self.catch_all.as_ref()) - } - - /// The catch-all ladder ([`ClusterNodeConfig::advertised_address`], else - /// the roster [`ClusterNodeConfig::ip`]) as configured, unparsed. Cluster - /// metadata publishes this verbatim when [`Self::advertised_for`] finds - /// nothing: the roster `ip` is only validated non-empty, and Docker - /// service names with underscores exist in the wild. - #[must_use] - pub fn raw_advertised_fallback(&self) -> &str { - self.config - .advertised_address - .as_deref() - .unwrap_or(&self.config.ip) + .unwrap_or(&self.catch_all) } /// Longest-prefix match over the boot-parsed selector networks. The @@ -594,6 +612,10 @@ pub enum AdvertisedAddress { } impl AdvertisedAddress { + pub fn is_unspecified(&self) -> bool { + matches!(self, Self::Ip(ip) if ip.is_unspecified()) + } + /// 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 { @@ -955,10 +977,23 @@ impl Validatable for ClusterConfig { return Err(ConfigurationError::InvalidConfigurationValue); } - if node.ip.trim().is_empty() { + // The roster ip is dialed verbatim for replica traffic and is + // never resolved, so no hostname can work here whatever its + // shape. + let node_ip = node.ip.parse::().map_err(|error| { eprintln!( - "Invalid cluster configuration: IP cannot be empty for node '{}'", - node.name + "Invalid cluster configuration: IP '{}' for node '{}' is not a literal IP \ + address: {error}; set node.advertised_address for the name clients dial", + node.ip, node.name + ); + ConfigurationError::InvalidConfigurationValue + })?; + + if node_ip.is_unspecified() { + eprintln!( + "Invalid cluster configuration: IP '{}' for node '{}' is the unspecified \ + address; declare the address peers and clients reach this node at", + node.ip, node.name ); return Err(ConfigurationError::InvalidConfigurationValue); } @@ -1019,13 +1054,20 @@ impl Validatable 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. + // all. It is the wider of the two - the roster `ip` above is + // held to a literal IP - so a node reachable only by name still + // publishes that name to clients. let client_address = match node.advertised_address.as_deref() { Some(advertised_address) => match advertised_address.parse::() { + Ok(address) if address.is_unspecified() => { + eprintln!( + "Invalid cluster configuration: advertised_address '{advertised_address}' for node '{}' \ + is the unspecified address, which tells a client which interfaces this node accepts \ + on rather than where to reach it; declare a routable address", + node.name + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } Ok(address) => Some(address), Err(error) => { eprintln!( @@ -1077,6 +1119,14 @@ impl Validatable for ClusterConfig { return Err(ConfigurationError::InvalidConfigurationValue); } let address = match selector.address.parse::() { + Ok(address) if address.is_unspecified() => { + eprintln!( + "Invalid cluster configuration: advertised_addresses address '{}' for node '{}' \ + is the unspecified address; declare the address clients in '{}' reach this node at", + selector.address, node.name, selector.client_cidr + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } Ok(address) => address, Err(error) => { eprintln!( @@ -1101,10 +1151,7 @@ impl Validatable for ClusterConfig { // override shadowing the same node's wider selector); a conflict // means some client wins both entries and would resolve both // nodes to one endpoint. The catch-all is an implicit - // match-everything-else selector, so it pools the same way. A - // roster ip that fails the strict parse skips the pool: it can - // never equal a parsed host, and two raw ips sharing host:port - // are already rejected by the bind-endpoint check above. + // match-everything-else selector, so it pools the same way. if let Some(address) = &client_address { let catch_all_clients = EffectiveClients::for_catch_all(&selector_ranges); for (name, port) in &client_ports { @@ -1605,7 +1652,7 @@ mod advertised_for_tests { } fn resolved(node: ClusterNodeConfig) -> ResolvedClusterNode { - node.into() + ResolvedClusterNode::try_from(node).expect("a roster node the validator would accept") } fn ip(address: &str) -> IpAddr { @@ -1617,7 +1664,7 @@ mod advertised_for_tests { let node = node_with_selectors(Vec::new()); assert_eq!( resolved(node).advertised_for(Some(ip("10.0.0.7"))), - Some(&AdvertisedAddress::Ip(ip("203.0.113.10"))) + &AdvertisedAddress::Ip(ip("203.0.113.10")) ); } @@ -1627,16 +1674,18 @@ mod advertised_for_tests { node.advertised_address = None; assert_eq!( resolved(node).advertised_for(Some(ip("10.0.0.7"))), - Some(&AdvertisedAddress::Ip(ip("10.0.1.5"))) + &AdvertisedAddress::Ip(ip("10.0.1.5")) ); } #[test] - fn is_none_when_no_fallback_parses() { + fn refuses_a_node_whose_ip_is_not_an_address() { + // Nothing downstream carries a fallback for an unparsable source, so + // the conversion is where such a node has to stop. let mut node = node_with_selectors(Vec::new()); node.advertised_address = None; node.ip = "iggy_node".to_owned(); - assert_eq!(resolved(node).advertised_for(Some(ip("10.0.0.7"))), None); + assert!(ResolvedClusterNode::try_from(node).is_err()); } #[test] @@ -1644,7 +1693,7 @@ mod advertised_for_tests { let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); assert_eq!( resolved(node).advertised_for(Some(ip("10.0.200.7"))), - Some(&AdvertisedAddress::Ip(ip("10.0.1.5"))) + &AdvertisedAddress::Ip(ip("10.0.1.5")) ); } @@ -1653,7 +1702,7 @@ mod advertised_for_tests { let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); assert_eq!( resolved(node).advertised_for(Some(ip("192.168.0.7"))), - Some(&AdvertisedAddress::Ip(ip("203.0.113.10"))) + &AdvertisedAddress::Ip(ip("203.0.113.10")) ); } @@ -1662,7 +1711,7 @@ mod advertised_for_tests { let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); assert_eq!( resolved(node).advertised_for(None), - Some(&AdvertisedAddress::Ip(ip("203.0.113.10"))) + &AdvertisedAddress::Ip(ip("203.0.113.10")) ); } @@ -1674,12 +1723,12 @@ mod advertised_for_tests { ])); assert_eq!( node.advertised_for(Some(ip("10.0.200.7"))), - Some(&AdvertisedAddress::Ip(ip("10.0.1.5"))), + &AdvertisedAddress::Ip(ip("10.0.1.5")), "the /16 must win over the /8 even though it is declared second" ); assert_eq!( node.advertised_for(Some(ip("10.9.0.7"))), - Some(&AdvertisedAddress::Ip(ip("10.255.255.1"))), + &AdvertisedAddress::Ip(ip("10.255.255.1")), "a client outside the /16 but inside the /8 must match the /8" ); } @@ -1696,7 +1745,7 @@ mod advertised_for_tests { ]); assert_eq!( resolved(node).advertised_for(Some(ip("10.0.200.7"))), - Some(&AdvertisedAddress::Ip(ip("10.0.1.5"))) + &AdvertisedAddress::Ip(ip("10.0.1.5")) ); } @@ -1706,7 +1755,7 @@ mod advertised_for_tests { let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); assert_eq!( resolved(node).advertised_for(Some(ip("::ffff:10.0.0.7"))), - Some(&AdvertisedAddress::Ip(ip("10.0.1.5"))) + &AdvertisedAddress::Ip(ip("10.0.1.5")) ); } @@ -1718,7 +1767,7 @@ mod advertised_for_tests { let node = node_with_selectors(vec![selector("::ffff:10.0.0.0/104", "10.0.1.5")]); assert_eq!( resolved(node).advertised_for(Some(ip("10.0.0.7"))), - Some(&AdvertisedAddress::Ip(ip("10.0.1.5"))) + &AdvertisedAddress::Ip(ip("10.0.1.5")) ); } @@ -1727,7 +1776,7 @@ mod advertised_for_tests { let node = node_with_selectors(vec![selector("2001:db8::/32", "2001:db8::1")]); assert_eq!( resolved(node).advertised_for(Some(ip("2001:db8::7"))), - Some(&AdvertisedAddress::Ip(ip("2001:db8::1"))) + &AdvertisedAddress::Ip(ip("2001:db8::1")) ); } @@ -1736,9 +1785,7 @@ mod advertised_for_tests { let node = node_with_selectors(vec![selector("10.0.0.0/16", "Broker.Internal.Example")]); assert_eq!( resolved(node).advertised_for(Some(ip("10.0.0.7"))), - Some(&AdvertisedAddress::Hostname( - "broker.internal.example".to_owned() - )) + &AdvertisedAddress::Hostname("broker.internal.example".to_owned()) ); } } @@ -1945,6 +1992,43 @@ mod cluster_validate_tests { assert!(c.validate().is_err()); } + #[test] + fn validate_rejects_an_unspecified_node_ip() { + for wildcard in ["0.0.0.0", "::"] { + let mut nodes = vec![node("n1", 0), node("n2", 1)]; + nodes[0].ip = wildcard.to_owned(); + assert!(cfg(nodes).validate().is_err(), "{wildcard} is not dialable"); + } + } + + #[test] + fn validate_rejects_a_hostname_node_ip() { + for hostname in ["iggy_leader", "node-1.example.com"] { + let mut nodes = vec![node("n1", 0), node("n2", 1)]; + nodes[0].ip = hostname.to_owned(); + assert!( + cfg(nodes).validate().is_err(), + "'{hostname}' must not pass as a roster ip" + ); + } + } + + #[test] + fn validate_rejects_an_unspecified_advertised_address() { + for wildcard in ["0.0.0.0", "::"] { + let mut nodes = vec![node("n1", 0), node("n2", 1)]; + nodes[0].advertised_address = Some(wildcard.to_owned()); + assert!(cfg(nodes).validate().is_err(), "{wildcard} is not dialable"); + } + } + + #[test] + fn validate_rejects_an_unspecified_selector_address() { + let mut nodes = vec![node("n1", 0), node("n2", 1)]; + nodes[0].advertised_addresses = vec![selector("10.0.0.0/16", "0.0.0.0")]; + assert!(cfg(nodes).validate().is_err()); + } + #[test] fn validate_rejects_out_of_range_replica_id() { // 2 nodes total, so id 2 is out of range. @@ -2156,19 +2240,6 @@ mod cluster_validate_tests { 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); diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs index 1705e861ba..c7fc8244b0 100644 --- a/core/configs/src/server_config/defaults.rs +++ b/core/configs/src/server_config/defaults.rs @@ -28,6 +28,7 @@ use super::cluster::{ }; use super::message_bus::MessageBusConfig; use super::metadata::MetadataConfig; +use super::node::NodeConfig; use super::partition::PartitionConfig; use super::quic::{QuicCertificateConfig, QuicConfig}; use super::server::ServerConfig; @@ -51,6 +52,7 @@ impl Default for ServerConfig { consumer_group: ConsumerGroupConfig::default(), data_maintenance: DataMaintenanceConfig::default(), heartbeat: HeartbeatConfig::default(), + node: NodeConfig::default(), personal_access_token: PersonalAccessTokenConfig::default(), system: Arc::new(ServerSystemConfig::default()), quic: QuicConfig::default(), diff --git a/core/configs/src/server_config/mod.rs b/core/configs/src/server_config/mod.rs index 718aee5cdd..c2ff98a353 100644 --- a/core/configs/src/server_config/mod.rs +++ b/core/configs/src/server_config/mod.rs @@ -26,6 +26,7 @@ pub mod defaults; pub mod displays; pub mod message_bus; pub mod metadata; +pub mod node; pub mod partition; pub mod quic; pub mod server; diff --git a/core/configs/src/server_config/node.rs b/core/configs/src/server_config/node.rs new file mode 100644 index 0000000000..938e490ef1 --- /dev/null +++ b/core/configs/src/server_config/node.rs @@ -0,0 +1,98 @@ +// 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. + +// This node's own client-facing identity, for the cluster-disabled server. + +use super::COMPONENT; +use super::cluster::AdvertisedAddress; +use crate::ConfigurationError; +use configs::ConfigEnv; +use iggy_common::Validatable; +use serde::{Deserialize, Serialize}; + +/// Named to match its roster counterpart: `advertised_address` here and +/// `cluster.nodes[*].advertised_address` there are the same setting for the +/// same question, and an operator moving between the two modes should not have +/// to learn a second spelling. +#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)] +#[serde(deny_unknown_fields)] +pub struct NodeConfig { + /// Client-facing address: a literal IP or a DNS hostname. `None` leaves + /// the server deriving one from its bind address. + #[serde(default)] + pub advertised_address: Option, +} + +impl Validatable for NodeConfig { + fn validate(&self) -> Result<(), ConfigurationError> { + let Some(address) = self.advertised_address.as_deref() else { + return Ok(()); + }; + + let parsed = address.parse::().map_err(|error| { + eprintln!("{COMPONENT} - node.advertised_address '{address}': {error}"); + ConfigurationError::InvalidConfigurationValue + })?; + + if parsed.is_unspecified() { + eprintln!( + "{COMPONENT} - node.advertised_address '{address}' is the unspecified address, \ + which tells a client which interfaces this node accepts on rather than where to \ + reach it; declare a routable address or leave it unset" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn advertised(address: &str) -> NodeConfig { + NodeConfig { + advertised_address: Some(address.to_owned()), + } + } + + #[test] + fn validate_accepts_an_unset_address() { + assert!(NodeConfig::default().validate().is_ok()); + } + + #[test] + fn validate_accepts_a_routable_address() { + assert!(advertised("203.0.113.10").validate().is_ok()); + assert!(advertised("broker-1.example.com").validate().is_ok()); + assert!(advertised("2001:db8::1").validate().is_ok()); + } + + #[test] + fn validate_rejects_an_unspecified_address() { + assert!(advertised("0.0.0.0").validate().is_err()); + assert!(advertised("::").validate().is_err()); + } + + #[test] + fn validate_rejects_an_unparsable_address() { + assert!(advertised("broker-1.example.com:8090").validate().is_err()); + assert!(advertised("10.0.0.256").validate().is_err()); + assert!(advertised("").validate().is_err()); + } +} diff --git a/core/configs/src/server_config/server.rs b/core/configs/src/server_config/server.rs index bd7d168b3a..5868afd637 100644 --- a/core/configs/src/server_config/server.rs +++ b/core/configs/src/server_config/server.rs @@ -19,6 +19,7 @@ use super::COMPONENT; use super::cluster::ClusterConfig; use super::message_bus::MessageBusConfig; use super::metadata::MetadataConfig; +use super::node::NodeConfig; use super::partition::PartitionConfig; use super::quic::QuicConfig; use super::tcp::TcpConfig; @@ -111,6 +112,7 @@ pub struct ServerConfig { pub consumer_group: ConsumerGroupConfig, pub data_maintenance: DataMaintenanceConfig, #[serde(default)] + pub node: NodeConfig, pub personal_access_token: PersonalAccessTokenConfig, pub heartbeat: HeartbeatConfig, pub system: Arc, diff --git a/core/configs/src/server_config/validators.rs b/core/configs/src/server_config/validators.rs index f8ff7b48a6..55357c1987 100644 --- a/core/configs/src/server_config/validators.rs +++ b/core/configs/src/server_config/validators.rs @@ -31,6 +31,7 @@ use crate::common::http::HMAC_JWT_ALGORITHMS; use crate::common::validators::SEGMENT_MAX_SIZE_BYTES; use err_trail::ErrContext; use iggy_common::{IggyExpiry, Validatable}; +use std::net::SocketAddr; /// compio-ws (tungstenite 0.29) `write_buffer_size` default. Used to /// evaluate the `max_write_buffer_size > write_buffer_size` invariant @@ -76,6 +77,10 @@ impl Validatable for ServerConfig { self.cluster.validate().error(|e: &ConfigurationError| { format!("{COMPONENT} (error: {e}) - failed to validate cluster config") })?; + self.node.validate().error(|e: &ConfigurationError| { + format!("{COMPONENT} (error: {e}) - failed to validate node config") + })?; + self.validate_client_facing_address()?; self.metadata.validate().error(|e: &ConfigurationError| { format!("{COMPONENT} (error: {e}) - failed to validate metadata config") })?; @@ -397,6 +402,38 @@ fn reject_unsupported(config: &ServerConfig) -> Result<(), ConfigurationError> { Ok(()) } +impl ServerConfig { + /// `tcp.address` must name a bind address, and a wildcard one must be + /// paired with a declared client-facing address. + fn validate_client_facing_address(&self) -> Result<(), ConfigurationError> { + let bind = self.tcp.address.parse::().map_err(|error| { + eprintln!( + "{COMPONENT} - tcp.address '{}' is not an address and port: {error}. The host \ + is required and must be a literal IP, so ':PORT' and 'hostname:PORT' are \ + both rejected; use 127.0.0.1:PORT for loopback or 0.0.0.0:PORT to accept on \ + every interface.", + self.tcp.address + ); + ConfigurationError::InvalidConfigurationValue + })?; + + if self.cluster.enabled || self.node.advertised_address.is_some() { + return Ok(()); + } + if !bind.ip().is_unspecified() { + return Ok(()); + } + + eprintln!( + "{COMPONENT} - tcp.address binds the wildcard {bind}, which says which interfaces this \ + node accepts on rather than where a client reaches it, so cluster metadata would carry \ + no address for this node. Set node.advertised_address to the address clients dial, or \ + bind a concrete address." + ); + Err(ConfigurationError::InvalidConfigurationValue) + } +} + #[cfg(test)] mod tests { use super::super::cluster::{ClusterNodeConfig, TransportPorts}; @@ -443,6 +480,56 @@ mod tests { ); } + #[test] + fn given_wildcard_bind_without_advertised_address_when_validating_should_reject() { + for wildcard in ["0.0.0.0:8090", "[::]:8090"] { + let config = config_with_override(&format!( + "[tcp]\naddress = \"{wildcard}\"\n[cluster]\nenabled = false\n" + )); + assert!( + config.validate().is_err(), + "{wildcard} names no address a client can dial" + ); + } + } + + #[test] + fn given_a_hostless_or_named_bind_address_when_validating_should_reject() { + for address in [":8090", "localhost:8090", "0.0.0.0", "not-an-address"] { + let config = config_with_override(&format!("[tcp]\naddress = \"{address}\"\n")); + assert!( + config.validate().is_err(), + "{address} does not name a bind address" + ); + } + } + + #[test] + fn given_wildcard_bind_with_advertised_address_when_validating_should_pass() { + let config = config_with_override( + "[tcp]\naddress = \"0.0.0.0:8090\"\n[cluster]\nenabled = false\n\ + [node]\nadvertised_address = \"broker-1.example.com\"\n", + ); + assert!(config.validate().is_ok()); + } + + #[test] + fn given_concrete_bind_without_advertised_address_when_validating_should_pass() { + let config = config_with_override( + "[tcp]\naddress = \"192.0.2.10:8090\"\n[cluster]\nenabled = false\n", + ); + assert!(config.validate().is_ok()); + } + + #[test] + fn given_clustered_wildcard_bind_without_advertised_address_when_validating_should_pass() { + // The roster answers the client-facing address per node, so the bind + // address is free to be a wildcard with nothing declared here. + let config = + config_with_override("[tcp]\naddress = \"0.0.0.0:8090\"\n[cluster]\nenabled = true\n"); + assert!(config.validate().is_ok()); + } + #[test] fn given_shipped_default_config_when_validating_should_pass() { let config: ServerConfig = Figment::new() diff --git a/core/integration/tests/server/cluster_metadata_advertised.rs b/core/integration/tests/server/cluster_metadata_advertised.rs new file mode 100644 index 0000000000..d1a3750ef6 --- /dev/null +++ b/core/integration/tests/server/cluster_metadata_advertised.rs @@ -0,0 +1,81 @@ +// 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. + +//! `node.advertised_address` on a cluster-disabled server: what a client is +//! told about the one node in the roster. +//! +//! Without a roster the server has only its own bind address to reason from, +//! and a bind address answers which interfaces it accepts on - never where a +//! client reaches it. Behind any NAT (published container ports, a Service, a +//! load balancer) the two are different addresses, and this setting is the +//! only way to state the second one. The bind-derived fallback and the empty +//! answer for a wildcard bind are pinned at unit level (`cluster_meta.rs`) +//! and across the SDKs in `bdd/`; what needs a real server is that a declared +//! address survives config load and reaches the wire. + +use iggy::prelude::*; +use integration::iggy_harness; + +const ADVERTISED_ADDRESS: &str = "broker-1.example.com"; + +// The harness runs a VSR cluster by default, where the roster answers the +// client-facing address per node and this setting is deliberately ignored. One +// node with clustering off is the shape that reaches the self-synthesized +// path: `--replica-id 0` stays valid without a cluster, any higher id does not. +#[iggy_harness( + cluster_nodes = 1, + server(cluster.enabled = false, node.advertised_address = "broker-1.example.com") +)] +async fn given_a_declared_advertised_address_when_getting_cluster_metadata_should_publish_it( + harness: &TestHarness, +) { + let client = harness + .node(0) + .tcp_client() + .expect("tcp client") + .with_root_login() + .connect() + .await + .expect("connect"); + + let metadata = client + .get_cluster_metadata() + .await + .expect("get cluster metadata"); + + assert_eq!( + metadata.nodes.len(), + 1, + "a cluster-disabled server reports itself alone, got {metadata}" + ); + assert!( + !metadata.name.is_empty(), + "the single-node label still names the cluster" + ); + let node = &metadata.nodes[0]; + // The harness binds a concrete loopback address, so this also pins the + // precedence: a declaration outranks an address the bind could vouch for, + // because only the declaration is a claim about reachability. + assert_eq!( + node.ip, ADVERTISED_ADDRESS, + "the declared address must reach the wire verbatim" + ); + assert_ne!( + node.endpoints.tcp, 0, + "the self node reports its real tcp port alongside the declared address" + ); +} diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs index f31ca33a02..7c4f440229 100644 --- a/core/integration/tests/server/mod.rs +++ b/core/integration/tests/server/mod.rs @@ -50,6 +50,9 @@ mod http_rbac; mod http_tls; // Binary GetClusterMetadata must serve the real roster from a VSR cluster. mod cluster_metadata_vsr; +// A declared node.advertised_address outranks the bind address a +// cluster-disabled server would otherwise publish. +mod cluster_metadata_advertised; // A metadata view change must persist the advanced view and recover it from disk // across a replica restart. mod cluster_view_durability_vsr; diff --git a/core/server/README.md b/core/server/README.md index f0150b32eb..1c4fc81c4b 100644 --- a/core/server/README.md +++ b/core/server/README.md @@ -12,6 +12,15 @@ cargo run --bin iggy-server --release The Docker image `apache/iggy:latest` ships the server together with the CLI; the `edge` tag tracks the latest development build. +The image binds every listener to `0.0.0.0` so the container is reachable from outside it. A wildcard bind says which interfaces accept connections, not where a client reaches the server, so the address to publish in cluster metadata has to be supplied and the server refuses to start without it. On a single host with published ports that address is `localhost`: + +```sh +docker run -p 3000:3000 -p 8090:8090 \ + -e IGGY_NODE_ADVERTISED_ADDRESS=localhost apache/iggy:latest +``` + +Use the hostname or load balancer address clients actually dial when they are not on the same host. The Helm chart derives it from the Service DNS name. + To run one node of a cluster, pass its replica ID from the `cluster.nodes` roster: ```sh @@ -27,7 +36,7 @@ Settings are read from [config.toml](config.toml), resolved relative to the work Any single value can be overridden with an `IGGY_`-prefixed environment variable that mirrors the TOML path: ```sh -IGGY_TCP_ADDRESS=0.0.0.0:8090 IGGY_HTTP_ENABLED=false cargo run --bin iggy-server +IGGY_TCP_ADDRESS=127.0.0.1:8090 IGGY_HTTP_ENABLED=false cargo run --bin iggy-server ``` Cluster membership, quorum and replica addressing live under `[cluster]`. diff --git a/core/server/config.toml b/core/server/config.toml index f420aeeb30..ae57a0c3d8 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -186,6 +186,17 @@ cert_file = "core/certs/iggy_cert.pem" # Path to the TLS key file. key_file = "core/certs/iggy_key.pem" +# This node's own client-facing identity, used when 'cluster.enabled' is +# false. +[node] +# A literal IP or a DNS hostname, without a port. Named to match its roster +# counterpart 'cluster.nodes.advertised_address', which answers the same +# question per node. The unspecified address ("0.0.0.0", "::") is rejected: it +# is the bind address's answer to a question clients are not asking. Commented +# out here because the shipped 'tcp.address' binds a concrete address and needs +# no declaration. +#advertised_address = "broker-1.example.com" + # TCP server configuration. [tcp] # Determines if the TCP server is active. @@ -680,7 +691,7 @@ ca_file = "" # # [[cluster.nodes]] # name = "iggy-node-1" -# ip = "10.0.1.5" # replica plane + last-resort fallback +# ip = "10.0.1.5" # replica plane, literal IP only # advertised_address = "203.0.113.10" # catch-all for unmatched clients # replica_id = 0 # ports = { tcp = 8090, http = 3000, tcp_replica = 9090 } diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 01ef2d1fab..46b296fa5e 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -16,7 +16,7 @@ // under the License. use crate::auth::warm_dummy_password_hash; -use crate::cluster_meta::ClusterRoster; +use crate::cluster_meta::{ClusterRoster, resolved_roster_nodes, self_advertised_address}; use crate::config_writer::write_current_config; use crate::dispatch::{ make_client_request_handler, make_deferred_client_request_handler, @@ -1703,23 +1703,32 @@ fn spawn_shutdown_watchdog( /// the shared [`ClusterRoster`] so the binary `GetClusterMetadata` read serves /// the real topology. `self_*` back only the cluster-disabled self-synthesis /// and carry the requested listener ports from the resolved topology, not the -/// bound ones (a `:0` wildcard is reported as 0). +/// bound ones (a `:0` wildcard is reported as 0). The self address resolves +/// through [`self_advertised_address`], which boot validation has already +/// guaranteed names somewhere a client can dial. fn build_cluster_roster( + shard_id: u16, config: &ServerConfig, topology: &TcpTopology, metadata_view: Arc, -) -> ClusterRoster { - ClusterRoster { +) -> Result { + let declared = config.node.advertised_address.as_deref(); + let self_advertised = self_advertised_address(declared, topology.client_listen_addr.ip()); + // The roster answers this per node, so a value here would be read by + // nobody. Silence would leave the operator believing it took effect. + // Every shard builds its own roster off the same config, so keep the + // operator-facing explanation to one line per process. + if declared.is_some() && config.cluster.enabled && shard_id == 0 { + warn!( + "node.advertised_address is set but cluster.enabled is true, so it is ignored; \ + the client-facing address of each node comes from its cluster.nodes entry" + ); + } + Ok(ClusterRoster { enabled: config.cluster.enabled, name: config.cluster.name.clone(), - nodes: config - .cluster - .nodes - .iter() - .cloned() - .map(Into::into) - .collect(), - self_ip: topology.client_listen_addr.ip().to_string(), + nodes: resolved_roster_nodes(&config.cluster).map_err(ServerError::Config)?, + self_advertised, self_ports: configs::cluster::TransportPorts { tcp: Some(topology.client_listen_addr.port()), quic: topology.quic_listen_addr.map(|addr| addr.port()), @@ -1728,7 +1737,7 @@ fn build_cluster_roster( tcp_replica: None, }, metadata_view, - } + }) } #[allow(clippy::too_many_arguments, clippy::too_many_lines)] @@ -1984,10 +1993,11 @@ async fn build_shard_for_thread( sessions .borrow_mut() .set_cluster_roster(Rc::new(build_cluster_roster( + shard_id, config, topology, metadata_view, - ))); + )?)); let shard_name = format!("server-shard-{shard_id}"); let built = IggyShardBuilder::new( ShardIdentity::new(shard_id, shard_name), @@ -2974,6 +2984,12 @@ async fn start_tcp_runtime( // reactor, so it binds independently. Shard-0 gating comes from the sole // caller of this function. if let Some(http_addr) = topology.http_listen_addr { + // One host for all four transports, resolved from the client-facing + // TCP bind so both listeners publish the same node address. + let self_advertised = self_advertised_address( + config.node.advertised_address.as_deref(), + topology.client_listen_addr.ip(), + ); let self_ports = configs::cluster::TransportPorts { tcp: config .tcp @@ -2991,6 +3007,7 @@ async fn start_tcp_runtime( config.personal_access_token.max_tokens_per_user, &config.cluster, Arc::clone(&config.system), + &self_advertised, self_ports, ) .await?; diff --git a/core/server/src/cluster_meta.rs b/core/server/src/cluster_meta.rs index 7372d3e597..021b6aadda 100644 --- a/core/server/src/cluster_meta.rs +++ b/core/server/src/cluster_meta.rs @@ -28,7 +28,8 @@ //! leader, but the full roster is still returned). The self-synthesized single //! node is the cluster-disabled fallback, shared by both callers. -use configs::cluster::{ResolvedClusterNode, TransportPorts}; +use configs::ConfigurationError; +use configs::cluster::{ClusterConfig, ResolvedClusterNode, TransportPorts}; use iggy_common::{ ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, TransportEndpoints, }; @@ -43,6 +44,34 @@ const SELF_NODE_NAME: &str = "iggy-node"; /// single-node label. const SINGLE_NODE_CLUSTER_NAME: &str = "single-node"; +pub fn self_advertised_address(declared: Option<&str>, bind: IpAddr) -> String { + declared.map_or_else(|| bind.to_string(), ToOwned::to_owned) +} + +/// Resolve the roster a [`ClusterRoster`] serves, which is only the configured +/// one while the cluster is enabled. Gated on the same flag the config +/// validator gates itself on: a disabled cluster leaves `cluster.nodes` +/// unvalidated, and a stale entry left there must not fail a boot that never +/// consults it. +/// +/// # Errors +/// +/// Returns [`ConfigurationError`] when an enabled roster carries a node whose +/// address does not parse, which boot validation rejects first. +pub fn resolved_roster_nodes( + cluster: &ClusterConfig, +) -> Result, ConfigurationError> { + if !cluster.enabled { + return Ok(Vec::new()); + } + cluster + .nodes + .iter() + .cloned() + .map(ResolvedClusterNode::try_from) + .collect() +} + /// Config-derived cluster topology reported by cluster-metadata reads. /// /// Copied out of `ClusterConfig` at listener/shard start so both handlers stay @@ -54,8 +83,9 @@ pub struct ClusterRoster { /// Roster nodes with selectors parsed once at roster build, so the /// per-request address resolution never re-parses config strings. pub nodes: Vec, - /// This node's own address, reported for the synthesized self node. - pub self_ip: String, + /// This node's own client-facing address, reported for the synthesized + /// self node (see [`self_advertised_address`]). + pub self_advertised: String, /// This node's own client ports for the same self node (`None` = transport /// disabled). pub self_ports: TransportPorts, @@ -69,15 +99,16 @@ pub struct ClusterRoster { pub const METADATA_VIEW_UNKNOWN: u64 = u64::MAX; impl ClusterRoster { - /// A cluster-disabled roster with no self address. Used as the pre-bootstrap - /// default before the real roster is installed; [`Self::cluster_metadata`] - /// on it synthesizes a bare single node. + /// A cluster-disabled roster with no self address. The pre-bootstrap + /// placeholder a [`crate::session_manager::SessionManager`] holds until + /// bootstrap installs the real roster, which happens before any listener + /// accepts, so its blank address is never served to a client. pub fn disabled() -> Self { Self { enabled: false, name: String::new(), nodes: Vec::new(), - self_ip: String::new(), + self_advertised: String::new(), self_ports: TransportPorts::default(), metadata_view: Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)), } @@ -131,12 +162,14 @@ impl ClusterRoster { } } + /// The cluster-disabled single node, carrying the address + /// [`self_advertised_address`] resolved. fn self_metadata(&self) -> ClusterMetadata { ClusterMetadata { name: SINGLE_NODE_CLUSTER_NAME.to_owned(), nodes: vec![ClusterNode { name: SELF_NODE_NAME.to_owned(), - ip: self.self_ip.clone(), + ip: self.self_advertised.clone(), endpoints: ports_to_endpoints(&self.self_ports), role: ClusterNodeRole::Leader, status: ClusterNodeStatus::Healthy, @@ -149,16 +182,11 @@ impl ClusterRoster { /// matching what boot validation compared and what redirect URLs render, so /// textual config variants of one address publish identical metadata. The /// per-client-network selectors, the catch-all `advertised_address`, and the -/// roster `ip` are consulted in that order ([`ResolvedClusterNode::advertised_for`]). -/// Metadata deliberately does NOT fail closed like the redirect path: a host -/// that parses as neither IP nor hostname (the roster `ip` is only validated -/// non-empty - Docker service names with underscores exist in the wild) -/// publishes verbatim via [`ResolvedClusterNode::raw_advertised_fallback`]. +/// roster `ip` are consulted in that order +/// ([`ResolvedClusterNode::advertised_for`]), which always resolves: a node +/// whose sources do not parse never becomes a [`ResolvedClusterNode`]. fn client_host(node: &ResolvedClusterNode, client_ip: Option) -> String { - node.advertised_for(client_ip).map_or_else( - || node.raw_advertised_fallback().to_owned(), - ToString::to_string, - ) + node.advertised_for(client_ip).to_string() } const fn role_for(primary_index: Option, replica_id: u8) -> ClusterNodeRole { @@ -198,8 +226,8 @@ mod tests { ClusterRoster { enabled: true, name: "test-cluster".to_owned(), - nodes: vec![node.into()], - self_ip: "127.0.0.1".to_owned(), + nodes: vec![ResolvedClusterNode::try_from(node).expect("valid roster node")], + self_advertised: "127.0.0.1".to_owned(), self_ports: TransportPorts::default(), metadata_view: Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)), } @@ -244,16 +272,6 @@ mod tests { } } - #[test] - fn cluster_metadata_passes_unparsable_replica_ip_verbatim() { - let mut node = node_config(None); - node.ip = "iggy_node".to_owned(); - - let metadata = roster_of(node).cluster_metadata(Some(0), None); - - assert_eq!(metadata.nodes[0].ip, "iggy_node"); - } - #[test] fn cluster_metadata_serves_the_selector_address_to_a_matching_client() { let mut node = node_config(Some("203.0.113.10".to_owned())); @@ -289,4 +307,76 @@ mod tests { assert_eq!(metadata.nodes[0].ip, "broker.internal.test"); } + + #[test] + fn resolved_roster_nodes_ignores_a_roster_a_disabled_cluster_never_reads() { + // The shipped config carries a roster with the cluster off, so an + // entry that stopped parsing must not fail a boot that never serves + // it: the validator skips those entries for the same reason. + let mut cluster = ClusterConfig { + enabled: false, + ..ClusterConfig::default() + }; + cluster.nodes[0].ip = "iggy-server".to_owned(); + + assert!( + resolved_roster_nodes(&cluster) + .expect("no roster to resolve") + .is_empty() + ); + } + + #[test] + fn resolved_roster_nodes_refuses_an_enabled_roster_that_does_not_parse() { + let mut cluster = ClusterConfig { + enabled: true, + ..ClusterConfig::default() + }; + cluster.nodes[0].ip = "iggy-server".to_owned(); + + assert!(resolved_roster_nodes(&cluster).is_err()); + } + + #[test] + fn self_advertised_address_prefers_a_declared_address() { + assert_eq!( + self_advertised_address(Some("broker-1.example.com"), "192.0.2.10".parse().unwrap()), + "broker-1.example.com" + ); + } + + #[test] + fn self_advertised_address_falls_back_to_the_bind_address() { + assert_eq!( + self_advertised_address(None, "192.0.2.10".parse().unwrap()), + "192.0.2.10" + ); + assert_eq!( + self_advertised_address(None, "2001:db8::1".parse().unwrap()), + "2001:db8::1" + ); + } + + #[test] + fn self_metadata_synthesizes_a_single_leader_node() { + let roster = ClusterRoster { + enabled: false, + name: String::new(), + nodes: Vec::new(), + self_advertised: "broker-1.example.com".to_owned(), + self_ports: TransportPorts { + tcp: Some(8090), + ..TransportPorts::default() + }, + metadata_view: Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)), + }; + + let metadata = roster.cluster_metadata(None, None); + + assert_eq!(metadata.nodes.len(), 1); + assert_eq!(metadata.nodes[0].ip, "broker-1.example.com"); + assert_eq!(metadata.nodes[0].endpoints.tcp, 8090); + assert_eq!(metadata.nodes[0].role, ClusterNodeRole::Leader); + assert_eq!(metadata.nodes[0].status, ClusterNodeStatus::Healthy); + } } diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index 9251bf925b..4097d05402 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -4974,8 +4974,13 @@ mod tests { let multi_node = Rc::new(ClusterRoster { enabled: true, name: "test-cluster".to_owned(), - nodes: vec![roster_node("node-0").into(), roster_node("node-1").into()], - self_ip: "127.0.0.1".to_owned(), + nodes: ["node-0", "node-1"] + .map(|name| { + configs::cluster::ResolvedClusterNode::try_from(roster_node(name)) + .expect("valid roster node") + }) + .to_vec(), + self_advertised: "127.0.0.1".to_owned(), self_ports: TransportPorts::default(), metadata_view: Arc::new(std::sync::atomic::AtomicU64::new( crate::cluster_meta::METADATA_VIEW_UNKNOWN, diff --git a/core/server/src/http.rs b/core/server/src/http.rs index 28c6ebeb01..3dc2e0a7c4 100644 --- a/core/server/src/http.rs +++ b/core/server/src/http.rs @@ -65,7 +65,7 @@ use tower_http::cors::{AllowOrigin, CorsLayer}; use tracing::{error, info, warn}; use crate::bootstrap::ServerShard; -use crate::cluster_meta::ClusterRoster; +use crate::cluster_meta::{ClusterRoster, resolved_roster_nodes}; use crate::http::handlers::{ change_password, create_cg, create_partitions, create_pat, create_stream, create_topic, create_user, delete_cg, delete_consumer_offset, delete_partitions, delete_pat, delete_segments, @@ -103,6 +103,7 @@ pub async fn start( max_tokens_per_user: u32, cluster: &ClusterConfig, system_config: Arc, + self_advertised: &str, self_ports: TransportPorts, ) -> Result<(), ServerError> { // In cluster mode with no configured JWT secret the signing key derives @@ -148,8 +149,8 @@ pub async fn start( roster: ClusterRoster { enabled: cluster.enabled, name: cluster.name.clone(), - nodes: cluster.nodes.iter().cloned().map(Into::into).collect(), - self_ip: bound_addr.ip().to_string(), + nodes: resolved_roster_nodes(cluster).map_err(ServerError::Config)?, + self_advertised: self_advertised.to_owned(), // The self node reports the live bound HTTP port; the other client // ports arrive resolved from the caller. self_ports: TransportPorts { diff --git a/core/server/src/http/error.rs b/core/server/src/http/error.rs index 45ef05315c..db3edcc5ac 100644 --- a/core/server/src/http/error.rs +++ b/core/server/src/http/error.rs @@ -555,7 +555,7 @@ pub(in crate::http) fn primary_http_socket( primary_index: u8, ) -> Option { let (node, http_port) = primary_node(roster, primary_index)?; - Some(SocketAddr::new(node.replica_ip()?, http_port)) + Some(SocketAddr::new(node.replica_ip(), http_port)) } /// Resolve the client-facing HTTP authority (`host:port`) for a redirect @@ -563,18 +563,16 @@ pub(in crate::http) fn primary_http_socket( /// match first, then the catch-all advertised address, then the private /// roster IP as the compatibility fallback. `AdvertisedAddress::authority` /// brackets IPv6 hosts and passes hostnames through, so the redirect URL -/// stays valid. This is the fail-closed caller: a host that is neither a -/// valid IP nor a valid hostname yields `None` and the redirect becomes a -/// 503 rather than a `Location` pointing at an unparsable target (cluster -/// metadata makes the opposite choice and publishes such a host verbatim). +/// stays valid. `None` here means the roster has no node at `primary_index` +/// or that node declares no HTTP port, never that its address failed to +/// parse: such a node never becomes a [`ResolvedClusterNode`]. fn primary_advertised_http_authority( roster: &ClusterRoster, primary_index: u8, client_ip: Option, ) -> Option { let (node, http_port) = primary_node(roster, primary_index)?; - let address = node.advertised_for(client_ip)?; - Some(address.authority(http_port)) + Some(node.advertised_for(client_ip).authority(http_port)) } fn primary_node(roster: &ClusterRoster, primary_index: u8) -> Option<(&ResolvedClusterNode, u16)> { @@ -614,8 +612,11 @@ mod tests { ClusterRoster { enabled: true, name: "test-cluster".to_owned(), - nodes: nodes.into_iter().map(Into::into).collect(), - self_ip: "127.0.0.1".to_owned(), + nodes: nodes + .into_iter() + .map(|node| ResolvedClusterNode::try_from(node).expect("valid roster node")) + .collect(), + self_advertised: "127.0.0.1".to_owned(), self_ports: TransportPorts::default(), metadata_view: std::sync::Arc::new(std::sync::atomic::AtomicU64::new( crate::cluster_meta::METADATA_VIEW_UNKNOWN, diff --git a/examples/csharp/README.md b/examples/csharp/README.md index 995859aff8..53d2afd5fc 100644 --- a/examples/csharp/README.md +++ b/examples/csharp/README.md @@ -16,7 +16,7 @@ You can also customize the server using environment variables: ```bash ## Example: Enable HTTP transport and set custom address -IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server +IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server ``` ## Basic Examples diff --git a/examples/go/README.md b/examples/go/README.md index d463dee8bc..3e6eed2f33 100644 --- a/examples/go/README.md +++ b/examples/go/README.md @@ -16,7 +16,7 @@ You can also customize the server using environment variables: ```bash ## Example: Enable HTTP transport and set custom address -IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server +IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server ``` You can run multiple producers and consumers simultaneously to observe how messages are distributed across clients. diff --git a/examples/java/README.md b/examples/java/README.md index 4507d90095..b506823735 100644 --- a/examples/java/README.md +++ b/examples/java/README.md @@ -39,7 +39,7 @@ You can also customize the server using environment variables: ```bash ## Example: set a custom TCP address -IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server +IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server ``` ## Basic Examples diff --git a/examples/node/README.md b/examples/node/README.md index 90447387ae..a49fc3f3f3 100644 --- a/examples/node/README.md +++ b/examples/node/README.md @@ -8,7 +8,8 @@ To run any example, first start the server with ```bash # Using latest release -docker run --rm -p 8080:8080 -p 3000:3000 -p 8090:8090 apache/iggy:latest +docker run --rm -p 8080:8080 -p 3000:3000 -p 8090:8090 \ + -e IGGY_NODE_ADVERTISED_ADDRESS=localhost apache/iggy:latest # Or build from source (recommended for development) cd ../../ && cargo run --bin iggy-server @@ -24,7 +25,7 @@ You can also customize the server using environment variables: ```bash ## Example: Enable HTTP transport and set custom address -IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server +IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server ``` and then install Node.js dependencies: diff --git a/examples/python/README.md b/examples/python/README.md index 7e5da180d4..4e5c0c740d 100644 --- a/examples/python/README.md +++ b/examples/python/README.md @@ -8,7 +8,8 @@ To run any example, first start the server with ```bash # Using latest release -docker run --rm -p 8080:8080 -p 3000:3000 -p 8090:8090 apache/iggy:latest +docker run --rm -p 8080:8080 -p 3000:3000 -p 8090:8090 \ + -e IGGY_NODE_ADVERTISED_ADDRESS=localhost apache/iggy:latest # Or build from source (recommended for development) cd ../../ && cargo run --bin iggy-server @@ -24,7 +25,7 @@ You can also customize the server using environment variables: ```bash ## Example: Enable HTTP transport and set custom address -IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server +IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server ``` and then install Python dependencies: diff --git a/examples/rust/README.md b/examples/rust/README.md index d14b5491f5..b7b296e77c 100644 --- a/examples/rust/README.md +++ b/examples/rust/README.md @@ -59,7 +59,7 @@ You can also customize the server using environment variables: ```bash ## Example: Enable HTTP transport and set custom address -IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server +IGGY_HTTP_ENABLED=true IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server ``` You can run multiple producers and consumers simultaneously to observe how messages are distributed across clients. Most examples support configurable options via the [Args](https://github.com/apache/iggy/blob/master/examples/rust/src/shared/args.rs) struct, including transport protocol, stream/topic/partition settings, consumer ID, message size, and more. diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs index 23001c9da5..e6cafc964d 100644 --- a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs +++ b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs @@ -291,6 +291,7 @@ private Dictionary BuildClusterEnvironment(IReadOnlyList if (!ClusterEnabled) { + environment["IGGY_NODE_ADVERTISED_ADDRESS"] = "127.0.0.1"; return environment; } diff --git a/foreign/java/external-processors/iggy-connector-flink/docker-compose.yml b/foreign/java/external-processors/iggy-connector-flink/docker-compose.yml index da7011c302..8b20600f8d 100644 --- a/foreign/java/external-processors/iggy-connector-flink/docker-compose.yml +++ b/foreign/java/external-processors/iggy-connector-flink/docker-compose.yml @@ -26,6 +26,7 @@ services: - "3000:3000" # HTTP API / Web UI environment: - IGGY_TCP_ADDRESS=0.0.0.0:8090 + - IGGY_NODE_ADVERTISED_ADDRESS=iggy - IGGY_HTTP_ADDRESS=0.0.0.0:3000 - IGGY_QUIC_ADDRESS=0.0.0.0:8080 - IGGY_SYSTEM_LOGGING_LEVEL=info diff --git a/foreign/java/external-processors/iggy-connector-pinot/docker-compose.yml b/foreign/java/external-processors/iggy-connector-pinot/docker-compose.yml index 8e50a9be38..49fed9a29f 100644 --- a/foreign/java/external-processors/iggy-connector-pinot/docker-compose.yml +++ b/foreign/java/external-processors/iggy-connector-pinot/docker-compose.yml @@ -27,6 +27,7 @@ services: environment: - IGGY_SYSTEM_LOGGING_LEVEL=info - IGGY_TCP_ADDRESS=0.0.0.0:8090 + - IGGY_NODE_ADVERTISED_ADDRESS=iggy - IGGY_HTTP_ENABLED=true - IGGY_HTTP_ADDRESS=0.0.0.0:3000 - IGGY_QUIC_ADDRESS=0.0.0.0:8080 diff --git a/foreign/php/README.md b/foreign/php/README.md index 73a31cfa42..d6bd9de37a 100644 --- a/foreign/php/README.md +++ b/foreign/php/README.md @@ -64,6 +64,7 @@ php -r 'var_dump(extension_loaded("iggy-php"));' docker run --rm --name iggy-php-test \ -p 8090:8090 \ -p 3000:3000 \ + -e IGGY_NODE_ADVERTISED_ADDRESS=localhost \ apache/iggy:latest ``` diff --git a/foreign/php/docker-compose.test.yml b/foreign/php/docker-compose.test.yml index 3831e39bde..19abcc2178 100644 --- a/foreign/php/docker-compose.test.yml +++ b/foreign/php/docker-compose.test.yml @@ -31,6 +31,7 @@ services: environment: - IGGY_HTTP_ADDRESS=0.0.0.0:3000 - IGGY_TCP_ADDRESS=0.0.0.0:8090 + - IGGY_NODE_ADVERTISED_ADDRESS=iggy-server - IGGY_QUIC_ADDRESS=0.0.0.0:8080 - IGGY_WEBSOCKET_ADDRESS=0.0.0.0:8092 - IGGY_ROOT_USERNAME=iggy diff --git a/foreign/python/docker-compose.test.yml b/foreign/python/docker-compose.test.yml index 78caed9929..665df9121c 100644 --- a/foreign/python/docker-compose.test.yml +++ b/foreign/python/docker-compose.test.yml @@ -36,6 +36,7 @@ services: environment: - IGGY_HTTP_ADDRESS=0.0.0.0:3000 - IGGY_TCP_ADDRESS=0.0.0.0:8090 + - IGGY_NODE_ADVERTISED_ADDRESS=iggy-server - IGGY_QUIC_ADDRESS=0.0.0.0:8080 - IGGY_WEBSOCKET_ADDRESS=0.0.0.0:8092 - IGGY_ROOT_USERNAME=iggy diff --git a/foreign/python/tests/test_tls.py b/foreign/python/tests/test_tls.py index f914e41466..1992f9aa34 100644 --- a/foreign/python/tests/test_tls.py +++ b/foreign/python/tests/test_tls.py @@ -66,6 +66,7 @@ def tls_container(): .with_env("IGGY_TCP_TLS_CERT_FILE", "/app/certs/iggy_cert.pem") .with_env("IGGY_TCP_TLS_KEY_FILE", "/app/certs/iggy_key.pem") .with_env("IGGY_TCP_ADDRESS", f"0.0.0.0:{CONTAINER_TCP_PORT}") + .with_env("IGGY_NODE_ADVERTISED_ADDRESS", "127.0.0.1") .with_volume_mapping(CERTS_DIR, "/app/certs", "ro") .with_kwargs(privileged=True) ) diff --git a/helm/charts/iggy/README.md b/helm/charts/iggy/README.md index 4e4cd42116..101cf47bf8 100644 --- a/helm/charts/iggy/README.md +++ b/helm/charts/iggy/README.md @@ -210,6 +210,13 @@ Ensure the server binds to `0.0.0.0` instead of `127.0.0.1`. This is configured * `IGGY_TCP_ADDRESS=0.0.0.0:8090` * `IGGY_QUIC_ADDRESS=0.0.0.0:8080` +A wildcard bind says which interfaces accept connections, not where clients +reach the pod, so the server also needs the address to publish in cluster +metadata and refuses to start without it. The chart sets +`IGGY_NODE_ADVERTISED_ADDRESS` to the in-cluster Service DNS name; override it +with `server.advertisedAddress` when clients arrive through a LoadBalancer or +an Ingress. + ## Accessing the Server ### Port Forward @@ -312,7 +319,8 @@ pre-commit install | podSecurityContext | object | `{"seccompProfile":{"type":"Unconfined"}}` | Pod security context (server uses io_uring, requires unconfined seccomp) | | resources | object | `{}` | Resource limits and requests for server | | securityContext | object | `{"capabilities":{"add":["IPC_LOCK"]}}` | Container security context (server requires IPC_LOCK for io_uring) | -| server | object | `{"affinity":{},"enabled":true,"env":[{"name":"RUST_LOG","value":"info"},{"name":"IGGY_HTTP_ADDRESS","value":"0.0.0.0:3000"},{"name":"IGGY_TCP_ADDRESS","value":"0.0.0.0:8090"},{"name":"IGGY_QUIC_ADDRESS","value":"0.0.0.0:8080"},{"name":"IGGY_WEBSOCKET_ADDRESS","value":"0.0.0.0:8092"}],"image":{"pullPolicy":"Always","repository":"apache/iggy","tag":"0.7.0"},"ingress":{"annotations":{},"className":"","enabled":false,"hosts":[{"host":"chart-example.local","paths":[{"path":"/","pathType":"ImplementationSpecific"}]}],"tls":[]},"nodeSelector":{},"persistence":{"accessMode":"ReadWriteOnce","annotations":{},"enabled":false,"existingClaim":"","size":"8Gi","storageClass":""},"ports":{"http":3000,"quic":8080,"tcp":8090},"replicaCount":1,"service":{"port":3000,"type":"ClusterIP"},"serviceMonitor":{"additionalLabels":{},"authorization":{},"enabled":false,"honorLabels":false,"interval":"30s","namespace":"","path":"/metrics","scrapeTimeout":"10s"},"tolerations":[],"users":{"root":{"createSecret":true,"existingSecret":{"name":"","passwordKey":"password","usernameKey":"username"},"password":"changeit","username":"iggy"}}}` | Iggy server configuration | +| server | object | `{"advertisedAddress":"","affinity":{},"enabled":true,"env":[{"name":"RUST_LOG","value":"info"},{"name":"IGGY_HTTP_ADDRESS","value":"0.0.0.0:3000"},{"name":"IGGY_TCP_ADDRESS","value":"0.0.0.0:8090"},{"name":"IGGY_QUIC_ADDRESS","value":"0.0.0.0:8080"},{"name":"IGGY_WEBSOCKET_ADDRESS","value":"0.0.0.0:8092"}],"image":{"pullPolicy":"Always","repository":"apache/iggy","tag":"0.7.0"},"ingress":{"annotations":{},"className":"","enabled":false,"hosts":[{"host":"chart-example.local","paths":[{"path":"/","pathType":"ImplementationSpecific"}]}],"tls":[]},"nodeSelector":{},"persistence":{"accessMode":"ReadWriteOnce","annotations":{},"enabled":false,"existingClaim":"","size":"8Gi","storageClass":""},"ports":{"http":3000,"quic":8080,"tcp":8090},"replicaCount":1,"service":{"port":3000,"type":"ClusterIP"},"serviceMonitor":{"additionalLabels":{},"authorization":{},"enabled":false,"honorLabels":false,"interval":"30s","namespace":"","path":"/metrics","scrapeTimeout":"10s"},"tolerations":[],"users":{"root":{"createSecret":true,"existingSecret":{"name":"","passwordKey":"password","usernameKey":"username"},"password":"changeit","username":"iggy"}}}` | Iggy server configuration | +| server.advertisedAddress | string | `""` | Client-facing address published in cluster metadata. | | server.affinity | object | `{}` | Affinity rules for server pods | | server.enabled | bool | `true` | Enable the Iggy server deployment | | server.env | list | `[{"name":"RUST_LOG","value":"info"},{"name":"IGGY_HTTP_ADDRESS","value":"0.0.0.0:3000"},{"name":"IGGY_TCP_ADDRESS","value":"0.0.0.0:8090"},{"name":"IGGY_QUIC_ADDRESS","value":"0.0.0.0:8080"},{"name":"IGGY_WEBSOCKET_ADDRESS","value":"0.0.0.0:8092"}]` | Environment variables for the server container | diff --git a/helm/charts/iggy/README.md.gotmpl b/helm/charts/iggy/README.md.gotmpl index a30365fe6c..f2f3029461 100644 --- a/helm/charts/iggy/README.md.gotmpl +++ b/helm/charts/iggy/README.md.gotmpl @@ -228,6 +228,13 @@ Ensure the server binds to `0.0.0.0` instead of `127.0.0.1`. This is configured * `IGGY_TCP_ADDRESS=0.0.0.0:8090` * `IGGY_QUIC_ADDRESS=0.0.0.0:8080` +A wildcard bind says which interfaces accept connections, not where clients +reach the pod, so the server also needs the address to publish in cluster +metadata and refuses to start without it. The chart sets +`IGGY_NODE_ADVERTISED_ADDRESS` to the in-cluster Service DNS name; override it +with `server.advertisedAddress` when clients arrive through a LoadBalancer or +an Ingress. + ## Accessing the Server ### Port Forward diff --git a/helm/charts/iggy/templates/deployment.yaml b/helm/charts/iggy/templates/deployment.yaml index 4f17b2f33d..b8635ecd78 100644 --- a/helm/charts/iggy/templates/deployment.yaml +++ b/helm/charts/iggy/templates/deployment.yaml @@ -89,6 +89,14 @@ spec: name: {{ include "iggy.fullname" . }}-root-credentials key: password {{- end }}{{- end}} + {{- $declaredInEnv := false }} + {{- range .Values.server.env }} + {{- if eq .name "IGGY_NODE_ADVERTISED_ADDRESS" }}{{- $declaredInEnv = true }}{{- end }} + {{- end }} + {{- if not $declaredInEnv }} + - name: IGGY_NODE_ADVERTISED_ADDRESS + value: {{ .Values.server.advertisedAddress | default (printf "%s.%s.svc.cluster.local" (include "iggy.fullname" .) .Release.Namespace) | quote }} + {{- end }} {{- if .Values.server.env }} {{- range .Values.server.env }} - name: {{ .name }} diff --git a/helm/charts/iggy/values.yaml b/helm/charts/iggy/values.yaml index 7e760ea547..88b91c10df 100644 --- a/helm/charts/iggy/values.yaml +++ b/helm/charts/iggy/values.yaml @@ -17,6 +17,8 @@ # -- Iggy server configuration server: + # -- Client-facing address published in cluster metadata. + advertisedAddress: "" # -- Enable the Iggy server deployment enabled: true # -- Number of server replicas diff --git a/web/README.md b/web/README.md index f2ef7cdbc3..e3dcc4224c 100644 --- a/web/README.md +++ b/web/README.md @@ -25,7 +25,8 @@ The [docker image](https://hub.docker.com/r/apache/iggy-web-ui) is available, an ``` ```sh - docker run -p 3000:3000 -p 8090:8090 apache/iggy:latest + docker run -p 3000:3000 -p 8090:8090 \ + -e IGGY_NODE_ADVERTISED_ADDRESS=localhost apache/iggy:latest ``` 2. **Clone the repository:** diff --git a/web/docker-compose.yml b/web/docker-compose.yml index dca555202b..22bc6b62c3 100644 --- a/web/docker-compose.yml +++ b/web/docker-compose.yml @@ -24,6 +24,7 @@ services: IGGY_HTTP_ADDRESS: 0.0.0.0:3000 IGGY_QUIC_ADDRESS: 0.0.0.0:8080 IGGY_TCP_ADDRESS: 0.0.0.0:8090 + IGGY_NODE_ADVERTISED_ADDRESS: iggy-server IGGY_WEBSOCKET_ADDRESS: 0.0.0.0:8092 cap_add: - SYS_NICE