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
//! 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::SystemTime;
use tracing::{info, warn};
use super::history::Closing;
use super::{Engine, Occupant};
use crate::api;
use crate::cache;
use crate::docker::{self, DockerApi, RunnerState, is_not_found};
use crate::github::GithubApi;
use crate::scheduler::SlotRef;
use crate::time;
impl<G: GithubApi, 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_cap(&container.id, &profile).await;
self.reseed_memory.insert(profile.clone(), memory);
self.reseeds.insert(
profile,
tokio::spawn(async move { cache::finish_reseed(&*docker, &id).await }),
);
}
}
/// The memory cap a container left by an earlier daemon runs with, which a reload since may have changed in the
/// config; the profile's configured cap only when Docker cannot tell.
async fn running_cap(&self, container: &str, profile: &str) -> u64 {
match self.docker.inspect(container).await {
Ok(RunnerState {
memory: Some(memory),
..
}) => memory,
_ => self
.config
.profiles
.get(profile)
.map_or(0, |settings| settings.memory.0),
}
}
/// 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 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 = 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,
started_at: SystemTime::now(),
memory: self.running_cap(&container.id, &profile).await,
// 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(),
};
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());
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
);
}
_ => {
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);
}
}
}
}
self.recover_history(&running).await;
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.
async fn recover_history(&mut self, running: &BTreeSet<String>) {
match self.store.open_jobs().await {
Ok(open) => {
for job in open {
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}"),
}
}
}