1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
//! Deployment stabilization polling.
//!
//! Provides a reusable function that waits for all services in a deployment to
//! reach their desired replica count and pass health checks, or time out.
//!
//! This module lives in `zlayer-agent` (a library crate) so that both the
//! runtime binary and the API server can share the same stabilization logic
//! instead of duplicating it.
use std::time::{Duration, Instant};
use crate::service::ServiceManager;
use zlayer_spec::{DeploymentSpec, Protocol, ScaleSpec};
/// Per-service health summary returned by stabilization polling.
#[derive(Debug, Clone, serde::Serialize)]
pub struct ServiceHealthSummary {
/// Service name
pub name: String,
/// Running replica count
pub running: u32,
/// Desired replica count from the spec
pub desired: u32,
/// Whether health checks are passing for all running replicas
pub healthy: bool,
/// Endpoint URLs for this service (e.g. "<http://localhost:8080>")
pub endpoints: Vec<String>,
}
/// Outcome of the stabilization wait.
///
/// This is intentionally decoupled from `DeploymentStatus` (which lives in
/// `zlayer-api`) to avoid circular dependencies. Callers should map this to
/// their own status types.
#[derive(Debug, Clone)]
pub enum StabilizationOutcome {
/// All services reached their desired state within the timeout.
Ready,
/// The timeout expired before all services stabilized.
TimedOut {
/// Human-readable description of which services were not ready.
message: String,
},
}
/// Result of waiting for a deployment to stabilize.
#[derive(Debug, Clone)]
pub struct StabilizationResult {
/// Whether stabilization succeeded or timed out
pub outcome: StabilizationOutcome,
/// Per-service health summaries (always populated regardless of outcome)
pub services: Vec<ServiceHealthSummary>,
}
/// Wait for all services in a deployment to reach their desired replica count
/// and pass health checks, or time out.
///
/// Polls every 500ms for up to `timeout`. Returns [`StabilizationOutcome::Ready`]
/// if all services reach their desired state, or [`StabilizationOutcome::TimedOut`]
/// if the timeout expires.
pub async fn wait_for_stabilization(
manager: &ServiceManager,
spec: &DeploymentSpec,
timeout: Duration,
) -> StabilizationResult {
let poll_interval = Duration::from_millis(500);
let start = Instant::now();
loop {
let mut all_ready = true;
let mut summaries = Vec::with_capacity(spec.services.len());
for (name, service_spec) in &spec.services {
// Jobs and cron jobs are run-to-completion workloads (triggered /
// scheduled), not scaled long-running services — they have no desired
// replica count to stabilize on. Skip them so a deployment containing
// a job doesn't time out waiting for a "running" replica that never
// exists (the job runs only when triggered).
if matches!(
service_spec.rtype,
zlayer_spec::ResourceType::Job | zlayer_spec::ResourceType::Cron
) {
continue;
}
let desired = match &service_spec.scale {
ScaleSpec::Fixed { replicas } => *replicas,
ScaleSpec::Adaptive { min, .. } => *min,
ScaleSpec::Manual => 0,
};
// Cluster-wide: sum running replicas + health across every node the
// service is placed on (distributed scaling places replicas on
// remote nodes; a leader-local count would read 0 and never
// stabilize). Falls back to the local view when not clustered.
let node_states = manager.cluster_service_states(name).await;
#[allow(clippy::cast_possible_truncation)]
let running: u32 = node_states.iter().map(|s| s.running).sum();
let healthy = if desired == 0 {
true // Manual scaling / 0 replicas is trivially healthy.
} else {
// Every node with replicas must report healthy; nodes running
// none are trivially healthy and don't drag the aggregate.
node_states.iter().all(|s| s.healthy)
};
let service_ready = running == desired && healthy;
if !service_ready && desired > 0 {
all_ready = false;
}
// Build endpoint URLs from the spec
let endpoints: Vec<String> = service_spec
.endpoints
.iter()
.map(|ep| {
let proto = match ep.protocol {
Protocol::Http => "http",
Protocol::Https => "https",
Protocol::Tcp => "tcp",
Protocol::Udp => "udp",
Protocol::Websocket => "ws",
};
format!("{}://localhost:{}", proto, ep.port)
})
.collect();
summaries.push(ServiceHealthSummary {
name: name.clone(),
running,
desired,
healthy,
endpoints,
});
}
if all_ready {
return StabilizationResult {
outcome: StabilizationOutcome::Ready,
services: summaries,
};
}
if start.elapsed() >= timeout {
// Build a failure message from unhealthy services, including
// the tail of each failing service's container logs so the
// user sees the real cause (e.g. "GLIBC_2.38 not found",
// "failed to prepare rootfs", "panicked at ...") instead of
// just "1/1 replicas, healthy=false".
let failing: Vec<&ServiceHealthSummary> = summaries
.iter()
.filter(|s| (s.running != s.desired || !s.healthy) && s.desired > 0)
.collect();
let mut parts: Vec<String> = Vec::with_capacity(failing.len());
for s in &failing {
let header = format!(
"{}: {}/{} replicas, healthy={}",
s.name, s.running, s.desired, s.healthy
);
// Try to fetch the last 20 log lines from this service's
// replicas. A miss (no containers, runtime error) falls
// back to just the header so we never block the error
// on log retrieval.
match manager.get_service_logs(&s.name, 20, None).await {
Ok(entries) if !entries.is_empty() => {
let body = entries
.iter()
.map(|e| format!(" {}", e.message))
.collect::<Vec<_>>()
.join("\n");
parts.push(format!("{header}\n logs:\n{body}"));
}
_ => parts.push(header),
}
}
let message = if parts.is_empty() {
"Stabilization timed out".to_string()
} else {
format!("Stabilization timed out:\n {}", parts.join("\n "))
};
return StabilizationResult {
outcome: StabilizationOutcome::TimedOut { message },
services: summaries,
};
}
tokio::time::sleep(poll_interval).await;
}
}