diff --git a/control_plane/src/storage.rs b/control_plane/src/storage.rs index 86c59d0ec2..1c17672880 100644 --- a/control_plane/src/storage.rs +++ b/control_plane/src/storage.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; use std::fs::File; -use std::io::{BufReader, Write}; +use std::io::{BufReader, Read, Write}; use std::net::TcpStream; use std::num::NonZeroU64; use std::path::PathBuf; @@ -533,17 +533,27 @@ impl PageServerNode { &self, tenant_id: ZTenantId, timeline_id: ZTimelineId, - tarfile: PathBuf, - lsn: Lsn, + base: (Lsn, PathBuf), + pg_wal: Option<(Lsn, PathBuf)>, ) -> anyhow::Result<()> { let mut client = self.pg_connection_config.connect(NoTls).unwrap(); - // Init reader - let file = File::open(tarfile)?; - let mut reader = BufReader::new(file); + // Init base reader + let (start_lsn, base_tarfile_path) = base; + let base_tarfile = File::open(base_tarfile_path)?; + let mut reader = BufReader::new(base_tarfile); + + // If there's any wal, extend the reader and end_lsn + let end_lsn = if let Some((end_lsn, wal_tarfile_path)) = pg_wal { + let wal_tarfile = File::open(wal_tarfile_path)?; + (&mut reader).chain(BufReader::new(wal_tarfile)); + end_lsn + } else { + start_lsn + }; // Init writer - let import_cmd = format!("import {tenant_id} {timeline_id} {lsn}"); + let import_cmd = format!("import {tenant_id} {timeline_id} {start_lsn} {end_lsn}"); let mut writer = client.copy_in(&import_cmd)?; // Stream reader -> writer diff --git a/neon_local/src/main.rs b/neon_local/src/main.rs index 0e48d6bd6c..253d3ccd1a 100644 --- a/neon_local/src/main.rs +++ b/neon_local/src/main.rs @@ -14,7 +14,7 @@ use safekeeper::defaults::{ DEFAULT_PG_LISTEN_PORT as DEFAULT_SAFEKEEPER_PG_PORT, }; use std::collections::{BTreeSet, HashMap}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::process::exit; use std::str::FromStr; use utils::{ @@ -165,9 +165,13 @@ fn main() -> Result<()> { .arg(timeline_id_arg.clone()) .arg(Arg::new("node-name").long("node-name").takes_value(true) .help("Name to assign to the imported timeline")) - .arg(Arg::new("tarfile").long("tarfile").takes_value(true) - .help("Basebackup tarfil to import")) - .arg(Arg::new("lsn").long("lsn").takes_value(true) + .arg(Arg::new("base-tarfile").long("base-tarfile").takes_value(true) + .help("Basebackup tarfile to import")) + .arg(Arg::new("base-lsn").long("base-lsn").takes_value(true) + .help("Lsn the basebackup starts at")) + .arg(Arg::new("wal-tarfile").long("wal-tarfile").takes_value(true) + .help("Wal to add after base")) + .arg(Arg::new("end-lsn").long("end-lsn").takes_value(true) .help("Lsn the basebackup ends at"))) ).subcommand( App::new("tenant") @@ -629,14 +633,26 @@ fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) - .expect("No timeline id provided"); let name = import_match.value_of("node-name") .ok_or_else(|| anyhow!("No node name provided"))?; - let tarfile = import_match.value_of("tarfile") - .ok_or_else(|| anyhow!("No tarfile provided"))?; - let lsn = Lsn::from_str(import_match.value_of("lsn") - .ok_or_else(|| anyhow!("No lsn provided"))?)?; + + // Parse base inputs + let base_tarfile = import_match.value_of("base-tarfile") + .map(|s| PathBuf::from_str(s).unwrap()) + .ok_or_else(|| anyhow!("No base-tarfile provided"))?; + let base_lsn = Lsn::from_str(import_match.value_of("base-lsn") + .ok_or_else(|| anyhow!("No base-lsn provided"))?)?; + let base = (base_lsn, base_tarfile); + + // Parse pg_wal inputs + let wal_tarfile = import_match.value_of("wal-tarfile") + .map(|s| PathBuf::from_str(s).unwrap()); + let end_lsn = import_match.value_of("end-lsn") + .map(|s| Lsn::from_str(s).unwrap()); + // TODO validate both or none are provided + let pg_wal = end_lsn.zip(wal_tarfile); let mut cplane = ComputeControlPlane::load(env.clone())?; println!("Importing timeline into pageserver ..."); - pageserver.timeline_import(tenant_id, timeline_id, tarfile.try_into()?, lsn)?; + pageserver.timeline_import(tenant_id, timeline_id, base, pg_wal)?; println!("Creating node for imported timeline ..."); env.register_branch_mapping(name.to_string(), tenant_id, timeline_id)?; cplane.new_node(tenant_id, name, timeline_id, None, None)?; diff --git a/pageserver/src/import_datadir.rs b/pageserver/src/import_datadir.rs index 083059cb7b..399e325898 100644 --- a/pageserver/src/import_datadir.rs +++ b/pageserver/src/import_datadir.rs @@ -434,65 +434,56 @@ fn import_wal( pub fn import_timeline_from_tar( tline: &mut DatadirTimeline, - reader: Reader, + mut reader: Reader, + base_lsn: Lsn, end_lsn: Lsn, ) -> Result<()> { - let mut ar = tar::Archive::new(reader); - - // HACK - let mut modification = tline.begin_modification(Lsn::from_str("0/3000028")?); + info!("importing base at {}", base_lsn); + let mut modification = tline.begin_modification(base_lsn); modification.init_empty()?; - let mut entries_iter = ar.entries()?; - - let mut pg_control: Option = None; - // Import base - for e in &mut entries_iter { - let mut entry = e.unwrap(); + for base_tar_entry in tar::Archive::new(&mut reader).entries()? { + let mut entry = base_tar_entry.unwrap(); let header = entry.header(); let file_path = header.path().unwrap().into_owned(); match header.entry_type() { tar::EntryType::Regular => { let mut buffer = Vec::new(); - // read the whole entry + // TODO stream it instead entry.read_to_end(&mut buffer).unwrap(); - if let Some(control) = import_file(&mut modification, &file_path.as_ref(), &buffer)? { - pg_control = Some(control); - } - } + import_file(&mut modification, &file_path.as_ref(), &buffer)?; + }, tar::EntryType::Directory => { info!("directory {:?}", file_path); if file_path.starts_with("pg_wal") { - info!("found pg_wal"); - break; + info!("found pg_wal in base lol"); } - } + }, _ => { panic!("tar::EntryType::?? {}", file_path.display()); - } + }, } } - let pg_control = pg_control.context("pg_control file not found")?; - let startpoint = Lsn(pg_control.checkPointCopy.redo); - info!("wal redo startpoint {}", startpoint); - modification.commit()?; // Set up walingest mutable state - let mut waldecoder = WalStreamDecoder::new(startpoint); - let mut segno = startpoint.segment_number(pg_constants::WAL_SEGMENT_SIZE); - let mut offset = startpoint.segment_offset(pg_constants::WAL_SEGMENT_SIZE); - let mut last_lsn = startpoint; - let mut walingest = WalIngest::new(tline, startpoint)?; + let mut waldecoder = WalStreamDecoder::new(base_lsn); + let mut segno = base_lsn.segment_number(pg_constants::WAL_SEGMENT_SIZE); + let mut offset = base_lsn.segment_offset(pg_constants::WAL_SEGMENT_SIZE); + let mut last_lsn = base_lsn; + let mut walingest = WalIngest::new(tline, base_lsn)?; // Ingest wal until end_lsn + info!("importing wal until {}", end_lsn); + let mut pg_wal_tar = tar::Archive::new(&mut reader); + let mut pg_wal_entries_iter = pg_wal_tar.entries()?; while last_lsn <= end_lsn { let bytes = { - let mut entry = entries_iter.next().expect("expected more wal")?; + let mut entry = pg_wal_entries_iter.next().expect("expected more wal")?; let header = entry.header(); let file_path = header.path().unwrap().into_owned(); @@ -501,10 +492,6 @@ pub fn import_timeline_from_tar( let mut buffer = Vec::new(); entry.read_to_end(&mut buffer).unwrap(); - if !file_path.starts_with("pg_wal") { - panic!("found non-wal file in wal section") - } - // TODO assert filename matches segno info!("processing wal file {:?}", file_path); @@ -536,15 +523,15 @@ pub fn import_timeline_from_tar( offset = 0; } - if last_lsn != startpoint { + if last_lsn != base_lsn { info!("reached end of WAL at {}", last_lsn); } else { info!("there was no WAL to import at {}", last_lsn); } // Log any extra unused files - for e in &mut entries_iter { - let mut entry = e.unwrap(); + for e in &mut pg_wal_entries_iter { + let entry = e.unwrap(); let header = entry.header(); let file_path = header.path().unwrap().into_owned(); info!("skipping {:?}", file_path); diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index bd901612f5..ada187b437 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -534,6 +534,7 @@ impl PageServerHandler { pgb: &mut PostgresBackend, tenant_id: ZTenantId, timeline_id: ZTimelineId, + base_lsn: Lsn, end_lsn: Lsn, ) -> anyhow::Result<()> { let _enter = info_span!("import", timeline = %timeline_id, tenant = %tenant_id).entered(); @@ -553,7 +554,7 @@ impl PageServerHandler { info!("importing basebackup"); pgb.write_message(&BeMessage::CopyInResponse)?; let reader = CopyInReader::new(pgb); - import_timeline_from_tar(&mut datadir_timeline, reader, end_lsn)?; + import_timeline_from_tar(&mut datadir_timeline, reader, base_lsn, end_lsn)?; info!("done"); Ok(()) @@ -838,14 +839,15 @@ impl postgres_backend::Handler for PageServerHandler { } else if query_string.starts_with("import ") { let (_, params_raw) = query_string.split_at("import ".len()); let params = params_raw.split_whitespace().collect::>(); - ensure!(params.len() == 3); + ensure!(params.len() == 4); let tenant = ZTenantId::from_str(params[0])?; let timeline = ZTimelineId::from_str(params[1])?; - let lsn = Lsn::from_str(params[2])?; + let base_lsn = Lsn::from_str(params[2])?; + let end_lsn = Lsn::from_str(params[2])?; self.check_permission(Some(tenant))?; - self.handle_import(pgb, tenant, timeline, lsn)?; + self.handle_import(pgb, tenant, timeline, base_lsn, end_lsn)?; pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?; } else if query_string.to_ascii_lowercase().starts_with("set ") { // important because psycopg2 executes "SET datestyle TO 'ISO'" diff --git a/test_runner/batch_others/test_import.py b/test_runner/batch_others/test_import.py index fca7f3b1c9..4a24da6e12 100644 --- a/test_runner/batch_others/test_import.py +++ b/test_runner/batch_others/test_import.py @@ -8,29 +8,24 @@ import json def test_import_from_vanilla(test_output_dir, pg_bin, vanilla_pg, neon_env_builder): + # Put data in vanilla pg + vanilla_pg.start() + vanilla_pg.safe_psql("create user cloud_admin with password 'postgres' superuser") + vanilla_pg.safe_psql("create table t as select generate_series(1,300000)") + assert vanilla_pg.safe_psql('select count(*) from t') == [(300000, )] - # HACK - basebackup_cache = "/home/bojan/tmp/ff/basebackup" - # basebackup_cache = None - + # Take basebackup basebackup_dir = os.path.join(test_output_dir, "basebackup") - if basebackup_cache: - basebackup_dir = basebackup_cache - else: - vanilla_pg.start() - vanilla_pg.safe_psql("create user cloud_admin with password 'postgres' superuser") - vanilla_pg.safe_psql("create table t as select generate_series(1,300000)") - assert vanilla_pg.safe_psql('select count(*) from t') == [(300000, )] - - vanilla_pg.safe_psql("CHECKPOINT") - os.mkdir(basebackup_dir) - pg_bin.run([ - "pg_basebackup", - # "-F", "tar", - "-d", vanilla_pg.connstr(), - "-D", basebackup_dir, - ]) + os.mkdir(basebackup_dir) + vanilla_pg.safe_psql("CHECKPOINT") + pg_bin.run([ + "pg_basebackup", + "-F", "tar", + "-d", vanilla_pg.connstr(), + "-D", basebackup_dir, + ]) + # Get start_lsn and end_lsn with open(os.path.join(basebackup_dir, "backup_manifest")) as f: manifest = json.load(f) start_lsn = manifest["WAL-Ranges"][0]["Start-LSN"] @@ -40,6 +35,7 @@ def test_import_from_vanilla(test_output_dir, pg_bin, vanilla_pg, neon_env_build tenant = uuid4() timeline = uuid4() + # Import to pageserver env = neon_env_builder.init_start() env.neon_cli.create_tenant(tenant) env.neon_cli.raw_cli([ @@ -48,89 +44,12 @@ def test_import_from_vanilla(test_output_dir, pg_bin, vanilla_pg, neon_env_build "--tenant-id", tenant.hex, "--timeline-id", timeline.hex, "--node-name", node_name, - # created manually with: tar -cvf basebackup.tar - "--tarfile", os.path.join(basebackup_dir, "basebackup.tar"), - "--lsn", end_lsn, - # "--tarfile", os.path.join(basebackup_dir, "base.tar"), - # "--lsn", start_lsn, + "--base-lsn", start_lsn, + "--base-tarfile", os.path.join(basebackup_dir, "base.tar"), + "--end-lsn", end_lsn, + "--wal-tarfile", os.path.join(basebackup_dir, "pg_wal.tar"), ]) + + # Check it worked pg = env.postgres.create_start(node_name, tenant_id=tenant) assert pg.safe_psql('select count(*) from t') == [(300000, )] - - -def test_import_from_neon(neon_env_builder, - port_distributor, - default_broker, - mock_s3_server, - test_output_dir, - pg_bin): - """Move a timeline to a new neon stack using pg_basebackup as interface.""" - node_name = "test_import" - source_repo_dir = Path(test_output_dir) / "source_repo" - destination_repo_dir = Path(test_output_dir) / "destination_repo" - basebackup_dir = Path(test_output_dir) / "basebackup" - basebackup_tar_path = Path(test_output_dir) / "basebackup.tar" - os.mkdir(basebackup_dir) - - # Create a repo, put some data in, take basebackup, and shut it down - with NeonEnvBuilder(source_repo_dir, port_distributor, default_broker, mock_s3_server) as builder: - - # Insert data - env = builder.init_start() - env.neon_cli.create_branch(node_name) - pg = env.postgres.create_start(node_name) - pg.safe_psql("create table t as select generate_series(1,300000)") - assert pg.safe_psql('select count(*) from t') == [(300000, )] - - # Get basebackup - lsn = pg.safe_psql('select pg_current_wal_flush_lsn()')[0][0] - tenant = pg.safe_psql("show neon.tenant_id")[0][0] - timeline = pg.safe_psql("show neon.timeline_id")[0][0] - timeline_dir = source_repo_dir / "tenants" / tenant / "timelines" / timeline - pg_bin.run(["pg_basebackup", "-d", pg.connstr(), "-D", str(basebackup_dir)]) - - # Pack basebackup into tar file (uncompressed) - with tarfile.open(basebackup_tar_path, "w") as tf: - # TODO match iteration order to what pageserver would do - tf.add(basebackup_dir) - - # Remove timeline - # env.pageserver.stop() - # shutil.rmtree(timeline_dir) - env.pageserver.http_client().timeline_detach(UUID(tenant), UUID(timeline)) - - # env.neon_cli.create_tenant(UUID(tenant)) - env.neon_cli.raw_cli([ - "timeline", - "import", - "--tenant-id", tenant, - "--timeline-id", timeline, - "--node-name", node_name, - "--tarfile", str(basebackup_tar_path), - "--lsn", lsn, - ]) - - # pg.stop_and_destroy() - # pg = env.postgres.create_start(node_name, tenant_id=UUID(tenant)) - - assert pg.safe_psql('select count(*) from t') == [(300000, )] - - - # XXX - return - - - # Create a new repo, load the basebackup into it, and check that data is there - with NeonEnvBuilder(destination_repo_dir, port_distributor, default_broker, mock_s3_server) as builder: - env = builder.init_start() - env.neon_cli.create_tenant(UUID(tenant)) - env.neon_cli.raw_cli([ - "timeline", - "import", - "--tenant-id", tenant, - "--timeline-id", timeline, - "--node-name", node_name, - "--tarfile", basebackup_tar_path, - ]) - pg = env.postgres.create_start(node_name, tenant_id=UUID(tenant)) - assert pg.safe_psql('select count(*) from t') == [(300000, )]