mirror of
https://github.com/neondatabase/neon.git
synced 2026-10-28 22:46:46 +00:00
WIP
This commit is contained in:
Bojan Serafimov
@@ -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
|
||||
|
||||
+25
-9
@@ -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)?;
|
||||
|
||||
@@ -434,65 +434,56 @@ fn import_wal<R: Repository>(
|
||||
|
||||
pub fn import_timeline_from_tar<R: Repository, Reader: Read>(
|
||||
tline: &mut DatadirTimeline<R>,
|
||||
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<ControlFileData> = 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<R: Repository, Reader: Read>(
|
||||
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<R: Repository, Reader: Read>(
|
||||
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);
|
||||
|
||||
@@ -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::<Vec<_>>();
|
||||
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'"
|
||||
|
||||
@@ -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 <all files, with pg_wal at the end>
|
||||
"--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, )]
|
||||
|
||||
Reference in New Issue
Block a user