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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
//! Startup recovery: adopts the runner and seed containers a previous daemon left, removes the rest, and reads what
//! the history left open. It changes nothing unless Docker lists both kinds of container, so it can be tried again;
//! it must not delete GitHub registrations: the first poll does.
use std::collections::BTreeSet;
use std::time::{Duration, SystemTime};
use tracing::{info, warn};
use super::history::Closing;
use super::{Engine, Occupant};
use crate::api;
use crate::cache;
use crate::docker::{self, DockerApi, is_not_found};
use crate::github::GithubApi;
use crate::record;
use crate::scheduler::SlotRef;
use crate::scheduler::warm::Attempt;
use crate::time;
/// The limits an adopted container runs with, and when it started.
struct Caps {
/// Bytes.
memory: u64,
/// Millicores.
cpus: u64,
/// When Docker started it; `None` when Docker cannot tell.
started_at: Option<SystemTime>,
}
impl<G: GithubApi + 'static, D: DockerApi + 'static> Engine<G, D> {
/// Takes over the seed containers a previous daemon left: a running one counts as its profile's reseed, so no
/// runner of the profile starts while it rewrites the snapshot and the slots' layers.
async fn recover_reseeds(&mut self, containers: Vec<docker::Container>) {
for container in containers {
let Some(profile) = container.labels.get("runnerd.profile").cloned() else {
continue;
};
// Only this daemon's own names: tests seed under other prefixes on the same Engine.
if container.name != self.names.seed(&profile) || !self.slots.contains_key(&profile) {
continue;
}
info!("{profile}: waiting for the reseed an earlier run started");
let docker = self.docker.clone();
let id = container.id.clone();
let memory = self.running_caps(&container.id, &profile).await.memory;
self.reseed_memory.insert(profile.clone(), memory);
self.reseeds.insert(
profile,
tokio::spawn(async move { cache::finish_reseed(&*docker, &id).await }),
);
}
}
/// The memory and CPU caps a container left by an earlier daemon runs with, which a reload since may have
/// changed in the config; the profile's configured caps only when Docker cannot tell. Also when it started.
async fn running_caps(&self, container: &str, profile: &str) -> Caps {
let state = self.docker.inspect(container).await.ok();
let configured = self.config.profiles.get(profile);
Caps {
memory: state
.as_ref()
.and_then(|state| state.memory)
.or_else(|| configured.map(|settings| settings.memory.0))
.unwrap_or(0),
cpus: state
.as_ref()
.and_then(|state| state.cpus)
.or_else(|| configured.map(|settings| u64::from(settings.cpus) * 1000))
.unwrap_or(0),
started_at: state
.and_then(|state| state.started_at)
.map(|ms| SystemTime::UNIX_EPOCH + Duration::from_millis(ms)),
}
}
/// What a runner container left running by an earlier daemon runs as, from its labels, Docker and its record.
async fn adopted(&self, container: &docker::Container, profile: &str) -> Occupant {
let label = |name: &str| container.labels.get(name).cloned().unwrap_or_default();
let caps = self.running_caps(&container.id, profile).await;
Occupant {
repository: label("runnerd.repository"),
runner_name: container.name.clone(),
runner_id: label("runnerd.runner-id").parse().unwrap_or(0),
container: container.id.clone(),
busy: true,
// Its uptime goes on from its container's start, not from this restart.
started_at: caps.started_at.unwrap_or_else(SystemTime::now),
memory: caps.memory,
cpus: caps.cpus,
// What it reserved before the restart, as its record kept it; its cap if the record has none.
admission: self
.store
.runner_admitted(&container.name)
.await
.ok()
.flatten(),
// Its capture reports it again; until then, or if it cannot, the history knows it.
job: self.store.runner_job(&container.name).await.ok().flatten(),
key: None,
labels: container
.labels
.get(super::runners::LABELS)
.map(|labels| labels.split(',').map(str::to_owned).collect()),
image: container.labels.get(super::runners::IMAGE).cloned(),
}
}
/// Adopts the runner and seed containers a previous daemon left running, and removes the runners that
/// stopped, or that run under this machine's names for no free slot. Their registrations, and any of this host's runners without a container, go at the first poll.
/// Changes nothing unless Docker lists both kinds, so it can be tried again.
pub(super) async fn recover(&mut self) -> Result<(), docker::Error> {
let seeds = self.docker.list("runnerd.seed").await?;
let containers = self.docker.list("runnerd.job").await?;
self.recover_reseeds(seeds).await;
// A shell's terminal does not outlive the daemon that served it. Shells carry their record's runner labels.
let (shells, containers): (Vec<_>, Vec<_>) = containers
.into_iter()
.partition(|container| container.labels.contains_key(record::SHELL_LABEL));
for shell in shells {
info!("removing {}, a shell an earlier run left", shell.name);
if let Err(error) = self.docker.remove(&shell.id).await {
warn!("cannot remove {}: {error}", shell.name);
}
}
self.list_records().await;
// A reload may have lowered the configured cap since the mirror started.
match self.docker.inspect(&self.names.mirror()).await {
Ok(state) if state.running => {
self.mirror_memory = state.memory.unwrap_or(self.mirror_memory);
}
Ok(_) => self.mirror_memory = 0,
Err(error) if is_not_found(&error) => self.mirror_memory = 0,
// Unknown: the configured cap.
Err(_) => {}
}
// Runner containers still running once recovery is done.
let mut running = BTreeSet::new();
for container in containers {
// Runner containers on this Engine are this machine's whatever their name, which older builds or
// another machine ID may have chosen: a running one is adopted when its labels name a free slot.
let ours = container.name.starts_with(&self.runner_prefix);
let label = |name: &str| container.labels.get(name).cloned().unwrap_or_default();
let profile = label("runnerd.profile");
let index = label("runnerd.slot")
.parse::<usize>()
.ok()
.and_then(|slot| slot.checked_sub(1));
let occupant = self.adopted(&container, &profile).await;
let known_repository = self.repositories.contains_key(&occupant.repository);
let free = index.and_then(|index| {
self.slots
.get_mut(&profile)?
.get_mut(index)
.filter(|slot| slot.is_none())
});
match free {
Some(slot) if container.running && known_repository => {
info!(
"adopted {} in {profile} slot {}",
occupant.runner_name,
index.map_or(0, |index| index + 1)
);
*slot = Some(occupant.clone());
running.insert(occupant.runner_name.clone());
// Its socket back in the directory its container mounts, if it does; its job's writes are
// decided again.
drop(self.attach_build_cache(&occupant.runner_name, &occupant.repository));
let slot = SlotRef {
profile,
index: index.unwrap_or(0),
};
self.observe(&slot, &occupant).await;
}
_ if !ours && container.running => {
running.insert(container.name.clone());
warn!(
"leaving {}: a runner container not named by this machine, for no free slot",
container.name
);
}
_ => {
if !container.running {
self.note_lost_record(&profile, occupant.job, &container.id)
.await;
}
info!("removing {}, left by an earlier run", container.name);
if let Err(error) = self.docker.remove(&container.id).await {
warn!("cannot remove {}: {error}", container.name);
}
}
}
}
if let Some(cache) = &self.build_cache {
let keep: Vec<&str> = running.iter().map(String::as_str).collect();
cache.forget_others(&keep);
}
self.recover_history(&running).await;
self.measure_docker_disk();
Ok(())
}
/// Reads what the history left open: queued jobs to watch, and jobs to close. A job whose runner's container
/// is still `running` is left to run: `forget` closes it once its container is gone. Any other lost its
/// container while no daemon watched: its runner is recorded removed now, unless it was before, as is every
/// other runner the history holds open whose container is not running, with a job or idle.
async fn recover_history(&mut self, running: &BTreeSet<String>) {
self.store
.runners_gone(running.iter().cloned().collect(), time::now_ms());
match self.store.open_jobs().await {
Ok(open) => {
for job in open {
// Its run is as old as the job, for the warm runners its remaining jobs get, not as new as
// this start.
let runs = job.state == api::JobState::Queued
|| job
.runner
.as_ref()
.is_some_and(|runner| running.contains(runner));
if runs {
self.note_job(
&job.repository,
Attempt {
run_id: job.run_id,
run_attempt: job.run_attempt,
},
job.id,
job.workflow.as_deref(),
job.queued_at,
);
}
if job.state == api::JobState::Queued {
self.queued_jobs.insert(job.id, job.repository);
continue;
}
if let Some(runner) = &job.runner {
if running.contains(runner) {
continue;
}
self.store.runner_removed(runner, time::now_ms());
}
self.closing.insert(job.id, Closing::now(&job.repository));
}
}
Err(error) => warn!("cannot read the job history: {error}"),
}
match self.store.mirror_log_until().await {
Ok(until) => self.mirror_until = until,
Err(error) => warn!("cannot read the job history: {error}"),
}
}
}