use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize)]
#[serde(deny_unknown_fields)]
pub struct Upstream {
pub name: String,
pub addresses: Vec<AddressSpec>,
#[serde(default)]
pub tls: Option<UpstreamTls>,
#[serde(default)]
pub lb_algorithm: LbAlgorithm,
#[serde(default)]
pub hash: Option<HashConfig>,
pub health: HealthConfig,
#[serde(default)]
pub resolve_interval_ms: u64,
#[serde(default = "default_request_timeout_ms")]
pub request_timeout_ms: u64,
#[serde(default = "default_max_retries")]
pub max_retries: u64,
#[serde(default)]
pub overall_timeout_ms: u64,
#[serde(default)]
pub circuit_breaker: CircuitBreaker,
#[serde(default)]
pub outlier_detection: OutlierDetection,
}
#[derive(Debug, Clone, Default, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct UpstreamTls {
#[serde(default)]
pub ca_path: Option<String>,
#[serde(default)]
pub sni: Option<String>,
#[serde(default)]
pub client_cert_path: Option<String>,
#[serde(default)]
pub client_key_path: Option<String>,
}
#[derive(
Debug, Clone, Copy, Default, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq,
)]
#[serde(rename_all = "snake_case")]
pub enum LbAlgorithm {
#[default]
RoundRobin,
LeastRequest,
Maglev,
}
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, PartialEq, Eq)]
#[serde(untagged)]
pub enum AddressSpec {
Bare(String),
Weighted(WeightedAddress),
}
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WeightedAddress {
pub address: String,
#[serde(default = "default_instance_weight")]
pub weight: u32,
}
impl AddressSpec {
pub fn address(&self) -> &str {
match self {
AddressSpec::Bare(a) => a,
AddressSpec::Weighted(w) => &w.address,
}
}
pub fn weight(&self) -> u32 {
match self {
AddressSpec::Bare(_) => default_instance_weight(),
AddressSpec::Weighted(w) => w.weight,
}
}
}
impl Serialize for AddressSpec {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
match self {
AddressSpec::Bare(addr) => serializer.serialize_str(addr),
AddressSpec::Weighted(w) if w.weight == default_instance_weight() => {
serializer.serialize_str(&w.address)
}
AddressSpec::Weighted(w) => w.serialize(serializer),
}
}
}
fn default_instance_weight() -> u32 {
1
}
pub(crate) const MAX_INSTANCE_WEIGHT: u32 = 1000;
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct HashConfig {
pub key: HashKeyKind,
#[serde(default)]
pub header: Option<String>,
#[serde(default = "default_hash_table_size")]
pub table_size: u32,
}
#[derive(Debug, Clone, Copy, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum HashKeyKind {
Header,
SourceIp,
}
fn default_hash_table_size() -> u32 {
65537
}
pub(crate) const MAX_HASH_TABLE_SIZE: u32 = 5_000_011;
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize)]
#[serde(deny_unknown_fields)]
pub struct OutlierDetection {
#[serde(default)]
pub consecutive_gateway_failures: u32,
#[serde(default = "default_base_ejection_time_ms")]
pub base_ejection_time_ms: u64,
#[serde(default = "default_max_ejection_percent")]
pub max_ejection_percent: u32,
}
impl Default for OutlierDetection {
fn default() -> Self {
Self {
consecutive_gateway_failures: 0,
base_ejection_time_ms: default_base_ejection_time_ms(),
max_ejection_percent: default_max_ejection_percent(),
}
}
}
fn default_base_ejection_time_ms() -> u64 {
30_000
}
fn default_max_ejection_percent() -> u32 {
10
}
#[derive(Debug, Clone, Default, Deserialize, schemars::JsonSchema, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CircuitBreaker {
#[serde(default)]
pub max_requests: u32,
}
#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct HealthConfig {
pub path: String,
#[serde(default = "default_interval_ms")]
pub interval_ms: u64,
#[serde(default = "default_timeout_ms")]
pub timeout_ms: u64,
#[serde(default = "default_healthy_threshold")]
pub healthy_threshold: u32,
#[serde(default = "default_unhealthy_threshold")]
pub unhealthy_threshold: u32,
#[serde(default)]
pub port: Option<u16>,
}
fn default_request_timeout_ms() -> u64 {
30_000
}
fn default_max_retries() -> u64 {
1
}
fn default_interval_ms() -> u64 {
2000
}
fn default_timeout_ms() -> u64 {
1000
}
fn default_healthy_threshold() -> u32 {
2
}
fn default_unhealthy_threshold() -> u32 {
3
}
#[cfg(test)]
mod tests {
use super::*;
use crate::error::ControlError;
use crate::manifest::Manifest;
#[test]
fn upstream_circuit_breaker_defaults_unlimited_and_parses() {
let m = Manifest::from_toml(
r#"
[[upstream]]
name = "a"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
"#,
)
.unwrap();
assert_eq!(
m.upstreams[0].circuit_breaker.max_requests, 0,
"an absent breaker is unlimited"
);
let m2 = Manifest::from_toml(
r#"
[[upstream]]
name = "a"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
[upstream.circuit_breaker]
max_requests = 64
"#,
)
.unwrap();
assert_eq!(m2.upstreams[0].circuit_breaker.max_requests, 64);
}
#[test]
fn upstream_tls_sni_defaults_absent_and_parses() {
let without = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
[upstream.tls]
"#,
)
.unwrap();
assert_eq!(without.upstreams[0].tls.as_ref().unwrap().sni, None);
let with = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
[upstream.tls]
sni = "backend.internal"
"#,
)
.unwrap();
assert_eq!(
with.upstreams[0].tls.as_ref().unwrap().sni.as_deref(),
Some("backend.internal")
);
assert_ne!(
without.content_hash().unwrap(),
with.content_hash().unwrap(),
"declaring sni must flip the content hash"
);
}
#[test]
fn upstream_requires_health() {
let parsed = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
"#,
);
assert!(
parsed.is_err(),
"an upstream without [upstream.health] is rejected"
);
}
#[test]
fn health_requires_path_but_defaults_the_rest() {
let no_path = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
interval_ms = 500
"#,
);
assert!(no_path.is_err(), "health without a probe path is rejected");
let m = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/up"
interval_ms = 250
timeout_ms = 100
healthy_threshold = 1
unhealthy_threshold = 5
"#,
)
.unwrap();
let h = &m.upstreams[0].health;
assert_eq!(h.interval_ms, 250);
assert_eq!(h.timeout_ms, 100);
assert_eq!(h.healthy_threshold, 1);
assert_eq!(h.unhealthy_threshold, 5);
}
#[test]
fn health_port_defaults_to_none_but_can_override_the_probe_port() {
let default_port = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
"#,
)
.unwrap();
assert_eq!(default_port.upstreams[0].health.port, None);
let overridden = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
port = 9100
"#,
)
.unwrap();
assert_eq!(overridden.upstreams[0].health.port, Some(9100));
}
#[test]
fn upstream_request_timeout_defaults_and_overrides() {
let defaulted = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
"#,
)
.unwrap();
assert_eq!(
defaulted.upstreams[0].request_timeout_ms, 30_000,
"an unspecified upstream timeout defaults to 30s"
);
let overridden = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
request_timeout_ms = 250
[upstream.health]
path = "/healthz"
"#,
)
.unwrap();
assert_eq!(overridden.upstreams[0].request_timeout_ms, 250);
assert_ne!(
defaulted.content_hash().unwrap(),
overridden.content_hash().unwrap(),
"changing the upstream timeout must flip the config version"
);
let disabled = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
request_timeout_ms = 0
[upstream.health]
path = "/healthz"
"#,
)
.unwrap();
assert_eq!(
disabled.upstreams[0].request_timeout_ms, 0,
"request_timeout_ms = 0 disables the timeout and must parse"
);
}
#[test]
fn upstream_max_retries_defaults_and_overrides() {
let defaulted = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
[upstream.health]
path = "/healthz"
"#,
)
.unwrap();
assert_eq!(
defaulted.upstreams[0].max_retries, 1,
"an unspecified upstream defaults to one retry"
);
let disabled = Manifest::from_toml(
r#"
[[upstream]]
name = "x"
addresses = ["127.0.0.1:9000"]
max_retries = 0
[upstream.health]
path = "/healthz"
"#,
)
.unwrap();
assert_eq!(disabled.upstreams[0].max_retries, 0, "0 disables retry");
assert_ne!(
defaulted.content_hash().unwrap(),
disabled.content_hash().unwrap(),
"changing max_retries must flip the config version"
);
}
fn upstream_toml(body: &str) -> Result<Manifest, ControlError> {
Manifest::from_toml(&format!(
"[[upstream]]\nname = \"u\"\n{body}\n[upstream.health]\npath = \"/healthz\"\n"
))
}
#[test]
fn lb_algorithm_defaults_round_robin_and_parses() {
let m = upstream_toml("addresses = [\"a:1\"]").unwrap();
assert_eq!(m.upstreams[0].lb_algorithm, LbAlgorithm::RoundRobin);
let lr = upstream_toml("addresses = [\"a:1\"]\nlb_algorithm = \"least_request\"").unwrap();
assert_eq!(lr.upstreams[0].lb_algorithm, LbAlgorithm::LeastRequest);
assert_ne!(
m.content_hash().unwrap(),
lr.content_hash().unwrap(),
"changing lb_algorithm must flip the config version"
);
}
#[test]
fn addresses_parse_bare_and_weighted_mixed() {
let m = upstream_toml(
"addresses = [\"a:1\", { address = \"b:2\", weight = 5 }, { address = \"c:3\" }]",
)
.unwrap();
let a = &m.upstreams[0].addresses;
assert_eq!(a[0].address(), "a:1");
assert_eq!(a[0].weight(), 1, "bare = weight 1");
assert_eq!(a[1].address(), "b:2");
assert_eq!(a[1].weight(), 5);
assert_eq!(a[2].address(), "c:3");
assert_eq!(a[2].weight(), 1, "weighted form defaults weight to 1");
}
#[test]
fn explicit_weight_one_hashes_like_a_bare_address() {
let bare = upstream_toml("addresses = [\"a:1\"]").unwrap();
let explicit = upstream_toml("addresses = [{ address = \"a:1\", weight = 1 }]").unwrap();
assert_eq!(
bare.content_hash().unwrap(),
explicit.content_hash().unwrap(),
"an explicit default weight must not change the config version"
);
let weighted = upstream_toml("addresses = [{ address = \"a:1\", weight = 3 }]").unwrap();
assert_ne!(
bare.content_hash().unwrap(),
weighted.content_hash().unwrap()
);
}
#[test]
fn maglev_hash_config_parses() {
let m = upstream_toml(
"addresses = [\"a:1\", \"b:2\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"header\"\nheader = \"x-user\"\ntable_size = 1009",
)
.unwrap();
let up = &m.upstreams[0];
assert_eq!(up.lb_algorithm, LbAlgorithm::Maglev);
let hash = up.hash.as_ref().unwrap();
assert_eq!(hash.key, HashKeyKind::Header);
assert_eq!(hash.header.as_deref(), Some("x-user"));
assert_eq!(hash.table_size, 1009);
assert!(up.validate_lb().is_ok());
}
#[test]
fn hash_table_size_defaults_to_65537() {
let m = upstream_toml(
"addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"",
)
.unwrap();
assert_eq!(m.upstreams[0].hash.as_ref().unwrap().table_size, 65537);
assert_eq!(
m.upstreams[0].hash.as_ref().unwrap().key,
HashKeyKind::SourceIp
);
}
#[test]
fn validate_lb_rejects_bad_configs() {
let w0 = upstream_toml("addresses = [{ address = \"a:1\", weight = 0 }]").unwrap();
assert!(w0.upstreams[0].validate_lb().is_err(), "weight 0 rejected");
let wbig = upstream_toml(&format!(
"addresses = [{{ address = \"a:1\", weight = {} }}]",
MAX_INSTANCE_WEIGHT + 1
))
.unwrap();
assert!(
wbig.upstreams[0].validate_lb().is_err(),
"over-cap weight rejected"
);
let no_hash = upstream_toml("addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"").unwrap();
assert!(
no_hash.upstreams[0].validate_lb().is_err(),
"maglev needs [upstream.hash]"
);
let stray_hash =
upstream_toml("addresses = [\"a:1\"]\n[upstream.hash]\nkey = \"source_ip\"").unwrap();
assert!(
stray_hash.upstreams[0].validate_lb().is_err(),
"[upstream.hash] only valid with maglev"
);
let no_header = upstream_toml(
"addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"header\"",
)
.unwrap();
assert!(
no_header.upstreams[0].validate_lb().is_err(),
"header key needs a name"
);
let nonprime = upstream_toml(
"addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"\ntable_size = 1000",
)
.unwrap();
assert!(
nonprime.upstreams[0].validate_lb().is_err(),
"table_size must be prime"
);
let too_small = upstream_toml(
"addresses = [\"a:1\", \"b:2\", \"c:3\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"\ntable_size = 2",
)
.unwrap();
assert!(
too_small.upstreams[0].validate_lb().is_err(),
"M must be >= instance count"
);
let ok = upstream_toml(
"addresses = [\"a:1\", \"b:2\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"\ntable_size = 97",
)
.unwrap();
assert!(ok.upstreams[0].validate_lb().is_ok());
let interval0 = Manifest::from_toml(
r#"
[[upstream]]
name = "u"
addresses = ["a:1"]
[upstream.health]
path = "/healthz"
interval_ms = 0
timeout_ms = 1000
"#,
)
.unwrap();
assert!(
interval0.upstreams[0].validate_lb().is_err(),
"interval_ms = 0 rejected"
);
let timeout0 = Manifest::from_toml(
r#"
[[upstream]]
name = "u"
addresses = ["a:1"]
[upstream.health]
path = "/healthz"
interval_ms = 1000
timeout_ms = 0
"#,
)
.unwrap();
assert!(
timeout0.upstreams[0].validate_lb().is_err(),
"timeout_ms = 0 rejected"
);
}
}