Skip to content

Commit 7f70f21

Browse files
committed
feat(canopy): send health_details (sizes, fixes, duration) with verification
Canopy now accepts an arbitrary `health_details` map on /restore-verification, and bestool-canopy 0.4.4 adds a generic request escape hatch. The typed RestoreVerification struct has no such field, so pgro serializes the typed report, splices in `health_details`, and POSTs the merged body via CanopyClient::request to the same endpoint. health_details (snake_case) carries: - sizes: per-database on-disk bytes (pg_database_size), keyed by db name. - fixes: an arbitrary jsonb map of the fix steps the restore applied (locale, reindex, reset_wal, recreated_pg_wal). Stored in _pgro.restore_info.fixes by the init script and forwarded verbatim, so adding a fix is one shell line + its flag — no schema or operator change. - restore_duration_sec: wall-clock from the restore CR's createdAt to report time. sizes + fixes come from one read-only connection to the restore's postgres (done in the switchover block, before any ephemeral teardown destroys the DB). Gathering is best-effort: on the failure path, or if postgres never came up, those pieces are omitted and the verification still sends. Duration is independent of postgres. Records pg_resetwal / pg_wal-recreation via flag files so they surface in fixes alongside the existing locale/reindex flags.
1 parent 30ab42c commit 7f70f21

6 files changed

Lines changed: 265 additions & 11 deletions

File tree

Cargo.lock

Lines changed: 3 additions & 3 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

Cargo.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ publish = false
88
[dependencies]
99
anyhow = "1.0.102"
1010
axum = "0.8.9"
11-
bestool-canopy = "0.4.3"
11+
bestool-canopy = "0.4.4"
1212
bestool-kopia = { version = "0.3.4", features = ["proxy"] }
1313
cronexpr = "1.5.0"
1414
futures = "0.3.31"

src/canopy.rs

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,36 @@ impl Client {
147147
.map_err(|err| Error::Canopy(format!("restore_verification: {err:?}")))
148148
}
149149

150+
/// Report a restore outcome with an arbitrary JSON body — used to include
151+
/// the `health_details` field, which the typed [`RestoreVerification`]
152+
/// struct doesn't carry. `body` should be the serialized verification
153+
/// plus any extra fields. Goes to the same `POST /restore-verification`
154+
/// endpoint via the generic request escape hatch; a non-2xx response is
155+
/// an error carrying the status + body.
156+
pub async fn restore_verification_json(&self, body: &serde_json::Value) -> Result<()> {
157+
let resp = self
158+
.inner
159+
.request(
160+
bestool_canopy::reqwest::Method::POST,
161+
&self.base_url,
162+
"/restore-verification",
163+
)
164+
.await
165+
.map_err(|err| Error::Canopy(format!("restore_verification request: {err:?}")))?
166+
.json(body)
167+
.send()
168+
.await
169+
.map_err(|err| Error::Canopy(format!("restore_verification send: {err:?}")))?;
170+
let status = resp.status();
171+
if !status.is_success() {
172+
let text = resp.text().await.unwrap_or_default();
173+
return Err(Error::Canopy(format!(
174+
"restore_verification returned {status}: {text}"
175+
)));
176+
}
177+
Ok(())
178+
}
179+
150180
/// Direct access to the public-mTLS base URL the client is configured
151181
/// against. The tailnet path uses its own hardcoded URL inside
152182
/// `bestool-canopy`.

src/controllers/canopy/verification.rs

Lines changed: 163 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,16 @@
77
88
use bestool_canopy::{Outcome, RestoreVerification};
99
use jiff::Timestamp;
10-
use kube::ResourceExt;
10+
use k8s_openapi::api::core::v1::Secret;
11+
use kube::{Api, ResourceExt};
1112
use serde::Deserialize;
13+
use serde_json::{Value, json};
1214
use tracing::{info, warn};
1315
use uuid::Uuid;
1416

1517
use crate::{
1618
context::Context,
17-
controllers::canopy::labels,
19+
controllers::{canopy::labels, postgres},
1820
types::{PostgresPhysicalReplica, PostgresPhysicalRestore},
1921
};
2022

@@ -125,11 +127,32 @@ pub async fn report(
125127
s3_received_payload_bytes: Some(stats.received_payload_bytes as i64),
126128
};
127129

128-
match canopy.restore_verification(&report).await {
130+
// Serialize the typed report, then splice in `health_details` — the
131+
// typed struct doesn't carry it, so we send via the arbitrary-JSON
132+
// path. If serialization somehow fails, fall back to the typed call so
133+
// the outcome still reaches canopy.
134+
let mut body = match serde_json::to_value(&report) {
135+
Ok(v) => v,
136+
Err(err) => {
137+
warn!(
138+
restore = %restore.name_any(),
139+
error = %err,
140+
"canopy verification: failed to serialize report; sending without health_details"
141+
);
142+
if let Err(err) = canopy.restore_verification(&report).await {
143+
warn!(restore = %restore.name_any(), error = %err, "canopy verification report failed");
144+
}
145+
return;
146+
}
147+
};
148+
body["health_details"] = gather_health_details(ctx, replica, restore).await;
149+
150+
match canopy.restore_verification_json(&body).await {
129151
Ok(()) => info!(
130152
replica = %replica.name_any(),
131153
restore = %restore.name_any(),
132154
?outcome,
155+
health_details = %body["health_details"],
133156
"canopy verification reported"
134157
),
135158
Err(err) => warn!(
@@ -140,3 +163,140 @@ pub async fn report(
140163
),
141164
}
142165
}
166+
167+
/// Best-effort gather of the `health_details` map (snake_case keys):
168+
/// `{ sizes: {<db>: bytes}, fixes: {reindex, locale}, restore_duration_sec }`.
169+
///
170+
/// `sizes` and `fixes` come from a single read-only connection to the
171+
/// restore's postgres (`sizes` from `pg_database_size`, `fixes` from the
172+
/// `_pgro.restore_info` flags the init recorded). Any connection or query
173+
/// failure — expected on the failure path, where postgres may never have
174+
/// come up — just omits that piece; the verification still sends.
175+
/// `restore_duration_sec` is the wall-clock from the restore CR's
176+
/// `createdAt` to now (≈ activation), independent of postgres.
177+
async fn gather_health_details(
178+
ctx: &Context,
179+
replica: &PostgresPhysicalReplica,
180+
restore: &PostgresPhysicalRestore,
181+
) -> Value {
182+
let duration_sec = restore
183+
.status
184+
.as_ref()
185+
.and_then(|s| s.created_at.as_ref())
186+
.map(|created| Timestamp::now().duration_since(created.0).as_secs().max(0) as u64);
187+
188+
let pg = match gather_from_postgres(ctx, replica, restore).await {
189+
Ok(parts) => Some(parts),
190+
// Expected on the failure path (postgres may never have come up) and
191+
// on ephemeral replicas racing teardown; the verification still sends.
192+
Err(err) => {
193+
warn!(
194+
replica = %replica.name_any(),
195+
restore = %restore.name_any(),
196+
error = %err,
197+
"canopy verification: could not gather sizes/fixes; reporting without them"
198+
);
199+
None
200+
}
201+
};
202+
203+
build_health_details(duration_sec, pg)
204+
}
205+
206+
/// Postgres-derived pieces of the health details: per-database sizes and
207+
/// the `fixes` map the restore init recorded. `fixes` is an arbitrary JSON
208+
/// object (`{locale, reindex, reset_wal, ...}`) forwarded verbatim, so new
209+
/// fix flags flow through without operator changes.
210+
struct PostgresHealth {
211+
sizes: Vec<(String, u64)>,
212+
fixes: Value,
213+
}
214+
215+
/// Assemble the `health_details` JSON from already-gathered parts. Pure, so
216+
/// the snake_case wire shape is unit-testable without a database. Omits any
217+
/// piece that wasn't gathered.
218+
fn build_health_details(duration_sec: Option<u64>, pg: Option<PostgresHealth>) -> Value {
219+
let mut details = serde_json::Map::new();
220+
if let Some(secs) = duration_sec {
221+
details.insert("restore_duration_sec".into(), json!(secs));
222+
}
223+
if let Some(pg) = pg {
224+
let sizes_obj: serde_json::Map<String, Value> = pg
225+
.sizes
226+
.into_iter()
227+
.map(|(name, bytes)| (name, json!(bytes)))
228+
.collect();
229+
details.insert("sizes".into(), Value::Object(sizes_obj));
230+
details.insert("fixes".into(), pg.fixes);
231+
}
232+
Value::Object(details)
233+
}
234+
235+
/// Connect to the restore's postgres (as the replica's reader user) and
236+
/// read the per-database sizes + fix flags. Returns
237+
/// `(sizes, (locale_fixed, reindex_done))`.
238+
async fn gather_from_postgres(
239+
ctx: &Context,
240+
replica: &PostgresPhysicalReplica,
241+
restore: &PostgresPhysicalRestore,
242+
) -> crate::error::Result<PostgresHealth> {
243+
let namespace = replica.namespace().unwrap_or_default();
244+
let restore_name = restore.name_any();
245+
246+
let secrets: Api<Secret> = Api::namespaced(ctx.client.clone(), &namespace);
247+
let reader_secret = secrets.get(&replica.creds_secret_name()).await?;
248+
let reader_user = postgres::read_secret_field(&reader_secret, "username")?;
249+
let reader_password = postgres::read_secret_field(&reader_secret, "password")?;
250+
251+
let conn = postgres::connect_to_restore(
252+
&ctx.client,
253+
&namespace,
254+
&restore_name,
255+
"postgres",
256+
&reader_user,
257+
&reader_password,
258+
ctx.use_port_forward(),
259+
)
260+
.await?;
261+
262+
let sizes = postgres::list_database_sizes(&conn.client).await?;
263+
let fixes = postgres::read_restore_fixes(&conn.client)
264+
.await
265+
.unwrap_or_else(|_| json!({}));
266+
Ok(PostgresHealth { sizes, fixes })
267+
}
268+
269+
#[cfg(test)]
270+
mod tests {
271+
use super::*;
272+
273+
#[test]
274+
fn health_details_shape_is_snake_case() {
275+
let v = build_health_details(
276+
Some(700),
277+
Some(PostgresHealth {
278+
sizes: vec![("tamanu".into(), 1_872_782), ("postgres".into(), 12_829)],
279+
fixes: json!({ "locale": true, "reindex": false, "reset_wal": false }),
280+
}),
281+
);
282+
assert_eq!(v["restore_duration_sec"], json!(700));
283+
assert_eq!(v["sizes"]["tamanu"], json!(1_872_782u64));
284+
assert_eq!(v["sizes"]["postgres"], json!(12_829u64));
285+
// fixes is forwarded verbatim, so new flags pass through untouched.
286+
assert_eq!(v["fixes"]["locale"], json!(true));
287+
assert_eq!(v["fixes"]["reindex"], json!(false));
288+
assert_eq!(v["fixes"]["reset_wal"], json!(false));
289+
}
290+
291+
#[test]
292+
fn health_details_omits_ungathered_parts() {
293+
// Failure path: no postgres connection, no createdAt.
294+
let v = build_health_details(None, None);
295+
assert_eq!(v, json!({}));
296+
// Duration known but postgres unreachable: sizes/fixes omitted.
297+
let v = build_health_details(Some(42), None);
298+
assert_eq!(v, json!({ "restore_duration_sec": 42 }));
299+
assert!(v.get("sizes").is_none());
300+
assert!(v.get("fixes").is_none());
301+
}
302+
}

src/controllers/postgres.rs

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -258,6 +258,49 @@ pub async fn database_size_on(pg: &tokio_postgres::Client) -> Result<u64> {
258258
Ok(size as u64)
259259
}
260260

261+
/// Per-database on-disk sizes in bytes, keyed by database name, excluding
262+
/// the `template0`/`template1` templates. Runs on an already-open
263+
/// connection (typically to the `postgres` database). Used to build the
264+
/// `sizes` map in the canopy restore-verification `health_details`.
265+
pub async fn list_database_sizes(pg: &tokio_postgres::Client) -> Result<Vec<(String, u64)>> {
266+
let rows = pg
267+
.query(
268+
"SELECT datname, pg_database_size(datname) FROM pg_database \
269+
WHERE datname NOT IN ('template0', 'template1') AND datallowconn \
270+
ORDER BY pg_database_size(datname) DESC",
271+
&[],
272+
)
273+
.await?;
274+
Ok(rows
275+
.iter()
276+
.map(|r| {
277+
let name: String = r.get(0);
278+
let size: i64 = r.get(1);
279+
(name, size.max(0) as u64)
280+
})
281+
.collect())
282+
}
283+
284+
/// Read the `fixes` map the restore's init recorded into
285+
/// `_pgro.restore_info` — an arbitrary JSON object like
286+
/// `{"locale": true, "reindex": false, "reset_wal": false, ...}`. Read as
287+
/// text and parsed here so we don't need tokio_postgres's serde_json
288+
/// feature. Absent row/column (older restores) reads as an empty object.
289+
/// Runs on an already-open connection to the `postgres` database.
290+
pub async fn read_restore_fixes(pg: &tokio_postgres::Client) -> Result<serde_json::Value> {
291+
let row = pg
292+
.query_opt(
293+
"SELECT coalesce(fixes::text, '{}') FROM _pgro.restore_info WHERE id = 1",
294+
&[],
295+
)
296+
.await?;
297+
let text: String = match row {
298+
Some(r) => r.get(0),
299+
None => return Ok(serde_json::json!({})),
300+
};
301+
Ok(serde_json::from_str(&text).unwrap_or_else(|_| serde_json::json!({})))
302+
}
303+
261304
/// Query the on-disk size of the given database (bytes) via `pg_database_size()`.
262305
pub async fn measure_database_size(
263306
client: &Client,

src/controllers/restore/builders.rs

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1209,6 +1209,7 @@ HBAEOF
12091209
if [ ! -d "$PGDATA/pg_wal" ]; then
12101210
echo "pg_wal directory missing (snapshot may be from a Windows host with WAL on a separate path), creating empty pg_wal..."
12111211
mkdir -p "$PGDATA/pg_wal"
1212+
touch /pgdata/fix-recreated-pg-wal
12121213
fi
12131214
12141215
# Run a postgres --single SQL command with a two-stage pg_resetwal -f
@@ -1252,6 +1253,7 @@ postgres_single_or_resetwal() {{
12521253
rm -f "$logfile"
12531254
pg_resetwal -f "$PGDATA"
12541255
touch /pgdata/needs-reindex-all
1256+
touch /pgdata/fix-reset-wal
12551257
echo "$sql_input" | postgres --single -D "$PGDATA" postgres
12561258
return $?
12571259
fi
@@ -1273,6 +1275,7 @@ postgres_single_or_resetwal() {{
12731275
rm -f "$logfile"
12741276
pg_resetwal -f "$PGDATA"
12751277
touch /pgdata/needs-reindex-all
1278+
touch /pgdata/fix-reset-wal
12761279
echo "$sql_input" | postgres --single -D "$PGDATA" postgres
12771280
}}
12781281
@@ -1368,13 +1371,29 @@ ALTER ROLE ${{ANALYTICS_USERNAME}} WITH SUPERUSER;
13681371
SQLEOF
13691372
fi
13701373
1374+
# Record which "fix" steps this restore had to apply, so the operator can
1375+
# read them back (SELECT from _pgro.restore_info) and forward them to
1376+
# canopy in the restore-verification health_details. Stored as a jsonb map
1377+
# so adding a new fix is one shell line here plus recording its flag — no
1378+
# schema change, no operator change (the operator forwards the map as-is).
1379+
# Each fix is keyed by a flag file the fix step touches:
1380+
# locale — the post-startup locale rewrite actually changed rows
1381+
# reindex — REINDEX ran (after pg_resetwal, or a locale rewrite)
1382+
# reset_wal — pg_resetwal -f ran (snapshot's trailing WAL was unusable)
1383+
# recreated_pg_wal — an empty pg_wal was created (Windows-host snapshot)
13711384
if [ -f /pgdata/needs-reindex ] || [ -f /pgdata/needs-reindex-all ]; then
13721385
PGRO_STAGE=restored
1386+
PGRO_REINDEX=true
13731387
else
13741388
PGRO_STAGE=ready
1389+
PGRO_REINDEX=false
13751390
fi
1391+
if [ "${{LOCALE_CHANGED:-0}}" != "0" ]; then PGRO_LOCALE=true; else PGRO_LOCALE=false; fi
1392+
if [ -f /pgdata/fix-reset-wal ]; then PGRO_RESET_WAL=true; else PGRO_RESET_WAL=false; fi
1393+
if [ -f /pgdata/fix-recreated-pg-wal ]; then PGRO_RECREATED_WAL=true; else PGRO_RECREATED_WAL=false; fi
1394+
PGRO_FIXES="{{\"locale\": ${{PGRO_LOCALE}}, \"reindex\": ${{PGRO_REINDEX}}, \"reset_wal\": ${{PGRO_RESET_WAL}}, \"recreated_pg_wal\": ${{PGRO_RECREATED_WAL}}}}"
13761395
1377-
echo "Writing restore metadata (stage=${{PGRO_STAGE}})..."
1396+
echo "Writing restore metadata (stage=${{PGRO_STAGE}} fixes=${{PGRO_FIXES}})..."
13781397
psql -U postgres -d postgres << SQLEOF
13791398
CREATE SCHEMA IF NOT EXISTS _pgro;
13801399
CREATE TABLE IF NOT EXISTS _pgro.restore_info (
@@ -1387,14 +1406,16 @@ CREATE TABLE IF NOT EXISTS _pgro.restore_info (
13871406
);
13881407
ALTER TABLE _pgro.restore_info ADD COLUMN IF NOT EXISTS stage text NOT NULL DEFAULT 'restored';
13891408
ALTER TABLE _pgro.restore_info ADD COLUMN IF NOT EXISTS last_transition_time timestamptz NOT NULL DEFAULT now();
1390-
INSERT INTO _pgro.restore_info (id, snapshot_id, snapshot_time, stage, last_transition_time)
1391-
VALUES (1, '${{PGRO_SNAPSHOT_ID}}', CASE WHEN '${{PGRO_SNAPSHOT_TIME}}' = '' THEN NULL ELSE '${{PGRO_SNAPSHOT_TIME}}'::timestamptz END, '${{PGRO_STAGE}}', now())
1409+
ALTER TABLE _pgro.restore_info ADD COLUMN IF NOT EXISTS fixes jsonb;
1410+
INSERT INTO _pgro.restore_info (id, snapshot_id, snapshot_time, stage, last_transition_time, fixes)
1411+
VALUES (1, '${{PGRO_SNAPSHOT_ID}}', CASE WHEN '${{PGRO_SNAPSHOT_TIME}}' = '' THEN NULL ELSE '${{PGRO_SNAPSHOT_TIME}}'::timestamptz END, '${{PGRO_STAGE}}', now(), '${{PGRO_FIXES}}'::jsonb)
13921412
ON CONFLICT (id) DO UPDATE
13931413
SET snapshot_id = EXCLUDED.snapshot_id,
13941414
snapshot_time = EXCLUDED.snapshot_time,
13951415
restored_at = now(),
13961416
stage = EXCLUDED.stage,
1397-
last_transition_time = now();
1417+
last_transition_time = now(),
1418+
fixes = EXCLUDED.fixes;
13981419
SQLEOF
13991420
14001421
echo "Stopping temporary postgres..."

0 commit comments

Comments
 (0)