NationTech/harmony
3
0
Fork 2
Code Issues 39 Pull Requests 11 Actions Packages Projects Releases 1 Wiki Activity

WIP: feat/tenant-backup #352

Draft
johnride wants to merge 8 commits from feat/tenant-backup into master
pull from: feat/tenant-backup
merge into: NationTech:master
Conversation 0 Commits 8 Files Changed 13 +2888
13 changed files with 2888 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
Cargo.lock generated
View File

@@ -4277,6 +4277,23 @@ dependencies = [
"thiserror 2.0.19", "thiserror 2.0.19",
] ]
[[package]]
name = "harmony-tenant-backup"
version = "0.1.0"
dependencies = [
"anyhow",
"chrono",
"clap",
"harmony",
"harmony_config",
"k8s-openapi",
"kube",
"schemars 0.8.22",
"serde",
"serde_json",
"tokio",
]
[[package]] [[package]]
name = "harmony_agent" name = "harmony_agent"
version = "0.1.0" version = "0.1.0"

28
examples/harmony_tenant_backup/Cargo.toml Normal file
View File

@@ -0,0 +1,28 @@
[package]
name = "harmony-tenant-backup"
edition = "2024"
version.workspace = true
readme.workspace = true
license.workspace = true
publish = false
[[bin]]
name = "harmony-tenant-backup"
path = "src/main.rs"
[[bin]]
name = "harmony-tenant-backup-setup"
path = "src/setup.rs"
[dependencies]
harmony = { path = "../../harmony" }
harmony_config = { path = "../../harmony_config" }
tokio = { workspace = true, features = ["process", "time"] }
kube = { workspace = true }
k8s-openapi = { workspace = true }
clap = { workspace = true }
anyhow = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
schemars = "0.8"
chrono = { workspace = true }

301
examples/harmony_tenant_backup/DESIGN.md Normal file
View File

@@ -0,0 +1,301 @@
# Tenant backup design
Status: this is the current plan. The code in this folder is an earlier
prototype (v0), kept for reference only. It does not implement this design;
see "What changes from v0".
A tenant on a shared cluster opts in to offsite backups of their own data. The
backup job runs in the tenant's namespace with an identity that can write
backups and can never delete them through the S3 API.
## Backup instance
The unit is a backup instance. It is created when a tenant enables backup and
owns one config, one bucket, one restic password and one writer identity. A
second instance, or a replacement, gets all four new. Nothing is shared
between instances or between tenants.
| Who | Does |
|---|---|
| Platform | `BackupInstanceScore`: bucket on the offsite RGW with versioning and Object Lock, a writer scoped to it, writer credentials into OpenBao at the tenant's path. Run by the platform team on request today, triggered by tenant self-service later. |
| Tenant | `TenantBackupScore` in their deploy code: schedule, CNPG clusters, buckets, PVCs. Data they can regenerate stays out. Generates the restic password in their own password manager. |
| Job | CronJob in the tenant namespace. Reads its instance config from OpenBao (see "Job identity"). |
Tenants are admins of their namespace. They can read the job's secrets, edit
or delete the CronJob. That only affects their own backups, which is
acceptable because backup is their choice.
## Storage and security
- The bucket is owned by the platform's RGW account. The writer is an IAM
user in that account with an allow-only policy: `GetObject`, `PutObject`,
`ListBucket`, `GetBucketLocation`, and `DeleteObject` on `locks/*`.
Everything else is denied by default. The bucket owner is never the writer:
an owner keeps full control through ACLs whatever the policy says.
- `PutObject` can overwrite, so the bucket has versioning and Object Lock in
compliance mode, 90 days by default. An overwrite or delete only adds a
version or a delete marker; the locked version stays. restic never rewrites
a key except `locks/*`. The 2025 median intrusion dwell time is 14 days, 26
when an outside party reports it (Mandiant M-Trends 2026); 90 days leaves
room for the slow tail. Cost: every locked version is stored for 90 days,
and a bucket can only be deleted 90 days after its last write. Compliance
retention can be extended, never shortened.
- One lifecycle rule, on `locks/` only, expires noncurrent lock versions and
stale delete markers after the retention period. No lifecycle rule ever
touches other prefixes; their old versions are what a rollback reads.
- A bucket quota caps growth. A full bucket fails the job loudly.
- Object Lock binds the S3 API only. The Admin Ops bucket purge,
`radosgw-admin bucket rm --purge-objects`, `radosgw-admin object rm` and raw
`rados rm` all delete without checking retention (only `rgw_op.cc`'s S3
delete paths call `verify_object_lock`). On the offsite cluster, Ceph
keyrings that reach the RGW pools and RGW users with `buckets=write` caps,
including the one Rook creates, are held outside routine operator access.
Object Lock protects against tenants and stolen S3 keys, not against the
storage operator.
- RGW accounts and IAM users need Ceph Squid (19.2) or later. On Reef the
writer is a plain RGW user with an inline policy on a bucket owned by
another user.
- Writer credentials reach one bucket. A tenant who reads them learns nothing
about another tenant.
- An offsite breach alone yields ciphertext without the key.
- The job never runs `restic forget` or `prune`.
- TLS is never skipped: `sslmode=verify-full` to Postgres, the RGW CA mounted
if it is private.
`BackupInstanceScore` proves its own policy before returning: the writer can
put an object, and cannot delete an object version or change the bucket
policy.
Not covered: whoever can read the instance's OpenBao path (OpenBao root, or
cluster-admin acting as the job) gets the password and the writer
credentials, and with them all history. They could already read the live
data. restic is symmetric (no write-only key exists in restic, rustic, kopia
or borg), so the job necessarily holds a key that reads all history.
## Restic password
The tenant generates the password in their own password manager, so the
offline copy exists before the backup does. Nothing on the platform side
generates it.
The password is a `#[config(secret)]` field of the instance's config, stored
by `harmony_config` like any other secret. On first apply, `TenantBackupScore`
calls `get_or_prompt` and the tenant pastes it. In CI a missing password
fails with "apply once interactively to set the restic password". On every
apply the Score checks that the password opens the existing repo, so a wrong
paste never creates a second repo or orphans history.
| Event | Outcome |
|---|---|
| Password lost from the tenant's vault, OpenBao alive | Read it back with `harmony_config`. Same instance. |
| Lost from both | Backups are unreadable, permanently. |
| Password leaked | Read exposure only. New instance; the old bucket is deleted once the new one has a verified snapshot. |
| Writer credentials reissued by the platform | Update the tenant's vault copy; the Score picks them up on next apply. |
The tenant's vault entry holds endpoint, bucket, region, writer credentials
and password: everything the restore below needs. The password is never
rotated in place and never escrowed by the platform.
## Job identity (open)
The job reads its config from OpenBao at run time, using its own short-lived
token rather than a stored secret. All auth goes through Zitadel today, and
Zitadel cannot exchange a Kubernetes ServiceAccount token for its own: token
exchange only accepts Zitadel-issued subject tokens, and the JWT IdP is a
browser flow. So one of these:
| | Holds | OpenBao trusts | Revocation | Cost |
|---|---|---|---|---|
| A. Second OpenBao JWT mount on the cluster's service-account issuer, static `jwt_validation_pubkeys` | Projected token, audience `openbao`, about 1 h | Zitadel, plus each cluster issuer limited to backup roles | Token TTL | Mount per cluster; one role per instance bound to `system:serviceaccount:<ns>:<sa>` and one path; update keys when the cluster rotates its signing key. `harmony_secret` reads the token file instead of signing with a Zitadel key. |
| B. Zitadel machine key per instance (v0) | Long-lived key in a namespace Secret | Zitadel only | Manual | None, but the stored secret returns |
| C. OpenBao `kubernetes` auth (TokenReview) | Projected token | Zitadel, plus each cluster API server | Immediate | OpenBao reaches each API server; new login method |
Leaning A. Revisit if Zitadel ships workload identity federation. Rejected:
copying the config into a namespace Secret at apply time, or External Secrets
Operator; both leave the secret in etcd, and ESO adds a cluster-wide
operator.
## The job
Two CronJobs, so a PVC that cannot mount never blocks the databases:
- `pg+s3`: Postgres dumps and tenant buckets.
- `pvc`: declared RWX volumes. Only created when the Score lists any.
Settings: `concurrencyPolicy: Forbid`, `startingDeadlineSeconds: 3600`,
`backoffLimit: 0` (a retry duplicates snapshots), `restartPolicy: Never`,
`activeDeadlineSeconds`, `failedJobsHistoryLimit: 3`. Scratch is an `emptyDir`
with `sizeLimit` and a matching `ephemeral-storage` request, staged one item
at a time. Memory leaves room for the restic index, which grows forever
without prune.
The Score emits one NetworkPolicy: egress from the job to the CNPG pods on
5432, the RGW on 443 and DNS; matching ingress on the CNPG pods. Shared OKD
projects often default-deny.
Every run ends with `restic check --read-data-subset=<weekday>/7`, so the
whole repo is read once a week. It is the only thing that notices a pack
overwritten with the writer's credentials. A killed run leaves a stale
non-exclusive lock that does not block the next backup, and orphan packs that
stay forever (bounded per run).
Exit is zero or non-zero; the log line says fatal or partial and why.
## Surfaces
- CNPG: the Score names the credential Secret per cluster, `<cluster>-app` by
default (CNPG disables superuser access by default). `pg_dumpall
--globals-only --no-role-passwords`, then `pg_dump --format=custom
--compress=0 --lock-wait-timeout=60s` per database over `<cluster>-rw`, then
`pg_restore --list` to reject an unreadable archive. No compression because
restic compresses and compressed dumps defeat deduplication. Superuser is
required only for a cluster with several databases owned by different
roles; the Score checks this at apply time. The image's `pg_dump` is the
highest Postgres major on the platform. RPO is the last successful run;
there is no PITR.
- Tenant S3 buckets: `rclone copy` to scratch with read-only source
credentials, then restic. Old versions on a versioned source bucket are not
copied.
- PVCs: RWX only, decided at apply time. A declared RWO PVC is a Score error;
on Ceph most PVCs are RBD RWO, so this surface mostly covers CephFS. The
volume is `readOnly` at the `persistentVolumeClaim` source, not just the
mount, with `fsGroupChangePolicy: OnRootMismatch`, so kubelet never chowns
the tenant's live data. Files the job's UID cannot read make the run
partial, not fatal. CNPG data PVCs are never listed; the dump covers them.
Each instance is one restic repo at the root of its bucket, with `--host`
pinned to `<cluster-id>-<ns>`. Tags on every snapshot: `run:`, `ns:`,
`surface:pg|s3|pvc`, and `cluster:`, `bucket:` or `pvc:`.
## Deploy and failure visibility
`TenantBackupScore` returns only after one Job created from the CronJob
template exits zero and its snapshot is listed. NetworkPolicy, region,
path-style and TLS mistakes fail the deploy with an actionable message instead
of a 3 AM surprise.
The Score emits a PrometheusRule for user-workload monitoring: alert when
`time() - kube_cronjob_status_last_successful_time` exceeds twice the
schedule interval. An optional dead-man's-switch URL is pinged on success
with the snapshot IDs, which also catches a deleted CronJob or a burned
cluster.
## Schedule, growth and retention
The schedule is a typed Score field: `Daily { at }` (default), `Hourly`, or
`Cron(String)` for anything else. There is no pruning, so the bucket only
grows; restic deduplication keeps each run to the chunks that changed, which
is why dumps are uncompressed. Each run logs the bytes it added, so the
tenant sees the real growth rate instead of an estimate.
To drop old backups the tenant opens a ticket. The platform disables the
writer key; the grace period, no shorter than the Object Lock retention,
starts then. The tenant provisions a new instance, and the platform deletes
the old bucket after the grace period.
## Restore
Restore needs only the tenant's vault entry, restic and rclone, never the
burned cluster. Pick
the snapshot by the S3 `LastModified` of `snapshots/<id>`, which the server
sets and Object Lock preserves, never by `latest` or restic's own timestamp:
someone holding the writer credentials and the password can write a fake
newer or backdated snapshot. restic 0.18 or later:
```bash
export RESTIC_REPOSITORY="s3:https://<target-endpoint>/<instance-bucket>"
export RESTIC_PASSWORD='<instance restic password>'
export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_DEFAULT_REGION=default
R=(restic -o s3.bucket-lookup=path)
"${R[@]}" snapshots --tag "ns:tenant-x"
# one database, into a NEW CNPG cluster, never over the broken one
"${R[@]}" restore <snapshot-id> --target /restore
psql -h new-cluster-rw -U postgres -f /restore/pg/app/globals.sql # first; "role exists" is fine
pg_restore -h new-cluster-rw -U postgres -C -d postgres /restore/pg/app/app.dump
# then reset app role passwords from the new cluster's -app Secret
"${R[@]}" restore <snapshot-id> --target /restore # bucket or PVC snapshot
```
The restore target needs superuser enabled on the new cluster. Dumps are
staged at `pg/<cluster>/<db>.dump` and `pg/<cluster>/globals.sql`; the job
keeps that layout fixed.
If the repo was overwritten, read it as it was before the incident without
changing anything in the bucket:
```bash
RCLONE_S3_VERSION_AT='<time before incident>' \
restic --no-lock -r rclone:offsite:<instance-bucket> restore <snapshot-id> --target /restore
```
Tell the tenant the cut time before starting: everything after the last dump
is lost. A Zitadel database is useless without its master key, which must
come from outside the cluster.
Drill: once before the schedule is trusted, then yearly or when the vault
owner changes. On a machine with no cluster access, using only the vault
entry:
`snapshots`, restore one dump, restore it into a scratch CNPG cluster, run the
app against it.
## Code layout
```
tenant_backup/ runtime binary, no harmony dep: dump, copy, restic, check
tenant_backup_deploy/ TenantBackupScore<T: K8sclient>, via harmony_cli
harmony modules/storage BackupInstanceScore<T: ObjectStorageAdmin>
harmony infra/rgw ObjectStorageAdmin for Ceph RGW (accounts, IAM users, Object Lock)
```
`ObjectStorageAdmin` is a capability, not an RGW binding. The existing
`ObjectBucketScore` cannot be reused: an ObjectBucketClaim cannot set Object
Lock. Estimated size 1.3 to 1.6k lines; v0's restic/rclone/pg wrappers and
naming helpers carry over.
## What changes from v0
The prototype gets the job shape right (stage locally, restic offsite, no
delete on the happy path). These parts go:
| v0 | This design |
|---|---|
| Writer credentials shared by every tenant on a cluster (`<cluster-id>/*`) | One writer per instance, IAM user in the platform account |
| Append-only by IAM alone | Versioning and Object Lock |
| `-setup` binary from a workstation | `TenantBackupScore` in the tenant's deploy code |
| Password minted by setup | Generated in the tenant's password manager, pasted once through `harmony_config` |
| Job logs in to OpenBao with a long-lived Zitadel key in the namespace | Short-lived projected token (pending) |
| `<cluster>-superuser` assumed | Named Secret, `<cluster>-app` by default |
| Compressed `-Fc` dumps | `--compress=0`, restic compresses |
| Monthly repo rotation | New instance, old bucket deleted after the grace period |
| Planner/worker split, snapshot-to-clone for RWO | Two CronJobs; RWX only |
| Runtime binary depends on harmony | Runtime and deploy crates split |
## Rejected
- One cluster-wide job with one repo and one password, reading every tenant's
secrets through a ClusterRole. It fits a single organisation, not tenants
who must not see each other's data, and one failing tenant fails everyone.
- An operator watching a backup CRD in each namespace. It reimplements what
CronJob already schedules. It may come back with tenant self-service or
per-tenant coverage status, behind the same Score.
- `age` per artifact to a tenant public key, instead of restic. The job would
hold no key that reads history, but every run is a full copy and hourly runs
become unaffordable. Anyone who controls the cluster already has the live
data.
- Password escrow, Shamir splits, HSM wrapping. The job must hold the
plaintext key at runtime, and each adds a key or a ceremony to lose.
- Dumping from the `-ro` replica, co-mounting RWO volumes by pod affinity,
`rclone check` after `rclone copy`: more failure modes than they remove at
this size.
## Open questions
1. Job identity (above).
CNPG's generated server certificate lists `<cluster>-rw`, `.<ns>`, `.<ns>.svc`
and `.<ns>.svc.<cluster-domain>` (`api/v1/cluster_funcs.go`
`GetClusterAltDNSNames`), so `verify-full` against `<cluster>-rw.<ns>.svc`
with `<cluster>-ca`/`ca.crt` works without extra configuration.

22
examples/harmony_tenant_backup/Dockerfile Normal file
View File

@@ -0,0 +1,22 @@
# Build first: cargo build --release -p harmony-tenant-backup
# Build context = target/release (COPY paths), e.g.
# docker build -f examples/harmony_tenant_backup/Dockerfile target/release
FROM docker.io/library/debian:13-slim
# The real pin is this image's digest (setup refuses tags). These two have
# upstream release cycles outside apt's: the forum bug was rclone 1.60->1.71.
ARG RESTIC_VERSION=0.18.1
ARG RCLONE_VERSION=v1.71.0
RUN apt-get update && apt-get install -y --no-install-recommends \
ca-certificates curl unzip bzip2 postgresql-client tini \
&& curl -fsSL "https://github.com/restic/restic/releases/download/v${RESTIC_VERSION}/restic_${RESTIC_VERSION}_linux_amd64.bz2" \
| bunzip2 > /usr/local/bin/restic && chmod +x /usr/local/bin/restic \
&& curl -fsSL "https://downloads.rclone.org/${RCLONE_VERSION}/rclone-${RCLONE_VERSION}-linux-amd64.zip" -o /tmp/r.zip \
&& unzip -j /tmp/r.zip '*/rclone' -d /usr/local/bin && chmod +x /usr/local/bin/rclone && rm /tmp/r.zip \
&& apt-get clean && rm -rf /var/lib/apt/lists/*
COPY harmony-tenant-backup /usr/local/bin/harmony-tenant-backup
USER 65534:65534
ENTRYPOINT ["/usr/bin/tini","--","/usr/local/bin/harmony-tenant-backup"]

136
examples/harmony_tenant_backup/README.md Normal file
View File

@@ -0,0 +1,136 @@
# harmony-tenant-backup
> **Reference only.** This is the v0 prototype. The current plan is
> [DESIGN.md](DESIGN.md), which replaces the setup binary, shared credentials
> and monthly rotation described below.
Backup one tenant namespace. Not restore. Not retention. Not PITR.
Nightly CronJob covering CNPG dumps, opt-in S3 buckets, and PVCs (direct
mount or snapshot-clone). One restic repo per namespace on the off-site RGW.
Two binaries, same split as `harmony-fleet` vs `harmony-fleet-deploy`:
- `harmony-tenant-backup run` — operate (in-cluster, or a laptop dry-run)
- `harmony-tenant-backup-setup` — install RBAC + CronJob (workstation)
Out of scope: OpenBao, Ceph, K8s objects, Zitadel.
## Data layout
- Repo: `s3:https://<target-endpoint>/<target-bucket>/<cluster-id>/<ns>`, or
`.../<YYYY-MM>` with `rotate_repo_monthly` (expiry = delete a whole prefix).
- Host pinned to `<cluster-id>-<ns>`. Tags: `run:`, `ns:`, `surface:pg|s3|pvc`,
plus `cluster:` / `bucket:` / `pvc:`.
## Runbook
1. Build and push the image:
```bash
cargo build --release -p harmony-tenant-backup
docker build -f examples/harmony_tenant_backup/Dockerfile \
-t harbor.example/harmony/ns-backup:v0 target/release
docker push harbor.example/harmony/ns-backup:v0
```
2. Config at `<mount>/data/harmony/<ns>/TenantBackup` (`NAMESPACE` maps to
store path `harmony/<ns>`):
```bash
bao kv put -mount=secret harmony/tenant-x/TenantBackup \
target_endpoint=https://rgw.site-b target_bucket=tenant-backups \
target_access_key=... target_secret_key=... target_region=default \
source_endpoint=https://rgw.site-A source_access_key=... \
source_secret_key=... rotate_repo_monthly=true
```
`target_*`: append-only at the target. `source_*`: read-only site-A, only
for buckets. `target_ca_pem`: private-CA target only. Never skip TLS.
Do not set `restic_password`: setup mints it into OpenBao for the job
and prints it once for the offsite kit (KeepassX). Never rotate it.
3. One vault credential in the namespace, plus the bucket opt-in:
```bash
kubectl create secret generic harmony-tenant-backup-zitadel-key \
-n tenant-x --from-file=key.json=<machine-key.json>
kubectl annotate ns tenant-x backup.platform/buckets=tenant-x-assets,tenant-x-uploads
```
4. Install:
```bash
OPENBAO_URL=... OPENBAO_JWT_ROLE=... HARMONY_SSO_URL=... HARMONY_ZITADEL_AUDIENCE=... \
HARMONY_ZITADEL_KEY_JSON="$(cat machine-key.json)" \
cargo run -p harmony-tenant-backup --bin harmony-tenant-backup-setup -- \
--namespace tenant-x --cluster-id site-a-prod \
--schedule "17 3 * * *" \
--image harbor.example/harmony/ns-backup@sha256:<digest>
```
Image must be digest-pinned. Stagger schedules per namespace.
5. Run now / watch:
```bash
kubectl create job backup-now --from=cronjob/harmony-tenant-backup -n tenant-x
kubectl logs -n tenant-x job/backup-now -f
```
Laptop rehearsal needs the same OpenBao login as setup, plus the
machine key as `HARMONY_ZITADEL_KEY_JSON` (or `HARMONY_ZITADEL_KEY_PATH`):
```bash
HARMONY_ZITADEL_KEY_JSON="$(cat machine-key.json)" \
harmony-tenant-backup run --namespace tenant-x --cluster-id site-a-prod \
--dry-run --scratch /tmp/htb
```
Exit codes: 0 clean, 2 fatal (no repo, no config), 3 partial. Setup exits 1.
## Restore
Fire restore uses the offsite kit (KeepassX), never OpenBao or the burned
cluster. Same kit as OpenBao unseal keys. Target S3 creds + restic ≥ 0.18:
```bash
export RESTIC_REPOSITORY="s3:https://rgw.site-b/tenant-backups/site-a-prod/tenant-x/2026-08"
export RESTIC_PASSWORD='<restic_password from KeepassX>'
export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_DEFAULT_REGION=default
R=(restic -o s3.bucket-lookup=path)
"${R[@]}" snapshots --tag "ns:tenant-x"
# one database, into a NEW CNPG cluster. Monthly rotation: repo path ends /YYYY-MM;
# "latest" is only that month. --tag cluster:<cnpg-cluster-name>
"${R[@]}" restore latest --tag surface:pg --tag cluster:zito --target /restore/pg/zito
pg_restore -h new-cluster-rw -U postgres -d app /restore/pg/zito/app.dump
psql -h new-cluster-rw -U postgres -f /restore/pg/zito/globals.sql
"${R[@]}" restore latest --tag bucket:tenant-x-assets --target /restore/assets
"${R[@]}" restore latest --tag pvc:media --target /restore/media
```
`globals.sql` is `pg_dumpall --globals-only`: without it, app roles are gone.
State the RPO to the tenant before starting.
## Retention
No `restic forget`/`prune`. Monthly rotation expires a whole prefix, by hand:
`rclone delete --rmdirs ":s3:<bucket>/<cluster>/<ns>/2026-05"`. A human on
the delete button is the design.
## Pilot checklist
1. `harmony-tenant-backup run --namespace <ns> --cluster-id ... --dry-run --scratch /tmp/htb`
2. NetworkPolicy on CNPG: planner cannot reach `<cluster>-rw`. First failure.
3. SigV4 = region mismatch; `SecondLevelDomainDenied` = path-style stripped.
4. emptyDir peak = largest dump or bucket (CronJob ships 5Gi).
5. `restic unlock` if a killed Job left a lock (`--retry-lock 5m` covers blips).
6. Restore one database and one PVC into a scratch namespace. Run the app.
## Development
`cargo test -p harmony-tenant-backup`. Acceptance is the pilot.
`src/common.rs` is the names both binaries share.

351
examples/harmony_tenant_backup/src/common.rs Normal file
View File

@@ -0,0 +1,351 @@
//! Shared contract: every name, label, and env var setup writes is what
//! the runtime reads.
use chrono::{DateTime, Utc};
use k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, SecretKeySelector};
use std::collections::BTreeMap;
/// Both binaries construct pod env; one place defines the shape.
pub fn env_var(name: &str, value: &str) -> EnvVar {
EnvVar {
name: name.into(),
value: Some(value.into()),
..Default::default()
}
}
/// Secret-as-env, not a volume: OKD restricted-v2 rejects a pinned
/// fsGroup, and without fsGroup a 0400 secret mount is unreadable.
pub fn env_from_secret(name: &str, secret: &str, key: &str) -> EnvVar {
EnvVar {
name: name.into(),
value_from: Some(EnvVarSource {
secret_key_ref: Some(SecretKeySelector {
name: secret.into(),
key: key.into(),
optional: Some(false),
}),
..Default::default()
}),
..Default::default()
}
}
pub const MANAGED_BY: &str = "app.kubernetes.io/managed-by";
pub const MANAGED_BY_VALUE: &str = "harmony-tenant-backup";
pub const SA_NAME: &str = "harmony-tenant-backup";
pub const CRONJOB_NAME: &str = "harmony-tenant-backup";
/// Comma-separated bucket list opt-in, set by the tenant/platform on the
/// namespace. Absent = "no S3 surface here" (and a coverage gap worth
/// reporting, never a silent skip).
pub const A_BUCKETS: &str = "backup.platform/buckets";
/// Per-PVC opt-out. `true` skips backup, on the record.
pub const A_PVC_DISABLED: &str = "backup.platform/disabled";
/// Non-secret OpenBao/Zitadel coordinates forwarded verbatim from the planner
/// environment into worker Jobs. The machine key is env-from-secret
/// (`HARMONY_ZITADEL_KEY_JSON`), not forwarded and not a volume.
pub const FORWARD_ENV: [&str; 4] = [
"OPENBAO_URL",
"OPENBAO_JWT_ROLE",
"HARMONY_SSO_URL",
"HARMONY_ZITADEL_AUDIENCE",
];
pub const ENV_NAMESPACE: &str = "NAMESPACE";
pub const ENV_CLUSTER_ID: &str = "CLUSTER_ID";
pub const ENV_WORKER_IMAGE: &str = "WORKER_IMAGE";
pub const ENV_RUN_ID: &str = "RUN_ID";
pub const ENV_SCRATCH: &str = "SCRATCH";
pub const DEFAULT_SCRATCH: &str = "/scratch";
pub const WORK_MOUNT: &str = "/mnt/data";
pub const ZITADEL_SECRET: &str = "harmony-tenant-backup-zitadel-key";
pub const ENV_ZITADEL_SECRET: &str = "ZITADEL_SECRET";
pub const ZITADEL_KEY_FILE: &str = "key.json";
pub const ENV_ZITADEL_KEY_JSON: &str = "HARMONY_ZITADEL_KEY_JSON";
pub const A_SCC_UID_RANGE: &str = "openshift.io/sa.scc.uid-range";
/// OKD `restricted-v2` injects fsGroup from this annotation and rejects
/// a pin outside the range. k3d has no SCC: pin 65534 to match the
/// image USER so emptyDir and PVC clones are readable.
pub fn volume_fsgroup(annotations: Option<&BTreeMap<String, String>>) -> Option<i64> {
match annotations.and_then(|a| a.get(A_SCC_UID_RANGE)) {
Some(_) => None,
None => Some(65534),
}
}
/// The planner's own CronJob budget; setup writes it, the runtime bounds its
/// patience by WORKER_DEADLINE_SECS.
pub const PLANNER_DEADLINE_SECS: i64 = 6 * 3600;
// One worker push may run the full 4h; the planner's per-job patience is the
// worker deadline plus slack, and the planner's own CronJob deadline must
// exceed snapshot + clone + worker budgets. Mismatching these orphans
// run-id-named snapshots when the pod is killed mid-cleanup.
pub const WORKER_DEADLINE_SECS: i64 = 4 * 3600;
/// Job/snapshot/clone names are DNS-1123: lowercase, digits, hyphen.
pub fn default_run_id(now: DateTime<Utc>) -> String {
now.format("%Y%m%dt%H%M%Sz").to_string()
}
pub fn dns1123_run_id(s: &str) -> Option<String> {
let s = s.to_ascii_lowercase();
let ok = (1..=32).contains(&s.len())
&& s.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-');
ok.then_some(s)
}
/// restic `--host`: never the pod name, or "restore latest" goes ambiguous
/// across rebuilt Jobs.
pub fn restic_host(cluster_id: &str, namespace: &str) -> String {
format!("{cluster_id}-{namespace}")
}
pub fn repo_prefix(cluster_id: &str, namespace: &str) -> String {
format!("{cluster_id}/{namespace}")
}
/// Tokens of the `backup.platform/buckets` annotation. Tenant-authored, so
/// hold them to S3's own grammar: a token escaping it could climb out of the
/// staging dir (`..`), or parse as an rclone connection-string option that
/// redirects the source credentials somewhere else.
pub fn valid_bucket_token(b: &str) -> bool {
let alnum = |c: char| c.is_ascii_lowercase() || c.is_ascii_digit();
!b.is_empty()
&& b.len() <= 63
&& b.chars().all(|c| alnum(c) || c == '.' || c == '-')
&& !b.contains("..")
&& b.starts_with(alnum)
&& b.ends_with(alnum)
}
/// restic repo URL. Monthly rotation puts each run in a fresh repo per
/// period; expiry then means deleting a whole prefix, never partial-repo
/// surgery on immutable shared packs.
pub fn repo_path(
endpoint: &str,
bucket: &str,
cluster_id: &str,
namespace: &str,
monthly: bool,
now: DateTime<Utc>,
) -> String {
// `s3:` prefix mandatory: a bare https:// location parses as the REST
// backend, silently writing nothing.
let base = format!(
"s3:{endpoint}/{bucket}/{}",
repo_prefix(cluster_id, namespace)
);
if monthly {
format!("{base}/{}", now.format("%Y-%m"))
} else {
base
}
}
/// restic backup args: one pinned host + tag scheme for every push, so
/// restores address snapshots exactly per the README's restore interface.
pub fn backup_tags(run_id: &str, namespace: &str, surface: &str) -> Vec<String> {
vec![
format!("run:{run_id}"),
format!("ns:{namespace}"),
format!("surface:{surface}"),
]
}
/// K8s names are DNS-1123 and 63 chars max: `<prefix>-<pvc>-<run_id>`,
/// truncated front-first on the volatile parts to stay collision-free
/// within a run.
pub fn run_object_name(prefix: &str, pvc: &str, run_id: &str) -> String {
let full = format!("{prefix}-{pvc}-{run_id}");
if full.len() <= 63 {
return full;
}
let keep = 63usize
.saturating_sub(prefix.len())
.saturating_sub(run_id.len())
.saturating_sub(2);
let mut cut = pvc.chars().take(keep).collect::<String>();
cut = cut.trim_matches(|c| c == '-' || c == '.').to_string();
format!("{prefix}-{cut}-{run_id}")
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PvcDecision {
/// Mount the PVC directly (RWX and unattached).
Direct,
/// RWO: VolumeSnapshot -> clone -> mount the clone.
SnapshotFirst,
Skip(&'static str),
/// Cannot be backed up by this mechanism; must surface as a failure.
Fail(&'static str),
}
/// Classify one PVC for backup. `cnpg_owner` is the `cnpg.io/cluster` label.
pub fn decide_pvc(
phase: &str,
cnpg_owner: Option<&str>,
disabled: bool,
block: bool,
access_mode: &str,
) -> PvcDecision {
if phase != "Bound" {
return PvcDecision::Skip("phase is not Bound");
}
if cnpg_owner.is_some() {
return PvcDecision::Skip("owned by a CNPG cluster (covered by pg_dump)");
}
if disabled {
return PvcDecision::Skip("backup.platform/disabled=true");
}
if block {
return PvcDecision::Fail(
"volumeMode=Block: restic cannot read it, needs app-native backup",
);
}
if access_mode == "ReadWriteMany" {
PvcDecision::Direct
} else {
// RWO/RWX-but-attached volumes block multi-attach: snapshot -> clone.
PvcDecision::SnapshotFirst
}
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::TimeZone;
fn t() -> DateTime<Utc> {
Utc.with_ymd_and_hms(2026, 9, 17, 3, 17, 0).unwrap()
}
#[test]
fn repo_path_plain_and_monthly() {
assert_eq!(
repo_path(
"https://rgw.site-b",
"tenant-backups",
"site-a-prod",
"tenant-x",
false,
t()
),
"s3:https://rgw.site-b/tenant-backups/site-a-prod/tenant-x"
);
assert_eq!(
repo_path(
"https://rgw.site-b",
"tenant-backups",
"site-a-prod",
"tenant-x",
true,
t()
),
"s3:https://rgw.site-b/tenant-backups/site-a-prod/tenant-x/2026-09"
);
}
#[test]
fn fsgroup_k3d_vs_okd() {
assert_eq!(volume_fsgroup(None), Some(65534));
let mut okd = BTreeMap::new();
okd.insert(A_SCC_UID_RANGE.into(), "1000700000/10000".into());
assert_eq!(volume_fsgroup(Some(&okd)), None);
}
#[test]
fn host_is_pinned_not_pod_name() {
assert_eq!(
restic_host("site-a-prod", "tenant-x"),
"site-a-prod-tenant-x"
);
}
#[test]
fn run_object_name_stays_dns_safe() {
let long_pvc = "a".repeat(80);
let n = run_object_name("htb-wk", &long_pvc, "20260917t031700z");
assert!(n.len() <= 63, "{n}");
assert!(n.starts_with("htb-wk-") && n.ends_with("-20260917t031700z"));
assert_eq!(
run_object_name("htb-snap", "media", "R1"),
"htb-snap-media-R1"
);
let long_run = "r".repeat(70);
let n = run_object_name("htb-wk", "media", &long_run);
assert!(
n.len() <= 70 + 20,
"no underflow, api rejects long ones: {n}"
);
}
#[test]
fn bucket_tokens_hold_to_s3_grammar() {
assert!(valid_bucket_token("tenant-x.assets-1"));
for bad in [
"",
"-lead",
"UPPER",
"../etc",
"a..b",
"bucket,opt=evil",
"x/y",
] {
assert!(!valid_bucket_token(bad), "{bad}");
}
}
#[test]
fn classification() {
assert_eq!(
decide_pvc("Bound", None, false, false, "ReadWriteOnce"),
PvcDecision::SnapshotFirst
);
assert_eq!(
decide_pvc("Bound", None, false, false, "ReadWriteMany"),
PvcDecision::Direct
);
assert!(matches!(
decide_pvc("Pending", None, false, false, "ReadWriteOnce"),
PvcDecision::Skip(_)
));
assert!(matches!(
decide_pvc("Bound", Some("zito"), false, false, "ReadWriteOnce"),
PvcDecision::Skip(_)
));
assert!(matches!(
decide_pvc("Bound", None, true, false, "ReadWriteOnce"),
PvcDecision::Skip(_)
));
assert!(matches!(
decide_pvc("Bound", None, false, true, "ReadWriteOnce"),
PvcDecision::Fail(_)
));
}
#[test]
fn run_id_is_dns1123() {
let id = default_run_id(t());
assert_eq!(id, "20260917t031700z");
assert_eq!(
dns1123_run_id("20260917T031700Z").as_deref(),
Some("20260917t031700z")
);
assert!(dns1123_run_id("Nope_Underscore").is_none());
assert!(dns1123_run_id(&"a".repeat(33)).is_none());
}
#[test]
fn tags_are_the_restore_addressing_scheme() {
assert_eq!(
backup_tags("20260917T031700Z", "tenant-x", "pg"),
vec!["run:20260917T031700Z", "ns:tenant-x", "surface:pg"]
);
}
}

345
examples/harmony_tenant_backup/src/exec.rs Normal file
View File

@@ -0,0 +1,345 @@
//! Subprocess wrappers: restic, rclone, postgres client. Credentials travel
//! via environment only, never argv.
use anyhow::{Context, Result, anyhow};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use tokio::process::Command;
/// Single place where the restic repo path + credentials are assembled, so
/// planner and worker can never disagree about where data lands.
pub fn build_restic(
cfg: &crate::TenantBackup,
cluster_id: &str,
ns: &str,
ca_file: Option<PathBuf>,
scratch: &Path,
) -> Restic {
Restic::new(
crate::common::repo_path(
&cfg.target_endpoint,
&cfg.target_bucket,
cluster_id,
ns,
cfg.rotate_repo_monthly,
chrono::Utc::now(),
),
cfg.restic_password.clone(),
cfg.target_access_key.clone(),
cfg.target_secret_key.clone(),
cfg.target_region.clone(),
ca_file,
scratch,
)
}
pub fn logline(msg: &str) {
eprintln!("[{}] {msg}", chrono::Utc::now().format("%H:%M:%S"));
}
pub async fn run_capture(program: &str, args: &[&str], env: &[(&str, &str)]) -> Result<String> {
let out = Command::new(program)
.args(args)
.envs(env.iter().copied())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.output()
.await
.with_context(|| format!("spawn {program}"))?;
if !out.status.success() {
return Err(anyhow!(
"{program} failed ({})\nstderr: {}",
out.status,
tail(&String::from_utf8_lossy(&out.stderr))
));
}
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
}
async fn run_quiet(program: &str, args: &[&str], env: &[(&str, &str)]) -> Result<()> {
run_capture(program, args, env).await.map(|_| ())
}
fn strs(v: &[String]) -> Vec<&str> {
v.iter().map(String::as_str).collect()
}
fn tail(s: &str) -> String {
s.trim()
.lines()
.rev()
.take(8)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect::<Vec<_>>()
.join("\n")
}
pub struct Restic {
repo: String,
password: String,
access_key: String,
secret_key: String,
region: Option<String>,
ca_file: Option<PathBuf>,
cache_dir: PathBuf,
}
impl Restic {
pub fn repo(&self) -> &str {
&self.repo
}
pub fn new(
repo: String,
password: String,
access_key: String,
secret_key: String,
region: Option<String>,
ca_file: Option<PathBuf>,
scratch: &Path,
) -> Self {
Self {
repo,
password,
access_key,
secret_key,
region,
ca_file,
cache_dir: scratch.join("cache"),
}
}
fn global_args(&self) -> Vec<String> {
let mut a = vec![
"-r".into(),
self.repo.clone(),
"-o".into(),
"s3.bucket-lookup=path".into(),
"--cache-dir".into(),
self.cache_dir.display().to_string(),
"--retry-lock".into(),
"5m".into(),
];
if let Some(ca) = &self.ca_file {
a.push(format!("--cacert={}", ca.display()));
}
a
}
fn env(&self) -> Vec<(&'static str, &str)> {
// restic's child-process contract. The value is TenantBackup.restic_password
// from harmony_config (OpenBao), never a RESTIC_PASSWORD on the pod.
let mut e = vec![
("RESTIC_PASSWORD", self.password.as_str()),
("AWS_ACCESS_KEY_ID", self.access_key.as_str()),
("AWS_SECRET_ACCESS_KEY", self.secret_key.as_str()),
];
if let Some(r) = &self.region {
// restic reads AWS_DEFAULT_REGION, not AWS_REGION.
e.push(("AWS_DEFAULT_REGION", r.as_str()));
}
e
}
pub async fn ensure_repo(&self, dry_run: bool) -> Result<()> {
let mut args = self.global_args();
args.extend(["cat".into(), "config".into()]);
if run_quiet("restic", &strs(&args), &self.env()).await.is_ok() {
return Ok(());
}
logline(&format!("initialising repo {}", self.repo));
if dry_run {
return Ok(());
}
let mut args = self.global_args();
args.push("init".into());
run_quiet("restic", &strs(&args), &self.env())
.await
.context("restic init")
}
/// The one push path: pinned host + `run:`/`ns:`/`surface:` tags.
#[allow(clippy::too_many_arguments)]
pub async fn backup(
&self,
host: &str,
run_id: &str,
namespace: &str,
surface: &str,
extra_tags: &[String],
path: &str,
dry_run: bool,
) -> Result<()> {
if dry_run {
logline(&format!("DRY-RUN: would push {path} ({surface})"));
return Ok(());
}
let mut args = self.global_args();
args.push("backup".into());
args.extend(["--host".into(), host.into()]);
for tag in crate::common::backup_tags(run_id, namespace, surface)
.into_iter()
.chain(extra_tags.iter().cloned())
{
args.extend(["--tag".into(), tag]);
}
args.push(path.into());
let out = run_capture("restic", &strs(&args), &self.env()).await?;
logline(&format!(
"pushed {path} ({surface}): {}",
tail(&out).lines().last().unwrap_or("ok")
));
Ok(())
}
pub async fn snapshots_tagged(&self, tag: &str) -> Result<()> {
let mut args = self.global_args();
args.extend(["snapshots".into(), "--tag".into(), tag.into()]);
let out = run_capture("restic", &strs(&args), &self.env()).await?;
eprint!("{out}");
Ok(())
}
}
/// Read-only tenant source on the site-A RGW (bucket credentials).
pub struct SrcS3 {
pub endpoint: String,
pub region: String,
pub access_key: String,
pub secret_key: String,
}
impl SrcS3 {
fn args(&self) -> Vec<String> {
vec![
"--s3-endpoint".into(),
self.endpoint.clone(),
"--s3-provider=Ceph".into(),
"--s3-region".into(),
self.region.clone(),
"--s3-force-path-style".into(),
"--s3-env-auth".into(),
"--s3-no-check-bucket".into(),
"--disable-http2".into(),
]
}
fn env(&self) -> Vec<(&'static str, &str)> {
vec![
("AWS_ACCESS_KEY_ID", self.access_key.as_str()),
("AWS_SECRET_ACCESS_KEY", self.secret_key.as_str()),
]
}
fn cmd<'a>(&self, sub: &'a str, paths_and_flags: &'a [String]) -> Vec<&'a str> {
let mut argv: Vec<&str> = vec![sub];
argv.extend(paths_and_flags.iter().map(String::as_str));
argv
}
/// `copy`, never `sync`: add/overwrite only, deletes never propagate.
pub async fn copy(&self, bucket: &str, dst: &Path) -> Result<()> {
let mut args = vec![":s3:".to_string() + bucket, dst.display().to_string()];
args.extend(self.args());
args.extend([
"--transfers".into(),
"8".into(),
"--checkers".into(),
"16".into(),
"--stats-one-line".into(),
"-v".into(),
"--ignore-checksum".into(),
]);
run_quiet("rclone", &self.cmd("copy", &args), &self.env())
.await
.context("rclone copy")
}
/// Count-only assertion against the staged copy (multi-site ETags differ
/// by construction, so content hashes would always "mismatch").
pub async fn check(&self, bucket: &str, dst: &Path) -> Result<()> {
let mut args = vec![":s3:".to_string() + bucket, dst.display().to_string()];
args.extend(self.args());
args.extend(["--one-way".into(), "--count".into()]);
run_quiet("rclone", &self.cmd("check", &args), &self.env())
.await
.context("rclone check mismatch: staged copy not trustworthy")
}
pub async fn list_dirs(&self, bucket: &str) -> Result<()> {
let mut args = vec![":s3:".to_string() + bucket];
args.extend(self.args());
let out = run_capture("rclone", &self.cmd("lsd", &args), &self.env()).await?;
eprint!("{out}");
Ok(())
}
}
/// All postgres wire calls go through the CNPG `<cluster>-rw` service with
/// TLS verified against the cluster CA secret.
pub struct Pg {
conn_base: String,
password: String,
}
impl Pg {
pub fn new(cluster: &str, namespace: &str, ca_file: &Path, password: String) -> Self {
let conn_base = format!(
"host={cluster}-rw.{namespace}.svc port=5432 user=postgres sslmode=verify-ca sslrootcert={}",
ca_file.display()
);
Self {
conn_base,
password,
}
}
fn conn(&self, db: &str) -> String {
format!("{} dbname={db}", self.conn_base)
}
fn env(&self) -> Vec<(&'static str, &str)> {
vec![("PGPASSWORD", self.password.as_str())]
}
pub async fn query(&self, db: &str, sql: &str) -> Result<String> {
run_capture("psql", &[&self.conn(db), "-Atc", sql], &self.env()).await
}
pub async fn dump_all_globals(&self, out_file: &Path) -> Result<()> {
run_quiet(
"pg_dumpall",
&[
&self.conn("postgres"),
"--globals-only",
"-f",
&out_file.display().to_string(),
],
&self.env(),
)
.await
.context("pg_dumpall --globals-only: a DB-only dump loses the app roles")
}
pub async fn dump_db(&self, db: &str, out_file: &Path) -> Result<()> {
run_quiet(
"pg_dump",
&[
&self.conn(db),
"--format=custom",
"--file",
&out_file.display().to_string(),
],
&self.env(),
)
.await
.context("pg_dump")
}
/// The artifact must be a readable archive, not just a file that exists.
pub async fn verify_dump(path: &Path) -> Result<()> {
run_quiet("pg_restore", &["--list", &path.display().to_string()], &[]).await
}
}

475
examples/harmony_tenant_backup/src/k8s.rs Normal file
View File

@@ -0,0 +1,475 @@
//! Cluster mechanics: discovery reads, VolumeSnapshot/clone lifecycle,
//! worker-Job creation and completion polling.
use crate::common::{
ENV_ZITADEL_KEY_JSON, MANAGED_BY, MANAGED_BY_VALUE, WORKER_DEADLINE_SECS, ZITADEL_KEY_FILE,
env_from_secret, env_var,
};
use anyhow::{Result, anyhow};
use k8s_openapi::api::core::v1::{
Capabilities, EmptyDirVolumeSource, PersistentVolumeClaim, PersistentVolumeClaimSpec,
PodSecurityContext, SeccompProfile, TypedLocalObjectReference, Volume, VolumeMount,
VolumeResourceRequirements,
};
use k8s_openapi::api::{batch::v1::Job, core::v1::SecurityContext};
use k8s_openapi::apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::ObjectMeta};
use kube::api::{
Api, ApiResource, DeleteParams, DynamicObject, ListParams, PostParams, PropagationPolicy,
};
use kube::{Client, ResourceExt};
use std::collections::BTreeMap;
use std::time::Duration;
use tokio::time::interval;
pub fn api_resource(group: &str, version: &str, kind: &str, plural: &str) -> ApiResource {
ApiResource {
group: group.into(),
version: version.into(),
api_version: format!("{group}/{version}"),
kind: kind.into(),
plural: plural.into(),
}
}
pub fn cnpg_clusters() -> ApiResource {
api_resource("postgresql.cnpg.io", "v1", "Cluster", "clusters")
}
pub fn volume_snapshots() -> ApiResource {
api_resource(
"snapshot.storage.k8s.io",
"v1",
"VolumeSnapshot",
"volumesnapshots",
)
}
pub fn volume_snapshot_classes() -> ApiResource {
api_resource(
"snapshot.storage.k8s.io",
"v1",
"VolumeSnapshotClass",
"volumesnapshotclasses",
)
}
fn sorted_names(list: kube::core::ObjectList<DynamicObject>) -> Vec<String> {
let mut names: Vec<String> = list
.iter()
.filter_map(|i| i.metadata.name.clone())
.collect();
names.sort();
names
}
pub async fn list_namespaced_names(
client: &Client,
ns: &str,
ar: &ApiResource,
) -> Result<Vec<String>> {
let api: Api<DynamicObject> = Api::namespaced_with(client.clone(), ns, ar);
Ok(sorted_names(api.list(&ListParams::default()).await?))
}
/// The one cluster-scoped name, if any. Picking silently when more than one
/// exists can route a snapshot to the wrong CSI driver.
pub async fn first_cluster_scoped_name(
client: &Client,
ar: &ApiResource,
) -> Result<Option<String>> {
let api: Api<DynamicObject> = Api::all_with(client.clone(), ar);
let names = sorted_names(api.list(&ListParams::default()).await?);
match names.as_slice() {
[] => Ok(None),
[one] => Ok(Some(one.clone())),
many => Err(anyhow!(
"{} has {} classes and none is configured; set snapshot_class (found {})",
ar.kind,
many.len(),
many.join(", ")
)),
}
}
pub async fn ns_annotations(client: &Client, ns: &str) -> Result<BTreeMap<String, String>> {
Ok(
Api::<k8s_openapi::api::core::v1::Namespace>::all(client.clone())
.get(ns)
.await?
.metadata
.annotations
.unwrap_or_default(),
)
}
pub async fn annotation(client: &Client, ns: &str, key: &str) -> Result<Option<String>> {
Ok(ns_annotations(client, ns).await?.get(key).cloned())
}
/// Restricted-v2 compatible container lock-down. No runAsUser: OKD assigns
/// the namespace UID; k3d uses the image USER.
pub fn container_security() -> SecurityContext {
SecurityContext {
allow_privilege_escalation: Some(false),
read_only_root_filesystem: Some(true),
run_as_non_root: Some(true),
capabilities: Some(Capabilities {
drop: Some(vec!["ALL".into()]),
..Default::default()
}),
seccomp_profile: Some(SeccompProfile {
type_: "RuntimeDefault".into(),
..Default::default()
}),
..Default::default()
}
}
pub fn pod_security(fs_group: Option<i64>) -> PodSecurityContext {
PodSecurityContext {
run_as_non_root: Some(true),
fs_group,
seccomp_profile: Some(SeccompProfile {
type_: "RuntimeDefault".into(),
..Default::default()
}),
..Default::default()
}
}
pub async fn secret_bytes(client: &Client, ns: &str, name: &str, key: &str) -> Result<Vec<u8>> {
let secret = Api::<k8s_openapi::api::core::v1::Secret>::namespaced(client.clone(), ns)
.get(name)
.await
.map_err(|e| anyhow!("secret {ns}/{name}: {e}"))?;
secret
.data
.as_ref()
.and_then(|d| d.get(key))
.map(|b| b.0.clone())
.ok_or_else(|| anyhow!("secret {ns}/{name} has no key '{key}'"))
}
fn managed_labels(run_id: &str) -> BTreeMap<String, String> {
BTreeMap::from([
(MANAGED_BY.to_string(), MANAGED_BY_VALUE.to_string()),
("run".to_string(), run_id.to_string()),
])
}
pub async fn create_volume_snapshot(
client: &Client,
ns: &str,
name: &str,
class: &str,
pvc: &str,
run_id: &str,
) -> Result<()> {
let ar = volume_snapshots();
let mut obj = DynamicObject::new(name, &ar);
obj.metadata.namespace = Some(ns.into());
obj.metadata.labels = Some(managed_labels(run_id));
obj.data["spec"] = serde_json::json!({
"volumeSnapshotClassName": class,
"source": { "persistentVolumeClaimName": pvc },
});
Api::<DynamicObject>::namespaced_with(client.clone(), ns, &ar)
.create(&PostParams::default(), &obj)
.await?;
Ok(())
}
pub async fn create_clone_pvc(
client: &Client,
ns: &str,
name: &str,
source: &PersistentVolumeClaim,
snap: &str,
run_id: &str,
) -> Result<()> {
let size = source
.status
.as_ref()
.and_then(|s| s.capacity.as_ref())
.and_then(|c| c.get("storage"))
.cloned()
.unwrap_or_else(|| Quantity("1Gi".into()));
let mut labels = managed_labels(run_id);
labels.insert("backup.platform/source-pvc".into(), source.name_any());
let pvc = PersistentVolumeClaim {
metadata: ObjectMeta {
name: Some(name.into()),
namespace: Some(ns.into()),
labels: Some(labels),
..Default::default()
},
spec: Some(PersistentVolumeClaimSpec {
access_modes: Some(vec!["ReadWriteOnce".into()]),
volume_mode: Some("Filesystem".into()),
storage_class_name: source
.spec
.as_ref()
.and_then(|s| s.storage_class_name.clone()),
resources: Some(VolumeResourceRequirements {
requests: Some(BTreeMap::from([("storage".into(), size)])),
..Default::default()
}),
data_source: Some(TypedLocalObjectReference {
api_group: Some("snapshot.storage.k8s.io".into()),
kind: "VolumeSnapshot".into(),
name: snap.into(),
}),
..Default::default()
}),
..Default::default()
};
Api::<PersistentVolumeClaim>::namespaced(client.clone(), ns)
.create(&PostParams::default(), &pvc)
.await?;
Ok(())
}
pub struct WorkerSpec<'a> {
pub image: &'a str,
pub ns: &'a str,
pub name: &'a str,
pub pvc_mount: &'a str,
pub pvc_logical: &'a str,
pub run_id: &'a str,
pub cluster_id: &'a str,
/// Same in-cluster Zitadel secret the planner uses (env-from, not a mount).
pub zitadel_secret: &'a str,
/// OpenBao/Zitadel coordinates forwarded from this process's environment.
pub forward_env: Vec<(&'a str, &'a str)>,
/// None on OKD (SCC assigns); Some(65534) on k3d so PVC clones are readable.
pub fs_group: Option<i64>,
}
pub fn worker_job(w: &WorkerSpec) -> Job {
let mut env = vec![
env_var(crate::common::ENV_NAMESPACE, w.ns),
env_var(crate::common::ENV_RUN_ID, w.run_id),
env_var(crate::common::ENV_CLUSTER_ID, w.cluster_id),
env_from_secret(ENV_ZITADEL_KEY_JSON, w.zitadel_secret, ZITADEL_KEY_FILE),
];
env.extend(w.forward_env.iter().map(|(k, v)| env_var(k, v)));
Job {
metadata: ObjectMeta {
name: Some(w.name.into()),
namespace: Some(w.ns.into()),
labels: Some(managed_labels(w.run_id)),
..Default::default()
},
spec: Some(k8s_openapi::api::batch::v1::JobSpec {
backoff_limit: Some(0),
active_deadline_seconds: Some(WORKER_DEADLINE_SECS),
// Keep finished worker pods long enough to read their logs the
// morning after, when the planner only reports FAILURES by name.
ttl_seconds_after_finished: Some(86400),
template: k8s_openapi::api::core::v1::PodTemplateSpec {
metadata: Some(ObjectMeta {
labels: Some(managed_labels(w.run_id)),
..Default::default()
}),
spec: Some(k8s_openapi::api::core::v1::PodSpec {
restart_policy: Some("Never".into()),
service_account_name: Some(crate::common::SA_NAME.into()),
security_context: Some(pod_security(w.fs_group)),
containers: vec![k8s_openapi::api::core::v1::Container {
name: "worker".into(),
image: Some(w.image.into()),
args: Some(vec!["worker".into(), "--pvc".into(), w.pvc_logical.into()]),
env: Some(env),
volume_mounts: Some(vec![
VolumeMount {
name: "data".into(),
mount_path: crate::common::WORK_MOUNT.into(),
read_only: Some(true),
..Default::default()
},
VolumeMount {
name: "scratch".into(),
mount_path: crate::common::DEFAULT_SCRATCH.into(),
..Default::default()
},
]),
resources: Some(k8s_openapi::api::core::v1::ResourceRequirements {
requests: Some(BTreeMap::from([
("cpu".into(), Quantity("100m".into())),
("memory".into(), Quantity("256Mi".into())),
])),
limits: Some(BTreeMap::from([(
"memory".into(),
Quantity("1Gi".into()),
)])),
..Default::default()
}),
security_context: Some(container_security()),
..Default::default()
}],
volumes: Some(vec![
Volume {
name: "data".into(),
persistent_volume_claim: Some(
k8s_openapi::api::core::v1::PersistentVolumeClaimVolumeSource {
claim_name: w.pvc_mount.into(),
read_only: Some(true),
},
),
..Default::default()
},
Volume {
name: "scratch".into(),
empty_dir: Some(EmptyDirVolumeSource {
size_limit: Some(Quantity("2Gi".into())),
..Default::default()
}),
..Default::default()
},
]),
..Default::default()
}),
},
..Default::default()
}),
..Default::default()
}
}
pub async fn spawn_and_wait_job(
client: &Client,
ns: &str,
job: &Job,
patience: Duration,
) -> Result<()> {
let api = Api::<Job>::namespaced(client.clone(), ns);
let name = job.metadata.name.clone().unwrap_or_default();
api.create(&PostParams::default(), job).await?;
wait_job(client, ns, &name, patience).await
}
pub async fn wait_job(client: &Client, ns: &str, name: &str, patience: Duration) -> Result<()> {
let api = Api::<Job>::namespaced(client.clone(), ns);
let mut tick = interval(Duration::from_secs(15));
let deadline = tokio::time::Instant::now() + patience;
loop {
tick.tick().await;
let job = api.get(name).await?;
let status = job.status.unwrap_or_default();
if status.succeeded.unwrap_or(0) >= 1 {
return Ok(());
}
if let Some(f) = status.conditions.as_ref().and_then(|cs| {
cs.iter()
.find(|c| c.type_ == "Failed" && c.status == "True")
}) {
return Err(anyhow!(
"job {name} failed: {}",
f.message
.as_deref()
.or(f.reason.as_deref())
.unwrap_or("unknown")
));
}
if tokio::time::Instant::now() > deadline {
return Err(anyhow!("job {name} did not finish within {patience:?}"));
}
}
}
pub async fn wait_snapshot_ready(
client: &Client,
ns: &str,
name: &str,
patience: Duration,
) -> Result<()> {
let ar = volume_snapshots();
let api: Api<DynamicObject> = Api::namespaced_with(client.clone(), ns, &ar);
let mut tick = interval(Duration::from_secs(10));
let deadline = tokio::time::Instant::now() + patience;
loop {
tick.tick().await;
let obj = api.get(name).await?;
if obj
.data
.pointer("/status/readyToUse")
.and_then(|v| v.as_bool())
== Some(true)
{
return Ok(());
}
if let Some(err) = obj.data.pointer("/status/error") {
return Err(anyhow!("volumesnapshot {name} failed: {err}"));
}
if tokio::time::Instant::now() > deadline {
return Err(anyhow!(
"volumesnapshot {name} not readyToUse within {patience:?}"
));
}
}
}
pub async fn wait_pvc_bound(
client: &Client,
ns: &str,
name: &str,
patience: Duration,
) -> Result<()> {
let api = Api::<PersistentVolumeClaim>::namespaced(client.clone(), ns);
let mut tick = interval(Duration::from_secs(10));
let deadline = tokio::time::Instant::now() + patience;
loop {
tick.tick().await;
let pvc = api.get(name).await?;
if pvc.status.as_ref().and_then(|s| s.phase.as_deref()) == Some("Bound") {
return Ok(());
}
if tokio::time::Instant::now() > deadline {
return Err(anyhow!("pvc {name} not Bound within {patience:?}"));
}
}
}
pub async fn delete_gone(
client: &Client,
ns: &str,
name: &str,
snapshot: bool,
patience: Duration,
) -> Result<()> {
let del = DeleteParams {
propagation_policy: Some(PropagationPolicy::Foreground),
..Default::default()
};
let outcome = if snapshot {
let ar = volume_snapshots();
let api: Api<DynamicObject> = Api::namespaced_with(client.clone(), ns, &ar);
let _ = api.delete(name, &del).await;
wait_gone(&api, name, patience).await
} else {
let api = Api::<PersistentVolumeClaim>::namespaced(client.clone(), ns);
let _ = api.delete(name, &del).await;
wait_gone(&api, name, patience).await
};
outcome.map_err(|e| anyhow!("pvc/snapshot {ns}/{name} not gone: {e}"))
}
async fn wait_gone<T>(api: &Api<T>, name: &str, patience: Duration) -> Result<()>
where
T: kube::Resource + Clone + serde::de::DeserializeOwned + std::fmt::Debug,
{
let mut tick = interval(Duration::from_secs(5));
let deadline = tokio::time::Instant::now() + patience;
loop {
tick.tick().await;
match api.get(name).await {
Err(kube::Error::Api(e)) if e.code == 404 => return Ok(()),
Err(e) => return Err(e.into()),
Ok(_) if tokio::time::Instant::now() > deadline => {
return Err(anyhow!("still present"));
}
Ok(_) => {}
}
}
}

144
examples/harmony_tenant_backup/src/lib.rs Normal file
View File

@@ -0,0 +1,144 @@
//! Shared contract + engine for the two binaries.
//! `harmony-tenant-backup` operates a namespace; `harmony-tenant-backup-setup`
//! installs the CronJob. Names live in `common` so they cannot drift.
pub mod common;
pub mod exec;
pub mod k8s;
pub mod planner;
pub mod worker;
pub use common::*;
use anyhow::{Context, Result};
use harmony_config::{Config, ConfigClient, EnvSource};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::Arc;
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Config)]
pub struct TenantBackup {
pub target_endpoint: String,
pub target_bucket: String,
#[config(secret)]
pub target_access_key: String,
#[config(secret)]
pub target_secret_key: String,
pub target_region: Option<String>,
/// Minted by setup. Nightly job reads OpenBao; fire restore uses the
/// offsite kit, never this vault.
#[serde(default)]
#[config(secret)]
pub restic_password: String,
/// Read-only tenant source on the site-A RGW; needed only when the
/// namespace declares buckets.
pub source_endpoint: Option<String>,
pub source_region: Option<String>,
#[config(secret)]
pub source_access_key: Option<String>,
#[config(secret)]
pub source_secret_key: Option<String>,
pub snapshot_class: Option<String>,
/// One restic repo per month; expiry = delete a whole prefix.
#[serde(default)]
pub rotate_repo_monthly: bool,
/// PEM, only when the target RGW serves a private CA cert. Never
/// disable TLS verification instead.
pub target_ca_pem: Option<String>,
}
const OPENBAO_HINT: &str = concat!(
"OpenBao unavailable: set OPENBAO_URL, OPENBAO_JWT_ROLE, HARMONY_SSO_URL, ",
"HARMONY_ZITADEL_AUDIENCE and HARMONY_ZITADEL_KEY_JSON (or HARMONY_ZITADEL_KEY_PATH)"
);
async fn openbao_store(namespace: &str) -> Result<Arc<dyn harmony_config::ConfigSource>> {
harmony_config::openbao_source(
&format!("harmony/{namespace}"),
None,
None,
None,
None,
None,
None,
)
.await
.context(OPENBAO_HINT)
}
/// No prompt fallback: a Job has no TTY. Store path: `harmony/<tenant-ns>`.
pub async fn load_config(namespace: &str) -> Result<TenantBackup> {
let store = openbao_store(namespace).await?;
let client = ConfigClient::new(vec![Arc::new(EnvSource), store]);
client.get::<TenantBackup>().await.map_err(Into::into)
}
fn mint_restic_password() -> Result<String> {
use std::io::Read;
let mut b = [0u8; 32];
std::fs::File::open("/dev/urandom")
.context("open /dev/urandom")?
.read_exact(&mut b)
.context("read /dev/urandom")?;
Ok(b.iter().fold(String::with_capacity(64), |mut s, x| {
use std::fmt::Write;
let _ = write!(s, "{x:02x}");
s
}))
}
/// If OpenBao has target_* but no restic_password, mint one, persist it for
/// the CronJob, and return it so setup can print the offsite-kit line.
/// Already set → None (never rotate; that would brick the repo).
pub async fn ensure_restic_password(namespace: &str) -> Result<Option<String>> {
let store = openbao_store(namespace).await?;
let client = ConfigClient::new(vec![store]);
let mut cfg = client
.get::<TenantBackup>()
.await
.context("no TenantBackup in OpenBao; bao kv put target_* first (omit restic_password)")?;
if !cfg.restic_password.is_empty() {
return Ok(None);
}
let pw = mint_restic_password()?;
cfg.restic_password = pw.clone();
client
.set(&cfg)
.await
.context("write restic_password to OpenBao")?;
Ok(Some(pw))
}
pub async fn load_runtime(
namespace: &str,
cluster_id: &str,
scratch: &Path,
) -> Result<(TenantBackup, exec::Restic)> {
std::fs::create_dir_all(scratch).with_context(|| {
format!(
"cannot create scratch dir {} (override with --scratch / SCRATCH=)",
scratch.display()
)
})?;
let cfg = load_config(namespace).await?;
if cfg.restic_password.is_empty() {
anyhow::bail!(
"empty restic_password: run harmony-tenant-backup-setup (mints into OpenBao and the offsite kit)"
);
}
let ca_file = match cfg
.target_ca_pem
.as_deref()
.filter(|s| !s.trim().is_empty())
{
Some(pem) => {
let f = scratch.join("target-ca.crt");
std::fs::write(&f, pem).with_context(|| format!("writing {}", f.display()))?;
Some(f)
}
None => None,
};
let restic = exec::build_restic(&cfg, cluster_id, namespace, ca_file, scratch);
Ok((cfg, restic))
}

114
examples/harmony_tenant_backup/src/main.rs Normal file
View File

@@ -0,0 +1,114 @@
//! Operate one tenant namespace. Not restore. Not retention. Not install
//! (`harmony-tenant-backup-setup` does that).
use anyhow::{Context, Result};
use chrono::Utc;
use clap::{Parser, Subcommand};
use harmony_tenant_backup::{
DEFAULT_SCRATCH, ENV_CLUSTER_ID, ENV_NAMESPACE, ENV_RUN_ID, ENV_SCRATCH, default_run_id,
dns1123_run_id, load_runtime, planner, worker,
};
use std::path::Path;
#[derive(Parser, Debug)]
#[command(
name = "harmony-tenant-backup",
about = "Backup one tenant namespace. Not restore. Not install."
)]
struct Cli {
#[command(subcommand)]
command: Command,
}
#[derive(Subcommand, Debug)]
enum Command {
/// Discover and push this namespace (pg, buckets, PVCs).
Run {
#[arg(short, long, env = ENV_NAMESPACE)]
namespace: String,
#[arg(long, env = ENV_CLUSTER_ID)]
cluster_id: String,
/// Connectivity checks and the plan, no writes.
#[arg(long)]
dry_run: bool,
#[arg(long, env = ENV_SCRATCH, default_value = DEFAULT_SCRATCH)]
scratch: String,
#[arg(long, env = ENV_RUN_ID)]
run_id: Option<String>,
},
/// Push the volume mounted at /mnt/data. Spawned by `run`, not operators.
#[command(hide = true)]
Worker {
/// Logical PVC name this worker pushes.
#[arg(long)]
pvc: String,
#[arg(short, long, env = ENV_NAMESPACE)]
namespace: String,
#[arg(long, env = ENV_CLUSTER_ID)]
cluster_id: String,
#[arg(long, env = ENV_SCRATCH, default_value = DEFAULT_SCRATCH)]
scratch: String,
#[arg(long, env = ENV_RUN_ID)]
run_id: Option<String>,
},
}
#[tokio::main]
async fn main() -> std::process::ExitCode {
match dispatch().await {
Ok(code) => code,
Err(e) => {
eprintln!("{e:#}");
2
}
}
.into()
}
async fn dispatch() -> Result<u8> {
match Cli::parse().command {
Command::Run {
namespace,
cluster_id,
dry_run,
scratch,
run_id,
} => {
let run_id = resolve_run_id(run_id)?;
let (cfg, restic) = load_runtime(&namespace, &cluster_id, Path::new(&scratch)).await?;
let client = kube::Client::try_default()
.await
.context("no Kubernetes access (in-cluster ServiceAccount or KUBECONFIG)")?;
planner::Run::new(
&client,
&cfg,
&namespace,
&cluster_id,
&run_id,
Path::new(&scratch),
restic,
dry_run,
)
.run()
.await
}
Command::Worker {
pvc,
namespace,
cluster_id,
scratch,
run_id,
} => {
let run_id = resolve_run_id(run_id)?;
let (_cfg, restic) = load_runtime(&namespace, &cluster_id, Path::new(&scratch)).await?;
worker::run(&restic, &cluster_id, &namespace, &run_id, &pvc).await
}
}
}
fn resolve_run_id(run_id: Option<String>) -> Result<String> {
let run_id = run_id.unwrap_or_else(|| default_run_id(Utc::now()));
dns1123_run_id(&run_id).ok_or_else(|| {
anyhow::anyhow!("--run-id must be a DNS-1123 label, 1-32 chars [a-z0-9-]; got {run_id}")
})
}

506
examples/harmony_tenant_backup/src/planner.rs Normal file
View File

@@ -0,0 +1,506 @@
//! Planner: one namespace's three surfaces — CNPG logical dumps, S3
//! buckets (staged then restic'd), PVCs (direct or snapshot-clone worker).
use crate::common::*;
use crate::exec::{Pg, Restic, SrcS3, logline};
use crate::k8s::{self, WorkerSpec};
use anyhow::{Context, Result, anyhow};
use k8s_openapi::api::core::v1::PersistentVolumeClaim;
use kube::{Api, Client, ResourceExt};
use std::path::Path;
use std::time::Duration;
pub struct Plan {
pub pg: usize,
pub buckets: usize,
pub pvc: usize,
pub pvc_skipped: usize,
pub failures: Vec<String>,
}
const SNAP_WAIT: Duration = Duration::from_secs(30 * 60);
const CLONE_WAIT: Duration = Duration::from_secs(15 * 60);
/// Patience = worker's own deadline + slack, so only a genuinely stuck pod
/// times out here rather than a slow-but-live push.
const JOB_WAIT: Duration = Duration::from_secs(WORKER_DEADLINE_SECS as u64 + 15 * 60);
const CLEANUP_WAIT: Duration = Duration::from_secs(5 * 60);
const DB_LIST_SQL: &str =
"select datname from pg_database where not datistemplate and datallowconn order by 1";
/// Everything one backup run needs, threaded as one value so ns/run_id can
/// never drift between surfaces.
pub struct Run<'a> {
client: &'a Client,
cfg: &'a crate::TenantBackup,
restic: Restic,
ns: &'a str,
cluster_id: &'a str,
run_id: &'a str,
scratch: &'a Path,
dry_run: bool,
fs_group: Option<i64>,
plan: Plan,
}
impl<'a> Run<'a> {
// Whole context of a run; the restic handle is built once by main so
// planner and worker share one repo resolution.
#[allow(clippy::too_many_arguments)]
pub fn new(
client: &'a Client,
cfg: &'a crate::TenantBackup,
ns: &'a str,
cluster_id: &'a str,
run_id: &'a str,
scratch: &'a Path,
restic: Restic,
dry_run: bool,
) -> Self {
Self {
client,
cfg,
restic,
ns,
cluster_id,
run_id,
scratch,
dry_run,
fs_group: None,
plan: Plan {
pg: 0,
buckets: 0,
pvc: 0,
pvc_skipped: 0,
failures: vec![],
},
}
}
pub async fn run(&mut self) -> Result<u8> {
let anns = k8s::ns_annotations(self.client, self.ns).await?;
self.fs_group = volume_fsgroup(Some(&anns));
self.restic
.ensure_repo(self.dry_run)
.await
.context("restic repo unavailable")?;
self.do_pg().await;
self.do_buckets().await;
self.do_pvcs().await;
println!(
"ns {} run {} pg {} buckets {} pvc {} skipped {} failures {}",
self.ns,
self.run_id,
self.plan.pg,
self.plan.buckets,
self.plan.pvc,
self.plan.pvc_skipped,
self.plan.failures.len(),
);
logline("snapshots this run:");
self.restic
.snapshots_tagged(&format!("run:{}", self.run_id))
.await
.ok();
if self.plan.failures.is_empty() {
Ok(0)
} else {
eprintln!("FAILURES:");
for f in &self.plan.failures {
eprintln!(" - {f}");
}
Ok(3)
}
}
fn fail(&mut self, msg: String) {
logline(&format!("FAILED: {msg}"));
self.plan.failures.push(msg);
}
fn fail_pvc(&mut self, msg: String) {
self.fail(msg);
self.plan.pvc_skipped += 1;
}
async fn do_pg(&mut self) {
let clusters =
match k8s::list_namespaced_names(self.client, self.ns, &k8s::cnpg_clusters()).await {
Ok(c) => c,
// The CRD may legitimately be absent (no CNPG here), but a real
// API error must not read as "no databases".
Err(e) if is_not_found(&e) => {
logline("no CNPG CRD present, skipping pg surface");
return;
}
Err(e) => return self.fail(format!("cannot list CNPG clusters: {e}")),
};
if clusters.is_empty() {
logline(&format!("no CNPG clusters in {}", self.ns));
return;
}
for c in clusters {
logline(&format!("pg: cluster {c}"));
let pg = match self.open_pg(&c).await {
Ok(pg) => pg,
Err(e) => {
self.fail(format!("pg/{c}: {e}"));
continue;
}
};
if let Err(e) = pg.query("postgres", "select 1").await {
self.fail(format!("cannot reach {c}-rw (NetworkPolicy? TLS?): {e}"));
continue;
}
if self.dry_run {
match pg.query("postgres", DB_LIST_SQL).await {
Ok(dbs) => logline(&format!(
"DRY-RUN: would dump {c}: {}",
dbs.replace('\n', ", ")
)),
Err(e) => self.fail(format!("cannot list databases for {c}: {e}")),
}
continue;
}
let out = self.scratch.join("pg").join(&c);
tokio::fs::create_dir_all(&out).await.ok();
// metadata before dumps: without LSN + timestamp a dump is
// undateable in an incident.
let lsn = pg
.query("postgres", "select pg_current_wal_lsn()")
.await
.unwrap_or_else(|_| "unknown".into());
let dumped_at = pg
.query("postgres", "select now()")
.await
.unwrap_or_default();
tokio::fs::write(out.join("dumped_at"), &dumped_at)
.await
.ok();
let meta = serde_json::json!({
"cluster": c, "ns": self.ns, "run": self.run_id, "lsn": lsn.trim(),
"repo": self.restic.repo(),
});
tokio::fs::write(out.join("metadata.json"), meta.to_string() + "\n")
.await
.ok();
let mut cluster_ok = true;
if pg.dump_all_globals(&out.join("globals.sql")).await.is_err() {
self.fail(format!(
"pg_dumpall --globals-only {c}: a DB-only dump loses the app roles"
));
cluster_ok = false;
}
let dbs = match pg.query("postgres", DB_LIST_SQL).await {
Ok(dbs) => dbs,
Err(e) => {
self.fail(format!("cannot list databases for {c}: {e}"));
tokio::fs::remove_dir_all(&out).await.ok();
continue;
}
};
let mut dumped = 0;
for db in dbs.lines().filter(|d| !d.is_empty()) {
let f = out.join(format!("{db}.dump"));
logline(&format!("pg: {c}/{db} -> {}", f.display()));
match pg.dump_db(db, &f).await {
Ok(()) if Pg::verify_dump(&f).await.is_ok() => dumped += 1,
Ok(()) => {
self.fail(format!(
"pg_restore --list rejected {c}/{db} (corrupt dump)"
));
tokio::fs::remove_file(&f).await.ok();
cluster_ok = false;
}
Err(e) => {
self.fail(format!("pg_dump {c}/{db}: {e}"));
tokio::fs::remove_file(&f).await.ok();
cluster_ok = false;
}
}
}
if cluster_ok {
match self
.restic
.backup(
&restic_host(self.cluster_id, self.ns),
self.run_id,
self.ns,
"pg",
&[format!("cluster:{c}")],
&out.display().to_string(),
false,
)
.await
{
Ok(()) => self.plan.pg += dumped,
Err(e) => self.fail(e.to_string()),
}
}
tokio::fs::remove_dir_all(&out).await.ok();
}
}
async fn open_pg(&self, cluster: &str) -> Result<Pg> {
let password = String::from_utf8_lossy(
&k8s::secret_bytes(
self.client,
self.ns,
&format!("{cluster}-superuser"),
"password",
)
.await?,
)
.trim_end()
.to_string();
let ca_dir = self.scratch.join("tls");
tokio::fs::create_dir_all(&ca_dir).await.ok();
let ca_file = ca_dir.join(format!("{cluster}-ca.crt"));
let ca =
k8s::secret_bytes(self.client, self.ns, &format!("{cluster}-ca"), "ca.crt").await?;
tokio::fs::write(&ca_file, ca).await?;
Ok(Pg::new(cluster, self.ns, &ca_file, password))
}
async fn do_buckets(&mut self) {
let list = match k8s::annotation(self.client, self.ns, A_BUCKETS).await {
Ok(Some(l)) if !l.trim().is_empty() => l,
// Coverage gap, not a skip: say so loudly.
Ok(_) => {
logline(&format!(
"no '{A_BUCKETS}' annotation on ns/{} — skipping buckets (THIS IS A COVERAGE GAP)",
self.ns
));
return;
}
Err(e) => return self.fail(format!("cannot read ns/{} annotations: {e}", self.ns)),
};
let (Some(endpoint), Some(access), Some(secret)) = (
&self.cfg.source_endpoint,
&self.cfg.source_access_key,
&self.cfg.source_secret_key,
) else {
return self.fail(format!(
"bucket source credentials incomplete in vault for {} (source_endpoint/source_access_key/source_secret_key)",
self.ns
));
};
let src = SrcS3 {
endpoint: endpoint.clone(),
region: self
.cfg
.source_region
.clone()
.unwrap_or_else(|| "default".into()),
access_key: access.clone(),
secret_key: secret.clone(),
};
for b in list.split(',').map(str::trim).filter(|b| !b.is_empty()) {
// Tenant-authored string reaching filesystem paths and rclone
// connection strings: reject anything outside S3's own grammar.
if !valid_bucket_token(b) {
self.fail(format!(
"invalid bucket token {b:?} in {A_BUCKETS} (want ^[a-z0-9][a-z0-9.-]{{0,61}}[a-z0-9]$)"
));
continue;
}
let dst = self.scratch.join("s3").join(b);
if self.dry_run {
logline(&format!("DRY-RUN: would stage and push s3 bucket {b}"));
if let Err(e) = src.list_dirs(b).await {
self.fail(format!("bucket {b} not reachable: {e}"));
}
continue;
}
logline(&format!("s3: staging {b} -> {}", dst.display()));
tokio::fs::create_dir_all(&dst).await.ok();
let staged = match src.copy(b, &dst).await {
Ok(()) => src.check(b, &dst).await,
Err(e) => Err(e),
};
let pushed = match staged {
Ok(()) => {
self.restic
.backup(
&restic_host(self.cluster_id, self.ns),
self.run_id,
self.ns,
"s3",
&[format!("bucket:{b}")],
&dst.display().to_string(),
false,
)
.await
}
Err(e) => Err(e),
};
match pushed {
Ok(()) => self.plan.buckets += 1,
Err(e) => self.fail(e.to_string()),
}
tokio::fs::remove_dir_all(&dst).await.ok();
}
}
async fn do_pvcs(&mut self) {
let pvcs = match Api::<PersistentVolumeClaim>::namespaced(self.client.clone(), self.ns)
.list(&Default::default())
.await
{
Ok(l) => l,
Err(e) => return self.fail(format!("cannot list PVCs: {e}")),
};
let mut class: Option<String> = self.cfg.snapshot_class.clone();
for pvc in pvcs {
let name = pvc.name_any();
let ann = pvc.metadata.annotations.clone().unwrap_or_default();
let labels = pvc.metadata.labels.clone().unwrap_or_default();
let spec = pvc.spec.clone().unwrap_or_default();
let phase = pvc
.status
.as_ref()
.and_then(|s| s.phase.clone())
.unwrap_or_else(|| "Unknown".into());
let access = spec
.access_modes
.as_ref()
.and_then(|a| a.first())
.cloned()
.unwrap_or_else(|| "?".into());
let outcome = match decide_pvc(
&phase,
labels.get("cnpg.io/cluster").map(String::as_str),
ann.get(A_PVC_DISABLED).map(|v| v.as_str()) == Some("true"),
spec.volume_mode.as_deref() == Some("Block"),
&access,
) {
PvcDecision::Skip(why) => {
logline(&format!("pvc/{name}: {why}, skipping"));
self.plan.pvc_skipped += 1;
continue;
}
PvcDecision::Fail(why) => {
self.fail_pvc(format!("pvc/{name}: {why}"));
continue;
}
PvcDecision::Direct => {
logline(&format!("pvc/{name} ({access}): mounting directly"));
if self.dry_run {
logline(&format!("DRY-RUN: worker on {name}"));
continue;
}
self.spawn_worker(&name, &name).await
}
PvcDecision::SnapshotFirst => {
if class.is_none() {
// A swallowed 403 here reads as "no class exists":
// propagate so the RBAC gap is visible in the log.
match k8s::first_cluster_scoped_name(
self.client,
&k8s::volume_snapshot_classes(),
)
.await
{
Ok(c) => class = c,
Err(e) => {
self.fail_pvc(format!("pvc/{name}: {e}"));
continue;
}
}
}
let Some(cls) = class.clone() else {
self.fail_pvc(format!(
"pvc/{name}: no VolumeSnapshotClass found (set snapshot_class in vault)"
));
continue;
};
if self.dry_run {
logline(&format!(
"DRY-RUN: snapshot + clone via {cls} + worker for {name}"
));
continue;
}
self.snapshot_worker(&pvc, &name, &cls).await
}
};
match outcome {
Ok(()) => self.plan.pvc += 1,
Err(e) => self.fail_pvc(format!("pvc/{name} ({access}): {e}")),
}
}
}
/// Snapshot -> clone -> worker push -> cleanup. The clone must be deleted
/// before the snapshot it references.
async fn snapshot_worker(
&mut self,
source: &PersistentVolumeClaim,
logical: &str,
class: &str,
) -> Result<()> {
let snap = run_object_name("htb-snap", logical, self.run_id);
let clone = run_object_name("htb-clone", logical, self.run_id);
k8s::create_volume_snapshot(self.client, self.ns, &snap, class, logical, self.run_id)
.await?;
let outcome = async {
k8s::wait_snapshot_ready(self.client, self.ns, &snap, SNAP_WAIT).await?;
k8s::create_clone_pvc(self.client, self.ns, &clone, source, &snap, self.run_id).await?;
k8s::wait_pvc_bound(self.client, self.ns, &clone, CLONE_WAIT).await?;
self.spawn_worker(&clone, logical).await
}
.await;
// Left-over snapshots and clones run into CSI quotas and bill; a
// failed cleanup is a run failure, not a shrug.
for res in [
k8s::delete_gone(self.client, self.ns, &clone, false, CLEANUP_WAIT).await,
k8s::delete_gone(self.client, self.ns, &snap, true, CLEANUP_WAIT).await,
] {
if let Err(e) = res {
self.fail(format!("pvc/{logical}: cleanup: {e}"));
}
}
outcome
}
async fn spawn_worker(&self, pvc_mount: &str, logical: &str) -> Result<()> {
let image = std::env::var(ENV_WORKER_IMAGE)
.context(format!("${ENV_WORKER_IMAGE} required to back up PVCs"))?;
let fwd: Vec<(String, String)> = FORWARD_ENV
.iter()
.filter_map(|k| std::env::var(k).ok().map(|v| ((*k).to_string(), v)))
.collect();
if fwd.len() != FORWARD_ENV.len() {
return Err(anyhow!(
"need all of {} in the planner env to spawn a PVC worker",
FORWARD_ENV.join(", ")
));
}
let fwd_refs: Vec<(&str, &str)> =
fwd.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
let job = k8s::worker_job(&WorkerSpec {
image: &image,
ns: self.ns,
name: &run_object_name("htb-wk", logical, self.run_id),
pvc_mount,
pvc_logical: logical,
run_id: self.run_id,
cluster_id: self.cluster_id,
zitadel_secret: &std::env::var(ENV_ZITADEL_SECRET)
.unwrap_or_else(|_| ZITADEL_SECRET.into()),
forward_env: fwd_refs,
fs_group: self.fs_group,
});
k8s::spawn_and_wait_job(self.client, self.ns, &job, JOB_WAIT).await
}
}
fn is_not_found(e: &anyhow::Error) -> bool {
matches!(e.downcast_ref::<kube::Error>(), Some(kube::Error::Api(a)) if a.code == 404)
}

410
examples/harmony_tenant_backup/src/setup.rs Normal file
View File

@@ -0,0 +1,410 @@
//! Install RBAC + CronJob into a tenant namespace. Not the nightly run.
use anyhow::{Context, Result, anyhow};
use clap::Parser;
use harmony::inventory::Inventory;
use harmony::modules::k8s::K8sBareTopology;
use harmony::modules::k8s::resource::K8sResourceScore;
use harmony::score::Score;
use harmony_tenant_backup::k8s::{container_security, pod_security};
use harmony_tenant_backup::{env_from_secret, env_var, *};
use k8s_openapi::api::batch::v1::{CronJob, CronJobSpec, JobSpec, JobTemplateSpec};
use k8s_openapi::api::core::v1::{
Container, EmptyDirVolumeSource, PodSpec, PodTemplateSpec, ResourceRequirements,
ServiceAccount, Volume, VolumeMount,
};
use k8s_openapi::api::rbac::v1::{
ClusterRole, ClusterRoleBinding, PolicyRule, Role, RoleBinding, RoleRef, Subject,
};
use k8s_openapi::apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::ObjectMeta};
use kube::{Api, Client};
use std::collections::BTreeMap;
#[derive(Parser, Debug)]
#[command(
name = "harmony-tenant-backup-setup",
about = "Install backup RBAC + CronJob into a tenant namespace. Not the nightly run."
)]
struct Cli {
/// Tenant namespace to install into.
#[arg(short, long)]
namespace: String,
/// Cron schedule, e.g. "17 3 * * *" (stagger per namespace, off-peak).
#[arg(long)]
schedule: String,
/// Backup image. MUST be digest-pinned (harbor.example/ns-backup@sha256:...).
#[arg(long, env = "BACKUP_IMAGE")]
image: String,
/// Stable cluster identifier used in the restic repo path and --host.
#[arg(long, env = ENV_CLUSTER_ID)]
cluster_id: String,
#[arg(long, default_value = ZITADEL_SECRET)]
zitadel_secret: String,
#[arg(long, env = "OPENBAO_URL")]
openbao_url: Option<String>,
#[arg(long, env = "OPENBAO_JWT_ROLE")]
openbao_jwt_role: Option<String>,
#[arg(long, env = "HARMONY_SSO_URL")]
sso_url: Option<String>,
#[arg(long, env = "HARMONY_ZITADEL_AUDIENCE")]
audience: Option<String>,
}
const KEY_FILE: &str = ZITADEL_KEY_FILE;
macro_rules! apply {
($topo:expr, $ns:expr, $obj:expr, $what:expr) => {{
let score = K8sResourceScore::single($obj, Some($ns.to_string()));
let outcome = Score::<K8sBareTopology>::create_interpret(&score)
.execute(&Inventory::empty(), $topo)
.await
.map_err(|e| anyhow!("apply {}: {e}", $what))?;
println!("{:<12} {}: {:?}", $what, $ns, outcome.status);
}};
}
macro_rules! apply_cluster {
($topo:expr, $obj:expr, $what:expr) => {{
let score = K8sResourceScore::single($obj, None);
let outcome = Score::<K8sBareTopology>::create_interpret(&score)
.execute(&Inventory::empty(), $topo)
.await
.map_err(|e| anyhow!("apply {}: {e}", $what))?;
println!("{:<12} cluster: {:?}", $what, outcome.status);
}};
}
#[tokio::main]
async fn main() -> Result<()> {
let cli = Cli::parse();
if !cli.image.contains("@sha256:") {
return Err(anyhow!(
"--image must be digest-pinned; tag-following is exactly the \
\"worked last week\" bug class this tool exists to avoid"
));
}
for (name, val) in [
("OPENBAO_URL", &cli.openbao_url),
("OPENBAO_JWT_ROLE", &cli.openbao_jwt_role),
("HARMONY_SSO_URL", &cli.sso_url),
("HARMONY_ZITADEL_AUDIENCE", &cli.audience),
] {
if val.is_none() {
return Err(anyhow!(
"{name} is required (flag or env): the backup workload reaches its vault through it"
));
}
}
if std::env::var("HARMONY_ZITADEL_KEY_JSON").is_err()
&& std::env::var("HARMONY_ZITADEL_KEY_PATH").is_err()
{
return Err(anyhow!(
"HARMONY_ZITADEL_KEY_JSON or HARMONY_ZITADEL_KEY_PATH required: \
setup mints restic_password into OpenBao for the CronJob"
));
}
let client = Client::try_default()
.await
.context("no kubeconfig / cluster access")?;
let ns_obj = Api::<k8s_openapi::api::core::v1::Namespace>::all(client.clone())
.get(&cli.namespace)
.await
.map_err(|e| {
if matches!(e, kube::Error::Api(ref a) if a.code == 404) {
anyhow!("namespace {} does not exist", cli.namespace)
} else {
anyhow!(
"cannot reach the cluster to check namespace {}: {e}",
cli.namespace
)
}
})?;
let fs_group = volume_fsgroup(ns_obj.metadata.annotations.as_ref());
let secret = Api::<k8s_openapi::api::core::v1::Secret>::namespaced(client, &cli.namespace)
.get(&cli.zitadel_secret)
.await
.with_context(|| {
format!(
"secret {}/{} not found. Create it with:\n \
kubectl create secret generic {} -n {} --from-file={KEY_FILE}=<zitadel machine key file>",
cli.namespace, cli.zitadel_secret, cli.zitadel_secret, cli.namespace
)
})?;
if secret
.data
.as_ref()
.is_none_or(|d| !d.contains_key(KEY_FILE))
{
return Err(anyhow!(
"secret {}/{} must contain key '{KEY_FILE}'",
cli.namespace,
cli.zitadel_secret
));
}
let topology = K8sBareTopology::from_kubeconfig("tenant-backup-setup")
.await
.map_err(|e| anyhow!(e))?;
let ns = &cli.namespace;
apply!(&topology, ns, service_account(), "ServiceAccount");
apply!(&topology, ns, role(), "Role");
apply!(&topology, ns, role_binding(ns), "RoleBinding");
apply_cluster!(&topology, cluster_role(ns), "ClusterRole");
apply_cluster!(&topology, cluster_role_binding(ns), "ClusterRoleBinding");
apply!(&topology, ns, cron_job(&cli, fs_group), "CronJob");
match ensure_restic_password(ns).await? {
Some(pw) => {
println!(
"\nOFFSITE KIT (KeepassX). Break-glass. Restore never uses OpenBao or this cluster."
);
println!(" ns={ns} restic_password={pw}");
println!(
" Put this in the same kit as OpenBao unseal keys. You will not see it again."
);
}
None => println!(
"\nrestic_password already in OpenBao; the offsite kit must already have it (not rotated)."
),
}
println!(
"installed: CronJob {CRONJOB_NAME} in {ns}, schedule {:?}, image {}",
cli.schedule, cli.image
);
println!("run now: kubectl create job backup-now --from=cronjob/{CRONJOB_NAME} -n {ns}");
println!("watch: kubectl logs -n {ns} job/backup-now -f");
Ok(())
}
fn managed_labels() -> BTreeMap<String, String> {
BTreeMap::from([(MANAGED_BY.to_string(), MANAGED_BY_VALUE.to_string())])
}
fn service_account() -> ServiceAccount {
ServiceAccount {
metadata: ObjectMeta {
name: Some(SA_NAME.into()),
labels: Some(managed_labels()),
..Default::default()
},
..Default::default()
}
}
fn role() -> Role {
Role {
metadata: ObjectMeta {
name: Some(SA_NAME.into()),
labels: Some(managed_labels()),
..Default::default()
},
rules: Some(vec![
PolicyRule {
// CNPG <cluster>-superuser password + <cluster>-ca cert, both
// by known name. get-only: a backup identity must not sweep
// the tenant's secrets.
api_groups: Some(vec!["".into()]),
resources: Some(vec!["secrets".into()]),
verbs: vec!["get".into()],
..Default::default()
},
PolicyRule {
api_groups: Some(vec!["".into()]),
resources: Some(vec!["persistentvolumeclaims".into()]),
verbs: vec!["get".into(), "list".into()],
..Default::default()
},
PolicyRule {
// The RWO clone only; delete is scoped here, never on secrets.
api_groups: Some(vec!["".into()]),
resources: Some(vec!["persistentvolumeclaims".into()]),
verbs: vec!["create".into(), "delete".into()],
..Default::default()
},
PolicyRule {
// VolumeSnapshotClass is cluster-scoped: a namespaced Role
// can never grant it (see the ClusterRole below).
api_groups: Some(vec!["snapshot.storage.k8s.io".into()]),
resources: Some(vec!["volumesnapshots".into()]),
verbs: vec![
"get".into(),
"list".into(),
"create".into(),
"delete".into(),
],
..Default::default()
},
PolicyRule {
api_groups: Some(vec!["batch".into()]),
resources: Some(vec!["jobs".into()]),
verbs: vec!["get".into(), "list".into(), "create".into()],
..Default::default()
},
PolicyRule {
api_groups: Some(vec!["postgresql.cnpg.io".into()]),
resources: Some(vec!["clusters".into()]),
verbs: vec!["get".into(), "list".into()],
..Default::default()
},
]),
}
}
/// Cluster-scoped reads a namespaced Role can never grant: this
/// namespace (bucket annotation) and VolumeSnapshotClass list.
fn cluster_role(ns: &str) -> ClusterRole {
ClusterRole {
metadata: ObjectMeta {
name: Some(format!("{SA_NAME}-{ns}")),
labels: Some(managed_labels()),
..Default::default()
},
rules: Some(vec![
PolicyRule {
api_groups: Some(vec!["".into()]),
resources: Some(vec!["namespaces".into()]),
resource_names: Some(vec![ns.into()]),
verbs: vec!["get".into()],
..Default::default()
},
PolicyRule {
api_groups: Some(vec!["snapshot.storage.k8s.io".into()]),
resources: Some(vec!["volumesnapshotclasses".into()]),
verbs: vec!["get".into(), "list".into()],
..Default::default()
},
]),
..Default::default()
}
}
fn cluster_role_binding(ns: &str) -> ClusterRoleBinding {
ClusterRoleBinding {
metadata: ObjectMeta {
name: Some(format!("{SA_NAME}-{ns}")),
labels: Some(managed_labels()),
..Default::default()
},
role_ref: RoleRef {
api_group: "rbac.authorization.k8s.io".into(),
kind: "ClusterRole".into(),
name: format!("{SA_NAME}-{ns}"),
},
subjects: Some(vec![Subject {
api_group: Some("".into()),
kind: "ServiceAccount".into(),
name: SA_NAME.into(),
namespace: Some(ns.into()),
}]),
}
}
fn role_binding(ns: &str) -> RoleBinding {
RoleBinding {
metadata: ObjectMeta {
name: Some(SA_NAME.into()),
labels: Some(managed_labels()),
..Default::default()
},
role_ref: RoleRef {
api_group: "rbac.authorization.k8s.io".into(),
kind: "Role".into(),
name: SA_NAME.into(),
},
subjects: Some(vec![Subject {
api_group: Some("".into()),
kind: "ServiceAccount".into(),
name: SA_NAME.into(),
namespace: Some(ns.into()),
}]),
}
}
fn cron_job(cli: &Cli, fs_group: Option<i64>) -> CronJob {
let mut env = vec![
env_var(ENV_NAMESPACE, &cli.namespace),
env_var(ENV_CLUSTER_ID, &cli.cluster_id),
env_var(ENV_WORKER_IMAGE, &cli.image),
env_var(ENV_ZITADEL_SECRET, &cli.zitadel_secret),
env_from_secret(ENV_ZITADEL_KEY_JSON, &cli.zitadel_secret, KEY_FILE),
];
for (k, v) in [
("OPENBAO_URL", &cli.openbao_url),
("OPENBAO_JWT_ROLE", &cli.openbao_jwt_role),
("HARMONY_SSO_URL", &cli.sso_url),
("HARMONY_ZITADEL_AUDIENCE", &cli.audience),
] {
if let Some(v) = v {
env.push(env_var(k, v));
}
}
CronJob {
metadata: ObjectMeta {
name: Some(CRONJOB_NAME.into()),
labels: Some(managed_labels()),
..Default::default()
},
spec: Some(CronJobSpec {
schedule: cli.schedule.clone(),
concurrency_policy: Some("Forbid".into()),
successful_jobs_history_limit: Some(3),
failed_jobs_history_limit: Some(3),
job_template: JobTemplateSpec {
metadata: Some(ObjectMeta {
labels: Some(managed_labels()),
..Default::default()
}),
spec: Some(JobSpec {
backoff_limit: Some(0),
active_deadline_seconds: Some(PLANNER_DEADLINE_SECS),
template: PodTemplateSpec {
spec: Some(PodSpec {
restart_policy: Some("Never".into()),
service_account_name: Some(SA_NAME.into()),
security_context: Some(pod_security(fs_group)),
containers: vec![Container {
name: "planner".into(),
image: Some(cli.image.clone()),
args: Some(vec!["run".into()]),
env: Some(env),
volume_mounts: Some(vec![VolumeMount {
name: "scratch".into(),
mount_path: DEFAULT_SCRATCH.into(),
..Default::default()
}]),
resources: Some(ResourceRequirements {
requests: Some(BTreeMap::from([
("cpu".into(), Quantity("200m".into())),
("memory".into(), Quantity("512Mi".into())),
])),
limits: Some(BTreeMap::from([(
"memory".into(),
Quantity("2Gi".into()),
)])),
..Default::default()
}),
security_context: Some(container_security()),
..Default::default()
}],
volumes: Some(vec![Volume {
name: "scratch".into(),
empty_dir: Some(EmptyDirVolumeSource {
size_limit: Some(Quantity("5Gi".into())),
..Default::default()
}),
..Default::default()
}]),
..Default::default()
}),
..Default::default()
},
..Default::default()
}),
},
..Default::default()
}),
status: None,
}
}

39
examples/harmony_tenant_backup/src/worker.rs Normal file
View File

@@ -0,0 +1,39 @@
//! Worker role: push one mounted, read-only volume into the restic repo.
//! Runs as a Job spawned by the planner, with the clone (or RWX PVC)
//! mounted at WORK_MOUNT and the OpenBao coordinates forwarded by the
//! planner.
use crate::common::{WORK_MOUNT, restic_host};
use crate::exec::{Restic, logline};
use anyhow::{Context, Result, anyhow};
pub async fn run(
restic: &Restic,
cluster_id: &str,
ns: &str,
run_id: &str,
logical_pvc: &str,
) -> Result<u8> {
if !tokio::fs::try_exists(WORK_MOUNT).await.unwrap_or(false) {
return Err(anyhow!("worker: {WORK_MOUNT} not mounted"));
}
restic
.ensure_repo(false)
.await
.context("restic repo unavailable")?;
logline(&format!(
"worker: pushing {WORK_MOUNT} as pvc/{logical_pvc}"
));
restic
.backup(
&restic_host(cluster_id, ns),
run_id,
ns,
"pvc",
&[format!("pvc:{logical_pvc}")],
WORK_MOUNT,
false,
)
.await?;
Ok(0)
}