diff --git a/fluss-gateway/src/config.rs b/fluss-gateway/src/config.rs index e1d66ff415..ecd2ff2871 100644 --- a/fluss-gateway/src/config.rs +++ b/fluss-gateway/src/config.rs @@ -22,10 +22,26 @@ //! ```yaml //! gateway.rest.listen: 0.0.0.0:8080 //! gateway.rest.write.max-request-bytes: 32MiB +//! gateway.clusters: default +//! gateway.cluster.default.bootstrap.servers: 127.0.0.1:9123 +//! gateway.cluster.default.connection.service.account: gateway_svc +//! gateway.cluster.default.client.writer.batch-size: 2MiB //! ``` //! //! Environment variable names are derived from these public keys; for example, -//! `gateway.rest.listen` becomes `FLUSS_GATEWAY__REST__LISTEN`. +//! `gateway.rest.listen` becomes `FLUSS_GATEWAY__REST__LISTEN`, and +//! `gateway.cluster.default.client.writer.batch-size` becomes +//! `FLUSS_GATEWAY__CLUSTER__DEFAULT__CLIENT__WRITER__BATCH_SIZE`. +//! +//! Two rules make the configuration deterministic before anything binds or connects. `gateway.clusters` +//! is authoritative, and it is so whether or not it was written: an absent list means the single implicit +//! `default` cluster, so a mistyped cluster ID fails startup instead of creating a cluster nothing routes +//! to. And `gateway.cluster..client.*` is an allowlist, not a passthrough: the native-client options +//! that would weaken the write guarantees the gateway advertises, or pin the authorization identity that +//! user identity mode supplies per request, are rejected rather than honoured. +//! +//! Credentials are redacted in diagnostics: typed fields carry them as [`Secret`], and the open +//! `client.*` namespace declares per option which values are sensitive. use serde::Deserialize; use serde::de::{self, Deserializer}; @@ -36,9 +52,44 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::path::Path; use std::time::Duration; +#[cfg(test)] +mod tests; + /// Environment variable prefix for overrides. pub const ENV_PREFIX: &str = "FLUSS_GATEWAY__"; +/// Stand-in printed wherever a credential would otherwise be rendered. +/// +/// Spelled the same as `Password.HIDDEN_CONTENT` on the Java side, so one deployment's server and gateway +/// logs redact identically. +const REDACTED: &str = "******"; + +/// A configuration value that must never reach a log line, an error, or a debug dump. +/// +/// The wrapper is the whole mechanism: [`Debug`] is the only way diagnostics render configuration, so +/// redacting it here covers every present and future diagnostic without each call site remembering to. +#[derive(Clone, PartialEq, Eq, Deserialize)] +#[serde(transparent)] +pub struct Secret(String); + +impl Secret { + /// Wraps a credential that reached the gateway as plain text, so it redacts from here on. + pub fn new(value: impl Into) -> Self { + Self(value.into()) + } + + /// Hands the credential to the component that authenticates with it. + pub fn expose(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for Secret { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(REDACTED) + } +} + /// A duration written as ``. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct ConfigDuration(Duration); @@ -200,6 +251,13 @@ const REST_MAX_REQUEST_BYTES_KEY: &str = "gateway.rest.write.max-request-bytes"; const METRICS_ENABLED_KEY: &str = "gateway.metrics.enabled"; const METRICS_LISTEN_KEY: &str = "gateway.metrics.exporter.prometheus.listen"; const SHUTDOWN_DRAIN_TIMEOUT_KEY: &str = "gateway.shutdown.drain-timeout"; +const CLUSTERS_KEY: &str = "gateway.clusters"; +const CLUSTER_KEY_PREFIX: &str = "gateway.cluster."; +const CLIENT_OPTION_PREFIX: &str = "client."; +const SECURITY_AUTHENTICATION_KEY: &str = "gateway.security.authentication"; +const SECURITY_USERS_KEY: &str = "gateway.security.users"; +const SECURITY_TOKENS_KEY: &str = "gateway.security.tokens"; +const SECURITY_TRUSTED_HEADER_NAME_KEY: &str = "gateway.security.trusted-header.name"; const DEFAULT_REST_LISTEN: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080); const DEFAULT_REST_HEADER_READ_TIMEOUT: ConfigDuration = ConfigDuration::from_secs(10); @@ -209,44 +267,180 @@ const DEFAULT_METRICS_ENABLED: bool = true; const DEFAULT_METRICS_LISTEN: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9095); const DEFAULT_SHUTDOWN_DRAIN_TIMEOUT: ConfigDuration = ConfigDuration::from_secs(30); +/// The default cluster ID, so a single-cluster deployment configures no cluster list at all. +const DEFAULT_CLUSTER_ID: &str = "default"; +const DEFAULT_BOOTSTRAP_SERVERS: &str = "127.0.0.1:9123"; +const DEFAULT_TRUSTED_HEADER_NAME: &str = "x-forwarded-user"; + +/// How a source value is converted before it reaches Serde. +/// +/// Most options are strings whose typed parsing lives in a `Deserialize` impl, but a bool or an integer +/// has to arrive as the matching YAML scalar, and a server list accepts either a sequence or one +/// comma-separated string. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ValueKind { + /// Parsed downstream by the field's own `Deserialize` impl (address, duration, enum, secret, text). + Text, + Bool, + /// A non-negative integer, which Serde cannot read from a string. + Integer, + /// An integer count of bytes or a byte-size string such as `32MiB`. + Bytes, + /// A YAML sequence or one comma-separated string. + ServerList, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] struct ConfigEntry { key: &'static str, internal_path: &'static str, + kind: ValueKind, } const CONFIG_ENTRIES: &[ConfigEntry] = &[ ConfigEntry { key: INSTANCE_ID_KEY, internal_path: "server.instance_id", + kind: ValueKind::Text, }, ConfigEntry { key: REST_LISTEN_KEY, internal_path: "server.rest.bind_address", + kind: ValueKind::Text, }, ConfigEntry { key: REST_HEADER_READ_TIMEOUT_KEY, internal_path: "server.rest.header_read_timeout", + kind: ValueKind::Text, }, ConfigEntry { key: REST_REQUEST_TIMEOUT_KEY, internal_path: "server.rest.request_timeout", + kind: ValueKind::Text, }, ConfigEntry { key: REST_MAX_REQUEST_BYTES_KEY, internal_path: "server.rest.max_body_bytes", + kind: ValueKind::Bytes, + }, + ConfigEntry { + key: "gateway.rest.write.max-rows", + internal_path: "request_limits.write_max_rows", + kind: ValueKind::Integer, + }, + ConfigEntry { + key: "gateway.rest.write.max-concurrent-requests", + internal_path: "request_limits.write_max_concurrent_requests", + kind: ValueKind::Integer, + }, + ConfigEntry { + key: "gateway.rest.lookup.max-keys", + internal_path: "request_limits.lookup_max_keys", + kind: ValueKind::Integer, + }, + ConfigEntry { + key: "gateway.rest.lookup.max-key-bytes", + internal_path: "request_limits.lookup_max_key_bytes", + kind: ValueKind::Bytes, + }, + ConfigEntry { + key: "gateway.rest.lookup.max-concurrent-requests", + internal_path: "request_limits.lookup_max_concurrent_requests", + kind: ValueKind::Integer, + }, + ConfigEntry { + key: "gateway.rest.prefix-lookup.max-prefixes", + internal_path: "request_limits.prefix_lookup_max_prefixes", + kind: ValueKind::Integer, + }, + ConfigEntry { + key: "gateway.rest.prefix-lookup.max-rows-per-prefix", + internal_path: "request_limits.prefix_lookup_max_rows_per_prefix", + kind: ValueKind::Integer, + }, + ConfigEntry { + key: "gateway.rest.prefix-lookup.max-concurrent-requests", + internal_path: "request_limits.prefix_lookup_max_concurrent_requests", + kind: ValueKind::Integer, }, ConfigEntry { key: METRICS_ENABLED_KEY, internal_path: "server.metrics.enabled", + kind: ValueKind::Bool, }, ConfigEntry { key: METRICS_LISTEN_KEY, internal_path: "server.metrics.bind_address", + kind: ValueKind::Text, }, ConfigEntry { key: SHUTDOWN_DRAIN_TIMEOUT_KEY, internal_path: "shutdown.drain_timeout", + kind: ValueKind::Text, + }, + ConfigEntry { + key: SECURITY_AUTHENTICATION_KEY, + internal_path: "security.authentication", + kind: ValueKind::Text, + }, + ConfigEntry { + key: SECURITY_USERS_KEY, + internal_path: "security.users", + kind: ValueKind::Text, + }, + ConfigEntry { + key: SECURITY_TOKENS_KEY, + internal_path: "security.tokens", + kind: ValueKind::Text, + }, + ConfigEntry { + key: SECURITY_TRUSTED_HEADER_NAME_KEY, + internal_path: "security.trusted_header_name", + kind: ValueKind::Text, + }, +]; + +/// The per-cluster options under `gateway.cluster..`, excluding the open `client.*` namespace. +const CLUSTER_ENTRIES: &[ConfigEntry] = &[ + ConfigEntry { + key: "bootstrap.servers", + internal_path: "bootstrap_servers", + kind: ValueKind::ServerList, + }, + ConfigEntry { + key: "connect-timeout", + internal_path: "connect_timeout", + kind: ValueKind::Text, + }, + ConfigEntry { + key: "request-timeout", + internal_path: "request_timeout", + kind: ValueKind::Text, + }, + ConfigEntry { + key: "connection.service.account", + internal_path: "service_account", + kind: ValueKind::Text, + }, + ConfigEntry { + key: "connection.service.secret", + internal_path: "service_secret", + kind: ValueKind::Text, + }, + ConfigEntry { + key: "connection.identity-mode", + internal_path: "identity_mode", + kind: ValueKind::Text, + }, + ConfigEntry { + key: "connection.max", + internal_path: "connection_max", + kind: ValueKind::Integer, + }, + ConfigEntry { + key: "connection.idle-timeout", + internal_path: "connection_idle_timeout", + kind: ValueKind::Text, }, ]; @@ -327,6 +521,520 @@ impl Default for MetricsServerConfig { } } +/// Deserializes a bootstrap list from either a YAML sequence or one comma-separated string. +fn deserialize_server_list<'de, D: Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + #[derive(Deserialize)] + #[serde(untagged)] + enum ListOrCsv { + List(Vec), + Csv(String), + } + + Ok(match ListOrCsv::deserialize(deserializer)? { + ListOrCsv::List(list) => list, + ListOrCsv::Csv(csv) => csv + .split(',') + .map(|entry| entry.trim().to_string()) + .collect(), + }) +} + +/// How the gateway derives the effective Fluss principal for one cluster. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)] +#[serde(rename_all = "lowercase")] +pub enum IdentityMode { + /// One shared connection authenticated as the configured service account. + #[default] + Service, + /// Authenticate as the service account and carry the request's principal as the authorization ID. + User, +} + +/// Connection and native-client settings for one Fluss cluster. +/// +/// The connections themselves arrive with the native backend; this task owns the typed, validated, and +/// redacted contract they will read. +#[derive(Clone, PartialEq, Deserialize)] +#[serde(deny_unknown_fields, default)] +pub struct ClusterConfig { + #[serde(deserialize_with = "deserialize_server_list")] + pub bootstrap_servers: Vec, + pub connect_timeout: ConfigDuration, + pub request_timeout: ConfigDuration, + /// Canonical account the gateway authenticates to Fluss with. An identity, not a credential, so it is + /// rendered in diagnostics: Fluss treats secrets and tokens as sensitive, not principal names. + pub service_account: Option, + /// Canonical credential for [`Self::service_account`]. + pub service_secret: Option, + pub identity_mode: IdentityMode, + /// Cap on pooled per-user connections. User identity mode only. + pub connection_max: Option, + /// Idle reclamation for pooled per-user connections. User identity mode only. + pub connection_idle_timeout: Option, + /// Validated `gateway.cluster..client.*` options, keyed without the `client.` prefix. + pub client_options: BTreeMap, +} + +impl Default for ClusterConfig { + fn default() -> Self { + Self { + bootstrap_servers: vec![DEFAULT_BOOTSTRAP_SERVERS.to_string()], + connect_timeout: ConfigDuration::from_secs(10), + request_timeout: ConfigDuration::from_secs(30), + service_account: None, + service_secret: None, + identity_mode: IdentityMode::Service, + connection_max: None, + connection_idle_timeout: None, + client_options: BTreeMap::new(), + } + } +} + +impl ClusterConfig { + /// Returns the account actually used, honouring the legacy last-wins override. + pub fn effective_service_account(&self) -> Option<&str> { + self.client_option(LEGACY_SERVICE_ACCOUNT_OPTION) + .or(self.service_account.as_deref()) + } + + /// Returns the credential actually used, honouring the legacy last-wins override. + pub fn effective_service_secret(&self) -> Option<&str> { + self.client_option(LEGACY_SERVICE_SECRET_OPTION) + .or_else(|| self.service_secret.as_ref().map(Secret::expose)) + } + + /// Reads one configured native-client option without its `client.` prefix. + pub fn client_option(&self, option: &str) -> Option<&str> { + self.client_options.get(option).map(String::as_str) + } +} + +impl fmt::Debug for ClusterConfig { + /// Renders the client options with each sensitive value replaced, rather than hiding all of them: the + /// configured tuning is what an operator needs to see, and only the credentials have to disappear. + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let client_options: BTreeMap<&str, &str> = self + .client_options + .iter() + .map(|(option, value)| { + let rendered = if client_option_is_sensitive(option) { + REDACTED + } else { + value.as_str() + }; + (option.as_str(), rendered) + }) + .collect(); + f.debug_struct("ClusterConfig") + .field("bootstrap_servers", &self.bootstrap_servers) + .field("connect_timeout", &self.connect_timeout) + .field("request_timeout", &self.request_timeout) + .field("service_account", &self.service_account) + .field("service_secret", &self.service_secret) + .field("identity_mode", &self.identity_mode) + .field("connection_max", &self.connection_max) + .field("connection_idle_timeout", &self.connection_idle_timeout) + .field("client_options", &client_options) + .finish() + } +} + +/// A validated native-client option. The variant records the conversion the native client needs. +/// +/// A credential stays wrapped in [`Secret`] rather than becoming a `Text`, so the parsed value redacts +/// exactly like the configured one did and no caller can print it back out by accident. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ClientOptionValue { + Text(String), + Secret(Secret), + Boolean(bool), + Integer(u64), + Millis(u64), + Bytes(u64), +} + +/// Superseded by `connection.service.account`; still honoured, with a warning. +const LEGACY_SERVICE_ACCOUNT_OPTION: &str = "security.sasl.username"; +/// Superseded by `connection.service.secret`; still honoured, with a warning. +const LEGACY_SERVICE_SECRET_OPTION: &str = "security.sasl.password"; + +/// How one `client.*` value is parsed and bounded. +/// +/// A bound is a fact about the native field the value lands in, so it is declared per option rather than +/// applied uniformly: the writer sizes are `i32` there, while the buffer size and the lookup counts are +/// `usize` and may legitimately exceed `i32::MAX` on a 64-bit target. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ClientOptionKind { + /// Passed through verbatim; any further constraint is a cross-field rule. + Text, + /// Verbatim like [`Self::Text`], but a credential: redacted everywhere it is rendered. + Secret, + Boolean, + /// An integer count the native field accepts within `min..=max`. + Count { + min: u64, + max: u64, + }, + /// A byte size, with the ceiling and the default of the native field. + Size { + max: u64, + default: u64, + }, + /// A duration, handed to the native client as milliseconds. + Duration, +} + +/// Ceilings of the native field types the values are stored in. +const NATIVE_I32_MAX: u64 = i32::MAX as u64; +const NATIVE_USIZE_MAX: u64 = usize::MAX as u64; + +const KIB: u64 = 1024; +const MIB: u64 = 1024 * KIB; + +impl ClientOptionKind { + /// Sensitivity follows from the kind, so no option can be declared a credential and rendered as text + /// at the same time. The Java side has to infer this from the key name instead + /// (`ConfigurationUtils.SENSITIVE_KEY_PARTS`, plus an allowlist for the keys that over-match). + fn is_sensitive(self) -> bool { + matches!(self, Self::Secret) + } +} + +/// One native-client option the gateway accepts under `gateway.cluster..client.`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ClientOptionSpec { + option: &'static str, + kind: ClientOptionKind, +} + +/// The `client.*` allowlist. Anything absent is rejected, so a native-client option the gateway has not +/// considered cannot reach a connection. +const CLIENT_OPTIONS: &[ClientOptionSpec] = &[ + ClientOptionSpec { + option: "security.protocol", + kind: ClientOptionKind::Text, + }, + ClientOptionSpec { + option: "security.sasl.mechanism", + kind: ClientOptionKind::Text, + }, + ClientOptionSpec { + option: LEGACY_SERVICE_ACCOUNT_OPTION, + kind: ClientOptionKind::Text, + }, + ClientOptionSpec { + option: LEGACY_SERVICE_SECRET_OPTION, + kind: ClientOptionKind::Secret, + }, + ClientOptionSpec { + option: "connect.timeout", + kind: ClientOptionKind::Duration, + }, + ClientOptionSpec { + option: "request.timeout", + kind: ClientOptionKind::Duration, + }, + ClientOptionSpec { + option: "writer.batch-size", + kind: ClientOptionKind::Size { + max: NATIVE_I32_MAX, + default: 2 * MIB, + }, + }, + ClientOptionSpec { + option: "writer.request-max-size", + kind: ClientOptionKind::Size { + max: NATIVE_I32_MAX, + default: 10 * MIB, + }, + }, + ClientOptionSpec { + option: "writer.buffer.memory-size", + kind: ClientOptionKind::Size { + max: NATIVE_USIZE_MAX, + default: 64 * MIB, + }, + }, + ClientOptionSpec { + option: "writer.buffer.wait-timeout", + kind: ClientOptionKind::Duration, + }, + ClientOptionSpec { + option: "writer.batch-timeout", + kind: ClientOptionKind::Duration, + }, + ClientOptionSpec { + option: "writer.dynamic-batch-size.enabled", + kind: ClientOptionKind::Boolean, + }, + ClientOptionSpec { + option: "writer.dynamic-batch-size.min", + kind: ClientOptionKind::Size { + max: NATIVE_I32_MAX, + default: 256 * KIB, + }, + }, + ClientOptionSpec { + option: "writer.kv-backpressure.max-throttle", + kind: ClientOptionKind::Duration, + }, + ClientOptionSpec { + option: "lookup.queue-size", + kind: ClientOptionKind::Count { + min: 1, + max: NATIVE_USIZE_MAX, + }, + }, + ClientOptionSpec { + option: "lookup.max-batch-size", + kind: ClientOptionKind::Count { + min: 1, + max: NATIVE_USIZE_MAX, + }, + }, + ClientOptionSpec { + option: "lookup.max-inflight-requests", + kind: ClientOptionKind::Count { + min: 1, + max: NATIVE_USIZE_MAX, + }, + }, + ClientOptionSpec { + option: "lookup.max-retries", + kind: ClientOptionKind::Count { + min: 0, + max: NATIVE_I32_MAX, + }, + }, + ClientOptionSpec { + option: "lookup.batch-timeout", + kind: ClientOptionKind::Duration, + }, +]; + +/// The `client.*` options the gateway refuses outright, each with the reason an operator sees. +/// +/// The first three would silently weaken the write guarantees the gateway advertises. The fourth is never +/// static: user identity mode sets the authorization ID per connection from the authenticated principal. +const RESERVED_CLIENT_OPTIONS: &[(&str, &str)] = &[ + ( + "writer.acks", + "is owned by the Gateway and cannot be overridden", + ), + ( + "writer.retries", + "is owned by the Gateway and cannot be overridden", + ), + ( + "writer.enable-idempotence", + "is owned by the Gateway and cannot be overridden", + ), + ( + "security.sasl.authorization-id", + "is supplied per connection in user identity mode and cannot be configured statically", + ), +]; + +/// True when the option's value must not appear in diagnostics. Unknown options are assumed sensitive, so +/// a value that failed validation cannot leak through a debug dump. +fn client_option_is_sensitive(option: &str) -> bool { + client_option_spec(option).is_none_or(|spec| spec.kind.is_sensitive()) +} + +/// The size the native client will use for `option`: the configured value, or the native default when the +/// deployment left it alone. +/// +/// Returns `None` for an option that is not a size, or for a configured value that failed validation and +/// is already reported on its own account. +/// +/// TODO: source the default from the native `Config::default()` and delegate the relationships to +/// `Config::validate_writer` once the gateway takes fluss-rust as a dependency; the declared defaults +/// exist only because it cannot be called from here yet. +fn effective_size(option: &str, configured: Option<&str>) -> Option { + let ClientOptionKind::Size { default, .. } = client_option_spec(option)?.kind else { + return None; + }; + match configured { + Some(raw) => match parse_client_option(option, raw) { + Ok(ClientOptionValue::Bytes(bytes)) => Some(bytes), + _ => None, + }, + None => Some(default), + } +} + +fn client_option_spec(option: &str) -> Option<&'static ClientOptionSpec> { + CLIENT_OPTIONS.iter().find(|spec| spec.option == option) +} + +/// Parses and bounds one `client.*` option before any connection exists, so a bad native-client value +/// fails startup instead of the first write. +/// +/// Relationships *between* options are not checked here, because a value is validated on its own: see +/// `GatewayConfig::validate_client_size_pairs`. +pub fn parse_client_option(option: &str, raw: &str) -> Result { + if let Some((_, reason)) = RESERVED_CLIENT_OPTIONS + .iter() + .find(|(reserved, _)| *reserved == option) + { + return Err(format!("client.{option} {reason}")); + } + let Some(spec) = client_option_spec(option) else { + return Err(format!( + "client.{option} is not a supported native-client option" + )); + }; + // No message echoes the value: this namespace also carries the legacy SASL credentials. + let malformed = + |expected: &str, reason: String| format!("client.{option}: expected {expected}: {reason}"); + match spec.kind { + ClientOptionKind::Text => Ok(ClientOptionValue::Text(raw.to_string())), + ClientOptionKind::Secret => Ok(ClientOptionValue::Secret(Secret::new(raw))), + ClientOptionKind::Boolean => raw.parse().map(ClientOptionValue::Boolean).map_err( + |error: std::str::ParseBoolError| malformed("true or false", error.to_string()), + ), + ClientOptionKind::Count { min, max } => { + let parsed = raw + .parse::() + .map_err(|error| malformed("a non-negative integer", error.to_string()))?; + if !(min..=max).contains(&parsed) { + return Err(format!("client.{option}: must be between {min} and {max}")); + } + Ok(ClientOptionValue::Integer(parsed)) + } + ClientOptionKind::Size { max, .. } => { + let size = ByteSize::parse(raw).map_err(|error| malformed("a byte size", error))?; + if size.bytes() > max { + return Err(format!("client.{option}: must not exceed {max} bytes")); + } + Ok(ClientOptionValue::Bytes(size.bytes())) + } + ClientOptionKind::Duration => { + let duration = + ConfigDuration::parse(raw).map_err(|error| malformed("a duration", error))?; + u64::try_from(duration.get().as_millis()) + .map(ClientOptionValue::Millis) + .map_err(|_| format!("client.{option}: does not fit in milliseconds")) + } + } +} + +/// How the gateway authenticates its own HTTP callers. +/// +/// The authenticators arrive with the REST authentication task; this task owns the typed contract. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)] +#[serde(rename_all = "kebab-case")] +pub enum AuthenticationMode { + /// Every caller is accepted and reported as an anonymous principal. + #[default] + Trust, + Password, + Token, + TrustedHeader, +} + +/// Client-to-gateway authentication settings. Every credential-bearing field is a [`Secret`]. +#[derive(Clone, PartialEq, Deserialize, Default)] +#[serde(deny_unknown_fields, default)] +pub struct SecurityConfig { + pub authentication: AuthenticationMode, + /// Password-mode user table; the entries embed password material. + pub users: Option, + /// Token-mode table; the entries are bearer tokens. + pub tokens: Option, + /// Trusted-header-mode header name, defaulting to `x-forwarded-user`. + pub trusted_header_name: Option, +} + +impl SecurityConfig { + /// Returns the header the trusted-header mode reads the principal from. + pub fn trusted_header_name(&self) -> &str { + self.trusted_header_name + .as_deref() + .unwrap_or(DEFAULT_TRUSTED_HEADER_NAME) + } +} + +impl fmt::Debug for SecurityConfig { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SecurityConfig") + .field("authentication", &self.authentication) + .field("users", &self.users) + .field("tokens", &self.tokens) + .field("trusted_header_name", &self.trusted_header_name) + .finish() + } +} + +/// Admission limits for the data-plane APIs, whose handlers arrive in later tasks. +#[derive(Debug, Clone, PartialEq, Deserialize)] +#[serde(deny_unknown_fields, default)] +pub struct RequestLimitsConfig { + pub write_max_rows: u32, + pub write_max_concurrent_requests: u32, + pub lookup_max_keys: u32, + pub lookup_max_key_bytes: ByteSize, + pub lookup_max_concurrent_requests: u32, + pub prefix_lookup_max_prefixes: u32, + pub prefix_lookup_max_rows_per_prefix: u32, + pub prefix_lookup_max_concurrent_requests: u32, +} + +impl Default for RequestLimitsConfig { + fn default() -> Self { + Self { + write_max_rows: 10_000, + write_max_concurrent_requests: 64, + lookup_max_keys: 128, + lookup_max_key_bytes: ByteSize::new(1024 * 1024), + lookup_max_concurrent_requests: 64, + prefix_lookup_max_prefixes: 16, + prefix_lookup_max_rows_per_prefix: 1000, + prefix_lookup_max_concurrent_requests: 32, + } + } +} + +impl RequestLimitsConfig { + fn validate(&self, problems: &mut Vec) { + for (key, value) in [ + ("gateway.rest.write.max-rows", self.write_max_rows), + ( + "gateway.rest.write.max-concurrent-requests", + self.write_max_concurrent_requests, + ), + ("gateway.rest.lookup.max-keys", self.lookup_max_keys), + ( + "gateway.rest.lookup.max-concurrent-requests", + self.lookup_max_concurrent_requests, + ), + ( + "gateway.rest.prefix-lookup.max-prefixes", + self.prefix_lookup_max_prefixes, + ), + ( + "gateway.rest.prefix-lookup.max-rows-per-prefix", + self.prefix_lookup_max_rows_per_prefix, + ), + ( + "gateway.rest.prefix-lookup.max-concurrent-requests", + self.prefix_lookup_max_concurrent_requests, + ), + ] { + if value == 0 { + problems.push(format!("{key} must be greater than zero")); + } + } + if self.lookup_max_key_bytes.bytes() == 0 { + problems + .push("gateway.rest.lookup.max-key-bytes must be greater than zero".to_string()); + } + } +} + /// Graceful-shutdown configuration. #[derive(Debug, Clone, PartialEq, Deserialize)] #[serde(deny_unknown_fields, default)] @@ -355,13 +1063,30 @@ impl ShutdownConfig { } /// The validated gateway configuration: everything the process needs before it binds a listener. -#[derive(Debug, Clone, PartialEq, Deserialize, Default)] +#[derive(Debug, Clone, PartialEq, Deserialize)] #[serde(deny_unknown_fields, default)] pub struct GatewayConfig { pub server: ServerConfig, + /// Every declared Fluss cluster, keyed by the ID used in REST paths. + pub clusters: BTreeMap, + pub security: SecurityConfig, + pub request_limits: RequestLimitsConfig, pub shutdown: ShutdownConfig, } +impl Default for GatewayConfig { + /// Defaults to the single `default` cluster, so a local deployment needs no cluster list. + fn default() -> Self { + Self { + server: ServerConfig::default(), + clusters: BTreeMap::from([(DEFAULT_CLUSTER_ID.to_string(), ClusterConfig::default())]), + security: SecurityConfig::default(), + request_limits: RequestLimitsConfig::default(), + shutdown: ShutdownConfig::default(), + } + } +} + impl GatewayConfig { /// Checks invariants, including values supplied programmatically. pub fn validate(&self) -> Result<(), ConfigError> { @@ -369,6 +1094,9 @@ impl GatewayConfig { self.server.rest.validate(&mut problems); self.shutdown.validate(&mut problems); self.validate_identity(&mut problems); + self.validate_clusters(&mut problems); + self.validate_security(&mut problems); + self.request_limits.validate(&mut problems); if problems.is_empty() { Ok(()) } else { @@ -376,6 +1104,187 @@ impl GatewayConfig { } } + /// Rejects any cluster the gateway could not connect with, before a listener is bound. + fn validate_clusters(&self, problems: &mut Vec) { + if self.clusters.is_empty() { + problems.push(format!("{CLUSTERS_KEY} must declare at least one cluster")); + } + for (id, cluster) in &self.clusters { + if !valid_cluster_id(id) { + problems.push(format!( + "cluster ID {id:?} must start with a lowercase letter and contain only \ + lowercase letters, digits, or underscores" + )); + } + if cluster.bootstrap_servers.is_empty() + || cluster + .bootstrap_servers + .iter() + .any(|server| server.trim().is_empty()) + { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.bootstrap.servers must list at least one non-blank server" + )); + } + validate_duration( + &format!("{CLUSTER_KEY_PREFIX}{id}.connect-timeout"), + cluster.connect_timeout.get(), + problems, + ); + validate_duration( + &format!("{CLUSTER_KEY_PREFIX}{id}.request-timeout"), + cluster.request_timeout.get(), + problems, + ); + if let Some(idle_timeout) = cluster.connection_idle_timeout { + validate_duration( + &format!("{CLUSTER_KEY_PREFIX}{id}.connection.idle-timeout"), + idle_timeout.get(), + problems, + ); + } + if cluster.connection_max == Some(0) { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.connection.max must be greater than zero" + )); + } + for (option, raw) in &cluster.client_options { + if let Err(problem) = parse_client_option(option, raw) { + problems.push(format!("{CLUSTER_KEY_PREFIX}{id}.{problem}")); + } + } + self.validate_credentials(id, cluster, problems); + self.validate_client_size_pairs(id, cluster, problems); + } + } + + /// Requires a service identity the gateway can actually authenticate with, wherever one is needed. + fn validate_credentials(&self, id: &str, cluster: &ClusterConfig, problems: &mut Vec) { + let account = cluster.effective_service_account(); + let secret = cluster.effective_service_secret(); + for (setting, value) in [("account", account), ("secret", secret)] { + if value.is_some_and(|value| value.trim().is_empty()) { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.connection.service.{setting} must not be blank" + )); + } + } + // A blank value is not a credential, so "complete" means usable rather than merely present. + let usable = |value: Option<&str>| value.is_some_and(|value| !value.trim().is_empty()); + let credentials_usable = usable(account) && usable(secret); + if account.is_some() != secret.is_some() { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.connection.service.account and connection.service.secret \ + must be set together" + )); + } + + let protocol = cluster.client_option("security.protocol"); + let sasl = protocol.is_some_and(|protocol| protocol.eq_ignore_ascii_case("sasl")); + if let Some(protocol) = protocol + && !sasl + && !protocol.eq_ignore_ascii_case("plaintext") + { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.client.security.protocol must be plaintext or sasl" + )); + } + if sasl && !credentials_usable { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.client.security.protocol sasl requires \ + connection.service.account and connection.service.secret" + )); + } + if let Some(mechanism) = cluster.client_option("security.sasl.mechanism") + && !mechanism.eq_ignore_ascii_case("plain") + { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.client.security.sasl.mechanism must be PLAIN" + )); + } + + if cluster.identity_mode == IdentityMode::User { + if !credentials_usable { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.connection.identity-mode user requires \ + connection.service.account and connection.service.secret" + )); + } + // Act-as travels as the SASL/PLAIN authorization ID, so without SASL there is nowhere to put + // the request's principal: Fluss would authorize the gateway's own identity for every caller. + // Requiring it to be declared rather than inferring it keeps a security-relevant choice visible. + if !sasl { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.connection.identity-mode user requires \ + client.security.protocol sasl, which carries the request principal as the SASL \ + authorization ID" + )); + } + } + } + + /// Rejects a writer size that cannot fit inside the one that has to hold it. + /// + /// Each side is the configured value or the native default, so overriding one size of a pair is caught + /// here rather than at the first write, which is the whole point of validating before a listener binds. + fn validate_client_size_pairs( + &self, + id: &str, + cluster: &ClusterConfig, + problems: &mut Vec, + ) { + let size = |option: &str| effective_size(option, cluster.client_option(option)); + for (smaller, larger) in [ + ("writer.batch-size", "writer.request-max-size"), + ("writer.batch-size", "writer.buffer.memory-size"), + ("writer.dynamic-batch-size.min", "writer.batch-size"), + ] { + if let (Some(smaller_bytes), Some(larger_bytes)) = (size(smaller), size(larger)) + && smaller_bytes > larger_bytes + { + problems.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.client.{smaller} must not exceed client.{larger}" + )); + } + } + } + + /// Requires the credential table the selected authentication mode reads. + fn validate_security(&self, problems: &mut Vec) { + let configured = |secret: &Option| { + secret + .as_ref() + .is_some_and(|value| !value.expose().trim().is_empty()) + }; + match self.security.authentication { + AuthenticationMode::Password if !configured(&self.security.users) => { + problems.push(format!( + "{SECURITY_USERS_KEY} must configure at least one user when \ + {SECURITY_AUTHENTICATION_KEY} is password" + )); + } + AuthenticationMode::Token if !configured(&self.security.tokens) => { + problems.push(format!( + "{SECURITY_TOKENS_KEY} must configure at least one token when \ + {SECURITY_AUTHENTICATION_KEY} is token" + )); + } + AuthenticationMode::TrustedHeader => { + let name = self.security.trusted_header_name(); + if name.is_empty() + || !name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) + { + problems.push(format!( + "{SECURITY_TRUSTED_HEADER_NAME_KEY} must be a legal HTTP header name" + )); + } + } + _ => {} + } + } + /// Rejects an unusable instance identity or a port clash between the two listeners. /// /// A non-loopback listener does **not** require an instance ID. Nothing the gateway returns is scoped to an @@ -414,8 +1323,47 @@ impl GatewayConfig { REST_LISTEN_KEY, self.server.rest.bind_address )); } + for (id, cluster) in &self.clusters { + let legacy: Vec<&str> = [LEGACY_SERVICE_ACCOUNT_OPTION, LEGACY_SERVICE_SECRET_OPTION] + .into_iter() + .filter(|option| cluster.client_options.contains_key(*option)) + .collect(); + // One warning per cluster, not per option: an operator migrates the pair together. + if !legacy.is_empty() { + warnings.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.client.{{{}}} is deprecated; use \ + connection.service.account and connection.service.secret. The legacy values \ + keep last-wins precedence", + legacy.join(",") + )); + } + if cluster.identity_mode == IdentityMode::Service + && (cluster.connection_max.is_some() || cluster.connection_idle_timeout.is_some()) + { + warnings.push(format!( + "{CLUSTER_KEY_PREFIX}{id}.connection.max and connection.idle-timeout are \ + ignored because connection.identity-mode is service" + )); + } + } warnings } + + /// Renders the whole configuration for diagnostics with every credential replaced. + /// + /// Redaction is a property of the types, so this is the plain [`Debug`] rendering; the method exists to + /// give callers something they can print without having to know that. + pub fn redacted_debug(&self) -> String { + format!("{self:?}") + } +} + +/// Cluster IDs appear in REST paths and in environment variable names, so they stay in the character set +/// that survives both: no hyphen, because the environment mapping already spends `_` on it. +fn valid_cluster_id(id: &str) -> bool { + let mut bytes = id.bytes(); + bytes.next().is_some_and(|first| first.is_ascii_lowercase()) + && bytes.all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') } /// True when two listeners cannot both bind: the addresses are equal, or either is a wildcard @@ -503,10 +1451,7 @@ fn insert_path(table: &mut Mapping, path: &str, value: Value) { } /// Attributes a typed error to the override that supplied the failing option. -fn attribute( - message: String, - overrides: &[(&'static str, &'static str, String, Value)], -) -> ConfigError { +fn attribute(message: String, overrides: &[(String, String, String, Value)]) -> ConfigError { for (_, key, origin, _) in overrides.iter().rev() { if message.starts_with(key) { return ConfigError::Parse(format!("{origin}: {message}")); @@ -528,6 +1473,16 @@ fn publicize_error_path(message: String) -> String { return format!("{}{reason}", entry.key); } } + // A cluster path carries the ID, so the public key is rebuilt rather than looked up. + if let Some(rest) = message.strip_prefix("clusters.") + && let Some((id, rest)) = rest.split_once('.') + { + for entry in CLUSTER_ENTRIES { + if let Some(reason) = rest.strip_prefix(entry.internal_path) { + return format!("{CLUSTER_KEY_PREFIX}{id}.{}{reason}", entry.key); + } + } + } message } @@ -535,15 +1490,32 @@ fn config_entry(key: &str) -> Option<&'static ConfigEntry> { CONFIG_ENTRIES.iter().find(|entry| entry.key == key) } +/// Derives the environment variable name from a public key: drop the `gateway.` prefix, uppercase each +/// dotted segment with `-` folded to `_`, and join the segments with `__`. fn environment_variable(key: &str) -> String { - let suffix = key - .strip_prefix("gateway.") - .expect("configuration keys use the gateway prefix") + let suffix = environment_suffix( + key.strip_prefix("gateway.") + .expect("configuration keys use the gateway prefix"), + ); + format!("{ENV_PREFIX}{suffix}") +} + +fn environment_suffix(dotted: &str) -> String { + dotted .split('.') .map(|segment| segment.replace('-', "_").to_ascii_uppercase()) .collect::>() - .join("__"); - format!("{ENV_PREFIX}{suffix}") + .join("__") +} + +/// Inverts [`environment_suffix`] for the open `client.*` namespace, whose option names are not a fixed +/// list. The mapping round-trips because option names use `-` inside a segment and `.` between segments. +fn environment_suffix_to_option(suffix: &str) -> String { + suffix + .split("__") + .map(|segment| segment.to_ascii_lowercase().replace('_', "-")) + .collect::>() + .join(".") } fn environment_entry(variable: &str) -> Option<&'static ConfigEntry> { @@ -553,26 +1525,212 @@ fn environment_entry(variable: &str) -> Option<&'static ConfigEntry> { } fn convert_environment_value(entry: &ConfigEntry, raw: &str) -> Result { - match entry.key { - METRICS_ENABLED_KEY => raw + match entry.kind { + ValueKind::Bool => raw .parse::() .map(Value::Bool) .map_err(|_| "expected true or false".to_string()), - _ => Ok(Value::String(raw.to_string())), + ValueKind::Integer => raw + .parse::() + .map(|number| Value::Number(number.into())) + .map_err(|_| "expected a non-negative integer".to_string()), + ValueKind::Text | ValueKind::Bytes | ValueKind::ServerList => { + Ok(Value::String(raw.to_string())) + } } } fn convert_file_value(entry: &ConfigEntry, value: &Value) -> Result { - match entry.key { - METRICS_ENABLED_KEY => scalar(value).cloned(), - REST_MAX_REQUEST_BYTES_KEY => match scalar(value)? { + // A quoted scalar is accepted wherever a bare one is, so an operator can move the same value between + // the file and the environment, where everything arrives as a string. + match entry.kind { + ValueKind::Bool => match scalar(value)? { + Value::Bool(_) => Ok(value.clone()), + Value::String(text) => text + .parse::() + .map(Value::Bool) + .map_err(|_| "expected true or false".to_string()), + _ => Err("expected true or false".to_string()), + }, + ValueKind::Integer => match scalar(value)? { + Value::Number(_) => Ok(value.clone()), + Value::String(text) => text + .parse::() + .map(|number| Value::Number(number.into())) + .map_err(|_| "expected a non-negative integer".to_string()), + _ => Err("expected a non-negative integer".to_string()), + }, + ValueKind::Bytes => match scalar(value)? { Value::Number(_) | Value::String(_) => Ok(value.clone()), _ => Err("expected an integer or byte-size string".to_string()), }, - _ => scalar_text(value).map(Value::String), + // A sequence is the natural YAML spelling; one comma-separated string stays accepted so the + // same value can come from the environment. + ValueKind::ServerList => match value { + Value::Sequence(_) => Ok(value.clone()), + _ => scalar_text(value).map(Value::String), + }, + ValueKind::Text => scalar_text(value).map(Value::String), } } +/// One recognised flat configuration key. +enum ResolvedKey { + /// `gateway.clusters`, the authoritative list of configurable cluster IDs. + ClusterDeclaration, + /// A fixed key, resolved to its dotted internal path. + Fixed(&'static ConfigEntry), + /// A per-cluster key under `gateway.cluster..`. + Cluster { + id: String, + entry: &'static ConfigEntry, + }, + /// A native-client option under `gateway.cluster..client.`. + ClientOption { id: String, option: String }, +} + +/// Resolves one public key, naming the key exactly as the operator wrote it when it is not recognised. +fn resolve_key(key: &str) -> Result { + let unknown = || ConfigError::Parse(format!("unknown configuration key: {key}")); + if key == CLUSTERS_KEY { + return Ok(ResolvedKey::ClusterDeclaration); + } + if let Some(entry) = config_entry(key) { + return Ok(ResolvedKey::Fixed(entry)); + } + let Some((id, suffix)) = key + .strip_prefix(CLUSTER_KEY_PREFIX) + .and_then(|rest| rest.split_once('.')) + else { + return Err(unknown()); + }; + if !valid_cluster_id(id) { + return Err(ConfigError::Parse(format!( + "invalid cluster ID in configuration key: {key}" + ))); + } + if let Some(option) = suffix.strip_prefix(CLIENT_OPTION_PREFIX) { + if option.is_empty() { + return Err(unknown()); + } + return Ok(ResolvedKey::ClientOption { + id: id.to_string(), + option: option.to_string(), + }); + } + CLUSTER_ENTRIES + .iter() + .find(|entry| entry.key == suffix) + .map(|entry| ResolvedKey::Cluster { + id: id.to_string(), + entry, + }) + .ok_or_else(unknown) +} + +/// Resolves one `FLUSS_GATEWAY__*` variable, which uses the same vocabulary as the file keys. +fn resolve_environment_variable(variable: &str) -> Result { + let unknown = || ConfigError::UnknownEnvKey(variable.to_string()); + let Some(suffix) = variable.strip_prefix(ENV_PREFIX) else { + return Err(unknown()); + }; + if suffix == environment_suffix("clusters") { + return Ok(ResolvedKey::ClusterDeclaration); + } + if let Some(entry) = environment_entry(variable) { + return Ok(ResolvedKey::Fixed(entry)); + } + let Some((id, rest)) = suffix + .strip_prefix("CLUSTER__") + .and_then(|rest| rest.split_once("__")) + else { + return Err(unknown()); + }; + let id = id.to_ascii_lowercase(); + if !valid_cluster_id(&id) { + return Err(unknown()); + } + if let Some(option) = rest.strip_prefix("CLIENT__") { + let option = environment_suffix_to_option(option); + if option.is_empty() { + return Err(unknown()); + } + return Ok(ResolvedKey::ClientOption { id, option }); + } + CLUSTER_ENTRIES + .iter() + .find(|entry| environment_suffix(entry.key) == rest) + .map(|entry| ResolvedKey::Cluster { id, entry }) + .ok_or_else(unknown) +} + +/// Reads the cluster IDs the file may configure, from a comma-separated string or a YAML sequence. +fn declared_cluster_ids(value: &Value) -> Result, ConfigError> { + let ids: Vec = match value { + Value::String(csv) => csv.split(',').map(|id| id.trim().to_string()).collect(), + Value::Sequence(items) => items + .iter() + .map(|item| { + item.as_str().map(str::to_string).ok_or_else(|| { + ConfigError::Parse(format!("{CLUSTERS_KEY}: entries must be strings")) + }) + }) + .collect::>()?, + _ => { + return Err(ConfigError::Parse(format!( + "{CLUSTERS_KEY}: expected a comma-separated string or a list" + ))); + } + }; + for id in &ids { + if !valid_cluster_id(id) { + return Err(ConfigError::Parse(format!( + "invalid cluster ID in {CLUSTERS_KEY}: {id:?}" + ))); + } + } + Ok(ids) +} + +/// Returns the mapping holding one cluster's fields, creating it when the cluster is first mentioned. +fn cluster_mapping<'a>(table: &'a mut Mapping, id: &str) -> &'a mut Mapping { + let clusters = table + .entry(Value::String("clusters".to_string())) + .or_insert_with(|| Value::Mapping(Mapping::new())); + if !clusters.is_mapping() { + *clusters = Value::Mapping(Mapping::new()); + } + let clusters = clusters.as_mapping_mut().expect("mapping inserted above"); + let cluster = clusters + .entry(Value::String(id.to_string())) + .or_insert_with(|| Value::Mapping(Mapping::new())); + if !cluster.is_mapping() { + *cluster = Value::Mapping(Mapping::new()); + } + cluster.as_mapping_mut().expect("mapping inserted above") +} + +/// Writes one native-client option. The option name keeps its dots, so it cannot go through +/// [`insert_path`], which would read them as nesting. +fn insert_client_option(table: &mut Mapping, id: &str, option: &str, raw: String) { + let cluster = cluster_mapping(table, id); + let options = cluster + .entry(Value::String("client_options".to_string())) + .or_insert_with(|| Value::Mapping(Mapping::new())); + if !options.is_mapping() { + *options = Value::Mapping(Mapping::new()); + } + options + .as_mapping_mut() + .expect("mapping inserted above") + .insert(Value::String(option.to_string()), Value::String(raw)); +} + +/// Writes one per-cluster field. +fn insert_cluster_value(table: &mut Mapping, id: &str, field: &str, value: Value) { + cluster_mapping(table, id).insert(Value::String(field.to_string()), value); +} + fn scalar(value: &Value) -> Result<&Value, String> { match value { Value::Bool(_) | Value::Number(_) | Value::String(_) => Ok(value), @@ -592,13 +1750,14 @@ fn scalar_text(value: &Value) -> Result { } } -/// Parses the flat-key YAML file into the nested mapping deserialized by [`GatewayConfig`]. -fn read_config_file(contents: &str) -> Result { +/// Parses the flat-key YAML file into the nested mapping deserialized by [`GatewayConfig`], along with +/// the `gateway.clusters` declaration when the file carries one. +fn read_config_file(contents: &str) -> Result<(Mapping, Option>), ConfigError> { let document: Value = serde_yaml_ng::from_str(contents).map_err(|e| ConfigError::Parse(e.to_string()))?; let mut table = Mapping::new(); if document.is_null() { - return Ok(table); + return Ok((table, None)); } let mapping = document.as_mapping().ok_or_else(|| { ConfigError::Parse( @@ -606,17 +1765,61 @@ fn read_config_file(contents: &str) -> Result { ) })?; + let mut declared = None; for (key, value) in mapping { let key = key .as_str() .ok_or_else(|| ConfigError::Parse("configuration keys must be strings".to_string()))?; - let entry = config_entry(key) - .ok_or_else(|| ConfigError::Parse(format!("unknown configuration key: {key}")))?; - let value = convert_file_value(entry, value) - .map_err(|reason| ConfigError::Parse(format!("{key}: {reason}")))?; - insert_path(&mut table, entry.internal_path, value); + let reason = |reason: String| ConfigError::Parse(format!("{key}: {reason}")); + match resolve_key(key)? { + ResolvedKey::ClusterDeclaration => declared = Some(declared_cluster_ids(value)?), + ResolvedKey::Fixed(entry) => { + let value = convert_file_value(entry, value).map_err(reason)?; + insert_path(&mut table, entry.internal_path, value); + } + ResolvedKey::Cluster { id, entry } => { + let value = convert_file_value(entry, value).map_err(reason)?; + insert_cluster_value(&mut table, &id, entry.internal_path, value); + } + ResolvedKey::ClientOption { id, option } => { + let raw = scalar_text(value).map_err(reason)?; + insert_client_option(&mut table, &id, &option, raw); + } + } + } + Ok((table, declared)) +} + +/// Makes every declared cluster exist and rejects any cluster configured without being declared. +/// +/// The declaration is authoritative whether or not it was written: an absent `gateway.clusters` means the +/// single implicit `default` cluster, so a mistyped ID fails startup instead of quietly creating a second +/// cluster that nothing routes to. +fn reconcile_declared_clusters( + table: &mut Mapping, + declared: Option<&[String]>, +) -> Result<(), ConfigError> { + let implicit = [DEFAULT_CLUSTER_ID.to_string()]; + let declared = declared.unwrap_or(&implicit); + for id in declared { + cluster_mapping(table, id); + } + let configured: Vec = table + .get(Value::String("clusters".to_string())) + .and_then(Value::as_mapping) + .into_iter() + .flat_map(Mapping::keys) + .filter_map(Value::as_str) + .map(str::to_string) + .collect(); + for id in configured { + if !declared.contains(&id) { + return Err(ConfigError::Parse(format!( + "{CLUSTER_KEY_PREFIX}{id}.* is configured but {id} is not declared in {CLUSTERS_KEY}" + ))); + } } - Ok(table) + Ok(()) } /// Loads configuration from all sources with precedence CLI > env > file > defaults. @@ -628,34 +1831,56 @@ pub fn load( cli: &CliOverrides, ) -> Result { let mut table = Mapping::new(); + let mut declared = None; if let Some(path) = path { let contents = std::fs::read_to_string(path) .map_err(|e| ConfigError::Io(format!("{}: {e}", path.display())))?; - table = read_config_file(&contents)?; + (table, declared) = read_config_file(&contents)?; } // Each override is kept with the source that wrote it, so a failure names what the operator wrote. - let mut overrides: Vec<(&'static str, &'static str, String, Value)> = Vec::new(); - for (key, raw) in env { - if !key.starts_with(ENV_PREFIX) { + let mut overrides: Vec<(String, String, String, Value)> = Vec::new(); + for (variable, raw) in env { + if !variable.starts_with(ENV_PREFIX) { continue; } - let entry = - environment_entry(key).ok_or_else(|| ConfigError::UnknownEnvKey(key.clone()))?; - overrides.push(( - entry.internal_path, - entry.key, - key.clone(), - convert_environment_value(entry, raw) - .map_err(|reason| ConfigError::Parse(format!("{key}: {}: {reason}", entry.key)))?, - )); + match resolve_environment_variable(variable)? { + ResolvedKey::ClusterDeclaration => { + declared = Some(declared_cluster_ids(&Value::String(raw.clone()))?); + } + ResolvedKey::Fixed(entry) => overrides.push(( + entry.internal_path.to_string(), + entry.key.to_string(), + variable.clone(), + convert_environment_value(entry, raw).map_err(|reason| { + ConfigError::Parse(format!("{variable}: {}: {reason}", entry.key)) + })?, + )), + ResolvedKey::Cluster { id, entry } => { + let public_key = format!("{CLUSTER_KEY_PREFIX}{id}.{}", entry.key); + let value = convert_environment_value(entry, raw).map_err(|reason| { + ConfigError::Parse(format!("{variable}: {public_key}: {reason}")) + })?; + overrides.push(( + format!("clusters.{id}.{}", entry.internal_path), + public_key, + variable.clone(), + value, + )); + } + // Client options are written straight in: the whole namespace lands in one map, so there is + // no typed path an error could be attributed to. + ResolvedKey::ClientOption { id, option } => { + insert_client_option(&mut table, &id, &option, raw.clone()); + } + } } if let Some(value) = &cli.bind_address { let entry = config_entry(REST_LISTEN_KEY).expect("REST listen option is registered"); overrides.push(( - entry.internal_path, - entry.key, + entry.internal_path.to_string(), + entry.key.to_string(), "--bind-address".to_string(), Value::String(value.clone()), )); @@ -665,6 +1890,8 @@ pub fn load( insert_path(&mut table, path, value.clone()); } + reconcile_declared_clusters(&mut table, declared.as_deref())?; + let config = deserialize_config(Value::Mapping(table)).map_err(|error| { let ConfigError::Parse(message) = error else { unreachable!("deserialization only creates parse errors") @@ -675,494 +1902,3 @@ pub fn load( config.validate()?; Ok(config) } - -#[cfg(test)] -mod tests { - use super::*; - use std::io::Write; - - fn no_env() -> BTreeMap { - BTreeMap::new() - } - - fn write_temp_config(contents: &str) -> tempfile::NamedTempFile { - let mut file = tempfile::NamedTempFile::new().expect("temp file"); - file.write_all(contents.as_bytes()).expect("write"); - file - } - - fn load_file(contents: &str) -> Result { - let file = write_temp_config(contents); - load(Some(file.path()), &no_env(), &CliOverrides::default()) - } - - fn problems(error: ConfigError) -> Vec { - match error { - ConfigError::Invalid(problems) => problems, - other => panic!("expected Invalid, got: {other:?}"), - } - } - - #[test] - fn defaults_when_no_sources() { - let config = load(None, &no_env(), &CliOverrides::default()).unwrap(); - assert_eq!( - config.server.rest.bind_address, - "127.0.0.1:8080".parse().unwrap() - ); - assert_eq!(config.server.rest.max_body_bytes.bytes(), 32 * 1024 * 1024); - assert_eq!( - config.server.rest.request_timeout.get(), - Duration::from_secs(30) - ); - assert!(config.server.metrics.enabled); - assert_eq!( - config.server.metrics.bind_address, - "127.0.0.1:9095".parse().unwrap() - ); - assert_eq!(config.shutdown.drain_timeout.get(), Duration::from_secs(30)); - assert!(config.warnings().is_empty()); - } - - #[test] - fn public_yaml_options_are_loaded() { - let config = load_file( - r#" - gateway.instance-id: gateway-1 - gateway.rest.listen: 0.0.0.0:8080 - gateway.rest.write.max-request-bytes: 32MiB - gateway.rest.write.request-timeout: 30s - gateway.metrics.enabled: true - gateway.metrics.exporter.prometheus.listen: 0.0.0.0:9095 - gateway.shutdown.drain-timeout: 10s - "#, - ) - .unwrap(); - assert_eq!(config.server.instance_id.as_deref(), Some("gateway-1")); - assert_eq!( - config.server.rest.bind_address, - "0.0.0.0:8080".parse().unwrap() - ); - assert_eq!(config.server.rest.max_body_bytes.bytes(), 32 * 1024 * 1024); - assert_eq!( - config.server.rest.request_timeout.get(), - Duration::from_secs(30) - ); - assert!(config.server.metrics.enabled); - assert_eq!( - config.server.metrics.bind_address, - "0.0.0.0:9095".parse().unwrap() - ); - assert_eq!(config.shutdown.drain_timeout.get(), Duration::from_secs(10)); - } - - #[test] - fn unknown_file_keys_name_the_original_key() { - for contents in [ - "gateway.rest.listenn: 0.0.0.0:8080\n", - "rest.listen: 0.0.0.0:8080\n", - "gateway.rest.lookup.max-keyz: 5\n", - "gateway.scan.cursor-ttl: 1m\n", - "gateway.tls.cert: /etc/tls.pem\n", - ] { - let error = load_file(contents).unwrap_err(); - assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); - let key = contents.split(':').next().unwrap(); - assert!(error.to_string().contains(key), "{key}: {error}"); - } - } - - #[test] - fn source_precedence_is_cli_then_env_then_file_then_defaults() { - let file = write_temp_config( - r#" - gateway.rest.listen: 127.0.0.1:18080 - gateway.metrics.enabled: true - "#, - ); - let mut env = no_env(); - env.insert( - "FLUSS_GATEWAY__REST__LISTEN".to_string(), - "127.0.0.1:28080".to_string(), - ); - env.insert( - "FLUSS_GATEWAY__METRICS__ENABLED".to_string(), - "false".to_string(), - ); - env.insert("PATH".to_string(), "/usr/bin".to_string()); - - let config = load( - Some(file.path()), - &env, - &CliOverrides { - bind_address: Some("127.0.0.1:38080".to_string()), - }, - ) - .unwrap(); - assert_eq!( - config.server.rest.bind_address, - "127.0.0.1:38080".parse().unwrap() - ); - assert!(!config.server.metrics.enabled); - } - - #[test] - fn missing_file_reported() { - let error = load( - Some(Path::new("/nonexistent/gateway.yaml")), - &no_env(), - &CliOverrides::default(), - ) - .unwrap_err(); - assert!(matches!(error, ConfigError::Io(_)), "got: {error:?}"); - } - - #[test] - fn malformed_file_reports_position() { - let error = load_file("gateway.rest.listen: [1\n").unwrap_err(); - assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); - assert!(error.to_string().contains("line"), "got: {error}"); - } - - #[test] - fn duplicate_flat_key_rejected() { - let error = - load_file("gateway.rest.listen: 127.0.0.1:8080\ngateway.rest.listen: 127.0.0.1:8081\n") - .unwrap_err(); - assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); - assert!(error.to_string().contains("duplicate"), "got: {error}"); - } - - #[test] - fn unknown_environment_variables_are_rejected() { - for key in [ - "FLUSS_GATEWAY__REST__LISTENN", - "FLUSS_GATEWAY__QUERY__ENABLED", - "FLUSS_GATEWAY__SERVER_REST__BIND_ADDRESS", - ] { - let mut env = no_env(); - env.insert(key.to_string(), "value".to_string()); - let error = load(None, &env, &CliOverrides::default()).unwrap_err(); - assert!( - matches!(error, ConfigError::UnknownEnvKey(_)), - "{key}: {error:?}" - ); - assert!(error.to_string().contains(key), "{key}: {error}"); - } - } - - #[test] - fn file_error_under_a_section_with_an_env_override_names_the_file() { - let file = write_temp_config("gateway.shutdown.drain-timeout: 0s\n"); - let mut env = no_env(); - env.insert( - "FLUSS_GATEWAY__REST__LISTEN".to_string(), - "127.0.0.1:28080".to_string(), - ); - let error = load(Some(file.path()), &env, &CliOverrides::default()).unwrap_err(); - assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); - assert!( - error.to_string().contains("gateway.shutdown.drain-timeout"), - "got: {error}" - ); - assert!( - !error.to_string().contains("FLUSS_GATEWAY__"), - "file problem misattributed to the env override: {error}" - ); - } - - #[test] - fn public_environment_options_are_loaded_by_type() { - let env = BTreeMap::from([ - ("FLUSS_GATEWAY__INSTANCE_ID".to_string(), "123".to_string()), - ( - "FLUSS_GATEWAY__REST__LISTEN".to_string(), - "127.0.0.1:18080".to_string(), - ), - ( - "FLUSS_GATEWAY__REST__WRITE__REQUEST_TIMEOUT".to_string(), - "5s".to_string(), - ), - ( - "FLUSS_GATEWAY__REST__WRITE__MAX_REQUEST_BYTES".to_string(), - "2MiB".to_string(), - ), - ( - "FLUSS_GATEWAY__METRICS__ENABLED".to_string(), - "false".to_string(), - ), - ( - "FLUSS_GATEWAY__METRICS__EXPORTER__PROMETHEUS__LISTEN".to_string(), - "127.0.0.1:19095".to_string(), - ), - ( - "FLUSS_GATEWAY__SHUTDOWN__DRAIN_TIMEOUT".to_string(), - "10s".to_string(), - ), - ]); - - let config = load(None, &env, &CliOverrides::default()).unwrap(); - assert_eq!(config.server.instance_id.as_deref(), Some("123")); - assert_eq!( - config.server.rest.bind_address, - "127.0.0.1:18080".parse().unwrap() - ); - assert_eq!( - config.server.rest.request_timeout.get(), - Duration::from_secs(5) - ); - assert_eq!(config.server.rest.max_body_bytes.bytes(), 2 * 1024 * 1024); - assert!(!config.server.metrics.enabled); - assert_eq!( - config.server.metrics.bind_address, - "127.0.0.1:19095".parse().unwrap() - ); - assert_eq!(config.shutdown.drain_timeout.get(), Duration::from_secs(10)); - } - - #[test] - fn invalid_env_value_names_the_variable() { - let mut env = no_env(); - env.insert( - "FLUSS_GATEWAY__REST__WRITE__MAX_REQUEST_BYTES".to_string(), - "many".to_string(), - ); - let error = load(None, &env, &CliOverrides::default()).unwrap_err(); - assert!( - error - .to_string() - .contains("FLUSS_GATEWAY__REST__WRITE__MAX_REQUEST_BYTES"), - "got: {error}" - ); - } - - #[test] - fn invalid_cli_value_names_the_flag() { - let cli = CliOverrides { - bind_address: Some("not-an-address".to_string()), - }; - let error = load(None, &no_env(), &cli).unwrap_err(); - assert!(error.to_string().contains("--bind-address"), "got: {error}"); - } - - #[test] - fn invalid_duration_rejected() { - for bad in ["0ms", "60", "60 s", "6.5s", "s", "60d", "-1s"] { - let error = - load_file(&format!("gateway.shutdown.drain-timeout: \"{bad}\"\n")).unwrap_err(); - assert!(matches!(error, ConfigError::Parse(_)), "{bad}: {error:?}"); - assert!( - error.to_string().contains("gateway.shutdown.drain-timeout"), - "{bad}: {error}" - ); - } - } - - #[test] - fn overflowing_duration_is_rejected_rather_than_saturated() { - for bad in [ - "18446744073709551615ms", - "18446744073709551615s", - "18446744073709551615m", - "18446744073709551615h", - ] { - let error = ConfigDuration::parse(bad).unwrap_err(); - assert!(error.contains("must not exceed"), "{bad}: {error}"); - } - assert_eq!( - ConfigDuration::parse("31536000s").unwrap().get(), - MAX_CONFIG_DURATION - ); - assert!(ConfigDuration::parse("31536001s").is_err()); - } - - #[test] - fn programmatically_constructed_durations_are_validated() { - let mut config = GatewayConfig::default(); - config.server.rest.request_timeout = ConfigDuration::from_millis(0); - config.shutdown.drain_timeout = - ConfigDuration::from_secs(MAX_CONFIG_DURATION.as_secs() + 1); - - let errors = problems(config.validate().unwrap_err()); - assert!( - errors.iter().any(|error| { - error == "gateway.rest.write.request-timeout must be greater than zero" - }), - "got: {errors:?}" - ); - assert!( - errors.iter().any(|error| { - error == "gateway.shutdown.drain-timeout must not exceed 31536000 seconds" - }), - "got: {errors:?}" - ); - } - - #[test] - fn programmatically_constructed_zero_byte_limit_is_validated() { - let mut config = GatewayConfig::default(); - config.server.rest.max_body_bytes = ByteSize::new(0); - - let errors = problems(config.validate().unwrap_err()); - assert_eq!( - errors, - vec!["gateway.rest.write.max-request-bytes must be greater than zero"] - ); - } - - #[test] - fn invalid_byte_size_rejected() { - for bad in ["0", "\"4Mb\"", "\"MiB\"", "-1", "\"1.5MiB\""] { - let error = - load_file(&format!("gateway.rest.write.max-request-bytes: {bad}\n")).unwrap_err(); - assert!(matches!(error, ConfigError::Parse(_)), "{bad}: {error:?}"); - assert!( - error - .to_string() - .contains("gateway.rest.write.max-request-bytes"), - "{bad}: {error}" - ); - } - } - - #[test] - fn metrics_address_must_differ_from_rest_address() { - let error = load_file( - "gateway.rest.listen: 127.0.0.1:9095\ngateway.metrics.exporter.prometheus.listen: 127.0.0.1:9095\n", - ) - .unwrap_err(); - assert!(problems(error).iter().any(|problem| { - problem.contains( - "gateway.metrics.exporter.prometheus.listen (127.0.0.1:9095) must differ from \ - gateway.rest.listen (127.0.0.1:9095)", - ) - })); - } - - /// Overlap detection covers the wildcard and dual-stack pairs that differ textually but cannot - /// both bind, plus the pairs that coexist. - #[test] - fn listener_overlap_covers_wildcards_and_dual_stack() { - let clashes = [ - ("0.0.0.0:8080", "127.0.0.1:8080"), - ("127.0.0.1:8080", "0.0.0.0:8080"), - ("0.0.0.0:8080", "0.0.0.0:8080"), - ("[::]:8080", "[::1]:8080"), - ("[::]:8080", "0.0.0.0:8080"), - ("127.0.0.1:8080", "[::]:8080"), - ]; - for (rest, metrics) in clashes { - let rest: SocketAddr = rest.parse().unwrap(); - let metrics: SocketAddr = metrics.parse().unwrap(); - assert!( - addresses_overlap(rest, metrics), - "{rest} and {metrics} cannot both bind" - ); - } - - let coexist = [ - ("127.0.0.1:8080", "192.168.1.2:8080"), - ("127.0.0.1:8080", "[::1]:8080"), - ("0.0.0.0:8080", "127.0.0.1:9095"), - ("0.0.0.0:0", "0.0.0.0:0"), - ("127.0.0.1:0", "0.0.0.0:8080"), - ]; - for (rest, metrics) in coexist { - let rest: SocketAddr = rest.parse().unwrap(); - let metrics: SocketAddr = metrics.parse().unwrap(); - assert!( - !addresses_overlap(rest, metrics), - "{rest} and {metrics} can coexist" - ); - } - } - - /// Two ephemeral listeners are not a clash: the OS hands out a different port to each. - #[test] - fn both_listeners_may_ask_for_an_ephemeral_port() { - let config = load_file( - "gateway.rest.listen: 127.0.0.1:0\ngateway.metrics.exporter.prometheus.listen: 127.0.0.1:0\n", - ) - .unwrap(); - assert_eq!(config.server.rest.bind_address.port(), 0); - } - - #[test] - fn non_loopback_bind_is_accepted_without_an_instance_id_but_warns() { - let config = load_file("gateway.rest.listen: 0.0.0.0:8080\n").unwrap(); - assert!(config.server.instance_id.is_none()); - assert_eq!(config.warnings().len(), 1); - assert!(config.warnings()[0].contains("not loopback")); - assert!( - config.warnings()[0].contains("accepts unauthenticated requests"), - "{:?}", - config.warnings() - ); - } - - #[test] - fn malformed_instance_id_rejected() { - let error = load_file("gateway.instance-id: has space\n").unwrap_err(); - assert!( - problems(error) - .iter() - .any(|problem| problem.contains("gateway.instance-id must be 1-128 ASCII")) - ); - } - - #[test] - fn duration_units() { - assert_eq!( - ConfigDuration::parse("250ms").unwrap().get(), - Duration::from_millis(250) - ); - assert_eq!( - ConfigDuration::parse("15m").unwrap().get(), - Duration::from_secs(900) - ); - assert_eq!( - ConfigDuration::parse("2h").unwrap().get(), - Duration::from_secs(7200) - ); - assert!(ConfigDuration::parse("0s").is_err()); - } - - #[test] - fn byte_size_units() { - assert_eq!(ByteSize::parse("512").unwrap().bytes(), 512); - assert_eq!(ByteSize::parse("512B").unwrap().bytes(), 512); - assert_eq!(ByteSize::parse("4KB").unwrap().bytes(), 4000); - assert_eq!(ByteSize::parse("4KiB").unwrap().bytes(), 4096); - assert_eq!(ByteSize::parse("1GiB").unwrap().bytes(), 1024 * 1024 * 1024); - assert!(ByteSize::parse("4TB").is_err()); - assert!(ByteSize::parse("0").is_err()); - } - - #[test] - fn options_are_complete_and_unambiguous() { - let mut public_keys = std::collections::BTreeSet::new(); - let mut internal_paths = std::collections::BTreeSet::new(); - let mut environment_variables = std::collections::BTreeSet::new(); - - for entry in CONFIG_ENTRIES { - assert!(entry.key.starts_with("gateway."), "{entry:?}"); - assert!( - public_keys.insert(entry.key), - "duplicate key: {}", - entry.key - ); - assert!( - internal_paths.insert(entry.internal_path), - "duplicate path: {}", - entry.internal_path - ); - assert!( - environment_variables.insert(environment_variable(entry.key)), - "duplicate environment variable for {}", - entry.key - ); - } - - assert_eq!(CONFIG_ENTRIES.len(), 8); - } -} diff --git a/fluss-gateway/src/config/tests.rs b/fluss-gateway/src/config/tests.rs new file mode 100644 index 0000000000..c57695d02e --- /dev/null +++ b/fluss-gateway/src/config/tests.rs @@ -0,0 +1,1375 @@ +// 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. + +//! Tests for the configuration vocabulary, its sources, and its validation. + +use super::*; +use std::io::Write; + +fn no_env() -> BTreeMap { + BTreeMap::new() +} + +fn write_temp_config(contents: &str) -> tempfile::NamedTempFile { + let mut file = tempfile::NamedTempFile::new().expect("temp file"); + file.write_all(contents.as_bytes()).expect("write"); + file +} + +fn load_file(contents: &str) -> Result { + let file = write_temp_config(contents); + load(Some(file.path()), &no_env(), &CliOverrides::default()) +} + +fn problems(error: ConfigError) -> Vec { + match error { + ConfigError::Invalid(problems) => problems, + other => panic!("expected Invalid, got: {other:?}"), + } +} + +#[test] +fn defaults_when_no_sources() { + let config = load(None, &no_env(), &CliOverrides::default()).unwrap(); + assert_eq!( + config.server.rest.bind_address, + "127.0.0.1:8080".parse().unwrap() + ); + assert_eq!(config.server.rest.max_body_bytes.bytes(), 32 * 1024 * 1024); + assert_eq!( + config.server.rest.request_timeout.get(), + Duration::from_secs(30) + ); + assert!(config.server.metrics.enabled); + assert_eq!( + config.server.metrics.bind_address, + "127.0.0.1:9095".parse().unwrap() + ); + assert_eq!(config.shutdown.drain_timeout.get(), Duration::from_secs(30)); + assert!(config.warnings().is_empty()); +} + +#[test] +fn public_yaml_options_are_loaded() { + let config = load_file( + r#" +gateway.instance-id: gateway-1 +gateway.rest.listen: 0.0.0.0:8080 +gateway.rest.write.max-request-bytes: 32MiB +gateway.rest.write.request-timeout: 30s +gateway.metrics.enabled: true +gateway.metrics.exporter.prometheus.listen: 0.0.0.0:9095 +gateway.shutdown.drain-timeout: 10s +"#, + ) + .unwrap(); + assert_eq!(config.server.instance_id.as_deref(), Some("gateway-1")); + assert_eq!( + config.server.rest.bind_address, + "0.0.0.0:8080".parse().unwrap() + ); + assert_eq!(config.server.rest.max_body_bytes.bytes(), 32 * 1024 * 1024); + assert_eq!( + config.server.rest.request_timeout.get(), + Duration::from_secs(30) + ); + assert!(config.server.metrics.enabled); + assert_eq!( + config.server.metrics.bind_address, + "0.0.0.0:9095".parse().unwrap() + ); + assert_eq!(config.shutdown.drain_timeout.get(), Duration::from_secs(10)); +} + +#[test] +fn unknown_file_keys_name_the_original_key() { + for contents in [ + "gateway.rest.listenn: 0.0.0.0:8080\n", + "rest.listen: 0.0.0.0:8080\n", + "gateway.rest.lookup.max-keyz: 5\n", + "gateway.scan.cursor-ttl: 1m\n", + "gateway.tls.cert: /etc/tls.pem\n", + ] { + let error = load_file(contents).unwrap_err(); + assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); + let key = contents.split(':').next().unwrap(); + assert!(error.to_string().contains(key), "{key}: {error}"); + } +} + +#[test] +fn source_precedence_is_cli_then_env_then_file_then_defaults() { + let file = write_temp_config( + r#" +gateway.rest.listen: 127.0.0.1:18080 +gateway.metrics.enabled: true +"#, + ); + let mut env = no_env(); + env.insert( + "FLUSS_GATEWAY__REST__LISTEN".to_string(), + "127.0.0.1:28080".to_string(), + ); + env.insert( + "FLUSS_GATEWAY__METRICS__ENABLED".to_string(), + "false".to_string(), + ); + env.insert("PATH".to_string(), "/usr/bin".to_string()); + + let config = load( + Some(file.path()), + &env, + &CliOverrides { + bind_address: Some("127.0.0.1:38080".to_string()), + }, + ) + .unwrap(); + assert_eq!( + config.server.rest.bind_address, + "127.0.0.1:38080".parse().unwrap() + ); + assert!(!config.server.metrics.enabled); +} + +#[test] +fn missing_file_reported() { + let error = load( + Some(Path::new("/nonexistent/gateway.yaml")), + &no_env(), + &CliOverrides::default(), + ) + .unwrap_err(); + assert!(matches!(error, ConfigError::Io(_)), "got: {error:?}"); +} + +#[test] +fn malformed_file_reports_position() { + let error = load_file("gateway.rest.listen: [1\n").unwrap_err(); + assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); + assert!(error.to_string().contains("line"), "got: {error}"); +} + +#[test] +fn duplicate_flat_key_rejected() { + let error = + load_file("gateway.rest.listen: 127.0.0.1:8080\ngateway.rest.listen: 127.0.0.1:8081\n") + .unwrap_err(); + assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); + assert!(error.to_string().contains("duplicate"), "got: {error}"); +} + +#[test] +fn unknown_environment_variables_are_rejected() { + for key in [ + "FLUSS_GATEWAY__REST__LISTENN", + "FLUSS_GATEWAY__QUERY__ENABLED", + "FLUSS_GATEWAY__SERVER_REST__BIND_ADDRESS", + ] { + let mut env = no_env(); + env.insert(key.to_string(), "value".to_string()); + let error = load(None, &env, &CliOverrides::default()).unwrap_err(); + assert!( + matches!(error, ConfigError::UnknownEnvKey(_)), + "{key}: {error:?}" + ); + assert!(error.to_string().contains(key), "{key}: {error}"); + } +} + +#[test] +fn file_error_under_a_section_with_an_env_override_names_the_file() { + let file = write_temp_config("gateway.shutdown.drain-timeout: 0s\n"); + let mut env = no_env(); + env.insert( + "FLUSS_GATEWAY__REST__LISTEN".to_string(), + "127.0.0.1:28080".to_string(), + ); + let error = load(Some(file.path()), &env, &CliOverrides::default()).unwrap_err(); + assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); + assert!( + error.to_string().contains("gateway.shutdown.drain-timeout"), + "got: {error}" + ); + assert!( + !error.to_string().contains("FLUSS_GATEWAY__"), + "file problem misattributed to the env override: {error}" + ); +} + +#[test] +fn public_environment_options_are_loaded_by_type() { + let env = BTreeMap::from([ + ("FLUSS_GATEWAY__INSTANCE_ID".to_string(), "123".to_string()), + ( + "FLUSS_GATEWAY__REST__LISTEN".to_string(), + "127.0.0.1:18080".to_string(), + ), + ( + "FLUSS_GATEWAY__REST__WRITE__REQUEST_TIMEOUT".to_string(), + "5s".to_string(), + ), + ( + "FLUSS_GATEWAY__REST__WRITE__MAX_REQUEST_BYTES".to_string(), + "2MiB".to_string(), + ), + ( + "FLUSS_GATEWAY__METRICS__ENABLED".to_string(), + "false".to_string(), + ), + ( + "FLUSS_GATEWAY__METRICS__EXPORTER__PROMETHEUS__LISTEN".to_string(), + "127.0.0.1:19095".to_string(), + ), + ( + "FLUSS_GATEWAY__SHUTDOWN__DRAIN_TIMEOUT".to_string(), + "10s".to_string(), + ), + ]); + + let config = load(None, &env, &CliOverrides::default()).unwrap(); + assert_eq!(config.server.instance_id.as_deref(), Some("123")); + assert_eq!( + config.server.rest.bind_address, + "127.0.0.1:18080".parse().unwrap() + ); + assert_eq!( + config.server.rest.request_timeout.get(), + Duration::from_secs(5) + ); + assert_eq!(config.server.rest.max_body_bytes.bytes(), 2 * 1024 * 1024); + assert!(!config.server.metrics.enabled); + assert_eq!( + config.server.metrics.bind_address, + "127.0.0.1:19095".parse().unwrap() + ); + assert_eq!(config.shutdown.drain_timeout.get(), Duration::from_secs(10)); +} + +#[test] +fn invalid_env_value_names_the_variable() { + let mut env = no_env(); + env.insert( + "FLUSS_GATEWAY__REST__WRITE__MAX_REQUEST_BYTES".to_string(), + "many".to_string(), + ); + let error = load(None, &env, &CliOverrides::default()).unwrap_err(); + assert!( + error + .to_string() + .contains("FLUSS_GATEWAY__REST__WRITE__MAX_REQUEST_BYTES"), + "got: {error}" + ); +} + +#[test] +fn invalid_cli_value_names_the_flag() { + let cli = CliOverrides { + bind_address: Some("not-an-address".to_string()), + }; + let error = load(None, &no_env(), &cli).unwrap_err(); + assert!(error.to_string().contains("--bind-address"), "got: {error}"); +} + +#[test] +fn invalid_duration_rejected() { + for bad in ["0ms", "60", "60 s", "6.5s", "s", "60d", "-1s"] { + let error = load_file(&format!("gateway.shutdown.drain-timeout: \"{bad}\"\n")).unwrap_err(); + assert!(matches!(error, ConfigError::Parse(_)), "{bad}: {error:?}"); + assert!( + error.to_string().contains("gateway.shutdown.drain-timeout"), + "{bad}: {error}" + ); + } +} + +#[test] +fn overflowing_duration_is_rejected_rather_than_saturated() { + for bad in [ + "18446744073709551615ms", + "18446744073709551615s", + "18446744073709551615m", + "18446744073709551615h", + ] { + let error = ConfigDuration::parse(bad).unwrap_err(); + assert!(error.contains("must not exceed"), "{bad}: {error}"); + } + assert_eq!( + ConfigDuration::parse("31536000s").unwrap().get(), + MAX_CONFIG_DURATION + ); + assert!(ConfigDuration::parse("31536001s").is_err()); +} + +#[test] +fn programmatically_constructed_durations_are_validated() { + let mut config = GatewayConfig::default(); + config.server.rest.request_timeout = ConfigDuration::from_millis(0); + config.shutdown.drain_timeout = ConfigDuration::from_secs(MAX_CONFIG_DURATION.as_secs() + 1); + + let errors = problems(config.validate().unwrap_err()); + assert!( + errors.iter().any(|error| { + error == "gateway.rest.write.request-timeout must be greater than zero" + }), + "got: {errors:?}" + ); + assert!( + errors.iter().any(|error| { + error == "gateway.shutdown.drain-timeout must not exceed 31536000 seconds" + }), + "got: {errors:?}" + ); +} + +#[test] +fn programmatically_constructed_zero_byte_limit_is_validated() { + let mut config = GatewayConfig::default(); + config.server.rest.max_body_bytes = ByteSize::new(0); + + let errors = problems(config.validate().unwrap_err()); + assert_eq!( + errors, + vec!["gateway.rest.write.max-request-bytes must be greater than zero"] + ); +} + +#[test] +fn invalid_byte_size_rejected() { + for bad in ["0", "\"4Mb\"", "\"MiB\"", "-1", "\"1.5MiB\""] { + let error = + load_file(&format!("gateway.rest.write.max-request-bytes: {bad}\n")).unwrap_err(); + assert!(matches!(error, ConfigError::Parse(_)), "{bad}: {error:?}"); + assert!( + error + .to_string() + .contains("gateway.rest.write.max-request-bytes"), + "{bad}: {error}" + ); + } +} + +#[test] +fn metrics_address_must_differ_from_rest_address() { + let error = load_file( + "gateway.rest.listen: 127.0.0.1:9095\ngateway.metrics.exporter.prometheus.listen: 127.0.0.1:9095\n", + ) + .unwrap_err(); + assert!(problems(error).iter().any(|problem| { + problem.contains( + "gateway.metrics.exporter.prometheus.listen (127.0.0.1:9095) must differ from \ + gateway.rest.listen (127.0.0.1:9095)", + ) + })); +} + +/// Overlap detection covers the wildcard and dual-stack pairs that differ textually but cannot +/// both bind, plus the pairs that coexist. +#[test] +fn listener_overlap_covers_wildcards_and_dual_stack() { + let clashes = [ + ("0.0.0.0:8080", "127.0.0.1:8080"), + ("127.0.0.1:8080", "0.0.0.0:8080"), + ("0.0.0.0:8080", "0.0.0.0:8080"), + ("[::]:8080", "[::1]:8080"), + ("[::]:8080", "0.0.0.0:8080"), + ("127.0.0.1:8080", "[::]:8080"), + ]; + for (rest, metrics) in clashes { + let rest: SocketAddr = rest.parse().unwrap(); + let metrics: SocketAddr = metrics.parse().unwrap(); + assert!( + addresses_overlap(rest, metrics), + "{rest} and {metrics} cannot both bind" + ); + } + + let coexist = [ + ("127.0.0.1:8080", "192.168.1.2:8080"), + ("127.0.0.1:8080", "[::1]:8080"), + ("0.0.0.0:8080", "127.0.0.1:9095"), + ("0.0.0.0:0", "0.0.0.0:0"), + ("127.0.0.1:0", "0.0.0.0:8080"), + ]; + for (rest, metrics) in coexist { + let rest: SocketAddr = rest.parse().unwrap(); + let metrics: SocketAddr = metrics.parse().unwrap(); + assert!( + !addresses_overlap(rest, metrics), + "{rest} and {metrics} can coexist" + ); + } +} + +/// Two ephemeral listeners are not a clash: the OS hands out a different port to each. +#[test] +fn both_listeners_may_ask_for_an_ephemeral_port() { + let config = load_file( + "gateway.rest.listen: 127.0.0.1:0\ngateway.metrics.exporter.prometheus.listen: 127.0.0.1:0\n", + ) + .unwrap(); + assert_eq!(config.server.rest.bind_address.port(), 0); +} + +#[test] +fn non_loopback_bind_is_accepted_without_an_instance_id_but_warns() { + let config = load_file("gateway.rest.listen: 0.0.0.0:8080\n").unwrap(); + assert!(config.server.instance_id.is_none()); + assert_eq!(config.warnings().len(), 1); + assert!(config.warnings()[0].contains("not loopback")); + assert!( + config.warnings()[0].contains("accepts unauthenticated requests"), + "{:?}", + config.warnings() + ); +} + +#[test] +fn malformed_instance_id_rejected() { + let error = load_file("gateway.instance-id: has space\n").unwrap_err(); + assert!( + problems(error) + .iter() + .any(|problem| problem.contains("gateway.instance-id must be 1-128 ASCII")) + ); +} + +#[test] +fn duration_units() { + assert_eq!( + ConfigDuration::parse("250ms").unwrap().get(), + Duration::from_millis(250) + ); + assert_eq!( + ConfigDuration::parse("15m").unwrap().get(), + Duration::from_secs(900) + ); + assert_eq!( + ConfigDuration::parse("2h").unwrap().get(), + Duration::from_secs(7200) + ); + assert!(ConfigDuration::parse("0s").is_err()); +} + +#[test] +fn byte_size_units() { + assert_eq!(ByteSize::parse("512").unwrap().bytes(), 512); + assert_eq!(ByteSize::parse("512B").unwrap().bytes(), 512); + assert_eq!(ByteSize::parse("4KB").unwrap().bytes(), 4000); + assert_eq!(ByteSize::parse("4KiB").unwrap().bytes(), 4096); + assert_eq!(ByteSize::parse("1GiB").unwrap().bytes(), 1024 * 1024 * 1024); + assert!(ByteSize::parse("4TB").is_err()); + assert!(ByteSize::parse("0").is_err()); +} + +fn cluster<'a>(config: &'a GatewayConfig, id: &str) -> &'a ClusterConfig { + config.clusters.get(id).expect("configured cluster") +} + +#[test] +fn a_single_default_cluster_needs_no_declaration() { + let config = load(None, &no_env(), &CliOverrides::default()).unwrap(); + assert_eq!(config.clusters.len(), 1); + assert_eq!( + cluster(&config, DEFAULT_CLUSTER_ID).bootstrap_servers, + [DEFAULT_BOOTSTRAP_SERVERS] + ); + assert_eq!( + cluster(&config, DEFAULT_CLUSTER_ID).identity_mode, + IdentityMode::Service + ); + assert_eq!(config.security.authentication, AuthenticationMode::Trust); + assert_eq!(config.request_limits, RequestLimitsConfig::default()); +} + +#[test] +fn typed_cluster_security_and_request_limit_options_are_loaded() { + let config = load_file( + "gateway.clusters: default, analytics\n\ + gateway.cluster.default.bootstrap.servers: [fluss-1:9123, fluss-2:9123]\n\ + gateway.cluster.default.connection.identity-mode: user\n\ + gateway.cluster.default.connection.service.account: gateway_svc\n\ + gateway.cluster.default.connection.service.secret: gw-pass\n\ + gateway.cluster.default.client.security.protocol: sasl\n\ + gateway.cluster.default.connection.max: 512\n\ + gateway.cluster.default.connection.idle-timeout: 10m\n\ + gateway.cluster.analytics.bootstrap.servers: analytics:9123,analytics-2:9123\n\ + gateway.cluster.analytics.connect-timeout: 5s\n\ + gateway.security.authentication: password\n\ + gateway.security.users: alice:secret\n\ + gateway.rest.write.max-rows: 500\n\ + gateway.rest.lookup.max-keys: 32\n\ + gateway.rest.lookup.max-key-bytes: 2MiB\n", + ) + .unwrap(); + + let default = cluster(&config, "default"); + assert_eq!(default.bootstrap_servers, ["fluss-1:9123", "fluss-2:9123"]); + assert_eq!(default.identity_mode, IdentityMode::User); + assert_eq!(default.effective_service_account(), Some("gateway_svc")); + assert_eq!(default.effective_service_secret(), Some("gw-pass")); + assert_eq!(default.connection_max, Some(512)); + assert_eq!( + default.connection_idle_timeout.map(ConfigDuration::get), + Some(Duration::from_secs(600)) + ); + // A comma-separated string is accepted so the same value can arrive from the environment. + assert_eq!( + cluster(&config, "analytics").bootstrap_servers, + ["analytics:9123", "analytics-2:9123"] + ); + assert_eq!( + cluster(&config, "analytics").connect_timeout.get(), + Duration::from_secs(5) + ); + assert_eq!(config.security.authentication, AuthenticationMode::Password); + assert_eq!(config.request_limits.write_max_rows, 500); + assert_eq!(config.request_limits.lookup_max_keys, 32); + assert_eq!( + config.request_limits.lookup_max_key_bytes.bytes(), + 2 * 1024 * 1024 + ); +} + +/// The declaration bounds which clusters may be configured, and it does so whether or not it was +/// written: with no `gateway.clusters`, the only configurable cluster is the implicit `default`. That is +/// what turns a mistyped cluster ID into a startup failure instead of an unreachable second cluster. +#[test] +fn declared_clusters_are_authoritative() { + for contents in [ + "gateway.clusters: default\n\ + gateway.cluster.analytics.bootstrap.servers: analytics:9123\n", + // No declaration at all: `analytics` is still not one of the allowed clusters. + "gateway.cluster.analytics.bootstrap.servers: analytics:9123\n", + "gateway.cluster.analytics.client.writer.batch-size: 2MiB\n", + ] { + let error = load_file(contents).unwrap_err(); + assert!( + error.to_string().contains("not declared"), + "{contents}: {error}" + ); + } + + // Declaring a cluster is enough to configure it; the rest of its settings default. + let config = load_file("gateway.clusters: default,analytics\n").unwrap(); + assert_eq!(config.clusters.len(), 2); + assert_eq!( + cluster(&config, "analytics").bootstrap_servers, + [DEFAULT_BOOTSTRAP_SERVERS] + ); + + // The implicit default needs no declaration, so a single-cluster deployment configures no list. + let config = load_file("gateway.cluster.default.bootstrap.servers: only:9123\n").unwrap(); + assert_eq!(config.clusters.keys().collect::>(), ["default"]); +} + +#[test] +fn malformed_cluster_ids_are_rejected() { + for contents in [ + "gateway.clusters: Default\n", + "gateway.clusters: 1st\n", + "gateway.clusters: eu-west\n", + "gateway.cluster.EU.bootstrap.servers: eu:9123\n", + ] { + let error = load_file(contents).unwrap_err(); + assert!( + error.to_string().contains("cluster ID"), + "{contents}: {error}" + ); + } +} + +#[test] +fn unknown_cluster_and_client_keys_name_the_original_key() { + for contents in [ + "gateway.cluster.default.bootstrap.serverz: fluss:9123\n", + "gateway.cluster.default.connection.identity: user\n", + "gateway.cluster.default.client.: 1\n", + "gateway.cluster.default: fluss:9123\n", + ] { + let error = load_file(contents).unwrap_err(); + assert!(matches!(error, ConfigError::Parse(_)), "got: {error:?}"); + let key = contents.split(':').next().unwrap(); + assert!(error.to_string().contains(key), "{key}: {error}"); + } +} + +#[test] +fn native_client_options_are_parsed_into_their_native_types() { + assert_eq!( + parse_client_option("writer.batch-size", "2MiB").unwrap(), + ClientOptionValue::Bytes(2 * 1024 * 1024) + ); + assert_eq!( + parse_client_option("writer.batch-timeout", "50ms").unwrap(), + ClientOptionValue::Millis(50) + ); + assert_eq!( + parse_client_option("writer.dynamic-batch-size.enabled", "false").unwrap(), + ClientOptionValue::Boolean(false) + ); + assert_eq!( + parse_client_option("lookup.max-retries", "3").unwrap(), + ClientOptionValue::Integer(3) + ); + assert_eq!( + parse_client_option("security.protocol", "sasl").unwrap(), + ClientOptionValue::Text("sasl".to_string()) + ); + + let config = load_file( + "gateway.cluster.default.client.writer.batch-size: 2MiB\n\ + gateway.cluster.default.client.lookup.max-retries: 3\n", + ) + .unwrap(); + let default = cluster(&config, "default"); + assert_eq!(default.client_option("writer.batch-size"), Some("2MiB")); + assert_eq!(default.client_option("lookup.max-retries"), Some("3")); +} + +/// The gateway advertises the write guarantees, so the client options that would weaken them, and the +/// authorization ID that user mode supplies per request, are refused rather than silently honoured. +#[test] +fn client_options_cannot_override_gateway_owned_guarantees_or_the_identity() { + for (option, value, expected) in [ + ("writer.acks", "0", "owned by the Gateway"), + ("writer.retries", "0", "owned by the Gateway"), + ("writer.enable-idempotence", "false", "owned by the Gateway"), + ( + "security.sasl.authorization-id", + "alice", + "cannot be configured statically", + ), + ( + "writer.unknown-knob", + "1", + "is not a supported native-client option", + ), + ] { + let error = parse_client_option(option, value).unwrap_err(); + assert!(error.contains(expected), "{option}: {error}"); + + let problems = problems( + load_file(&format!( + "gateway.cluster.default.client.{option}: {value}\n" + )) + .unwrap_err(), + ); + assert!( + problems.iter().any(|problem| { + problem.starts_with("gateway.cluster.default.client.") && problem.contains(option) + }), + "{option}: {problems:?}" + ); + } +} + +#[test] +fn out_of_range_client_values_fail_before_startup() { + for (option, value) in [ + ("lookup.queue-size", "0"), + ("lookup.max-batch-size", "-1"), + ("lookup.max-retries", "-1"), + ("writer.request-max-size", "3GiB"), + ("writer.batch-timeout", "0s"), + ("writer.dynamic-batch-size.enabled", "maybe"), + ] { + let error = parse_client_option(option, value).unwrap_err(); + assert!(error.starts_with(&format!("client.{option}")), "{error}"); + assert!( + load_file(&format!( + "gateway.cluster.default.client.{option}: \"{value}\"\n" + )) + .is_err(), + "accepted client.{option} = {value}" + ); + } +} + +#[test] +fn cross_field_cluster_and_security_constraints_fail_before_startup() { + for contents in [ + // An account without its secret, and the reverse. + "gateway.cluster.default.connection.service.account: gateway_svc\n", + "gateway.cluster.default.connection.service.secret: gw-pass\n", + // User identity mode has no service account to authenticate the pool with. + "gateway.cluster.default.connection.identity-mode: user\n", + // SASL without credentials, and an unsupported protocol or mechanism. + "gateway.cluster.default.client.security.protocol: sasl\n", + "gateway.cluster.default.client.security.protocol: ssl\n", + "gateway.cluster.default.client.security.sasl.mechanism: SCRAM-SHA-256\n", + "gateway.cluster.default.connection.max: 0\n", + "gateway.cluster.default.bootstrap.servers: \" \"\n", + // The mode's credential table is missing. + "gateway.security.authentication: password\n", + "gateway.security.authentication: token\n", + "gateway.security.authentication: trusted-header\n\ + gateway.security.trusted-header.name: \"bad header\"\n", + "gateway.rest.lookup.max-keys: 0\n", + "gateway.rest.prefix-lookup.max-prefixes: 0\n", + ] { + assert!(load_file(contents).is_err(), "accepted: {contents}"); + } + + assert!( + load_file( + "gateway.cluster.default.connection.identity-mode: user\n\ + gateway.cluster.default.connection.service.account: gateway_svc\n\ + gateway.cluster.default.connection.service.secret: gw-pass\n\ + gateway.cluster.default.client.security.protocol: SASL\n\ + gateway.cluster.default.client.security.sasl.mechanism: PLAIN\n" + ) + .is_ok() + ); +} + +/// The legacy SASL options stay usable and keep winning, because silently changing which credential a +/// running deployment authenticates with would be worse than the deprecation. +#[test] +fn legacy_credentials_win_and_warn_once_per_cluster() { + let config = load_file( + "gateway.clusters: default,analytics\n\ + gateway.cluster.default.connection.service.account: canonical-user\n\ + gateway.cluster.default.connection.service.secret: canonical-secret\n\ + gateway.cluster.default.client.security.sasl.username: legacy-user\n\ + gateway.cluster.default.client.security.sasl.password: legacy-secret\n\ + gateway.cluster.analytics.client.security.sasl.username: other-user\n\ + gateway.cluster.analytics.client.security.sasl.password: other-secret\n", + ) + .unwrap(); + + let default = cluster(&config, "default"); + assert_eq!(default.effective_service_account(), Some("legacy-user")); + assert_eq!(default.effective_service_secret(), Some("legacy-secret")); + + let deprecations: Vec = config + .warnings() + .into_iter() + .filter(|warning| warning.contains("is deprecated")) + .collect(); + assert_eq!(deprecations.len(), 2, "one per cluster: {deprecations:?}"); + for warning in &deprecations { + for secret in [ + "canonical-user", + "canonical-secret", + "legacy-user", + "legacy-secret", + "other-user", + "other-secret", + ] { + assert!(!warning.contains(secret), "leaked {secret}: {warning}"); + } + } +} + +#[test] +fn pool_settings_warn_when_the_identity_mode_ignores_them() { + let config = load_file( + "gateway.cluster.default.connection.max: 8\n\ + gateway.cluster.default.connection.idle-timeout: 5m\n", + ) + .unwrap(); + assert!( + config + .warnings() + .iter() + .any(|warning| warning.contains("ignored because connection.identity-mode is service")), + "{:?}", + config.warnings() + ); +} + +#[test] +fn diagnostics_redact_every_credential_and_keep_the_identities() { + let config = load_file( + "gateway.cluster.default.connection.service.account: canonical-user\n\ + gateway.cluster.default.connection.service.secret: canonical-secret\n\ + gateway.cluster.default.client.security.sasl.username: legacy-user\n\ + gateway.cluster.default.client.security.sasl.password: legacy-secret\n\ + gateway.cluster.default.client.writer.batch-size: 4MiB\n\ + gateway.security.authentication: password\n\ + gateway.security.users: alice:user-secret\n\ + gateway.security.tokens: token-secret:alice\n", + ) + .unwrap(); + + for diagnostic in [config.redacted_debug(), format!("{config:?}")] { + for credential in [ + "canonical-secret", + "legacy-secret", + "user-secret", + "token-secret", + ] { + assert!( + !diagnostic.contains(credential), + "leaked {credential}: {diagnostic}" + ); + } + // The identities and the tuning stay readable: they are what an operator came to look at. + for readable in ["canonical-user", "legacy-user", "4MiB"] { + assert!(diagnostic.contains(readable), "{readable}: {diagnostic}"); + } + assert!(diagnostic.contains(REDACTED), "{diagnostic}"); + } + + // The credentials are still reachable by the components that authenticate with them. + assert_eq!( + config.security.users.as_ref().map(Secret::expose), + Some("alice:user-secret") + ); + assert_eq!( + cluster(&config, "default").effective_service_secret(), + Some("legacy-secret") + ); +} + +/// A configuration error is what an operator sees on stderr, so it must name the option without +/// quoting any credential the file happens to carry. +#[test] +fn configuration_errors_never_quote_a_credential() { + let file = write_temp_config( + "gateway.security.authentication: token\n\ + gateway.security.tokens: do-not-leak\n\ + gateway.cluster.default.connection.service.secret: also-secret\n\ + gateway.cluster.default.client.writer.acks: 0\n", + ); + let error = load(Some(file.path()), &no_env(), &CliOverrides::default()).unwrap_err(); + let rendered = error.to_string(); + assert!(rendered.contains("writer.acks"), "{rendered}"); + assert!(!rendered.contains("do-not-leak"), "{rendered}"); + assert!(!rendered.contains("also-secret"), "{rendered}"); +} + +/// The same precedence statement for the two dynamic namespaces, where the environment name is derived +/// rather than registered: for every per-cluster option and every allowed client option, setting the +/// environment variable must be indistinguishable from having written that value in the file. +#[test] +fn the_environment_overrides_the_file_for_every_cluster_and_client_option() { + // Keeps every variant loadable: user identity mode needs usable credentials over SASL. The key + // under test is removed from the base so the file never carries it twice. + let base = [ + ("connection.service.account", "base-account"), + ("connection.service.secret", "base-secret"), + ("client.security.protocol", "sasl"), + ]; + + let mut cases: Vec<(String, String, &str, &str)> = Vec::new(); + for entry in CLUSTER_ENTRIES { + let values = match entry.kind { + ValueKind::ServerList => ("file-host:9123", "env-host:9123"), + ValueKind::Integer => ("11", "22"), + ValueKind::Text if entry.key == "connection.identity-mode" => ("service", "user"), + ValueKind::Text if entry.key.ends_with("timeout") => ("11s", "22s"), + ValueKind::Text => ("file-value", "env-value"), + ValueKind::Bool | ValueKind::Bytes => { + unreachable!("no per-cluster option uses {:?}", entry.kind) + } + }; + cases.push(( + entry.key.to_string(), + environment_suffix(entry.key), + values.0, + values.1, + )); + } + for spec in CLIENT_OPTIONS { + let values = match spec.kind { + // The only legal value is PLAIN, so the two spellings differ only in case. + _ if spec.option == "security.sasl.mechanism" => ("PLAIN", "plain"), + _ if spec.option == "security.protocol" => ("plaintext", "sasl"), + ClientOptionKind::Text | ClientOptionKind::Secret => ("file-value", "env-value"), + ClientOptionKind::Boolean => ("true", "false"), + ClientOptionKind::Count { .. } => ("5", "6"), + // Both values have to keep the size relationships intact against the native defaults. + ClientOptionKind::Size { .. } if spec.option.ends_with("dynamic-batch-size.min") => { + ("1MiB", "2MiB") + } + ClientOptionKind::Size { .. } => ("4MiB", "8MiB"), + ClientOptionKind::Duration => ("11s", "22s"), + }; + cases.push(( + format!("{CLIENT_OPTION_PREFIX}{}", spec.option), + format!("CLIENT__{}", environment_suffix(spec.option)), + values.0, + values.1, + )); + } + + for (key, env_suffix, file_value, env_value) in cases { + let contents = |value: &str| { + base.iter() + .filter(|(base_key, _)| *base_key != key) + .map(|(base_key, base_value)| (*base_key, *base_value)) + .chain([(key.as_str(), value)]) + .map(|(key, value)| { + format!("{CLUSTER_KEY_PREFIX}{DEFAULT_CLUSTER_ID}.{key}: \"{value}\"\n") + }) + .collect::() + }; + let load_valid = |contents: &str, env: &BTreeMap| { + let file = write_temp_config(contents); + load(Some(file.path()), env, &CliOverrides::default()) + .unwrap_or_else(|error| panic!("{key}: {error}\n{contents}")) + }; + + let env = BTreeMap::from([( + format!( + "{ENV_PREFIX}CLUSTER__{}__{env_suffix}", + DEFAULT_CLUSTER_ID.to_ascii_uppercase() + ), + env_value.to_string(), + )]); + let from_file = load_valid(&contents(file_value), &no_env()); + let overridden = load_valid(&contents(file_value), &env); + let as_written = load_valid(&contents(env_value), &no_env()); + + assert_ne!( + from_file, overridden, + "{key} ignores its environment variable" + ); + assert_eq!( + overridden, as_written, + "{key} from the environment differs from the same value in the file" + ); + } +} + +/// The reserved options are refused from the environment exactly as they are from the file, and the +/// fixed request limits are reachable there too. +#[test] +fn the_environment_is_held_to_the_same_client_option_rules_as_the_file() { + let mut env = BTreeMap::from([( + "FLUSS_GATEWAY__REST__LOOKUP__MAX_KEYS".to_string(), + "16".to_string(), + )]); + let config = load(None, &env, &CliOverrides::default()).unwrap(); + assert_eq!(config.request_limits.lookup_max_keys, 16); + + env.insert( + "FLUSS_GATEWAY__CLUSTER__DEFAULT__CLIENT__WRITER__ACKS".to_string(), + "0".to_string(), + ); + let error = load(None, &env, &CliOverrides::default()).unwrap_err(); + assert!(error.to_string().contains("writer.acks"), "got: {error}"); +} + +#[test] +fn the_environment_can_declare_clusters() { + let file = write_temp_config("gateway.cluster.analytics.bootstrap.servers: eu:9123\n"); + let mut env = no_env(); + env.insert( + "FLUSS_GATEWAY__CLUSTERS".to_string(), + "analytics".to_string(), + ); + let config = load(Some(file.path()), &env, &CliOverrides::default()).unwrap(); + assert_eq!(config.clusters.keys().collect::>(), ["analytics"]); + + env.insert("FLUSS_GATEWAY__CLUSTERS".to_string(), "default".to_string()); + let error = load(Some(file.path()), &env, &CliOverrides::default()).unwrap_err(); + assert!(error.to_string().contains("not declared"), "got: {error}"); +} + +#[test] +fn unknown_cluster_environment_variables_are_rejected() { + for variable in [ + "FLUSS_GATEWAY__CLUSTER__DEFAULT__BOOTSTRAP__SERVERZ", + "FLUSS_GATEWAY__CLUSTER__DEFAULT", + "FLUSS_GATEWAY__CLUSTER__1ST__BOOTSTRAP__SERVERS", + ] { + let mut env = no_env(); + env.insert(variable.to_string(), "value".to_string()); + let error = load(None, &env, &CliOverrides::default()).unwrap_err(); + assert!( + matches!(error, ConfigError::UnknownEnvKey(_)), + "{variable}: {error:?}" + ); + assert!(error.to_string().contains(variable), "{variable}: {error}"); + } +} + +/// A bad per-cluster value is reported with the public key rather than the internal Serde path, and an +/// environment override additionally names the variable the operator set. +#[test] +fn a_bad_cluster_value_names_the_public_key_and_its_source() { + let error = load_file("gateway.cluster.default.request-timeout: 0s\n").unwrap_err(); + assert!( + error + .to_string() + .contains("gateway.cluster.default.request-timeout"), + "got: {error}" + ); + + let env = BTreeMap::from([( + "FLUSS_GATEWAY__CLUSTER__DEFAULT__CONNECT_TIMEOUT".to_string(), + "soon".to_string(), + )]); + let rendered = load(None, &env, &CliOverrides::default()) + .unwrap_err() + .to_string(); + assert!( + rendered.contains("FLUSS_GATEWAY__CLUSTER__DEFAULT__CONNECT_TIMEOUT"), + "{rendered}" + ); + assert!( + rendered.contains("gateway.cluster.default.connect-timeout"), + "{rendered}" + ); +} + +#[test] +fn programmatically_constructed_clusters_are_validated() { + let mut config = GatewayConfig::default(); + config.clusters.clear(); + let errors = problems(config.validate().unwrap_err()); + assert!( + errors + .iter() + .any(|error| error == "gateway.clusters must declare at least one cluster"), + "got: {errors:?}" + ); + + let mut config = GatewayConfig::default(); + config + .clusters + .get_mut(DEFAULT_CLUSTER_ID) + .expect("default cluster") + .identity_mode = IdentityMode::User; + let errors = problems(config.validate().unwrap_err()); + assert!( + errors + .iter() + .any(|error| error.contains("identity-mode user requires")), + "got: {errors:?}" + ); +} + +/// The `client.*` namespace is open, so its environment mapping cannot be checked key by key like the +/// fixed vocabulary. It round-trips only while option names keep `-` inside a segment and `.` between +/// segments: an option name containing `_` would come back from the environment as a different name and +/// silently configure the wrong option. Exhaustive over the allowlist, so adding such a name fails here. +#[test] +fn every_client_option_round_trips_through_the_environment() { + for spec in CLIENT_OPTIONS { + let suffix = environment_suffix(spec.option); + assert_eq!( + environment_suffix_to_option(&suffix), + spec.option, + "{} does not survive the environment mapping", + spec.option + ); + assert!( + !spec.option.contains('_'), + "{} must spell words with '-', which the environment mapping reserves '_' for", + spec.option + ); + let file_key = format!( + "{CLUSTER_KEY_PREFIX}{DEFAULT_CLUSTER_ID}.{CLIENT_OPTION_PREFIX}{}", + spec.option + ); + assert!( + matches!(resolve_key(&file_key), Ok(ResolvedKey::ClientOption { .. })), + "{file_key} is not reachable as a file key" + ); + } +} + +/// Sensitivity is declared per option, so the declaration is what has to be right. Fluss decides it on +/// the Java side from these same substrings, which is the cross-check: any option whose name looks like +/// a credential must be marked, and the allowlist and the refusals must not overlap. +#[test] +fn client_option_sensitivity_and_refusals_are_declared_consistently() { + for spec in CLIENT_OPTIONS { + let looks_sensitive = ["password", "secret", "token"] + .iter() + .any(|part| spec.option.contains(part)); + assert_eq!( + spec.kind.is_sensitive(), + looks_sensitive, + "{}: the kind and the name disagree about being a credential", + spec.option + ); + assert!( + !RESERVED_CLIENT_OPTIONS + .iter() + .any(|(reserved, _)| *reserved == spec.option), + "{} is both allowed and refused", + spec.option + ); + } + // An option the gateway never validated must not be rendered on the chance that it holds a secret. + assert!(client_option_is_sensitive("some.unknown.option")); + assert!(client_option_is_sensitive(LEGACY_SERVICE_SECRET_OPTION)); + assert!(!client_option_is_sensitive(LEGACY_SERVICE_ACCOUNT_OPTION)); + + // A parsed credential stays wrapped, so printing the parse result cannot leak it either. + let parsed = parse_client_option(LEGACY_SERVICE_SECRET_OPTION, "legacy-secret").unwrap(); + assert_eq!( + parsed, + ClientOptionValue::Secret(Secret::new("legacy-secret")) + ); + assert!(!format!("{parsed:?}").contains("legacy-secret")); + assert!(format!("{parsed:?}").contains(REDACTED)); +} + +/// Precedence is stated once for the whole vocabulary rather than sampled on one key, so a per-kind +/// conversion that only reads one source cannot hide: every option is driven from the file, then +/// overridden from the environment, and the environment value has to win in the loaded config. +#[test] +fn the_environment_overrides_the_file_for_every_option() { + for entry in CONFIG_ENTRIES { + let (file_value, env_value) = match entry.kind { + ValueKind::Bool => ("true", "false"), + ValueKind::Integer => ("11", "22"), + ValueKind::Bytes => ("1MiB", "2MiB"), + ValueKind::ServerList => ("file-host:9123", "env-host:9123"), + ValueKind::Text => match entry.key { + REST_LISTEN_KEY => ("127.0.0.1:11111", "127.0.0.1:22222"), + METRICS_LISTEN_KEY => ("127.0.0.1:11112", "127.0.0.1:22223"), + REST_HEADER_READ_TIMEOUT_KEY + | REST_REQUEST_TIMEOUT_KEY + | SHUTDOWN_DRAIN_TIMEOUT_KEY => ("11s", "22s"), + // Both modes must be valid on their own: the file value is loaded without the + // environment override, and password and token modes need a credential table. + SECURITY_AUTHENTICATION_KEY => ("trusted-header", "trust"), + _ => ("file-value", "env-value"), + }, + }; + + let file = write_temp_config(&format!("{}: \"{file_value}\"\n", entry.key)); + let from_file = load(Some(file.path()), &no_env(), &CliOverrides::default()) + .unwrap_or_else(|error| panic!("{}: {error}", entry.key)); + let env = BTreeMap::from([(environment_variable(entry.key), env_value.to_string())]); + let from_env = load(Some(file.path()), &env, &CliOverrides::default()) + .unwrap_or_else(|error| panic!("{}: {error}", entry.key)); + + assert_ne!( + from_file, from_env, + "{} ignores its environment variable", + entry.key + ); + let only_env = load(None, &env, &CliOverrides::default()) + .unwrap_or_else(|error| panic!("{}: {error}", entry.key)); + assert_eq!( + from_env, only_env, + "{} lets the file value survive the environment override", + entry.key + ); + } +} + +/// User identity mode is only safe when the connection can actually carry the request's principal, so +/// the credentials must be usable *and* SASL must be selected. Both were previously satisfied by a +/// `Some("")` credential over the default PLAINTEXT, which authorizes the gateway's own identity for +/// every caller instead of failing. +#[test] +fn user_identity_mode_requires_usable_credentials_over_sasl() { + let user_mode = "gateway.cluster.default.connection.identity-mode: user\n"; + let credentials = "gateway.cluster.default.connection.service.account: gateway_svc\n\ + gateway.cluster.default.connection.service.secret: gw-pass\n"; + let sasl = "gateway.cluster.default.client.security.protocol: sasl\n"; + + for (contents, expected) in [ + ( + format!("{user_mode}{credentials}"), + "requires client.security.protocol sasl", + ), + ( + format!("{user_mode}{sasl}"), + "requires connection.service.account", + ), + ( + format!( + "{user_mode}{sasl}gateway.cluster.default.connection.service.account: \"\"\n\ + gateway.cluster.default.connection.service.secret: \" \"\n" + ), + "must not be blank", + ), + ( + format!( + "{user_mode}{sasl}{credentials}\ + gateway.cluster.default.client.security.sasl.mechanism: SCRAM-SHA-256\n" + ), + "must be PLAIN", + ), + ] { + let problems = problems(load_file(&contents).unwrap_err()); + assert!( + problems.iter().any(|problem| problem.contains(expected)), + "expected {expected:?} for:\n{contents}got: {problems:?}" + ); + } + + // The complete, coherent form is accepted. + assert!(load_file(&format!("{user_mode}{sasl}{credentials}")).is_ok()); + // Service mode needs no SASL: it authenticates as itself, with no principal to propagate. + assert!(load_file("gateway.cluster.default.connection.identity-mode: service\n").is_ok()); +} + +/// A size that cannot fit inside the size holding it fails before a listener binds, whether the operator +/// set both sides or only one: leaving the other at its native default is the common way to break a pair. +#[test] +fn writer_size_pairs_must_fit_including_against_the_native_defaults() { + for (contents, rejected) in [ + // Both sides configured. + ( + "gateway.cluster.default.client.writer.batch-size: 2MiB\n\ + gateway.cluster.default.client.writer.request-max-size: 1MiB\n", + true, + ), + ( + "gateway.cluster.default.client.writer.dynamic-batch-size.min: 4MiB\n\ + gateway.cluster.default.client.writer.batch-size: 2MiB\n", + true, + ), + // One side only: the other is the native default, 10MiB request-max and 2MiB batch-size. + ( + "gateway.cluster.default.client.writer.batch-size: 128MiB\n", + true, + ), + ( + "gateway.cluster.default.client.writer.dynamic-batch-size.min: 3MiB\n", + true, + ), + ( + "gateway.cluster.default.client.writer.request-max-size: 1MiB\n", + true, + ), + // Coherent against the defaults, and coherent as a pair. + ( + "gateway.cluster.default.client.writer.batch-size: 4MiB\n", + false, + ), + ( + "gateway.cluster.default.client.writer.batch-size: 32MiB\n\ + gateway.cluster.default.client.writer.request-max-size: 64MiB\n", + false, + ), + ] { + let result = load_file(contents); + assert_eq!( + result.is_err(), + rejected, + "unexpected outcome for:\n{contents}" + ); + if rejected { + assert!( + problems(result.unwrap_err()) + .iter() + .any(|problem| problem.contains("must not exceed client.")), + "{contents}" + ); + } + } +} + +/// A value is bounded by the native field it lands in, not by one blanket ceiling: the writer sizes are +/// `i32` there, while the buffer size and the lookup counts are `usize` and may exceed `i32::MAX`. +#[test] +fn client_option_bounds_follow_the_native_field_type() { + let over_i32 = u64::from(i32::MAX as u32) + 1; + + for option in ["writer.batch-size", "writer.request-max-size"] { + let error = parse_client_option(option, &format!("{over_i32}")).unwrap_err(); + assert!(error.contains("must not exceed"), "{option}: {error}"); + } + assert_eq!( + parse_client_option("writer.buffer.memory-size", "4GiB").unwrap(), + ClientOptionValue::Bytes(4 * 1024 * 1024 * 1024) + ); + for option in [ + "lookup.queue-size", + "lookup.max-batch-size", + "lookup.max-inflight-requests", + ] { + assert_eq!( + parse_client_option(option, &format!("{over_i32}")).unwrap(), + ClientOptionValue::Integer(over_i32), + "{option} is stored as usize and must accept this" + ); + } + let error = parse_client_option("lookup.max-retries", &format!("{over_i32}")).unwrap_err(); + assert!(error.contains("must be between 0 and"), "{error}"); +} + +/// The declared native defaults are a copy of the client's, so they must at least satisfy the +/// relationships the client enforces; a mistyped copy shows up here and not as a rejected valid file. +#[test] +fn the_declared_native_size_defaults_are_coherent() { + let default = |option| effective_size(option, None).expect("a declared size default"); + let batch = default("writer.batch-size"); + assert!(batch <= default("writer.request-max-size")); + assert!(batch <= default("writer.buffer.memory-size")); + assert!(default("writer.dynamic-batch-size.min") <= batch); +} + +#[test] +fn options_are_complete_and_unambiguous() { + let mut public_keys = std::collections::BTreeSet::new(); + let mut internal_paths = std::collections::BTreeSet::new(); + let mut environment_variables = std::collections::BTreeSet::new(); + + for entry in CONFIG_ENTRIES { + assert!(entry.key.starts_with("gateway."), "{entry:?}"); + assert!( + public_keys.insert(entry.key), + "duplicate key: {}", + entry.key + ); + assert!( + internal_paths.insert(entry.internal_path), + "duplicate path: {}", + entry.internal_path + ); + assert!( + environment_variables.insert(environment_variable(entry.key)), + "duplicate environment variable for {}", + entry.key + ); + } + + assert_eq!(CONFIG_ENTRIES.len(), 20); +} + +/// The per-cluster vocabulary shares the environment namespace with `client.*`, so its keys have to +/// stay distinct from each other and unreachable through the client prefix. +#[test] +fn cluster_options_are_complete_and_unambiguous() { + let mut keys = std::collections::BTreeSet::new(); + let mut fields = std::collections::BTreeSet::new(); + let mut suffixes = std::collections::BTreeSet::new(); + + for entry in CLUSTER_ENTRIES { + assert!(!entry.key.starts_with("gateway."), "{entry:?}"); + assert!( + !entry.key.starts_with(CLIENT_OPTION_PREFIX), + "{} collides with the client namespace", + entry.key + ); + assert!(keys.insert(entry.key), "duplicate key: {}", entry.key); + assert!( + fields.insert(entry.internal_path), + "duplicate field: {}", + entry.internal_path + ); + assert!( + suffixes.insert(environment_suffix(entry.key)), + "duplicate environment suffix for {}", + entry.key + ); + } + + assert_eq!(CLUSTER_ENTRIES.len(), 8); +} diff --git a/fluss-gateway/src/lifecycle.rs b/fluss-gateway/src/lifecycle.rs index 03a304e1c3..4d810304e8 100644 --- a/fluss-gateway/src/lifecycle.rs +++ b/fluss-gateway/src/lifecycle.rs @@ -245,6 +245,9 @@ pub async fn start(config: GatewayConfig) -> Result { /// Binds the listeners, installs the router, and spawns every process-owned task. async fn start_internal(config: GatewayConfig) -> Result { + // The effective configuration is what an operator needs when a deployment misbehaves, and the + // redacted rendering is the only form of it that is safe to write to a log. + log::debug!("effective configuration: {}", config.redacted_debug()); for warning in config.warnings() { log::warn!("{warning}"); } diff --git a/fluss-gateway/tests/process.rs b/fluss-gateway/tests/process.rs index 80ca730d99..346346ad5b 100644 --- a/fluss-gateway/tests/process.rs +++ b/fluss-gateway/tests/process.rs @@ -23,6 +23,7 @@ mod support; +use std::io::Read; use std::time::Duration; use support::{ChildGuard, await_http_ok, binary, free_port, write_config}; @@ -40,6 +41,98 @@ async fn an_invalid_configuration_fails_before_binding_with_exit_code_2() { ); } +/// A rejected `client.*` option is what an operator hits when migrating a native-client configuration, and +/// the file that carries it usually carries credentials too: the process must name the option on stderr +/// while leaking none of them. +#[tokio::test] +async fn a_restricted_client_option_fails_before_binding_without_leaking_credentials() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("gateway.yaml"); + std::fs::write( + &path, + "gateway.cluster.default.connection.service.account: gateway-user\n\ + gateway.cluster.default.connection.service.secret: canonical-secret\n\ + gateway.cluster.default.client.security.sasl.password: legacy-secret\n\ + gateway.security.authentication: token\n\ + gateway.security.tokens: token-secret:alice\n\ + gateway.cluster.default.client.writer.acks: 0\n", + ) + .expect("write"); + + let output = binary().arg("--config").arg(&path).output().expect("run"); + assert_eq!(output.status.code(), Some(2)); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("writer.acks"), "{stderr}"); + for credential in ["canonical-secret", "legacy-secret", "token-secret"] { + assert!( + !stderr.contains(credential), + "stderr leaked {credential}: {stderr}" + ); + } +} + +/// The deprecated SASL options keep working, so the only signal an operator gets is the startup warning: +/// it has to reach stderr, once per cluster, with the credentials redacted. Debug logging is on, so the +/// effective-configuration dump is on stderr too and is held to the same rule. +#[tokio::test] +async fn legacy_credentials_emit_one_redacted_warning_per_cluster() { + let dir = tempfile::tempdir().expect("tempdir"); + let port = free_port(); + let path = dir.path().join("gateway.yaml"); + std::fs::write( + &path, + format!( + "gateway.rest.listen: 127.0.0.1:{port}\n\ + gateway.metrics.enabled: false\n\ + gateway.cluster.default.connection.service.account: canonical-user\n\ + gateway.cluster.default.connection.service.secret: canonical-secret\n\ + gateway.cluster.default.client.security.sasl.username: legacy-user\n\ + gateway.cluster.default.client.security.sasl.password: legacy-secret\n\ + gateway.security.authentication: token\n\ + gateway.security.tokens: token-secret:alice\n" + ), + ) + .expect("write"); + + // Only the few startup log lines are piped, so the pipe buffer cannot fill and stall the child. + let child = binary() + .arg("--config") + .arg(&path) + .env("RUST_LOG", "debug") + .stderr(std::process::Stdio::piped()) + .spawn() + .expect("spawn"); + let mut guard = ChildGuard(child); + assert!( + await_http_ok( + &format!("http://127.0.0.1:{port}/health"), + Duration::from_secs(15) + ) + .await, + "health" + ); + guard.send_sigterm(); + let status = guard.wait_for_exit(Duration::from_secs(35)).await; + assert_eq!(status.code(), Some(0)); + + let mut stderr = String::new(); + guard + .0 + .stderr + .take() + .expect("piped stderr") + .read_to_string(&mut stderr) + .expect("read stderr"); + assert_eq!(stderr.matches("is deprecated").count(), 1, "{stderr}"); + assert!(stderr.contains("effective configuration"), "{stderr}"); + for credential in ["canonical-secret", "legacy-secret", "token-secret"] { + assert!( + !stderr.contains(credential), + "stderr leaked {credential}: {stderr}" + ); + } +} + #[tokio::test] async fn the_binary_starts_serves_health_and_drains_on_sigterm_with_exit_code_0() { let dir = tempfile::tempdir().expect("tempdir");