use super::{
ProcessHandle, ProcessObserver, ProcessStatus, ReadinessAttemptEvidence, ReadinessEvidence,
ReadinessFailure, ReadinessFailureKind, ReadinessObserver, SystemProcessRuntime,
TargetRegistryMatchEvidence,
};
use crate::interrupt;
use crate::operation_bound::{AttemptBound, OperationBound, OperationTerminalCause, Remaining};
use crate::plan::{ProcessEndpointPlan, ReadinessPlan, TargetRegistryExpectedTarget};
use std::thread;
use std::time::Duration;
const POLL_INTERVAL: Duration = Duration::from_millis(100);
const MAX_PROBE_INTERVAL: Duration = Duration::from_secs(5);
pub(super) const READINESS_ATTEMPT_CAP: Duration = Duration::from_millis(250);
#[derive(Debug, thiserror::Error)]
pub(super) enum ReadinessProbeError {
#[error("readiness operation deadline expired")]
Deadline,
#[error("bounded readiness attempt was unexpectedly unbounded")]
UnexpectedUnbounded,
#[error("{label} request failed: {source}")]
Request {
label: String,
#[source]
source: reqwest::Error,
},
#[error("{label} returned HTTP {status}")]
HttpStatus { label: String, status: u16 },
#[error("{label} returned invalid JSON: {source}")]
InvalidJson {
label: String,
#[source]
source: serde_json::Error,
},
#[error("target registry observation mismatch: {details}")]
RegistryMismatch { details: String },
}
pub(super) fn ensure_alive(status: ProcessStatus) -> Result<(), ReadinessFailure> {
if !status.queried {
return Err(readiness_failure(
ReadinessFailureKind::Exited,
status
.error
.unwrap_or_else(|| "failed to query server process group".to_owned()),
));
}
if !status.alive {
return Err(readiness_failure(
ReadinessFailureKind::Exited,
status
.error
.unwrap_or_else(|| "server process group exited before readiness".to_owned()),
));
}
Ok(())
}
fn readiness_failure(kind: ReadinessFailureKind, message: String) -> ReadinessFailure {
ReadinessFailure {
kind,
message,
timing: None,
diagnostic_attempts: Vec::new(),
}
}
pub(super) fn timed_readiness_failure(
mut failure: ReadinessFailure,
bound: &OperationBound,
terminal_cause: OperationTerminalCause,
diagnostic_attempts: Vec<ReadinessAttemptEvidence>,
) -> ReadinessFailure {
failure.timing = Some(bound.timing("before_readiness_wait", terminal_cause));
failure.diagnostic_attempts = diagnostic_attempts;
failure
}
pub(super) fn wait_http_ready<R: ProcessObserver>(
runtime: &R,
handle: &ProcessHandle,
endpoint: &ProcessEndpointPlan,
path: &str,
timeout_seconds: Option<u64>,
on_probe_failure: &mut dyn FnMut(&str),
) -> Result<ReadinessEvidence, ReadinessFailure> {
let bound = timeout_seconds
.map(|seconds| OperationBound::finite(Duration::from_secs(seconds)))
.unwrap_or_else(OperationBound::unbounded);
let url = format!("http://{}:{}{}", endpoint.host, endpoint.port, path);
let mut attempts = 0_u32;
let mut diagnostic_attempts = Vec::new();
let mut probe_interval = POLL_INTERVAL;
loop {
ensure_readiness_active(&bound, timeout_seconds, "no readiness probe completed").map_err(
|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts.clone(),
)
},
)?;
if interrupt::received() {
return Err(timed_readiness_failure(
readiness_failure(
ReadinessFailureKind::Interrupted,
"server startup was interrupted".to_owned(),
),
&bound,
OperationTerminalCause::Interrupted,
diagnostic_attempts,
));
}
let status = runtime.status_with_bound(handle, &bound);
ensure_readiness_active(
&bound,
timeout_seconds,
"the server process status attempt did not complete in time",
)
.map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts.clone(),
)
})?;
if interrupt::received() {
return Err(timed_readiness_failure(
readiness_failure(
ReadinessFailureKind::Interrupted,
"server startup was interrupted".to_owned(),
),
&bound,
OperationTerminalCause::Interrupted,
diagnostic_attempts,
));
}
ensure_alive(status).map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::Failed,
diagnostic_attempts.clone(),
)
})?;
attempts = attempts.saturating_add(1);
let attempt = probe_http_attempt(&endpoint.host, endpoint.port, path, &bound);
let effective_bound_ms = attempt.effective_bound_ms;
let last_error = match attempt.outcome {
Ok(()) => {
diagnostic_attempts = vec![ReadinessAttemptEvidence {
operation: "http_readiness".to_owned(),
effective_bound_ms,
succeeded: true,
error: None,
}];
let ready_unix_ms = unix_time_millis().map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::Failed,
diagnostic_attempts.clone(),
)
})?;
ensure_readiness_active(
&bound,
timeout_seconds,
"the readiness response completed after the deadline",
)
.map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts.clone(),
)
})?;
return Ok(ReadinessEvidence::Http {
url,
attempts,
ready_unix_ms,
timing: bound
.timing("before_readiness_wait", OperationTerminalCause::Succeeded),
diagnostic_attempts,
});
}
Err(error) => {
let error = error.to_string();
diagnostic_attempts = vec![ReadinessAttemptEvidence {
operation: "http_readiness".to_owned(),
effective_bound_ms,
succeeded: false,
error: Some(error.clone()),
}];
error
}
};
on_probe_failure(&last_error);
if bound.is_expired() {
let timeout_seconds = timeout_seconds.unwrap_or_default();
return Err(timed_readiness_failure(
readiness_failure(
ReadinessFailureKind::Timeout,
format!(
"server did not become ready within {timeout_seconds} seconds; last probe error: {last_error}"
),
),
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts,
));
}
sleep_within_readiness(&bound, probe_interval);
probe_interval = (probe_interval * 2).min(MAX_PROBE_INTERVAL);
}
}
pub(super) struct HttpTargetRegistryProbe<'a> {
pub(super) readiness_path: &'a str,
pub(super) registry_path: &'a str,
pub(super) targets_field: &'a str,
pub(super) target_url_field: &'a str,
pub(super) target_role_field: &'a str,
pub(super) target_healthy_field: &'a str,
pub(super) target_bootstrap_port_field: &'a str,
pub(super) expected_targets: &'a [TargetRegistryExpectedTarget],
}
fn sleep_within_readiness(bound: &OperationBound, cadence: Duration) {
match bound.remaining() {
Remaining::Finite(remaining) => thread::sleep(cadence.min(remaining)),
Remaining::Expired => {}
Remaining::Unbounded => thread::sleep(cadence),
}
}
fn ensure_readiness_active(
bound: &OperationBound,
timeout_seconds: Option<u64>,
last_error: &str,
) -> Result<(), ReadinessFailure> {
if !bound.is_expired() {
return Ok(());
}
Err(readiness_failure(
ReadinessFailureKind::Timeout,
format!(
"server did not become ready within {} seconds; last probe error: {last_error}",
timeout_seconds.unwrap_or_default()
),
))
}
fn attempt_remaining(attempt: &AttemptBound) -> Result<Duration, ReadinessProbeError> {
match attempt.remaining() {
Remaining::Finite(remaining) => Ok(remaining),
Remaining::Expired => Err(ReadinessProbeError::Deadline),
Remaining::Unbounded => Err(ReadinessProbeError::UnexpectedUnbounded),
}
}
pub(super) fn wait_http_target_registry_ready(
status: impl Fn(&OperationBound) -> ProcessStatus,
endpoint: &ProcessEndpointPlan,
probe: HttpTargetRegistryProbe<'_>,
timeout_seconds: Option<u64>,
on_probe_failure: &mut dyn FnMut(&str),
) -> Result<ReadinessEvidence, ReadinessFailure> {
let bound = timeout_seconds
.map(|seconds| OperationBound::finite(Duration::from_secs(seconds)))
.unwrap_or_else(OperationBound::unbounded);
let readiness_url = format!(
"http://{}:{}{}",
endpoint.host, endpoint.port, probe.readiness_path
);
let registry_url = format!(
"http://{}:{}{}",
endpoint.host, endpoint.port, probe.registry_path
);
let mut attempts = 0_u32;
let mut diagnostic_attempts = Vec::new();
let mut probe_interval = POLL_INTERVAL;
loop {
ensure_readiness_active(&bound, timeout_seconds, "no readiness probe completed").map_err(
|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts.clone(),
)
},
)?;
if interrupt::received() {
return Err(timed_readiness_failure(
readiness_failure(
ReadinessFailureKind::Interrupted,
"server startup was interrupted".to_owned(),
),
&bound,
OperationTerminalCause::Interrupted,
diagnostic_attempts,
));
}
let process_status = status(&bound);
ensure_readiness_active(
&bound,
timeout_seconds,
"the server process status attempt did not complete in time",
)
.map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts.clone(),
)
})?;
if interrupt::received() {
return Err(timed_readiness_failure(
readiness_failure(
ReadinessFailureKind::Interrupted,
"server startup was interrupted".to_owned(),
),
&bound,
OperationTerminalCause::Interrupted,
diagnostic_attempts,
));
}
ensure_alive(process_status).map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::Failed,
diagnostic_attempts.clone(),
)
})?;
attempts = attempts.saturating_add(1);
let public_attempt =
probe_http_attempt(&endpoint.host, endpoint.port, probe.readiness_path, &bound);
let public_effective_bound_ms = public_attempt.effective_bound_ms;
let last_error = match public_attempt.outcome {
Ok(()) => {
let registry_attempt =
probe_target_registry_attempt(&endpoint.host, endpoint.port, &probe, &bound);
let registry_effective_bound_ms = registry_attempt.effective_bound_ms;
match registry_attempt.outcome {
Ok(matched_targets) => {
diagnostic_attempts = vec![
ReadinessAttemptEvidence {
operation: "public_http_readiness".to_owned(),
effective_bound_ms: public_effective_bound_ms,
succeeded: true,
error: None,
},
ReadinessAttemptEvidence {
operation: "target_registry".to_owned(),
effective_bound_ms: registry_effective_bound_ms,
succeeded: true,
error: None,
},
];
let ready_unix_ms = unix_time_millis().map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::Failed,
diagnostic_attempts.clone(),
)
})?;
ensure_readiness_active(
&bound,
timeout_seconds,
"the target registry response completed after the deadline",
)
.map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts.clone(),
)
})?;
return Ok(ReadinessEvidence::HttpTargetRegistry {
readiness_url,
registry_url,
attempts,
ready_unix_ms,
matched_targets,
timing: bound
.timing("before_readiness_wait", OperationTerminalCause::Succeeded),
diagnostic_attempts,
});
}
Err(error) => {
let error = error.to_string();
diagnostic_attempts = vec![
ReadinessAttemptEvidence {
operation: "public_http_readiness".to_owned(),
effective_bound_ms: public_effective_bound_ms,
succeeded: true,
error: None,
},
ReadinessAttemptEvidence {
operation: "target_registry".to_owned(),
effective_bound_ms: registry_effective_bound_ms,
succeeded: false,
error: Some(error.clone()),
},
];
error
}
}
}
Err(error) => {
let error = error.to_string();
diagnostic_attempts = vec![ReadinessAttemptEvidence {
operation: "public_http_readiness".to_owned(),
effective_bound_ms: public_effective_bound_ms,
succeeded: false,
error: Some(error.clone()),
}];
format!("public readiness probe failed: {error}")
}
};
on_probe_failure(&last_error);
if bound.is_expired() {
let timeout_seconds = timeout_seconds.unwrap_or_default();
return Err(timed_readiness_failure(
readiness_failure(
ReadinessFailureKind::Timeout,
format!(
"server did not become ready within {timeout_seconds} seconds; last probe error: {last_error}"
),
),
&bound,
OperationTerminalCause::TimedOut,
diagnostic_attempts,
));
}
sleep_within_readiness(&bound, probe_interval);
probe_interval = (probe_interval * 2).min(MAX_PROBE_INTERVAL);
}
}
fn probe_target_registry_attempt(
host: &str,
port: u16,
probe: &HttpTargetRegistryProbe<'_>,
bound: &OperationBound,
) -> ProbeAttempt<Vec<TargetRegistryMatchEvidence>> {
let response =
probe_http_json_attempt(host, port, probe.registry_path, "target registry", bound);
let effective_bound_ms = response.effective_bound_ms;
let outcome = response
.outcome
.and_then(|response| match_target_registry(&response, probe, bound));
ProbeAttempt {
effective_bound_ms,
outcome,
}
}
pub(super) fn match_target_registry(
response: &serde_json::Value,
probe: &HttpTargetRegistryProbe<'_>,
bound: &OperationBound,
) -> Result<Vec<TargetRegistryMatchEvidence>, ReadinessProbeError> {
readiness_remaining(bound)?;
let targets = response
.get(probe.targets_field)
.and_then(serde_json::Value::as_array)
.ok_or_else(|| ReadinessProbeError::RegistryMismatch {
details: format!(
"target registry response has no array field {:?}",
probe.targets_field
),
})?;
let mut evidence = Vec::with_capacity(probe.expected_targets.len());
for expected in probe.expected_targets {
readiness_remaining(bound)?;
let matches: Vec<&serde_json::Map<String, serde_json::Value>> = targets
.iter()
.filter_map(serde_json::Value::as_object)
.filter(|target| {
target
.get(probe.target_url_field)
.and_then(serde_json::Value::as_str)
== Some(expected.url.as_str())
&& target
.get(probe.target_role_field)
.and_then(serde_json::Value::as_str)
== Some(expected.role.as_str())
})
.collect();
let target = match matches.as_slice() {
[] => {
return Err(ReadinessProbeError::RegistryMismatch {
details: format!(
"target registry has no {:?} target at {:?}",
expected.role, expected.url
),
});
}
[target] => *target,
_ => {
return Err(ReadinessProbeError::RegistryMismatch {
details: format!(
"target registry has multiple {:?} targets at {:?}",
expected.role, expected.url
),
});
}
};
let healthy = target
.get(probe.target_healthy_field)
.and_then(serde_json::Value::as_bool)
.ok_or_else(|| ReadinessProbeError::RegistryMismatch {
details: format!(
"target registry entry for {:?} at {:?} has no boolean {:?} field",
expected.role, expected.url, probe.target_healthy_field
),
})?;
if !healthy {
return Err(ReadinessProbeError::RegistryMismatch {
details: format!(
"target registry entry for {:?} at {:?} is not healthy",
expected.role, expected.url
),
});
}
let bootstrap_port = match target.get(probe.target_bootstrap_port_field) {
None | Some(serde_json::Value::Null) => None,
Some(value) => {
let port = value.as_u64().and_then(|port| u16::try_from(port).ok());
Some(port.ok_or_else(|| ReadinessProbeError::RegistryMismatch {
details: format!(
"target registry entry for {:?} at {:?} has invalid {:?}",
expected.role, expected.url, probe.target_bootstrap_port_field
),
})?)
}
};
if let Some(expected_port) = expected.bootstrap_port
&& bootstrap_port != Some(expected_port)
{
return Err(ReadinessProbeError::RegistryMismatch {
details: format!(
"target registry entry for {:?} at {:?} has bootstrap port {bootstrap_port:?}, expected {expected_port}",
expected.role, expected.url
),
});
}
evidence.push(TargetRegistryMatchEvidence {
url: expected.url.clone(),
role: expected.role.clone(),
healthy,
bootstrap_port,
});
}
readiness_remaining(bound)?;
Ok(evidence)
}
struct ProbeAttempt<T> {
effective_bound_ms: u64,
outcome: Result<T, ReadinessProbeError>,
}
#[cfg(test)]
pub(super) fn probe_http(
host: &str,
port: u16,
path: &str,
bound: &OperationBound,
) -> Result<(), ReadinessProbeError> {
probe_http_attempt(host, port, path, bound).outcome
}
fn probe_http_attempt(
host: &str,
port: u16,
path: &str,
bound: &OperationBound,
) -> ProbeAttempt<()> {
let attempt = bound.attempt(Some(READINESS_ATTEMPT_CAP));
let effective_bound_ms = attempt.configured_ms().unwrap_or_default();
let outcome = (|| {
let url = format!("http://{host}:{port}{path}");
let timeout = attempt_remaining(&attempt)?;
let client = reqwest::blocking::Client::builder()
.timeout(timeout)
.connect_timeout(timeout.min(Duration::from_secs(2)))
.redirect(reqwest::redirect::Policy::none())
.no_proxy()
.build()
.map_err(|source| readiness_request_error("readiness", source))?;
let response = client
.get(&url)
.send()
.map_err(|source| readiness_request_error("readiness", source))?;
let status = response.status().as_u16();
if (200..300).contains(&status) {
Ok(())
} else {
Err(ReadinessProbeError::HttpStatus {
label: "readiness".to_owned(),
status,
})
}
})();
ProbeAttempt {
effective_bound_ms,
outcome,
}
}
#[cfg(test)]
pub(super) fn probe_http_json(
host: &str,
port: u16,
path: &str,
label: &str,
bound: &OperationBound,
) -> Result<serde_json::Value, ReadinessProbeError> {
probe_http_json_attempt(host, port, path, label, bound).outcome
}
fn probe_http_json_attempt(
host: &str,
port: u16,
path: &str,
label: &str,
bound: &OperationBound,
) -> ProbeAttempt<serde_json::Value> {
let attempt = bound.attempt(Some(READINESS_ATTEMPT_CAP));
let effective_bound_ms = attempt.configured_ms().unwrap_or_default();
let outcome = (|| {
let url = format!("http://{host}:{port}{path}");
let timeout = attempt_remaining(&attempt)?;
let client = reqwest::blocking::Client::builder()
.timeout(timeout)
.connect_timeout(timeout.min(Duration::from_secs(2)))
.redirect(reqwest::redirect::Policy::none())
.no_proxy()
.build()
.map_err(|source| readiness_request_error(label, source))?;
let response = client
.get(&url)
.send()
.map_err(|source| readiness_request_error(label, source))?;
let status = response.status().as_u16();
if !(200..300).contains(&status) {
return Err(ReadinessProbeError::HttpStatus {
label: label.to_owned(),
status,
});
}
let body = response
.bytes()
.map_err(|source| readiness_request_error(label, source))?;
let value =
serde_json::from_slice(&body).map_err(|source| ReadinessProbeError::InvalidJson {
label: label.to_owned(),
source,
})?;
readiness_remaining(bound)?;
Ok(value)
})();
ProbeAttempt {
effective_bound_ms,
outcome,
}
}
fn readiness_request_error(label: &str, source: reqwest::Error) -> ReadinessProbeError {
if source.is_timeout() {
ReadinessProbeError::Deadline
} else {
ReadinessProbeError::Request {
label: label.to_owned(),
source,
}
}
}
fn readiness_remaining(bound: &OperationBound) -> Result<(), ReadinessProbeError> {
match bound.remaining() {
Remaining::Expired => Err(ReadinessProbeError::Deadline),
Remaining::Finite(_) | Remaining::Unbounded => Ok(()),
}
}
pub(super) fn unix_time_millis() -> Result<u64, ReadinessFailure> {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(crate::operation_bound::duration_millis)
.map_err(|error| {
readiness_failure(
ReadinessFailureKind::Exited,
format!("system clock is before Unix epoch: {error}"),
)
})
}
impl ReadinessObserver for SystemProcessRuntime {
fn wait_ready(
&self,
handle: &ProcessHandle,
endpoint: &ProcessEndpointPlan,
readiness: &ReadinessPlan,
on_probe_failure: &mut dyn FnMut(&str),
) -> Result<ReadinessEvidence, ReadinessFailure> {
match readiness {
ReadinessPlan::ProcessAlive => {
let bound = OperationBound::unbounded();
ensure_alive(self.status(handle)).map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::Failed,
Vec::new(),
)
})?;
Ok(ReadinessEvidence::ProcessAlive {
ready_unix_ms: unix_time_millis().map_err(|failure| {
timed_readiness_failure(
failure,
&bound,
OperationTerminalCause::Failed,
Vec::new(),
)
})?,
timing: bound.timing(
"before_process_alive_check",
OperationTerminalCause::Succeeded,
),
})
}
ReadinessPlan::Http {
path,
timeout_seconds,
..
} => wait_http_ready(
self,
handle,
endpoint,
path,
*timeout_seconds,
on_probe_failure,
),
ReadinessPlan::HttpTargetRegistry {
readiness_path,
registry_path,
targets_field,
target_url_field,
target_role_field,
target_healthy_field,
target_bootstrap_port_field,
expected_targets,
timeout_seconds,
..
} => wait_http_target_registry_ready(
|bound| self.status_with_bound(handle, bound),
endpoint,
HttpTargetRegistryProbe {
readiness_path,
registry_path,
targets_field,
target_url_field,
target_role_field,
target_healthy_field,
target_bootstrap_port_field,
expected_targets,
},
*timeout_seconds,
on_probe_failure,
),
}
}
}