From cb19f95872d064dc5ca380c090adfe5b61db67a4 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Wed, 11 Oct 2023 18:08:18 +0200 Subject: [PATCH] WIP: DR from S3 object versioning --- Cargo.lock | 25 +- s3_scrubber/Cargo.toml | 6 +- s3_scrubber/src/lib.rs | 1 - s3_scrubber/src/main.rs | 14 + .../restore_tenant_from_object_versioning.rs | 499 ++++++++++++++++++ 5 files changed, 541 insertions(+), 4 deletions(-) create mode 100644 s3_scrubber/src/restore_tenant_from_object_versioning.rs diff --git a/Cargo.lock b/Cargo.lock index be3f179d5f..21d83cb7fc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -570,9 +570,9 @@ dependencies = [ [[package]] name = "aws-smithy-types" -version = "0.56.0" +version = "0.56.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eed0a94eefd845a2a78677f1b72f02fa75802d38f7f59be675add140279aa8bf" +checksum = "d90dbc8da2f6be461fa3c1906b20af8f79d14968fe47f2b7d29d086f62a51728" dependencies = [ "base64-simd", "itoa", @@ -2420,6 +2420,16 @@ dependencies = [ "winapi", ] +[[package]] +name = "nu-ansi-term" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84" +dependencies = [ + "overload", + "winapi", +] + [[package]] name = "num-bigint" version = "0.4.3" @@ -2652,6 +2662,12 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4030760ffd992bef45b0ae3f10ce1aba99e33464c90d14dd7c039884963ddc7a" +[[package]] +name = "overload" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" + [[package]] name = "pagectl" version = "0.1.0" @@ -3748,9 +3764,11 @@ dependencies = [ "aws-config", "aws-sdk-s3", "aws-smithy-http", + "aws-smithy-types", "aws-types", "bincode", "bytes", + "camino", "chrono", "clap", "crc32c", @@ -3758,9 +3776,11 @@ dependencies = [ "futures-util", "hex", "histogram", + "humantime", "itertools", "pageserver", "rand", + "remote_storage", "reqwest", "serde", "serde_json", @@ -4942,6 +4962,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30a651bc37f915e81f087d86e62a18eec5f79550c7faff886f7090b4ea757c77" dependencies = [ "matchers", + "nu-ansi-term", "once_cell", "regex", "serde", diff --git a/s3_scrubber/Cargo.toml b/s3_scrubber/Cargo.toml index f3ea6e222c..4410c00475 100644 --- a/s3_scrubber/Cargo.toml +++ b/s3_scrubber/Cargo.toml @@ -33,9 +33,13 @@ reqwest = { workspace = true, default-features = false, features = ["rustls-tls" aws-config = { workspace = true, default-features = false, features = ["rustls", "credentials-sso"] } pageserver = { path = "../pageserver" } +remote_storage = { path = "../libs/remote_storage" } tracing.workspace = true -tracing-subscriber.workspace = true +tracing-subscriber = { version = "0.3.17", features = ["ansi"] } clap.workspace = true tracing-appender = "0.2" histogram = "0.7" +humantime.workspace = true +camino.workspace = true +aws-smithy-types = "0.56.1" diff --git a/s3_scrubber/src/lib.rs b/s3_scrubber/src/lib.rs index 072cdf1d93..4013664932 100644 --- a/s3_scrubber/src/lib.rs +++ b/s3_scrubber/src/lib.rs @@ -4,7 +4,6 @@ pub mod delete_batch_producer; pub mod metadata_stream; mod s3_deletion; pub mod scan_metadata; - use std::env; use std::fmt::Display; use std::time::Duration; diff --git a/s3_scrubber/src/main.rs b/s3_scrubber/src/main.rs index 3c60723f88..88e05605cb 100644 --- a/s3_scrubber/src/main.rs +++ b/s3_scrubber/src/main.rs @@ -16,6 +16,8 @@ use tracing::{info, warn}; use clap::{Parser, Subcommand, ValueEnum}; +mod restore_tenant_from_object_versioning; + #[derive(Parser)] #[command(author, version, about, long_about = None)] #[command(arg_required_else_help(true))] @@ -59,6 +61,9 @@ enum Command { skip_validation: bool, }, ScanMetadata {}, + RestoreTenantFromObjectVersioningMostRecentIndexPart( + restore_tenant_from_object_versioning::Command, + ), } async fn tidy( @@ -247,5 +252,14 @@ async fn main() -> anyhow::Result<()> { } } }, + Command::RestoreTenantFromObjectVersioningMostRecentIndexPart(arg) => { + match restore_tenant_from_object_versioning::doit(arg).await { + Err(e) => { + tracing::error!("Failed: {e:?}"); + Err(e) + } + Ok(()) => Ok(()), + } + } } } diff --git a/s3_scrubber/src/restore_tenant_from_object_versioning.rs b/s3_scrubber/src/restore_tenant_from_object_versioning.rs new file mode 100644 index 0000000000..a6c7d126dc --- /dev/null +++ b/s3_scrubber/src/restore_tenant_from_object_versioning.rs @@ -0,0 +1,499 @@ +//! Restore pageserver state from S3 object versioning. +//! +//! This sub-cmmand allows restoring a tenant's pageserver S3 state from S3 object versioning. +//! +//! # Instructions +//! +//! - Create empty directory `./restore` +//! - Run command +//! ``` +//! SSO_ACCOUNT_ID=... REGION=... \ +//! BUCKET=neon-{prod,staging}-storage-... \ +//! cargo run -p s3_scrubber \ +//! restore-tenant-from-object-versioning-most-recent-index-part \ +//! TENANT_TO_RESTORE \ +//! ./restore +//! timeline-list TIMELINE_TO_RESTORE TIMELINE_TO_RESTORE ... +//! ``` +//! - `./restore` now contains the timeline state referenced by the latest `index_part.json`s of the +//! specified timelines in the `timeline-list`` argument +//! - Use `cargo neon` to start a pageserver +//! - `rm -rf .neon` +//! - `cargo neon init` +//! - `sed -i 's/.*control_plane_api.*/#\1/` .neon/config +//! - configure the pageserver remote storage to point to the restore directory, using text editor +//! ``` +//! [remote_storage] +//! local_path = "/path/to/restore/pageserver/v1" +//! ```` +//! - `cargo neon start` +//! - make sure attaching the tenant works +//! - `curl -X POST localhost:9898/v1/tenant/TENANT_TO_RESTORE/attach` +//! - check `curl -X GET localhost:9898/v1/tenant/TENANT_TO_RESTORE | jq` +//! - for each timeline $timeline_id to restore: +//! - `cargo neon mappings map --branch-name restore --tenant-id TENANT_TO_RESTORE --timeline-id $timeline_id` +//! - `cargo neon endpoint create --tenant-id TENANT_TO_RESTORE --timeline-id $timeline_id restore-$timeline_id` +//! - `cargo neon endpoint start --tenant-id TENANT_TO_RESTORE restore-$timeline_id` +//! - dump database to somewhere using like so: `./pg_install/v15/bin/pg_dump postgresql://cloud_admin@127.0.0.1:55432/neondb > $timeline_id.dump` +//! - `cargo neon endpoint stop --tenant-id TENANT_TO_RESTORE restore-$timeline_id` +//! +//! - Use the pg_dump files to restore the database into a new Neon project. +//! +//! # Limitations & Future Work +//! +//! Just restoring Pageserver S3 state restores a consistent state at an LSN that is NOT THE LAST COMMIT LSN. +//! The reason is that Pageserver uploads layers to S3 with implementation-specific delays that are optimized for day-to-day operation. +//! +//! If we still had the Safekeeper WAL, we could restore the Safekeeper S3 state in a similar way. +//! In that case, we wouldn't need the `pg_dump` step. +//! We would simply attach the tenant to safekeepers and pageservers. +//! When attaching to Safekeeper, we would need to tell it that PS remote_consistent_lsn is the restore-point-LSN, +//! i.e., the LSNs in the restored index_part.json's in Pageserver S3 state. +//! Pageserver attach would pick up the restored state from S3, and the Safekeeper & Pageserver would +//! resume normal operation as if the clock had been wound back to restore-point-LSN. + +use std::{ + collections::{HashMap, HashSet}, + num::{NonZeroU32, NonZeroUsize}, + sync::Arc, +}; + +use anyhow::Context; + +use aws_sdk_s3::operation::list_object_versions::ListObjectVersionsOutput; +use aws_types::region::Region; +use camino::Utf8PathBuf; +use pageserver::tenant::{ + remote_timeline_client::index::LayerFileMetadata, TENANTS_SEGMENT_NAME, + TENANT_DELETED_MARKER_FILE_NAME, TIMELINES_SEGMENT_NAME, +}; +use remote_storage::{GenericRemoteStorage, RemotePath, RemoteStorageConfig, RemoteStorageKind}; +use s3_scrubber::{init_logging, init_s3_client, BucketConfig}; +use tokio::io::AsyncReadExt; +use tracing::{debug, info, info_span, Instrument}; +use utils::id::{TenantId, TimelineId}; + +#[derive(Debug, Clone, clap::Subcommand)] +enum ResurrectTimelines { + TimelineList { timeline_ids: Vec }, + // AllTimelinesDeletedAfter { timestamp: humantime::Timestamp }, +} + +#[derive(clap::Args)] +pub(crate) struct Command { + tenant_id: TenantId, + dest_dir: Utf8PathBuf, + #[clap(short, long)] + dry_run: bool, + #[clap(subcommand)] + timelines: ResurrectTimelines, +} + +pub(crate) async fn doit(args: Command) -> anyhow::Result<()> { + let _logging_guard = { + let log_prefix = format!("restore_tenant_from_object_versioning_{}", args.tenant_id); + let dry_suffix = if args.dry_run { "__dry" } else { "" }; + let file_name = { + format!( + "{}_{}{}.log", + log_prefix, + chrono::Utc::now().format("%Y_%m_%d__%H_%M_%S"), + dry_suffix, + ) + }; + init_logging(&file_name) + }; + + let restore_dst = if tokio::fs::try_exists(&args.dest_dir).await? { + anyhow::bail!("destination directory already exists: {}", args.dest_dir,); + } else { + GenericRemoteStorage::from_config(&RemoteStorageConfig { + max_concurrent_syncs: NonZeroUsize::new(100).unwrap(), + max_sync_errors: NonZeroU32::new(1).unwrap(), // ???? would want so specify 0 + storage: RemoteStorageKind::LocalFs(args.dest_dir.clone()), + }) + .context("instantiate restore destination")? + }; + + let bucket_config = BucketConfig::from_env()?; + + let bucket_region = Region::new(bucket_config.region); + let delimiter = "/".to_string(); + let s3_client = Arc::new(init_s3_client(bucket_config.sso_account_id, bucket_region)); + let tenant_root = [ + "pageserver", + "v1", + TENANTS_SEGMENT_NAME, + &args.tenant_id.to_string(), + ] + .join(&delimiter); + + let tenant_delete_marker = [&tenant_root, TENANT_DELETED_MARKER_FILE_NAME].join(&delimiter); + + // - Ensure the prefix is empty when ignoring existence of versions. + // - Ensure the tenant delete marker key is part of `DeleteMarkers`. If it isn't, the tenant hasn't finished deletion yet and we should let pageservers complete it first. + // - Restore each index_part.json based on the version in DeleteMarkers as well as the layers it references. For the layers, also use the version in DeleteMarkers and ensure it is the latest. + // - Remove the deleted_at mark for the specified timelines. + // + // Notes: + // - The restore will happen in-place because it's hard to change tenant/timeline ids. + // - The restore could be interrupted mid-way. + // - Hence, separate plan-making and plan-execution. + + async { + info!("send request"); + let res = s3_client + .list_objects_v2() + .bucket(&bucket_config.bucket) + .prefix(&tenant_root) + .send() + .await?; + + info!(response=?res, "got response"); + + if res.key_count() > 0 { + anyhow::bail!("tenant prefix is not empty in S3"); + } + if res.is_truncated() { + unimplemented!("can this even happen") + } + Ok(()) + } + .instrument(info_span!("ensure prefix empty")) + .await + .context("ensure prefix is empty")?; + + async { + info!("send request"); + let res = s3_client + .list_object_versions() + .bucket(&bucket_config.bucket) + .prefix(&tenant_delete_marker) + .send() + .await?; + debug!(response=?res, "got response"); + + if res.is_truncated() { + unimplemented!("can this even happen") + } + + let markers = res + .delete_markers() + .context("expected delete marker in response")?; + + if markers.len() != 1 { + anyhow::bail!("expected exactly one delete marker because we create and delete the marker exactly once, got {}", markers.len()); + } + + if !markers[0].is_latest() { + anyhow::bail!("expected delete marker to have IsLatest set: {:?}", markers[0]); + } + + Ok(()) + } + .instrument(info_span!( + "ensure tenant delete marker exists in DeleteMarkers", + tenant_delete_marker, + )) + .await + .context("ensure tenant delete marker exists in DeleteMarkers")?; + + let timelines = match args.timelines { + ResurrectTimelines::TimelineList { timeline_ids } => timeline_ids, + }; + + // Fetch all the information we need to execute the restore. + let version_responses_by_timeline = async { + let mut out: HashMap>> = Default::default(); + for tl in &timelines { + async { + let timeline_prefix = [tenant_root.as_str(), TIMELINES_SEGMENT_NAME , &tl.to_string()].join(&delimiter); + let mut next_key_marker = None; + let mut next_version_id_marker = None; + loop { + info!("sending request"); + let res: ListObjectVersionsOutput = s3_client.list_object_versions() + .bucket(&bucket_config.bucket) + .prefix(&timeline_prefix) + .set_key_marker(next_key_marker.take()) + .set_version_id_marker(next_version_id_marker.take()) + .send() + .await?; + + let res = Arc::new(res); + out.entry(*tl).or_default().push(Arc::clone(&res)); + + info!("got response"); + match res.versions() { + Some(versions) => { + for version in versions { + info!("version: {:?}", version); + } + } + None => { + info!("no versions"); + } + } + match res.delete_markers() { + Some(markers) => { + for marker in markers { + info!("delete marker: {:?}", marker); + } + } + None => { + info!("no delete markers"); + } + } + + if !res.is_truncated() { + break; + } + next_key_marker = res.next_key_marker().map(|s| s.to_string()); + next_version_id_marker = res.next_version_id_marker().map(|s| s.to_string()); + if let (None, None) = (&next_key_marker, &next_version_id_marker) { + anyhow::bail!("s3 returned is_truncated=true but neither next_key_marker nor next_version_id_marker are set"); + } + } + Ok(()) + }.instrument(info_span!("timeline", timeline_id=%tl)).await?; + } + anyhow::Ok(out) + }.instrument(info_span!("list all object versions and delete markers")).await?; + #[derive(Debug)] + struct LatestVersion { + key: String, + last_modified: aws_smithy_types::DateTime, + version_id: String, + } + let find_latest_version_based_on_delete_marker_last_modified = |tl: &TimelineId, key: &str| { + let restore_version_delete_marker = { + let mut candidates = Vec::new(); + for res in &version_responses_by_timeline[tl] { + let Some(markers) = res.delete_markers() else { + continue; + }; + for marker in markers { + if !marker.is_latest() { + continue; + } + if marker.key().unwrap() != key { + continue; + } + candidates.push(LatestVersion { + key: marker.key().unwrap().to_owned(), + last_modified: marker.last_modified().unwrap().clone(), + version_id: marker.version_id().unwrap().to_owned(), + }); + } + } + info!(?candidates, "marker candidates"); + if candidates.len() != 1 { + anyhow::bail!("expected exactly one IsLatest, got {}", candidates.len()); + } + candidates.pop().unwrap() + }; + info!(?restore_version_delete_marker, "found marker"); + + // There's no way to get the latest version from the delete marker. + // But, we observe (can't find written guarantee) that the Delete Marker's "Last Modified" is >= the latest version. + // So, find latest version based on that. + let restore_version = { + let mut candidates = Vec::new(); + for res in &version_responses_by_timeline[tl] { + let Some(versions) = res.versions() else { + continue; + }; + for version in versions { + if version.key().unwrap() != restore_version_delete_marker.key { + continue; + } + candidates.push(LatestVersion { + key: version.key().unwrap().to_owned(), + last_modified: version.last_modified().unwrap().clone(), + version_id: version.version_id().unwrap().to_owned(), + }); + } + } + candidates.sort_by_key(|v| v.last_modified.clone()); + info!(?candidates, "version candidates"); + if candidates.is_empty() { + anyhow::bail!( + "expected at least one version matching the delete marker's key, got none" + ); + } + { + let mut uniq = HashSet::new(); + for v in &candidates { + if !uniq.insert(v.last_modified.clone()) { + anyhow::bail!("last_modified timestamps are not unique, don't know which version to pick"); + } + } + } + candidates.pop().unwrap() // we sorted ascending, so, pop() is the latest + }; + anyhow::Ok(restore_version) + }; + + let latest_index_part_versions: HashMap = { + let span = info_span!("find index part version"); + let _enter = span.enter(); + + // The latest index part for a deleted tenant is always a DeletedMarker + let mut out = HashMap::new(); + for tl in &timelines { + let span = info_span!("timeline", timeline_id=%tl); + let _enter = span.enter(); + let restore_version = find_latest_version_based_on_delete_marker_last_modified( + tl, + // TODO: support generation numbers + &[ + &tenant_root, + TIMELINES_SEGMENT_NAME, + &tl.to_string(), + pageserver::tenant::IndexPart::FILE_NAME, + ] + .join(&delimiter), + )?; + out.insert(*tl, restore_version); + } + out + }; + + let index_part_contents: HashMap = async { + let mut out = HashMap::new(); + + for tl in &timelines { + async { + let v = &latest_index_part_versions[tl]; + + let mut body_buf = Vec::new(); + info!("send request"); + let res = s3_client + .get_object() + .bucket(&bucket_config.bucket) + .key(&v.key) + .version_id(&v.version_id) + .send() + .await?; + info!(?res, "got response header"); + + res.body + .into_async_read() + .read_to_end(&mut body_buf) + .await?; + + let body_buf = String::from_utf8(body_buf)?; + info!(body_buf, "received response body"); + + let mut index_part: pageserver::tenant::IndexPart = + serde_json::from_str(&body_buf)?; + info!(?index_part, "parsed index part"); + + let deleted_at = index_part.deleted_at.take(); + info!( + ?deleted_at, + "removing deleted_at field from index part, previous value logged here" + ); + + let updated_buf = serde_json::to_vec(&index_part)?; + let updated_buf_len = updated_buf.len(); + info!("uploading modified index part to restore_dst"); + restore_dst + .upload( + std::io::Cursor::new(updated_buf), + updated_buf_len, + &RemotePath::from_string(&v.key).unwrap(), + None, + ) + .await + .context("upload modified index part to restore_dst")?; + + out.insert(*tl, index_part); + + anyhow::Ok(()) + } + .instrument(info_span!("timeline", timeline_id=%tl)) + .await?; + } + + anyhow::Ok(out) + } + .instrument(info_span!("get index part contents")) + .await + .context("get index part contents")?; + + async { + for (tl, index_part) in &index_part_contents { + async { + for (layer_file_name, layer_md) in &index_part.layer_metadata { + let layer_md: LayerFileMetadata = layer_md.into(); + async { + // TODO: support generations + let layer_file_key = [ + &tenant_root, + TIMELINES_SEGMENT_NAME, + &tl.to_string(), + &layer_file_name.file_name(), + ] + .join(&delimiter); + + // The latest index parts naturally reference the latest layers. + // So, a deleted tenant's latest layers are the ones in DeleteMarkers. + // + // If we want to support restoring from not-latest index part, this will require more work. + // The idea is to + // 1. every index_part.json that we upload contains a strongly monotonically increasing sequence number + // 2. every image layer that we upload is S3-metadata-tagged with the sequence number of the IndexPart + // in which it first appeared. + // This allows to recover the correct layer object version, even if we have a bug that overwrites layers. + let restore_version = + find_latest_version_based_on_delete_marker_last_modified( + tl, + &layer_file_key, + )?; + + // TODO: teach RemoteStorage copy operation so we can use s3_client.copy_object() + async { + let res = s3_client + .get_object() + .bucket(&bucket_config.bucket) + .key(&restore_version.key) + .version_id(&restore_version.version_id) + .send() + .await + .context("get object header")?; + // TODO: instead of file_size(), do actual data integrity checking. + restore_dst + .upload( + res.body.into_async_read(), + layer_md.file_size().try_into().unwrap(), + &RemotePath::from_string(&restore_version.key).unwrap(), + None, + ) + .await + .context("download-body-and-upload")?; + anyhow::Ok(()) + } + .instrument(info_span!("copy", layer_file_name=%layer_file_name)) + .await?; + + anyhow::Ok(()) + } + .instrument(info_span!("layer", layer_file_name=%layer_file_name)) + .await?; + } + + anyhow::Ok(()) + } + .instrument(info_span!("timeline", timeline_id=%tl)) + .await?; + } + + anyhow::Ok(()) + } + .instrument(info_span!("download layer files into restore_dst")) + .await + .context("download layers into restore dst")?; + + Ok(()) +}