isb_apps/volume_backup/
export.rs

1//! A volume backup's run, for [`crate::backup`]: hook, snapshot, a
2//! temporary volume copied from the snapshot, incus' export of it
3//! (uncompressed tar) compressed and streamed to the bucket, then the
4//! temporary volume, the export and the snapshot deleted.
5//!
6//! The snapshot is what makes it consistent: the export is of a volume
7//! nothing writes to, taken right after the hook. incus writes its export to
8//! its own backups directory on the host while it is read; that copy goes
9//! when the run ends (and incus expires it after a day if isb could not).
10
11use std::io::{Read, Write};
12
13use serde_json::{Value, json};
14
15use super::{incus, model, org_pool, pre_snapshot};
16use crate::backup::{BackupFile, BackupSpec, Backups, Compression, Destination};
17use crate::client::Client;
18use crate::error::{Error, Result};
19use crate::jobs::RunLog;
20use crate::org::OrgId;
21use crate::volume::VolumeInfo;
22
23/// The incus backup name an export uses on the temporary volume.
24const EXPORT: &str = "isb";
25
26/// One volume backup: export, upload, verify, prune.
27pub fn backup_once(
28    bk: &Backups,
29    org: &OrgId,
30    spec: &BackupSpec,
31    volume: &str,
32    log: &mut RunLog,
33) -> Result<Value> {
34    let d = bk.destination_get(org, &spec.destination)?;
35    let c = bk.client(org, &d)?;
36    let (oc, pool) = org_pool(bk.apps().client(), org)?;
37    let info = incus::volume(&oc, &pool, volume)?;
38    let settings = super::load_settings(bk.state_dir(), org, volume);
39    let prefix = crate::backup::backup_prefix(&d, org, &spec.name);
40    let t = crate::stack::now_secs() as i64;
41    let key = model::volume_key(&prefix, volume, t, spec.compression);
42    let started = std::time::Instant::now();
43    let st = model::stamp(t);
44    let snap = format!("{}{st}", model::BACKUP_SNAPSHOT);
45    super::allow_snapshots(bk.apps().client(), org)?;
46    let hooks = pre_snapshot(&oc, &info, &settings, ("backup", &snap), log)?;
47    log.line(&format!(
48        "isb: snapshotting {volume} as {snap}, exporting it to s3://{}/{key}",
49        d.bucket
50    ));
51    incus::snapshot_create(&oc, &pool, volume, &snap, "isb backup")?;
52    let res = export_snapshot(
53        &oc,
54        &pool,
55        (volume, &snap, &model::export_volume(volume, &st)),
56        (&c, &key, spec.compression),
57        log,
58    );
59    if let Err(e) = incus::snapshot_delete(&oc, &pool, volume, &snap) {
60        log.line(&format!("isb: deleting snapshot {snap}: {e}"));
61    }
62    let (raw, size) = res?;
63    log.line(&format!(
64        "isb: uploaded {size} bytes ({raw} before compression) in {:.1}s",
65        started.elapsed().as_secs_f64()
66    ));
67    match c.head(&key)? {
68        Some(n) if n == size => log.line(&format!("isb: verified: {key} is {n} bytes")),
69        other => {
70            return Err(Error::invalid(format!(
71                "verify {key}: uploaded {size} bytes, HEAD says {other:?}"
72            )));
73        }
74    }
75    let pruned = prune(&c, &prefix, spec.keep as usize, log)?;
76    Ok(json!({
77        "key": key, "size": size, "dump_bytes": raw, "volume": volume,
78        "destination": d.name, "pruned": pruned, "hooks": hooks,
79    }))
80}
81
82fn prune(
83    c: &crate::s3::Client,
84    prefix: &str,
85    keep: usize,
86    log: &mut RunLog,
87) -> Result<Vec<String>> {
88    let mut pruned = Vec::new();
89    for k in model::select_volume_prune(prefix, &c.list(prefix)?, keep) {
90        match c.delete(&k) {
91            Ok(()) => {
92                log.line(&format!("isb: pruned {k}"));
93                pruned.push(k);
94            }
95            Err(e) => log.line(&format!("isb: prune {k}: {e}")),
96        }
97    }
98    Ok(pruned)
99}
100
101/// Copy `snap` of `volume` to `tmp`, export `tmp` into `key`, delete `tmp`.
102fn export_snapshot(
103    oc: &Client,
104    pool: &str,
105    (volume, snap, tmp): (&str, &str, &str),
106    to: (&crate::s3::Client, &str, Compression),
107    log: &mut RunLog,
108) -> Result<(u64, u64)> {
109    incus::copy_from_snapshot(
110        oc,
111        pool,
112        (volume, snap),
113        tmp,
114        &[(model::KEY_TEMPORARY, "true".into())],
115    )?;
116    let res = upload(oc, pool, tmp, to);
117    if let Err(e) = incus::backup_delete(oc, pool, tmp, EXPORT) {
118        if !e.is_not_found() {
119            log.line(&format!("isb: deleting the export of {tmp}: {e}"));
120        }
121    }
122    if let Err(e) = incus::delete_volume(oc, pool, tmp) {
123        log.line(&format!("isb: deleting {tmp}: {e}"));
124    }
125    res
126}
127
128/// Stream incus' export of `tmp` through the compressor to the bucket.
129/// Returns the tarball's size and the object's.
130fn upload(
131    oc: &Client,
132    pool: &str,
133    tmp: &str,
134    (c, key, compression): (&crate::s3::Client, &str, Compression),
135) -> Result<(u64, u64)> {
136    let mut r = incus::export(oc, pool, tmp, EXPORT)?;
137    let mut w = crate::backup::compressor(compression, c.upload(key, compression.content_type()));
138    let mut buf = vec![0u8; 256 << 10];
139    let mut raw = 0u64;
140    loop {
141        let n = match r.read(&mut buf) {
142            Ok(0) => break,
143            Ok(n) => n,
144            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
145            Err(e) => return Err(Error::invalid(format!("reading the export: {e}"))),
146        };
147        raw += n as u64;
148        w.write_all(&buf[..n])
149            .map_err(|e| Error::invalid(format!("upload: {e}")))?;
150    }
151    if raw == 0 {
152        return Err(Error::invalid("the export was empty"));
153    }
154    let upload = w
155        .finish()
156        .map_err(|e| Error::invalid(format!("compress: {e}")))?;
157    Ok((raw, upload.finish()?))
158}
159
160/// A volume backup's objects in its bucket, newest first.
161pub fn files(bk: &Backups, org: &OrgId, d: &Destination, prefix: &str) -> Result<Vec<BackupFile>> {
162    let c = bk.client(org, d)?;
163    let mut out: Vec<(i64, BackupFile)> = c
164        .list(prefix)?
165        .into_iter()
166        .filter_map(|o| {
167            let (volume, t, compression) = model::parse_volume_key(prefix, &o.key)?;
168            Some((
169                t,
170                BackupFile {
171                    key: o.key,
172                    size: o.size,
173                    taken_at: crate::cron::rfc3339(t),
174                    engine: None,
175                    volume: Some(volume),
176                    compression,
177                },
178            ))
179        })
180        .collect();
181    out.sort_by_key(|a| std::cmp::Reverse(a.0));
182    Ok(out.into_iter().map(|(_, f)| f).collect())
183}
184
185/// What a volume backup's source must be: a volume in the org.
186pub fn check_source(bk: &Backups, org: &OrgId, volume: &str) -> Result<VolumeInfo> {
187    model::validate_volume_name(volume)?;
188    let (oc, pool) = org_pool(bk.apps().client(), org)?;
189    incus::volume(&oc, &pool, volume)
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195
196    #[test]
197    fn upload_and_retention_against_a_fake_s3() {
198        let (ep, _seen) = crate::s3::tests::fake_s3("sk");
199        let c = crate::s3::tests::client(&ep, "sk");
200        let p = "isb/acme/home/";
201        for k in [
202            "isb/acme/home/ws_home-20261001T000000Z.volume.tar.gz",
203            "isb/acme/home/ws_home-20261002T000000Z.volume.tar.zst",
204            "isb/acme/home/ws_home-20261003T000000Z.volume.tar.gz",
205            "isb/acme/home/notes.txt",
206            "isb/acme/home/db-20261001T000000Z.postgres.gz",
207        ] {
208            c.put(k, b"x").unwrap();
209        }
210        // A compressed stream into an upload, as a run writes it.
211        let key = model::volume_key(p, "ws_home", 1_791_158_400, Compression::Zstd);
212        let mut w =
213            crate::backup::compressor(Compression::Zstd, c.upload(&key, "application/zstd"));
214        w.write_all(&[7u8; 100_000]).unwrap();
215        let size = w.finish().unwrap().finish().unwrap();
216        assert_eq!(c.head(&key).unwrap(), Some(size));
217        let mut log = RunLog::sink();
218        let gone = prune(&c, p, 2, &mut log).unwrap();
219        assert_eq!(gone.len(), 2, "{gone:?}");
220        let left: Vec<String> = c.list(p).unwrap().into_iter().map(|o| o.key).collect();
221        for k in [
222            key.as_str(),
223            "isb/acme/home/ws_home-20261003T000000Z.volume.tar.gz",
224            "isb/acme/home/notes.txt",
225            "isb/acme/home/db-20261001T000000Z.postgres.gz",
226        ] {
227            assert!(left.iter().any(|l| l == k), "{k} kept: {left:?}");
228        }
229        // It reads back through the decompressor, as a restore does.
230        let (_, body) = c.get(&key).unwrap();
231        let mut back = Vec::new();
232        crate::backup::decompressor(Compression::Zstd, body)
233            .unwrap()
234            .read_to_end(&mut back)
235            .unwrap();
236        assert_eq!(back, vec![7u8; 100_000]);
237    }
238}
isb_apps/volume_backup/export.rs

isb_apps/volume_backup/
export.rs
