OXYGENHUB
◆ v0.9.9 oxygenhub / crates / hub / src / images.rs
15.6 KiB · Rust · last changed Peer-held images: the hub keeps runners' manifests of images it lacks and places their jobs where a peer in the site holds them (closes #342; pins OxygenFlow #94) 9 days ago source blame history raw
92a797b2 Andy McCall ↶ 1 
//! Images by hash (oxygenflow#39, OxygenFlow docs/jetstream.md §8): rootfs
2 
//! images kept site-wide for runners to fetch, so nobody copies an image
3 
//! between machines. An image arrives as chunks (Oxygen's FastCDC, sent
4 
//! only when this hub lacks them) and then its manifest, which is accepted
5 
//! only when every chunk is here and the image they make has the hash it
6 
//! is named by. Chunks are kept zstd-compressed under
7 
//! `<data>/images/chunks`, shared by every image that has them.
8
9 
use crate::db::{now_ms, Pool};
10 
use jet::images::{is_hash, Manifest};
11 
use std::io::Write;
12 
use std::path::{Path, PathBuf};
13
14 
/// Where this hub keeps images.
15 
#[derive(Clone, Debug)]
16 
pub struct Store {
17 
    root: PathBuf,
18 
}
19
20 
impl Store {
21 
    pub fn new(data: &Path) -> Store {
22 
        Store { root: data.join("images") }
23 
    }
24
25 
    fn chunk_path(&self, hash: &str) -> PathBuf {
26 
        self.root.join("chunks").join(&hash[..2]).join(hash)
27 
    }
28
29 
    fn manifest_path(&self, hash: &str) -> PathBuf {
30 
        self.root.join("manifests").join(format!("{hash}.json"))
31 
    }
32
33 
    pub fn has_chunk(&self, hash: &str) -> bool {
34 
        is_hash(hash) && self.chunk_path(hash).exists()
35 
    }
36
37 
    /// Keep chunk `hash`, whose bytes the caller has checked.
38 
    pub fn put_chunk(&self, hash: &str, bytes: &[u8]) -> std::io::Result<()> {
39 
        let path = self.chunk_path(hash);
40 
        if path.exists() {
41 
            return Ok(());
42 
        }
43 
        std::fs::create_dir_all(path.parent().expect("a chunk's directory"))?;
44 
        let tmp = path.with_extension(format!("tmp{}", std::process::id()));
45 
        let mut f = std::fs::File::create(&tmp)?;
46 
        f.write_all(&zstd::encode_all(bytes, 3)?)?;
47 
        f.sync_all()?;
48 
        std::fs::rename(&tmp, &path)
49 
    }
50
51 
    pub fn chunk(&self, hash: &str) -> std::io::Result<Vec<u8>> {
52 
        if !is_hash(hash) {
53 
            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "not a chunk name"));
54 
        }
55 
        zstd::decode_all(std::fs::File::open(self.chunk_path(hash))?)
56 
    }
57
58 
    pub fn manifest(&self, hash: &str) -> Option<Manifest> {
59 
        if !is_hash(hash) {
60 
            return None;
61 
        }
62 
        serde_json::from_slice(&std::fs::read(self.manifest_path(hash)).ok()?).ok()
63 
    }
64
65 
    /// Accept manifest `m`: well formed, every chunk here, and the image
66 
    /// they make hashing to its name. Reads the whole image once.
67 
    pub fn verify_and_keep(&self, m: &Manifest) -> Result<(), String> {
68 
        m.check().map_err(|e| e.to_string())?;
69 
        let mut whole = blake3::Hasher::new();
70 
        for c in &m.chunks {
71 
            if c.is_zeros() {
72 
                whole.update(&vec![0u8; c.len()]);
73 
                continue;
74 
            }
75 
            let bytes = self.chunk(c.hash()).map_err(|_| format!("chunk {} was not sent; send it first", &c.hash()[..12]))?;
76 
            if bytes.len() != c.len() {
77 
                return Err(format!("chunk {} is {} bytes, not {}", &c.hash()[..12], bytes.len(), c.len()));
78 
            }
79 
            whole.update(&bytes);
80 
        }
81 
        let got = whole.finalize().to_hex().to_string();
82 
        if got != m.hash {
83 
            return Err(format!("the chunks make an image named {}, not {}", &got[..12], &m.hash[..12]));
84 
        }
93501da2 Andy McCall ↶ 85 
        self.keep_manifest(m)
86 
    }
87
951aed50 Andy McCall ↶ 88 
    /// Keep manifest `m` of an image only runners hold (oxygenflow#94): well
89 
    /// formed, its chunks not here, so not checked against its hash; a
90 
    /// fetcher checks the image it assembles against the pinned hash.
91 
    /// Never over a manifest already kept.
92 
    pub fn keep_peer_manifest(&self, m: &Manifest) -> Result<(), String> {
93 
        m.check().map_err(|e| e.to_string())?;
94 
        if self.manifest_path(&m.hash).exists() {
95 
            return Ok(());
96 
        }
97 
        self.keep_manifest(m)
98 
    }
99
93501da2 Andy McCall ↶ 100 
    /// Write manifest `m`, whose image is here.
101 
    fn keep_manifest(&self, m: &Manifest) -> Result<(), String> {
92a797b2 Andy McCall ↶ 102 
        let path = self.manifest_path(&m.hash);
103 
        std::fs::create_dir_all(path.parent().expect("the manifests directory")).map_err(|e| e.to_string())?;
104 
        let tmp = path.with_extension("json.tmp");
105 
        std::fs::write(&tmp, serde_json::to_vec(m).map_err(|e| e.to_string())?).map_err(|e| e.to_string())?;
106 
        std::fs::rename(&tmp, &path).map_err(|e| e.to_string())
107 
    }
93501da2 Andy McCall ↶ 108
109 
    /// What image `m` is for, read from its own files (OxygenFlow spec
110 
    /// §11.2, oxygenhub#173): its shell's ELF header, found through its ext4
111 
    /// out of the kept chunks, a few of them, not the whole image. `None`
112 
    /// when it is not ext4 or has no shell to read.
113 
    pub fn read_arch(&self, m: &Manifest) -> Option<&'static str> {
114 
        let mut starts = Vec::with_capacity(m.chunks.len());
115 
        let mut end = 0u64;
116 
        for c in &m.chunks {
117 
            starts.push(end);
118 
            end += c.len() as u64;
119 
        }
120 
        let mut cache: std::collections::HashMap<usize, Vec<u8>> = std::collections::HashMap::new();
121 
        let short = || std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "past the image's end");
122 
        jet::rootfs::arch(&mut |at: u64, buf: &mut [u8]| {
123 
            let mut done = 0usize;
124 
            while done < buf.len() {
125 
                let pos = at + done as u64;
126 
                let i = starts.partition_point(|&s| s <= pos).checked_sub(1).ok_or_else(short)?;
127 
                let c = &m.chunks[i];
128 
                let within = (pos - starts[i]) as usize;
129 
                if within >= c.len() {
130 
                    return Err(short());
131 
                }
132 
                let n = (c.len() - within).min(buf.len() - done);
133 
                if c.is_zeros() {
134 
                    buf[done..done + n].fill(0);
135 
                } else {
136 
                    if !cache.contains_key(&i) {
137 
                        if cache.len() >= 32 {
138 
                            cache.clear();
139 
                        }
140 
                        cache.insert(i, self.chunk(c.hash())?);
141 
                    }
142 
                    let bytes = &cache[&i];
143 
                    if bytes.len() != c.len() {
144 
                        return Err(short());
145 
                    }
146 
                    buf[done..done + n].copy_from_slice(&bytes[within..within + n]);
147 
                }
148 
                done += n;
149 
            }
150 
            Ok(())
151 
        })
152 
    }
153
154 
    /// Keep image `m`, checked as [`Store::verify_and_keep`], with the
155 
    /// architecture its own files name: the manifest's word is taken only
156 
    /// when the image cannot be read. The manifest as kept.
157 
    pub fn keep_reading_arch(&self, m: &Manifest) -> Result<Manifest, String> {
158 
        self.verify_and_keep(m)?;
159 
        match self.read_arch(m) {
160 
            Some(a) if m.arch.as_deref() != Some(a) => {
161 
                let mut kept = m.clone();
162 
                kept.arch = Some(a.to_string());
163 
                self.keep_manifest(&kept)?;
164 
                Ok(kept)
165 
            }
166 
            _ => Ok(m.clone()),
167 
        }
168 
    }
169
170 
    /// Record `arch` in kept image `hash`'s manifest.
171 
    fn set_arch(&self, hash: &str, arch: &str) -> Result<(), String> {
172 
        let mut m = self.manifest(hash).ok_or("no such image")?;
173 
        m.arch = Some(arch.to_string());
174 
        self.keep_manifest(&m)
175 
    }
92a797b2 Andy McCall ↶ 176 
}
177
951aed50 Andy McCall ↶ 178 
/// Record an image this hub now keeps, its chunks and all: one only
179 
/// runners held before becomes the hub's.
92a797b2 Andy McCall ↶ 180 
pub async fn record(pool: &Pool, m: &Manifest, uploader: &str) -> sqlx::Result<()> {
181 
    sqlx::query(
5d90e0e0 Andy McCall 182 
        "INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES ($1, $2, $3, $4, $5, $6) \
951aed50 Andy McCall ↶ 183 
         ON CONFLICT (hash) DO UPDATE SET source = 'hub', holder = '', \
184 
         arch = CASE WHEN excluded.arch != '' THEN excluded.arch ELSE jet_images.arch END",
92a797b2 Andy McCall ↶ 185 
    )
186 
    .bind(&m.hash)
187 
    .bind(m.size as i64)
188 
    .bind(m.arch.clone().unwrap_or_default())
189 
    .bind(m.chunks.len() as i64)
190 
    .bind(uploader)
191 
    .bind(now_ms())
192 
    .execute(pool)
193 
    .await?;
194 
    if let Some(a) = m.arch.as_ref().filter(|a| !a.is_empty()) {
195 
        crate::db::flow::record_image_archs(pool, &[(m.hash.clone(), a.clone())].into_iter().collect()).await?;
196 
    }
197 
    Ok(())
93501da2 Andy McCall ↶ 198 
}
199
200 
/// Learn what every image is for (oxygenhub#173), once at start: read
201 
/// each kept image that has no architecture from its own files, then give
202 
/// any image still unknown the architecture its jobs passed on. A runner's
203 
/// own report, read from the image it holds, wins over what was learned.
204 
/// How many images were read.
205 
pub async fn learn_archs(app: &crate::Shared) -> sqlx::Result<usize> {
206 
    let mut read = 0;
207 
    for k in kept(&app.db).await?.into_iter().filter(|k| k.arch.is_empty()) {
208 
        let store = app.images();
209 
        let hash = k.hash.clone();
210 
        let arch = tokio::task::spawn_blocking(move || {
211 
            let a = store.read_arch(&store.manifest(&hash)?)?;
212 
            store.set_arch(&hash, a).ok()?;
213 
            Some(a)
214 
        })
215 
        .await
216 
        .ok()
217 
        .flatten();
218 
        if let Some(a) = arch {
5d90e0e0 Andy McCall 219 
            sqlx::query("UPDATE jet_images SET arch = $1 WHERE hash = $2").bind(a).bind(&k.hash).execute(&app.db).await?;
93501da2 Andy McCall ↶ 220 
            crate::db::flow::record_image_archs(&app.db, &[(k.hash.clone(), a.to_string())].into_iter().collect()).await?;
221 
            read += 1;
222 
        }
223 
    }
224 
    crate::db::flow::learn_archs_from_runs(&app.db).await?;
225 
    Ok(read)
951aed50 Andy McCall ↶ 226 
}
227
228 
/// Record an image only runners hold, whose manifest `holder` sent
229 
/// (oxygenflow#94). Never over one the hub keeps.
230 
pub async fn record_peer(pool: &Pool, m: &Manifest, holder: &str) -> sqlx::Result<()> {
231 
    sqlx::query(
232 
        "INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms, source, holder) VALUES ($1, $2, $3, $4, $5, $6, 'peer', $7) \
233 
         ON CONFLICT (hash) DO NOTHING",
234 
    )
235 
    .bind(&m.hash)
236 
    .bind(m.size as i64)
237 
    .bind(m.arch.clone().unwrap_or_default())
238 
    .bind(m.chunks.len() as i64)
239 
    .bind(holder)
240 
    .bind(now_ms())
241 
    .bind(holder)
242 
    .execute(pool)
243 
    .await?;
244 
    if let Some(a) = m.arch.as_ref().filter(|a| !a.is_empty()) {
245 
        crate::db::flow::record_image_archs(pool, &[(m.hash.clone(), a.clone())].into_iter().collect()).await?;
246 
    }
247 
    Ok(())
92a797b2 Andy McCall ↶ 248 
}
249
951aed50 Andy McCall ↶ 250 
/// Every image this hub keeps a manifest of, by hash: its own and those
251 
/// only runners hold.
92a797b2 Andy McCall ↶ 252 
pub async fn held(pool: &Pool) -> sqlx::Result<std::collections::HashSet<String>> {
253 
    Ok(sqlx::query_scalar::<_, String>("SELECT hash FROM jet_images").fetch_all(pool).await?.into_iter().collect())
254 
}
6ca0b971 Andy McCall ↶ 255
951aed50 Andy McCall ↶ 256 
/// Where the images jobs need can come from (oxygenflow#94): the hub's
257 
/// own, pushed and verified, and those only runners hold, each with the
258 
/// swarm sites of the online runners holding it now.
259 
#[derive(Debug, Default)]
260 
pub struct Reach {
261 
    pub hub: std::collections::HashSet<String>,
262 
    pub peers: std::collections::HashMap<String, Vec<String>>,
263 
    /// Each runner in a swarm now, and its site.
264 
    pub sites: std::collections::HashMap<i64, String>,
265 
}
266
267 
impl Reach {
268 
    /// Whether a runner that fetches images can get `image`: from the hub,
269 
    /// or from a runner of its own site.
270 
    pub fn fetchable(&self, runner: i64, image: &str) -> bool {
271 
        self.hub.contains(image) || self.sites.get(&runner).is_some_and(|s| self.peers.get(image).is_some_and(|v| v.contains(s)))
272 
    }
273
274 
    /// Every image `runner` can fetch.
275 
    pub fn for_runner(&self, runner: i64) -> std::collections::HashSet<String> {
276 
        let mut out = self.hub.clone();
277 
        out.extend(self.peers.keys().filter(|i| self.fetchable(runner, i)).cloned());
278 
        out
279 
    }
280
281 
    /// Whether only runners hold `image`.
282 
    pub fn peer_only(&self, image: &str) -> bool {
283 
        !self.hub.contains(image) && self.peers.contains_key(image)
284 
    }
285 
}
286
287 
/// [`Reach`] now.
288 
pub async fn reach(app: &crate::Shared) -> sqlx::Result<Reach> {
289 
    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, source FROM jet_images").fetch_all(&app.db).await?;
290 
    let mut reach = Reach::default();
291 
    for (hash, source) in rows {
292 
        if source == "peer" {
293 
            reach.peers.insert(hash, Vec::new());
294 
        } else {
295 
            reach.hub.insert(hash);
296 
        }
297 
    }
298 
    for r in crate::db::runners::list(&app.db).await? {
299 
        if r.state == "disabled" || !r.online() {
300 
            continue;
301 
        }
302 
        let Some(addr) = app.swarm.addr(r.id) else { continue };
303 
        for image in r.image_list().unwrap_or_default() {
304 
            if let Some(sites) = reach.peers.get_mut(&image.to_ascii_lowercase()) {
305 
                if !sites.contains(&addr.site) {
306 
                    sites.push(addr.site.clone());
307 
                }
308 
            }
309 
        }
310 
        reach.sites.insert(r.id, addr.site);
311 
    }
312 
    Ok(reach)
313 
}
314
6ca0b971 Andy McCall ↶ 315 
/// An image this hub keeps (`jet_images`).
316 
#[derive(sqlx::FromRow)]
317 
pub struct Kept {
318 
    pub hash: String,
319 
    pub size: i64,
320 
    pub arch: String,
321 
    pub chunks: i64,
322 
    pub uploader: String,
323 
    pub created_ms: i64,
951aed50 Andy McCall ↶ 324 
    /// 'hub', or 'peer' for one only runners hold (oxygenflow#94).
325 
    pub source: String,
326 
    pub holder: String,
6ca0b971 Andy McCall ↶ 327 
}
328
329 
pub async fn kept(pool: &Pool) -> sqlx::Result<Vec<Kept>> {
951aed50 Andy McCall ↶ 330 
    sqlx::query_as("SELECT hash, size, arch, chunks, uploader, created_ms, source, holder FROM jet_images").fetch_all(pool).await
6ca0b971 Andy McCall ↶ 331 
}
332
333 
/// One repository's use of one image: its jobs since `since_ms`, and when
334 
/// it last ran one.
335 
#[derive(sqlx::FromRow)]
336 
pub struct Use {
337 
    pub image: String,
338 
    pub owner: String,
339 
    pub name: String,
340 
    pub jobs: i64,
341 
    pub last_ms: i64,
342 
}
343
344 
/// What each repository's jobs ran on since `since_ms`, newest first.
345 
pub async fn uses(pool: &Pool, since_ms: i64) -> sqlx::Result<Vec<Use>> {
3b1bc27f Andy McCall 346 
    let rootfs = crate::db::json_text("j.sandbox", "rootfs");
347 
    sqlx::query_as(&format!(
348 
        "SELECT lower({rootfs}) AS image, p.owner, p.name, COUNT(*) AS jobs,
6ca0b971 Andy McCall ↶ 349 
                MAX(COALESCE(j.started_ms, r.created_ms)) AS last_ms
350 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
3b1bc27f Andy McCall 351 
         WHERE r.created_ms >= $1 AND COALESCE({rootfs}, '') != ''
352 
         GROUP BY lower({rootfs}), p.id, p.owner, p.name ORDER BY last_ms DESC"
353 
    ))
6ca0b971 Andy McCall ↶ 354 
    .bind(since_ms)
355 
    .fetch_all(pool)
356 
    .await
357 
}
358
16ee5d19 Andy McCall ↶ 359 
/// What one repository's jobs ran on since `since_ms`, newest first
360 
/// (its Images page, oxygenhub#254).
361 
pub async fn uses_of(pool: &Pool, repo_id: i64, since_ms: i64) -> sqlx::Result<Vec<Use>> {
362 
    let rootfs = crate::db::json_text("j.sandbox", "rootfs");
363 
    sqlx::query_as(&format!(
364 
        "SELECT lower({rootfs}) AS image, p.owner, p.name, COUNT(*) AS jobs,
365 
                MAX(COALESCE(j.started_ms, r.created_ms)) AS last_ms
366 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
367 
         WHERE r.created_ms >= $1 AND p.id = $2 AND COALESCE({rootfs}, '') != ''
368 
         GROUP BY lower({rootfs}), p.id, p.owner, p.name ORDER BY last_ms DESC"
369 
    ))
370 
    .bind(since_ms)
371 
    .bind(repo_id)
372 
    .fetch_all(pool)
373 
    .await
374 
}
375
6ca0b971 Andy McCall ↶ 376 
/// The architecture runners reported for each image (`flow_images`).
377 
pub async fn reported_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {
378 
    Ok(sqlx::query_as::<_, (String, String)>("SELECT hash, arch FROM flow_images").fetch_all(pool).await?.into_iter().collect())
951aed50 Andy McCall ↶ 379 
}
380
381 
#[cfg(test)]
382 
mod reach_tests {
383 
    use super::Reach;
384
385 
    #[test]
386 
    fn an_image_only_runners_hold_is_fetchable_in_their_site_alone() {
387 
        let mut r = Reach::default();
388 
        r.hub.insert("hub-image".into());
389 
        r.peers.insert("peer-image".into(), vec!["home".into()]);
390 
        r.sites.insert(1, "home".into());
391 
        r.sites.insert(2, "cloud".into());
392 
        assert!(r.fetchable(1, "hub-image") && r.fetchable(2, "hub-image") && r.fetchable(3, "hub-image"));
393 
        assert!(r.fetchable(1, "peer-image"), "a runner of the holder's site");
394 
        assert!(!r.fetchable(2, "peer-image"), "another site");
395 
        assert!(!r.fetchable(3, "peer-image"), "a runner in no swarm");
396 
        assert!(r.peer_only("peer-image") && !r.peer_only("hub-image"));
397 
        assert_eq!(r.for_runner(1).len(), 2);
398 
        assert_eq!(r.for_runner(2).len(), 1);
399 
    }
6ca0b971 Andy McCall ↶ 400 
}