0.29.7
This commit is contained in:
1 parent
ad95bfb050
commit
1431dd47f0
62 files changed
+1360
-198
No files matched your search
@@ -345,8 +345,10 @@ pub async fn health_loop(app: Weak<crate::runtime::App>, shutdown: CancellationT
|
||||
_ = interval.tick() => {
|
||||
let Some(app) = app.upgrade() else { break; };
|
||||
let snapshot = app.config.load_full();
|
||||
// Own an Arc for each candidate. JoinSet tasks require an
|
||||
// owned, 'static member rather than a reference into snapshot.
|
||||
let members: Vec<_> = snapshot.pools.iter().flat_map(|pool|
|
||||
pool.members.iter().map(move |member| (member, &pool.settings.health_check))).collect();
|
||||
pool.members.iter().map(move |member| (Arc::clone(member), pool.settings.health_for(&member.definition)))).collect();
|
||||
for (member, _) in &members {
|
||||
member.state.sample_traffic();
|
||||
// Time-based pool recycling must not depend on receiving a
|
||||
@@ -361,9 +363,9 @@ pub async fn health_loop(app: Weak<crate::runtime::App>, shutdown: CancellationT
|
||||
for offset in 0..members.len() {
|
||||
if checks.len() >= 16 { break; }
|
||||
let index = (start + offset) % members.len();
|
||||
let (member, health) = members[index];
|
||||
let (member, health) = &members[index];
|
||||
if member.definition.enabled && member.state.claim_probe(health) {
|
||||
checks.spawn(probe(member.clone(), health.clone()));
|
||||
checks.spawn(probe(Arc::clone(member), health.clone()));
|
||||
cursor = (index + 1) % members.len();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -216,6 +216,25 @@ impl Default for Settings {
|
||||
}
|
||||
}
|
||||
impl Settings {
|
||||
/// Resolve active health per member; a disabled monitor never probes.
|
||||
pub fn health_for(&self, member: &PoolMember) -> HealthCheck {
|
||||
let mut health = member.health_check.clone().unwrap_or_else(|| self.health_check.clone());
|
||||
if health.host.is_none() && health.mode == HealthMode::Http {
|
||||
health.host = self.health_check.host.clone();
|
||||
}
|
||||
if !member.health_check_enabled {
|
||||
health.mode = HealthMode::Off;
|
||||
}
|
||||
health
|
||||
}
|
||||
/// Monitoring off also excludes passive ejection; failures remain logged.
|
||||
pub fn passive_for(&self, member: &PoolMember) -> PassiveHealth {
|
||||
let mut passive = self.passive_health.clone();
|
||||
if !member.health_check_enabled {
|
||||
passive.enabled = false;
|
||||
}
|
||||
passive
|
||||
}
|
||||
pub fn validate(&self) -> Result<()> {
|
||||
let h = &self.health_check;
|
||||
let k = &self.keepalive;
|
||||
|
||||
+12
-12
@@ -21,7 +21,7 @@ impl Pool {
|
||||
.zip(backends)
|
||||
.map(|(definition, backend)| {
|
||||
Arc::new(Member {
|
||||
state: Arc::new(BackendState::new(settings.health_check.mode)),
|
||||
state: Arc::new(BackendState::new(settings.health_for(&definition).mode)),
|
||||
definition,
|
||||
backend,
|
||||
token: rand::thread_rng()
|
||||
@@ -54,13 +54,13 @@ impl Pool {
|
||||
let same_origin = serde_json::to_vec(&member.definition.backend()).ok()
|
||||
== serde_json::to_vec(&old.definition.backend()).ok();
|
||||
let same_health =
|
||||
serde_json::to_vec(&(&self.settings.health_check, &self.settings.passive_health))
|
||||
.ok()
|
||||
== serde_json::to_vec(&(
|
||||
&previous.settings.health_check,
|
||||
&previous.settings.passive_health,
|
||||
))
|
||||
.ok();
|
||||
serde_json::to_vec(&(
|
||||
self.settings.health_for(&member.definition),
|
||||
self.settings.passive_for(&member.definition),
|
||||
)).ok() == serde_json::to_vec(&(
|
||||
previous.settings.health_for(&old.definition),
|
||||
previous.settings.passive_for(&old.definition),
|
||||
)).ok();
|
||||
let same_clients = member.backend.client.reusable_with(&old.backend.client)
|
||||
&& member
|
||||
.backend
|
||||
@@ -195,8 +195,8 @@ impl Pool {
|
||||
Some(Selection {
|
||||
lease: Arc::new(Lease {
|
||||
member: member.clone(),
|
||||
health: self.settings.health_check.clone(),
|
||||
passive_health: self.settings.passive_health.clone(),
|
||||
health: self.settings.health_for(&member.definition),
|
||||
passive_health: self.settings.passive_for(&member.definition),
|
||||
error_recorded: AtomicBool::new(false),
|
||||
}),
|
||||
cookie: None,
|
||||
@@ -336,8 +336,8 @@ impl Pool {
|
||||
Some(Selection {
|
||||
lease: Arc::new(Lease {
|
||||
member: selected.clone(),
|
||||
health: self.settings.health_check.clone(),
|
||||
passive_health: self.settings.passive_health.clone(),
|
||||
health: self.settings.health_for(&selected.definition),
|
||||
passive_health: self.settings.passive_for(&selected.definition),
|
||||
error_recorded: AtomicBool::new(false),
|
||||
}),
|
||||
cookie,
|
||||
|
||||
@@ -6,13 +6,14 @@ impl Pool {
|
||||
.map(|m| {
|
||||
let h = lock(&m.state.health);
|
||||
let failure = lock(&m.state.last_failure);
|
||||
let member_health = self.settings.health_for(&m.definition);
|
||||
let status = if !m.definition.enabled {
|
||||
"disabled"
|
||||
} else if m.definition.drain {
|
||||
"draining"
|
||||
} else if h.healthy {
|
||||
"up"
|
||||
} else if !h.checked && self.settings.health_check.mode != HealthMode::Off {
|
||||
} else if !h.checked && member_health.mode != HealthMode::Off {
|
||||
"checking"
|
||||
} else {
|
||||
"down"
|
||||
@@ -27,6 +28,8 @@ impl Pool {
|
||||
item.insert("url".into(), json!(m.definition.url));
|
||||
item.insert("weight".into(), json!(m.definition.weight));
|
||||
item.insert("backup".into(), json!(m.definition.backup));
|
||||
item.insert("health_check_enabled".into(), json!(m.definition.health_check_enabled));
|
||||
item.insert("health_mode".into(), json!(member_health.mode));
|
||||
item.insert(
|
||||
"max_concurrent_requests".into(),
|
||||
json!(m.definition.max_concurrent_requests),
|
||||
|
||||
@@ -52,6 +52,29 @@ fn select(pool: &Pool) -> Selection {
|
||||
.unwrap()
|
||||
}
|
||||
#[test]
|
||||
fn per_backend_monitoring_can_be_disabled_or_overridden() {
|
||||
let mut defaults: crate::vhost::PoolMember = serde_json::from_value(serde_json::json!({
|
||||
"id": "error_404", "url": "http://127.0.0.1:8282", "pool_eligible": false
|
||||
})).unwrap();
|
||||
let mut settings = Settings::default();
|
||||
settings.passive_health.enabled = true;
|
||||
assert!(defaults.health_check_enabled);
|
||||
assert_eq!(settings.health_for(&defaults).mode, HealthMode::Tcp);
|
||||
assert!(settings.passive_for(&defaults).enabled);
|
||||
defaults.health_check_enabled = false;
|
||||
assert_eq!(settings.health_for(&defaults).mode, HealthMode::Off);
|
||||
assert!(!settings.passive_for(&defaults).enabled);
|
||||
assert!(!defaults.pool_eligible); // routing intent is independent
|
||||
defaults.health_check_enabled = true;
|
||||
defaults.health_check = Some(HealthCheck { mode: HealthMode::Http, path: "/healthz".into(), ..Default::default() });
|
||||
let effective = settings.health_for(&defaults);
|
||||
assert_eq!(effective.mode, HealthMode::Http);
|
||||
assert_eq!(effective.path, "/healthz");
|
||||
assert_eq!(effective.host, settings.health_check.host);
|
||||
let restored: crate::vhost::PoolMember = serde_json::from_str(&serde_json::to_string(&defaults).unwrap()).unwrap();
|
||||
assert_eq!(settings.health_for(&restored).path, "/healthz");
|
||||
}
|
||||
#[test]
|
||||
fn default_health_check_uses_tcp() {
|
||||
assert_eq!(HealthMode::default(), HealthMode::Tcp);
|
||||
assert_eq!(HealthCheck::default().mode, HealthMode::Tcp);
|
||||
|
||||
Reference in new issue
Block a user