diff --git a/Cargo.lock b/Cargo.lock index 08e6961408..99a936dd40 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1245,16 +1245,19 @@ name = "control_plane" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "camino", "clap", "comfy-table", "compute_api", + "futures", "git-version", "hex", "hyper", "nix 0.26.2", "once_cell", "pageserver_api", + "pageserver_client", "postgres", "postgres_backend", "postgres_connection", @@ -1268,6 +1271,8 @@ dependencies = [ "tar", "thiserror", "tokio", + "tokio-postgres", + "tokio-util", "toml", "tracing", "url", @@ -3172,6 +3177,19 @@ dependencies = [ "workspace_hack", ] +[[package]] +name = "pageserver_client" +version = "0.1.0" +dependencies = [ + "async-trait", + "pageserver_api", + "reqwest", + "serde", + "thiserror", + "utils", + "workspace_hack", +] + [[package]] name = "parking" version = "2.1.1" diff --git a/Cargo.toml b/Cargo.toml index 496a9d7839..b44544d626 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "control_plane", "pageserver", "pageserver/ctl", + "pageserver/client", "proxy", "safekeeper", "storage_broker", @@ -182,6 +183,7 @@ compute_api = { version = "0.1", path = "./libs/compute_api/" } consumption_metrics = { version = "0.1", path = "./libs/consumption_metrics/" } metrics = { version = "0.1", path = "./libs/metrics/" } pageserver_api = { version = "0.1", path = "./libs/pageserver_api/" } +pageserver_client = { path = "./pageserver/client" } postgres_backend = { version = "0.1", path = "./libs/postgres_backend/" } postgres_connection = { version = "0.1", path = "./libs/postgres_connection/" } postgres_ffi = { version = "0.1", path = "./libs/postgres_ffi/" } diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 7ccddc161e..898ad05add 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -6,9 +6,11 @@ license.workspace = true [dependencies] anyhow.workspace = true +async-trait.workspace = true camino.workspace = true clap.workspace = true comfy-table.workspace = true +futures.workspace = true git-version.workspace = true nix.workspace = true once_cell.workspace = true @@ -24,10 +26,11 @@ tar.workspace = true thiserror.workspace = true toml.workspace = true tokio.workspace = true +tokio-postgres.workspace = true +tokio-util.workspace = true url.workspace = true -# Note: Do not directly depend on pageserver or safekeeper; use pageserver_api or safekeeper_api -# instead, so that recompile times are better. pageserver_api.workspace = true +pageserver_client.workspace = true postgres_backend.workspace = true safekeeper_api.workspace = true postgres_connection.workspace = true diff --git a/control_plane/src/attachment_service.rs b/control_plane/src/attachment_service.rs index 822ac7d8a6..731c05809e 100644 --- a/control_plane/src/attachment_service.rs +++ b/control_plane/src/attachment_service.rs @@ -9,7 +9,7 @@ pub struct AttachmentService { env: LocalEnv, listen: String, path: PathBuf, - client: reqwest::blocking::Client, + client: reqwest::Client, } const COMMAND: &str = "attachment_service"; @@ -53,7 +53,7 @@ impl AttachmentService { env: env.clone(), path, listen, - client: reqwest::blocking::ClientBuilder::new() + client: reqwest::ClientBuilder::new() .build() .expect("Failed to construct http client"), } @@ -64,7 +64,7 @@ impl AttachmentService { .expect("non-Unicode path") } - pub fn start(&self) -> anyhow::Result { + pub async fn start(&self) -> anyhow::Result { let path_str = self.path.to_string_lossy(); background_process::start_process( @@ -73,10 +73,11 @@ impl AttachmentService { &self.env.attachment_service_bin(), ["-l", &self.listen, "-p", &path_str], [], - background_process::InitialPidFile::Create(&self.pid_file()), + background_process::InitialPidFile::Create(self.pid_file()), // TODO: a real status check - || Ok(true), + || async move { anyhow::Ok(true) }, ) + .await } pub fn stop(&self, immediate: bool) -> anyhow::Result<()> { @@ -84,7 +85,7 @@ impl AttachmentService { } /// Call into the attach_hook API, for use before handing out attachments to pageservers - pub fn attach_hook( + pub async fn attach_hook( &self, tenant_id: TenantId, pageserver_id: NodeId, @@ -104,16 +105,16 @@ impl AttachmentService { node_id: Some(pageserver_id), }; - let response = self.client.post(url).json(&request).send()?; + let response = self.client.post(url).json(&request).send().await?; if response.status() != StatusCode::OK { return Err(anyhow!("Unexpected status {}", response.status())); } - let response = response.json::()?; + let response = response.json::().await?; Ok(response.gen) } - pub fn inspect(&self, tenant_id: TenantId) -> anyhow::Result> { + pub async fn inspect(&self, tenant_id: TenantId) -> anyhow::Result> { use hyper::StatusCode; let url = self @@ -126,12 +127,12 @@ impl AttachmentService { let request = InspectRequest { tenant_id }; - let response = self.client.post(url).json(&request).send()?; + let response = self.client.post(url).json(&request).send().await?; if response.status() != StatusCode::OK { return Err(anyhow!("Unexpected status {}", response.status())); } - let response = response.json::()?; + let response = response.json::().await?; Ok(response.attachment) } } diff --git a/control_plane/src/background_process.rs b/control_plane/src/background_process.rs index 26fc08fc8f..20fa3af9b8 100644 --- a/control_plane/src/background_process.rs +++ b/control_plane/src/background_process.rs @@ -44,15 +44,15 @@ const NOTICE_AFTER_RETRIES: u64 = 50; /// Argument to `start_process`, to indicate whether it should create pidfile or if the process creates /// it itself. -pub enum InitialPidFile<'t> { +pub enum InitialPidFile { /// Create a pidfile, to allow future CLI invocations to manipulate the process. - Create(&'t Utf8Path), + Create(Utf8PathBuf), /// The process will create the pidfile itself, need to wait for that event. - Expect(&'t Utf8Path), + Expect(Utf8PathBuf), } /// Start a background child process using the parameters given. -pub fn start_process( +pub async fn start_process( process_name: &str, datadir: &Path, command: &Path, @@ -62,7 +62,8 @@ pub fn start_process( process_status_check: F, ) -> anyhow::Result where - F: Fn() -> anyhow::Result, + F: Fn() -> Fut, + Fut: std::future::Future>, AI: IntoIterator, A: AsRef, // Not generic AsRef, otherwise empty `envs` prevents type inference @@ -89,7 +90,7 @@ where let filled_cmd = fill_remote_storage_secrets_vars(fill_rust_env_vars(background_command)); filled_cmd.envs(envs); - let pid_file_to_check = match initial_pid_file { + let pid_file_to_check = match &initial_pid_file { InitialPidFile::Create(path) => { pre_exec_create_pidfile(filled_cmd, path); path @@ -107,7 +108,7 @@ where ); for retries in 0..RETRIES { - match process_started(pid, Some(pid_file_to_check), &process_status_check) { + match process_started(pid, pid_file_to_check, &process_status_check).await { Ok(true) => { println!("\n{process_name} started, pid: {pid}"); return Ok(spawned_process); @@ -316,22 +317,20 @@ where cmd } -fn process_started( +async fn process_started( pid: Pid, - pid_file_to_check: Option<&Utf8Path>, + pid_file_to_check: &Utf8Path, status_check: &F, ) -> anyhow::Result where - F: Fn() -> anyhow::Result, + F: Fn() -> Fut, + Fut: std::future::Future>, { - match status_check() { - Ok(true) => match pid_file_to_check { - Some(pid_file_path) => match pid_file::read(pid_file_path)? { - PidFileRead::NotExist => Ok(false), - PidFileRead::LockedByOtherProcess(pid_in_file) => Ok(pid_in_file == pid), - PidFileRead::NotHeldByAnyProcess(_) => Ok(false), - }, - None => Ok(true), + match status_check().await { + Ok(true) => match pid_file::read(pid_file_to_check)? { + PidFileRead::NotExist => Ok(false), + PidFileRead::LockedByOtherProcess(pid_in_file) => Ok(pid_in_file == pid), + PidFileRead::NotHeldByAnyProcess(_) => Ok(false), }, Ok(false) => Ok(false), Err(e) => anyhow::bail!("process failed to start: {e}"), diff --git a/control_plane/src/bin/neon_local.rs b/control_plane/src/bin/neon_local.rs index 6f0b929ac6..03e69010f7 100644 --- a/control_plane/src/bin/neon_local.rs +++ b/control_plane/src/bin/neon_local.rs @@ -120,15 +120,20 @@ fn main() -> Result<()> { let mut env = LocalEnv::load_config().context("Error loading config")?; let original_env = env.clone(); + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let subcommand_result = match sub_name { - "tenant" => handle_tenant(sub_args, &mut env), - "timeline" => handle_timeline(sub_args, &mut env), - "start" => handle_start_all(sub_args, &env), + "tenant" => rt.block_on(handle_tenant(sub_args, &mut env)), + "timeline" => rt.block_on(handle_timeline(sub_args, &mut env)), + "start" => rt.block_on(handle_start_all(sub_args, &env)), "stop" => handle_stop_all(sub_args, &env), - "pageserver" => handle_pageserver(sub_args, &env), - "attachment_service" => handle_attachment_service(sub_args, &env), - "safekeeper" => handle_safekeeper(sub_args, &env), - "endpoint" => handle_endpoint(sub_args, &env), + "pageserver" => rt.block_on(handle_pageserver(sub_args, &env)), + "attachment_service" => rt.block_on(handle_attachment_service(sub_args, &env)), + "safekeeper" => rt.block_on(handle_safekeeper(sub_args, &env)), + "endpoint" => rt.block_on(handle_endpoint(sub_args, &env)), "mappings" => handle_mappings(sub_args, &mut env), "pg" => bail!("'pg' subcommand has been renamed to 'endpoint'"), _ => bail!("unexpected subcommand {sub_name}"), @@ -269,12 +274,13 @@ fn print_timeline( /// Returns a map of timeline IDs to timeline_id@lsn strings. /// Connects to the pageserver to query this information. -fn get_timeline_infos( +async fn get_timeline_infos( env: &local_env::LocalEnv, tenant_id: &TenantId, ) -> Result> { Ok(get_default_pageserver(env) - .timeline_list(tenant_id)? + .timeline_list(tenant_id) + .await? .into_iter() .map(|timeline_info| (timeline_info.timeline_id, timeline_info)) .collect()) @@ -373,11 +379,14 @@ fn pageserver_config_overrides(init_match: &ArgMatches) -> Vec<&str> { .collect() } -fn handle_tenant(tenant_match: &ArgMatches, env: &mut local_env::LocalEnv) -> anyhow::Result<()> { +async fn handle_tenant( + tenant_match: &ArgMatches, + env: &mut local_env::LocalEnv, +) -> anyhow::Result<()> { let pageserver = get_default_pageserver(env); match tenant_match.subcommand() { Some(("list", _)) => { - for t in pageserver.tenant_list()? { + for t in pageserver.tenant_list().await? { println!("{} {:?}", t.id, t.state); } } @@ -394,12 +403,16 @@ fn handle_tenant(tenant_match: &ArgMatches, env: &mut local_env::LocalEnv) -> an // We must register the tenant with the attachment service, so // that when the pageserver restarts, it will be re-attached. let attachment_service = AttachmentService::from_env(env); - attachment_service.attach_hook(tenant_id, pageserver.conf.id)? + attachment_service + .attach_hook(tenant_id, pageserver.conf.id) + .await? } else { None }; - pageserver.tenant_create(tenant_id, generation, tenant_conf)?; + pageserver + .tenant_create(tenant_id, generation, tenant_conf) + .await?; println!("tenant {tenant_id} successfully created on the pageserver"); // Create an initial timeline for the new tenant @@ -409,14 +422,16 @@ fn handle_tenant(tenant_match: &ArgMatches, env: &mut local_env::LocalEnv) -> an .copied() .context("Failed to parse postgres version from the argument string")?; - let timeline_info = pageserver.timeline_create( - tenant_id, - new_timeline_id, - None, - None, - Some(pg_version), - None, - )?; + let timeline_info = pageserver + .timeline_create( + tenant_id, + new_timeline_id, + None, + None, + Some(pg_version), + None, + ) + .await?; let new_timeline_id = timeline_info.timeline_id; let last_record_lsn = timeline_info.last_record_lsn; @@ -450,6 +465,7 @@ fn handle_tenant(tenant_match: &ArgMatches, env: &mut local_env::LocalEnv) -> an pageserver .tenant_config(tenant_id, tenant_conf) + .await .with_context(|| format!("Tenant config failed for tenant with id {tenant_id}"))?; println!("tenant {tenant_id} successfully configured on the pageserver"); } @@ -458,7 +474,7 @@ fn handle_tenant(tenant_match: &ArgMatches, env: &mut local_env::LocalEnv) -> an let new_pageserver = get_pageserver(env, matches)?; let new_pageserver_id = new_pageserver.conf.id; - migrate_tenant(env, tenant_id, new_pageserver)?; + migrate_tenant(env, tenant_id, new_pageserver).await?; println!("tenant {tenant_id} migrated to {}", new_pageserver_id); } @@ -468,13 +484,13 @@ fn handle_tenant(tenant_match: &ArgMatches, env: &mut local_env::LocalEnv) -> an Ok(()) } -fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) -> Result<()> { +async fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) -> Result<()> { let pageserver = get_default_pageserver(env); match timeline_match.subcommand() { Some(("list", list_match)) => { let tenant_id = get_tenant_id(list_match, env)?; - let timelines = pageserver.timeline_list(&tenant_id)?; + let timelines = pageserver.timeline_list(&tenant_id).await?; print_timelines_tree(timelines, env.timeline_name_mappings())?; } Some(("create", create_match)) => { @@ -490,14 +506,16 @@ fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) - let new_timeline_id_opt = parse_timeline_id(create_match)?; - let timeline_info = pageserver.timeline_create( - tenant_id, - new_timeline_id_opt, - None, - None, - Some(pg_version), - None, - )?; + let timeline_info = pageserver + .timeline_create( + tenant_id, + new_timeline_id_opt, + None, + None, + Some(pg_version), + None, + ) + .await?; let new_timeline_id = timeline_info.timeline_id; let last_record_lsn = timeline_info.last_record_lsn; @@ -542,7 +560,9 @@ fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) - let mut cplane = ComputeControlPlane::load(env.clone())?; println!("Importing timeline into pageserver ..."); - pageserver.timeline_import(tenant_id, timeline_id, base, pg_wal, pg_version)?; + pageserver + .timeline_import(tenant_id, timeline_id, base, pg_wal, pg_version) + .await?; env.register_branch_mapping(name.to_string(), tenant_id, timeline_id)?; println!("Creating endpoint for imported timeline ..."); @@ -578,14 +598,16 @@ fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) - .map(|lsn_str| Lsn::from_str(lsn_str)) .transpose() .context("Failed to parse ancestor start Lsn from the request")?; - let timeline_info = pageserver.timeline_create( - tenant_id, - None, - start_lsn, - Some(ancestor_timeline_id), - None, - None, - )?; + let timeline_info = pageserver + .timeline_create( + tenant_id, + None, + start_lsn, + Some(ancestor_timeline_id), + None, + None, + ) + .await?; let new_timeline_id = timeline_info.timeline_id; let last_record_lsn = timeline_info.last_record_lsn; @@ -604,7 +626,7 @@ fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) - Ok(()) } -fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> { +async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> { let (sub_name, sub_args) = match ep_match.subcommand() { Some(ep_subcommand_data) => ep_subcommand_data, None => bail!("no endpoint subcommand provided"), @@ -614,10 +636,12 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<( match sub_name { "list" => { let tenant_id = get_tenant_id(sub_args, env)?; - let timeline_infos = get_timeline_infos(env, &tenant_id).unwrap_or_else(|e| { - eprintln!("Failed to load timeline info: {}", e); - HashMap::new() - }); + let timeline_infos = get_timeline_infos(env, &tenant_id) + .await + .unwrap_or_else(|e| { + eprintln!("Failed to load timeline info: {}", e); + HashMap::new() + }); let timeline_name_mappings = env.timeline_name_mappings(); @@ -791,7 +815,9 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<( }; println!("Starting existing endpoint {endpoint_id}..."); - endpoint.start(&auth_token, safekeepers, remote_ext_config)?; + endpoint + .start(&auth_token, safekeepers, remote_ext_config) + .await?; } "reconfigure" => { let endpoint_id = sub_args @@ -809,7 +835,7 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<( } else { None }; - endpoint.reconfigure(pageserver_id)?; + endpoint.reconfigure(pageserver_id).await?; } "stop" => { let endpoint_id = sub_args @@ -875,11 +901,12 @@ fn get_pageserver(env: &local_env::LocalEnv, args: &ArgMatches) -> Result Result<()> { +async fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> { match sub_match.subcommand() { Some(("start", subcommand_args)) => { if let Err(e) = get_pageserver(env, subcommand_args)? .start(&pageserver_config_overrides(subcommand_args)) + .await { eprintln!("pageserver start failed: {e}"); exit(1); @@ -906,7 +933,10 @@ fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul exit(1); } - if let Err(e) = pageserver.start(&pageserver_config_overrides(subcommand_args)) { + if let Err(e) = pageserver + .start(&pageserver_config_overrides(subcommand_args)) + .await + { eprintln!("pageserver start failed: {e}"); exit(1); } @@ -920,14 +950,17 @@ fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul exit(1); } - if let Err(e) = pageserver.start(&pageserver_config_overrides(subcommand_args)) { + if let Err(e) = pageserver + .start(&pageserver_config_overrides(subcommand_args)) + .await + { eprintln!("pageserver start failed: {e}"); exit(1); } } Some(("status", subcommand_args)) => { - match get_pageserver(env, subcommand_args)?.check_status() { + match get_pageserver(env, subcommand_args)?.check_status().await { Ok(_) => println!("Page server is up and running"), Err(err) => { eprintln!("Page server is not available: {}", err); @@ -942,11 +975,14 @@ fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul Ok(()) } -fn handle_attachment_service(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> { +async fn handle_attachment_service( + sub_match: &ArgMatches, + env: &local_env::LocalEnv, +) -> Result<()> { let svc = AttachmentService::from_env(env); match sub_match.subcommand() { Some(("start", _start_match)) => { - if let Err(e) = svc.start() { + if let Err(e) = svc.start().await { eprintln!("start failed: {e}"); exit(1); } @@ -987,7 +1023,7 @@ fn safekeeper_extra_opts(init_match: &ArgMatches) -> Vec { .collect() } -fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> { +async fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> { let (sub_name, sub_args) = match sub_match.subcommand() { Some(safekeeper_command_data) => safekeeper_command_data, None => bail!("no safekeeper subcommand provided"), @@ -1005,7 +1041,7 @@ fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul "start" => { let extra_opts = safekeeper_extra_opts(sub_args); - if let Err(e) = safekeeper.start(extra_opts) { + if let Err(e) = safekeeper.start(extra_opts).await { eprintln!("safekeeper start failed: {}", e); exit(1); } @@ -1031,7 +1067,7 @@ fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul } let extra_opts = safekeeper_extra_opts(sub_args); - if let Err(e) = safekeeper.start(extra_opts) { + if let Err(e) = safekeeper.start(extra_opts).await { eprintln!("safekeeper start failed: {}", e); exit(1); } @@ -1044,15 +1080,15 @@ fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul Ok(()) } -fn handle_start_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow::Result<()> { +async fn handle_start_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow::Result<()> { // Endpoints are not started automatically - broker::start_broker_process(env)?; + broker::start_broker_process(env).await?; // Only start the attachment service if the pageserver is configured to need it if env.control_plane_api.is_some() { let attachment_service = AttachmentService::from_env(env); - if let Err(e) = attachment_service.start() { + if let Err(e) = attachment_service.start().await { eprintln!("attachment_service start failed: {:#}", e); try_stop_all(env, true); exit(1); @@ -1061,7 +1097,10 @@ fn handle_start_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow for ps_conf in &env.pageservers { let pageserver = PageServerNode::from_env(env, ps_conf); - if let Err(e) = pageserver.start(&pageserver_config_overrides(sub_match)) { + if let Err(e) = pageserver + .start(&pageserver_config_overrides(sub_match)) + .await + { eprintln!("pageserver {} start failed: {:#}", ps_conf.id, e); try_stop_all(env, true); exit(1); @@ -1070,7 +1109,7 @@ fn handle_start_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow for node in env.safekeepers.iter() { let safekeeper = SafekeeperNode::from_env(env, node); - if let Err(e) = safekeeper.start(vec![]) { + if let Err(e) = safekeeper.start(vec![]).await { eprintln!("safekeeper {} start failed: {:#}", safekeeper.id, e); try_stop_all(env, false); exit(1); diff --git a/control_plane/src/broker.rs b/control_plane/src/broker.rs index 6be865cc2e..f40705863b 100644 --- a/control_plane/src/broker.rs +++ b/control_plane/src/broker.rs @@ -11,7 +11,7 @@ use camino::Utf8PathBuf; use crate::{background_process, local_env}; -pub fn start_broker_process(env: &local_env::LocalEnv) -> anyhow::Result<()> { +pub async fn start_broker_process(env: &local_env::LocalEnv) -> anyhow::Result<()> { let broker = &env.broker; let listen_addr = &broker.listen_addr; @@ -19,15 +19,15 @@ pub fn start_broker_process(env: &local_env::LocalEnv) -> anyhow::Result<()> { let args = [format!("--listen-addr={listen_addr}")]; - let client = reqwest::blocking::Client::new(); + let client = reqwest::Client::new(); background_process::start_process( "storage_broker", &env.base_data_dir, &env.storage_broker_bin(), args, [], - background_process::InitialPidFile::Create(&storage_broker_pid_file_path(env)), - || { + background_process::InitialPidFile::Create(storage_broker_pid_file_path(env)), + || async { let url = broker.client_url(); let status_url = url.join("status").with_context(|| { format!("Failed to append /status path to broker endpoint {url}") @@ -36,12 +36,13 @@ pub fn start_broker_process(env: &local_env::LocalEnv) -> anyhow::Result<()> { .get(status_url) .build() .with_context(|| format!("Failed to construct request to broker endpoint {url}"))?; - match client.execute(request) { + match client.execute(request).await { Ok(resp) => Ok(resp.status().is_success()), Err(_) => Ok(false), } }, ) + .await .context("Failed to spawn storage_broker subprocess")?; Ok(()) } diff --git a/control_plane/src/endpoint.rs b/control_plane/src/endpoint.rs index a566f03db9..071f22dc2b 100644 --- a/control_plane/src/endpoint.rs +++ b/control_plane/src/endpoint.rs @@ -464,7 +464,7 @@ impl Endpoint { } } - pub fn start( + pub async fn start( &self, auth_token: &Option, safekeepers: Vec, @@ -587,7 +587,7 @@ impl Endpoint { const MAX_ATTEMPTS: u32 = 10 * 30; // Wait up to 30 s loop { attempt += 1; - match self.get_status() { + match self.get_status().await { Ok(state) => { match state.status { ComputeStatus::Init => { @@ -629,8 +629,8 @@ impl Endpoint { } // Call the /status HTTP API - pub fn get_status(&self) -> Result { - let client = reqwest::blocking::Client::new(); + pub async fn get_status(&self) -> Result { + let client = reqwest::Client::new(); let response = client .request( @@ -641,16 +641,17 @@ impl Endpoint { self.http_address.port() ), ) - .send()?; + .send() + .await?; // Interpret the response let status = response.status(); if !(status.is_client_error() || status.is_server_error()) { - Ok(response.json()?) + Ok(response.json().await?) } else { // reqwest does not export its error construction utility functions, so let's craft the message ourselves let url = response.url().to_owned(); - let msg = match response.text() { + let msg = match response.text().await { Ok(err_body) => format!("Error: {}", err_body), Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url), }; @@ -658,7 +659,7 @@ impl Endpoint { } } - pub fn reconfigure(&self, pageserver_id: Option) -> Result<()> { + pub async fn reconfigure(&self, pageserver_id: Option) -> Result<()> { let mut spec: ComputeSpec = { let spec_path = self.endpoint_path().join("spec.json"); let file = std::fs::File::open(spec_path)?; @@ -687,7 +688,7 @@ impl Endpoint { spec.pageserver_connstring = Some(format!("postgresql://no_user@{host}:{port}")); } - let client = reqwest::blocking::Client::new(); + let client = reqwest::Client::new(); let response = client .post(format!( "http://{}:{}/configure", @@ -698,14 +699,15 @@ impl Endpoint { "{{\"spec\":{}}}", serde_json::to_string_pretty(&spec)? )) - .send()?; + .send() + .await?; let status = response.status(); if !(status.is_client_error() || status.is_server_error()) { Ok(()) } else { let url = response.url().to_owned(); - let msg = match response.text() { + let msg = match response.text().await { Ok(err_body) => format!("Error: {}", err_body), Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url), }; diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs index 981ddd3e98..7d490016bf 100644 --- a/control_plane/src/pageserver.rs +++ b/control_plane/src/pageserver.rs @@ -6,28 +6,24 @@ //! use std::borrow::Cow; use std::collections::HashMap; -use std::fs::File; -use std::io::{BufReader, Write}; + +use std::io; +use std::io::Write; use std::num::NonZeroU64; use std::path::PathBuf; use std::process::{Child, Command}; use std::time::Duration; -use std::{io, result}; use anyhow::{bail, Context}; use camino::Utf8PathBuf; -use pageserver_api::models::{ - self, LocationConfig, TenantInfo, TenantLocationConfigRequest, TimelineInfo, -}; +use futures::SinkExt; +use pageserver_api::models::{self, LocationConfig, TenantInfo, TimelineInfo}; use pageserver_api::shard::TenantShardId; +use pageserver_client::mgmt_api; use postgres_backend::AuthType; use postgres_connection::{parse_host_port, PgConnectionConfig}; -use reqwest::blocking::{Client, RequestBuilder, Response}; -use reqwest::{IntoUrl, Method}; -use thiserror::Error; use utils::auth::{Claims, Scope}; use utils::{ - http::error::HttpErrorBody, id::{TenantId, TimelineId}, lsn::Lsn, }; @@ -38,45 +34,6 @@ use crate::{background_process, local_env::LocalEnv}; /// Directory within .neon which will be used by default for LocalFs remote storage. pub const PAGESERVER_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/pageserver"; -#[derive(Error, Debug)] -pub enum PageserverHttpError { - #[error("Reqwest error: {0}")] - Transport(#[from] reqwest::Error), - - #[error("Error: {0}")] - Response(String), -} - -impl From for PageserverHttpError { - fn from(e: anyhow::Error) -> Self { - Self::Response(e.to_string()) - } -} - -type Result = result::Result; - -pub trait ResponseErrorMessageExt: Sized { - fn error_from_body(self) -> Result; -} - -impl ResponseErrorMessageExt for Response { - fn error_from_body(self) -> Result { - let status = self.status(); - if !(status.is_client_error() || status.is_server_error()) { - return Ok(self); - } - - // reqwest does not export its error construction utility functions, so let's craft the message ourselves - let url = self.url().to_owned(); - Err(PageserverHttpError::Response( - match self.json::() { - Ok(err_body) => format!("Error: {}", err_body.msg), - Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url), - }, - )) - } -} - // // Control routines for pageserver. // @@ -87,8 +44,7 @@ pub struct PageServerNode { pub pg_connection_config: PgConnectionConfig, pub conf: PageServerConf, pub env: LocalEnv, - pub http_client: Client, - pub http_base_url: String, + pub http_client: mgmt_api::Client, } impl PageServerNode { @@ -100,8 +56,19 @@ impl PageServerNode { pg_connection_config: PgConnectionConfig::new_host_port(host, port), conf: conf.clone(), env: env.clone(), - http_client: Client::new(), - http_base_url: format!("http://{}/v1", conf.listen_http_addr), + http_client: mgmt_api::Client::new( + format!("http://{}", conf.listen_http_addr), + { + match conf.http_auth_type { + AuthType::Trust => None, + AuthType::NeonJWT => Some( + env.generate_auth_token(&Claims::new(None, Scope::PageServerApi)) + .unwrap(), + ), + } + } + .as_deref(), + ), } } @@ -182,8 +149,8 @@ impl PageServerNode { .expect("non-Unicode path") } - pub fn start(&self, config_overrides: &[&str]) -> anyhow::Result { - self.start_node(config_overrides, false) + pub async fn start(&self, config_overrides: &[&str]) -> anyhow::Result { + self.start_node(config_overrides, false).await } fn pageserver_init(&self, config_overrides: &[&str]) -> anyhow::Result<()> { @@ -224,7 +191,12 @@ impl PageServerNode { Ok(()) } - fn start_node(&self, config_overrides: &[&str], update_config: bool) -> anyhow::Result { + async fn start_node( + &self, + config_overrides: &[&str], + update_config: bool, + ) -> anyhow::Result { + // TODO: using a thread here because start_process() is not async but we need to call check_status() let datadir = self.repo_path(); print!( "Starting pageserver node {} at '{}' in {:?}", @@ -232,7 +204,7 @@ impl PageServerNode { self.pg_connection_config.raw_address(), datadir ); - io::stdout().flush()?; + io::stdout().flush().context("flush stdout")?; let datadir_path_str = datadir.to_str().with_context(|| { format!( @@ -244,20 +216,23 @@ impl PageServerNode { if update_config { args.push(Cow::Borrowed("--update-config")); } - background_process::start_process( "pageserver", &datadir, &self.env.pageserver_bin(), args.iter().map(Cow::as_ref), self.pageserver_env_variables()?, - background_process::InitialPidFile::Expect(&self.pid_file()), - || match self.check_status() { - Ok(()) => Ok(true), - Err(PageserverHttpError::Transport(_)) => Ok(false), - Err(e) => Err(anyhow::anyhow!("Failed to check node status: {e}")), + background_process::InitialPidFile::Expect(self.pid_file()), + || async { + let st = self.check_status().await; + match st { + Ok(()) => Ok(true), + Err(mgmt_api::Error::ReceiveBody(_)) => Ok(false), + Err(e) => Err(anyhow::anyhow!("Failed to check node status: {e}")), + } }, ) + .await } fn pageserver_basic_args<'a>( @@ -303,7 +278,12 @@ impl PageServerNode { background_process::stop_process(immediate, "pageserver", &self.pid_file()) } - pub fn page_server_psql_client(&self) -> anyhow::Result { + pub async fn page_server_psql_client( + &self, + ) -> anyhow::Result<( + tokio_postgres::Client, + tokio_postgres::Connection, + )> { let mut config = self.pg_connection_config.clone(); if self.conf.pg_auth_type == AuthType::NeonJWT { let token = self @@ -311,36 +291,18 @@ impl PageServerNode { .generate_auth_token(&Claims::new(None, Scope::PageServerApi))?; config = config.set_password(Some(token)); } - Ok(config.connect_no_tls()?) + Ok(config.connect_no_tls().await?) } - fn http_request(&self, method: Method, url: U) -> anyhow::Result { - let mut builder = self.http_client.request(method, url); - if self.conf.http_auth_type == AuthType::NeonJWT { - let token = self - .env - .generate_auth_token(&Claims::new(None, Scope::PageServerApi))?; - builder = builder.bearer_auth(token) - } - Ok(builder) + pub async fn check_status(&self) -> mgmt_api::Result<()> { + self.http_client.status().await } - pub fn check_status(&self) -> Result<()> { - self.http_request(Method::GET, format!("{}/status", self.http_base_url))? - .send()? - .error_from_body()?; - Ok(()) + pub async fn tenant_list(&self) -> mgmt_api::Result> { + self.http_client.list_tenants().await } - pub fn tenant_list(&self) -> Result> { - Ok(self - .http_request(Method::GET, format!("{}/tenant", self.http_base_url))? - .send()? - .error_from_body()? - .json()?) - } - - pub fn tenant_create( + pub async fn tenant_create( &self, new_tenant_id: TenantId, generation: Option, @@ -418,23 +380,10 @@ impl PageServerNode { if !settings.is_empty() { bail!("Unrecognized tenant settings: {settings:?}") } - self.http_request(Method::POST, format!("{}/tenant", self.http_base_url))? - .json(&request) - .send()? - .error_from_body()? - .json::>() - .with_context(|| { - format!("Failed to parse tenant creation response for tenant id: {new_tenant_id:?}") - })? - .context("No tenant id was found in the tenant creation response") - .and_then(|tenant_id_string| { - tenant_id_string.parse().with_context(|| { - format!("Failed to parse response string as tenant id: '{tenant_id_string}'") - }) - }) + Ok(self.http_client.tenant_create(&request).await?) } - pub fn tenant_config( + pub async fn tenant_config( &self, tenant_id: TenantId, mut settings: HashMap<&str, &str>, @@ -513,54 +462,30 @@ impl PageServerNode { bail!("Unrecognized tenant settings: {settings:?}") } - self.http_request(Method::PUT, format!("{}/tenant/config", self.http_base_url))? - .json(&models::TenantConfigRequest { tenant_id, config }) - .send()? - .error_from_body()?; + self.http_client + .tenant_config(&models::TenantConfigRequest { tenant_id, config }) + .await?; Ok(()) } - pub fn location_config( + pub async fn location_config( &self, tenant_id: TenantId, config: LocationConfig, flush_ms: Option, ) -> anyhow::Result<()> { - let req_body = TenantLocationConfigRequest { tenant_id, config }; - - let path = format!( - "{}/tenant/{}/location_config", - self.http_base_url, tenant_id - ); - let path = if let Some(flush_ms) = flush_ms { - format!("{}?flush_ms={}", path, flush_ms.as_millis()) - } else { - path - }; - - self.http_request(Method::PUT, path)? - .json(&req_body) - .send()? - .error_from_body()?; - - Ok(()) + Ok(self + .http_client + .location_config(tenant_id, config, flush_ms) + .await?) } - pub fn timeline_list(&self, tenant_id: &TenantId) -> anyhow::Result> { - let timeline_infos: Vec = self - .http_request( - Method::GET, - format!("{}/tenant/{}/timeline", self.http_base_url, tenant_id), - )? - .send()? - .error_from_body()? - .json()?; - - Ok(timeline_infos) + pub async fn timeline_list(&self, tenant_id: &TenantId) -> anyhow::Result> { + Ok(self.http_client.list_timelines(*tenant_id).await?) } - pub fn timeline_create( + pub async fn timeline_create( &self, tenant_id: TenantId, new_timeline_id: Option, @@ -571,29 +496,14 @@ impl PageServerNode { ) -> anyhow::Result { // If timeline ID was not specified, generate one let new_timeline_id = new_timeline_id.unwrap_or(TimelineId::generate()); - - self.http_request( - Method::POST, - format!("{}/tenant/{}/timeline", self.http_base_url, tenant_id), - )? - .json(&models::TimelineCreateRequest { + let req = models::TimelineCreateRequest { new_timeline_id, ancestor_start_lsn, ancestor_timeline_id, pg_version, existing_initdb_timeline_id, - }) - .send()? - .error_from_body()? - .json::>() - .with_context(|| { - format!("Failed to parse timeline creation response for tenant id: {tenant_id}") - })? - .with_context(|| { - format!( - "No timeline id was found in the timeline creation response for tenant {tenant_id}" - ) - }) + }; + Ok(self.http_client.timeline_create(tenant_id, &req).await?) } /// Import a basebackup prepared using either: @@ -605,7 +515,7 @@ impl PageServerNode { /// * `timeline_id` - id to assign to imported timeline /// * `base` - (start lsn of basebackup, path to `base.tar` file) /// * `pg_wal` - if there's any wal to import: (end lsn, path to `pg_wal.tar`) - pub fn timeline_import( + pub async fn timeline_import( &self, tenant_id: TenantId, timeline_id: TimelineId, @@ -613,36 +523,60 @@ impl PageServerNode { pg_wal: Option<(Lsn, PathBuf)>, pg_version: u32, ) -> anyhow::Result<()> { - let mut client = self.page_server_psql_client()?; + let (client, conn) = self.page_server_psql_client().await?; + // The connection object performs the actual communication with the database, + // so spawn it off to run on its own. + tokio::spawn(async move { + if let Err(e) = conn.await { + eprintln!("connection error: {}", e); + } + }); + tokio::pin!(client); // Init base reader let (start_lsn, base_tarfile_path) = base; - let base_tarfile = File::open(base_tarfile_path)?; - let mut base_reader = BufReader::new(base_tarfile); + let base_tarfile = tokio::fs::File::open(base_tarfile_path).await?; + let base_tarfile = tokio_util::io::ReaderStream::new(base_tarfile); // Init wal reader if necessary let (end_lsn, wal_reader) = if let Some((end_lsn, wal_tarfile_path)) = pg_wal { - let wal_tarfile = File::open(wal_tarfile_path)?; - let wal_reader = BufReader::new(wal_tarfile); + let wal_tarfile = tokio::fs::File::open(wal_tarfile_path).await?; + let wal_reader = tokio_util::io::ReaderStream::new(wal_tarfile); (end_lsn, Some(wal_reader)) } else { (start_lsn, None) }; - // Import base - let import_cmd = format!( - "import basebackup {tenant_id} {timeline_id} {start_lsn} {end_lsn} {pg_version}" - ); - let mut writer = client.copy_in(&import_cmd)?; - io::copy(&mut base_reader, &mut writer)?; - writer.finish()?; + let copy_in = |reader, cmd| { + let client = &client; + async move { + let writer = client.copy_in(&cmd).await?; + let writer = std::pin::pin!(writer); + let mut writer = writer.sink_map_err(|e| { + std::io::Error::new(std::io::ErrorKind::Other, format!("{e}")) + }); + let mut reader = std::pin::pin!(reader); + writer.send_all(&mut reader).await?; + writer.into_inner().finish().await?; + anyhow::Ok(()) + } + }; + // Import base + copy_in( + base_tarfile, + format!( + "import basebackup {tenant_id} {timeline_id} {start_lsn} {end_lsn} {pg_version}" + ), + ) + .await?; // Import wal if necessary - if let Some(mut wal_reader) = wal_reader { - let import_cmd = format!("import wal {tenant_id} {timeline_id} {start_lsn} {end_lsn}"); - let mut writer = client.copy_in(&import_cmd)?; - io::copy(&mut wal_reader, &mut writer)?; - writer.finish()?; + if let Some(wal_reader) = wal_reader { + copy_in( + wal_reader, + format!("import wal {tenant_id} {timeline_id} {start_lsn} {end_lsn}"), + ) + .await?; } Ok(()) diff --git a/control_plane/src/safekeeper.rs b/control_plane/src/safekeeper.rs index a8baa0ac53..4026ef0eb9 100644 --- a/control_plane/src/safekeeper.rs +++ b/control_plane/src/safekeeper.rs @@ -13,7 +13,6 @@ use std::{io, result}; use anyhow::Context; use camino::Utf8PathBuf; use postgres_connection::PgConnectionConfig; -use reqwest::blocking::{Client, RequestBuilder, Response}; use reqwest::{IntoUrl, Method}; use thiserror::Error; use utils::{http::error::HttpErrorBody, id::NodeId}; @@ -34,12 +33,14 @@ pub enum SafekeeperHttpError { type Result = result::Result; +#[async_trait::async_trait] pub trait ResponseErrorMessageExt: Sized { - fn error_from_body(self) -> Result; + async fn error_from_body(self) -> Result; } -impl ResponseErrorMessageExt for Response { - fn error_from_body(self) -> Result { +#[async_trait::async_trait] +impl ResponseErrorMessageExt for reqwest::Response { + async fn error_from_body(self) -> Result { let status = self.status(); if !(status.is_client_error() || status.is_server_error()) { return Ok(self); @@ -48,7 +49,7 @@ impl ResponseErrorMessageExt for Response { // reqwest does not export its error construction utility functions, so let's craft the message ourselves let url = self.url().to_owned(); Err(SafekeeperHttpError::Response( - match self.json::() { + match self.json::().await { Ok(err_body) => format!("Error: {}", err_body.msg), Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url), }, @@ -69,7 +70,7 @@ pub struct SafekeeperNode { pub pg_connection_config: PgConnectionConfig, pub env: LocalEnv, - pub http_client: Client, + pub http_client: reqwest::Client, pub http_base_url: String, } @@ -80,7 +81,7 @@ impl SafekeeperNode { conf: conf.clone(), pg_connection_config: Self::safekeeper_connection_config(conf.pg_port), env: env.clone(), - http_client: Client::new(), + http_client: reqwest::Client::new(), http_base_url: format!("http://127.0.0.1:{}/v1", conf.http_port), } } @@ -103,7 +104,7 @@ impl SafekeeperNode { .expect("non-Unicode path") } - pub fn start(&self, extra_opts: Vec) -> anyhow::Result { + pub async fn start(&self, extra_opts: Vec) -> anyhow::Result { print!( "Starting safekeeper at '{}' in '{}'", self.pg_connection_config.raw_address(), @@ -191,13 +192,16 @@ impl SafekeeperNode { &self.env.safekeeper_bin(), &args, [], - background_process::InitialPidFile::Expect(&self.pid_file()), - || match self.check_status() { - Ok(()) => Ok(true), - Err(SafekeeperHttpError::Transport(_)) => Ok(false), - Err(e) => Err(anyhow::anyhow!("Failed to check node status: {e}")), + background_process::InitialPidFile::Expect(self.pid_file()), + || async { + match self.check_status().await { + Ok(()) => Ok(true), + Err(SafekeeperHttpError::Transport(_)) => Ok(false), + Err(e) => Err(anyhow::anyhow!("Failed to check node status: {e}")), + } }, ) + .await } /// @@ -216,7 +220,7 @@ impl SafekeeperNode { ) } - fn http_request(&self, method: Method, url: U) -> RequestBuilder { + fn http_request(&self, method: Method, url: U) -> reqwest::RequestBuilder { // TODO: authentication //if self.env.auth_type == AuthType::NeonJWT { // builder = builder.bearer_auth(&self.env.safekeeper_auth_token) @@ -224,10 +228,12 @@ impl SafekeeperNode { self.http_client.request(method, url) } - pub fn check_status(&self) -> Result<()> { + pub async fn check_status(&self) -> Result<()> { self.http_request(Method::GET, format!("{}/{}", self.http_base_url, "status")) - .send()? - .error_from_body()?; + .send() + .await? + .error_from_body() + .await?; Ok(()) } } diff --git a/control_plane/src/tenant_migration.rs b/control_plane/src/tenant_migration.rs index fbb0358158..79df108896 100644 --- a/control_plane/src/tenant_migration.rs +++ b/control_plane/src/tenant_migration.rs @@ -19,11 +19,11 @@ use utils::{ }; /// Given an attached pageserver, retrieve the LSN for all timelines -fn get_lsns( +async fn get_lsns( tenant_id: TenantId, pageserver: &PageServerNode, ) -> anyhow::Result> { - let timelines = pageserver.timeline_list(&tenant_id)?; + let timelines = pageserver.timeline_list(&tenant_id).await?; Ok(timelines .into_iter() .map(|t| (t.timeline_id, t.last_record_lsn)) @@ -32,13 +32,13 @@ fn get_lsns( /// Wait for the timeline LSNs on `pageserver` to catch up with or overtake /// `baseline`. -fn await_lsn( +async fn await_lsn( tenant_id: TenantId, pageserver: &PageServerNode, baseline: HashMap, ) -> anyhow::Result<()> { loop { - let latest = match get_lsns(tenant_id, pageserver) { + let latest = match get_lsns(tenant_id, pageserver).await { Ok(l) => l, Err(e) => { println!( @@ -84,7 +84,7 @@ fn await_lsn( /// - Coordinate attach/secondary/detach on pageservers /// - call into attachment_service for generations /// - reconfigure compute endpoints to point to new attached pageserver -pub fn migrate_tenant( +pub async fn migrate_tenant( env: &LocalEnv, tenant_id: TenantId, dest_ps: PageServerNode, @@ -108,16 +108,18 @@ pub fn migrate_tenant( } } - let previous = attachment_service.inspect(tenant_id)?; + let previous = attachment_service.inspect(tenant_id).await?; let mut baseline_lsns = None; if let Some((generation, origin_ps_id)) = &previous { let origin_ps = PageServerNode::from_env(env, env.get_pageserver_conf(*origin_ps_id)?); if origin_ps_id == &dest_ps.conf.id { println!("🔁 Already attached to {origin_ps_id}, freshening..."); - let gen = attachment_service.attach_hook(tenant_id, dest_ps.conf.id)?; + let gen = attachment_service + .attach_hook(tenant_id, dest_ps.conf.id) + .await?; let dest_conf = build_location_config(LocationConfigMode::AttachedSingle, gen, None); - dest_ps.location_config(tenant_id, dest_conf, None)?; + dest_ps.location_config(tenant_id, dest_conf, None).await?; println!("✅ Migration complete"); return Ok(()); } @@ -126,20 +128,24 @@ pub fn migrate_tenant( let stale_conf = build_location_config(LocationConfigMode::AttachedStale, Some(*generation), None); - origin_ps.location_config(tenant_id, stale_conf, Some(Duration::from_secs(10)))?; + origin_ps + .location_config(tenant_id, stale_conf, Some(Duration::from_secs(10))) + .await?; - baseline_lsns = Some(get_lsns(tenant_id, &origin_ps)?); + baseline_lsns = Some(get_lsns(tenant_id, &origin_ps).await?); } - let gen = attachment_service.attach_hook(tenant_id, dest_ps.conf.id)?; + let gen = attachment_service + .attach_hook(tenant_id, dest_ps.conf.id) + .await?; let dest_conf = build_location_config(LocationConfigMode::AttachedMulti, gen, None); println!("🔁 Attaching to pageserver {}", dest_ps.conf.id); - dest_ps.location_config(tenant_id, dest_conf, None)?; + dest_ps.location_config(tenant_id, dest_conf, None).await?; if let Some(baseline) = baseline_lsns { println!("🕑 Waiting for LSN to catch up..."); - await_lsn(tenant_id, &dest_ps, baseline)?; + await_lsn(tenant_id, &dest_ps, baseline).await?; } let cplane = ComputeControlPlane::load(env.clone())?; @@ -149,7 +155,7 @@ pub fn migrate_tenant( "🔁 Reconfiguring endpoint {} to use pageserver {}", endpoint_name, dest_ps.conf.id ); - endpoint.reconfigure(Some(dest_ps.conf.id))?; + endpoint.reconfigure(Some(dest_ps.conf.id)).await?; } } @@ -159,7 +165,7 @@ pub fn migrate_tenant( } let other_ps = PageServerNode::from_env(env, other_ps_conf); - let other_ps_tenants = other_ps.tenant_list()?; + let other_ps_tenants = other_ps.tenant_list().await?; // Check if this tenant is attached let found = other_ps_tenants @@ -181,7 +187,9 @@ pub fn migrate_tenant( "💤 Switching to secondary mode on pageserver {}", other_ps.conf.id ); - other_ps.location_config(tenant_id, secondary_conf, None)?; + other_ps + .location_config(tenant_id, secondary_conf, None) + .await?; } println!( @@ -189,7 +197,7 @@ pub fn migrate_tenant( dest_ps.conf.id ); let dest_conf = build_location_config(LocationConfigMode::AttachedSingle, gen, None); - dest_ps.location_config(tenant_id, dest_conf, None)?; + dest_ps.location_config(tenant_id, dest_conf, None).await?; println!("✅ Migration complete"); diff --git a/libs/postgres_connection/src/lib.rs b/libs/postgres_connection/src/lib.rs index 35cb1a2691..ccf9108895 100644 --- a/libs/postgres_connection/src/lib.rs +++ b/libs/postgres_connection/src/lib.rs @@ -163,8 +163,18 @@ impl PgConnectionConfig { } /// Connect using postgres protocol with TLS disabled. - pub fn connect_no_tls(&self) -> Result { - postgres::Config::from(self.to_tokio_postgres_config()).connect(postgres::NoTls) + pub async fn connect_no_tls( + &self, + ) -> Result< + ( + tokio_postgres::Client, + tokio_postgres::Connection, + ), + postgres::Error, + > { + self.to_tokio_postgres_config() + .connect(postgres::NoTls) + .await } } diff --git a/pageserver/client/Cargo.toml b/pageserver/client/Cargo.toml new file mode 100644 index 0000000000..4bd36185a6 --- /dev/null +++ b/pageserver/client/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "pageserver_client" +version = "0.1.0" +edition.workspace = true +license.workspace = true + +[dependencies] +pageserver_api.workspace = true +thiserror.workspace = true +async-trait.workspace = true +reqwest.workspace = true +utils.workspace = true +serde.workspace = true +workspace_hack = { version = "0.1", path = "../../workspace_hack" } diff --git a/pageserver/client/src/lib.rs b/pageserver/client/src/lib.rs new file mode 100644 index 0000000000..3963fd466c --- /dev/null +++ b/pageserver/client/src/lib.rs @@ -0,0 +1 @@ +pub mod mgmt_api; diff --git a/pageserver/client/src/mgmt_api.rs b/pageserver/client/src/mgmt_api.rs new file mode 100644 index 0000000000..821c238082 --- /dev/null +++ b/pageserver/client/src/mgmt_api.rs @@ -0,0 +1,172 @@ +use pageserver_api::models::*; +use reqwest::{IntoUrl, Method}; +use utils::{ + http::error::HttpErrorBody, + id::{TenantId, TimelineId}, +}; + +#[derive(Debug)] +pub struct Client { + mgmt_api_endpoint: String, + authorization_header: Option, + client: reqwest::Client, +} + +#[derive(thiserror::Error, Debug)] +pub enum Error { + #[error("receive body: {0}")] + ReceiveBody(reqwest::Error), + + #[error("receive error body: {0}")] + ReceiveErrorBody(String), + + #[error("pageserver API: {0}")] + ApiError(String), +} + +pub type Result = std::result::Result; + +#[async_trait::async_trait] +pub trait ResponseErrorMessageExt: Sized { + async fn error_from_body(self) -> Result; +} + +#[async_trait::async_trait] +impl ResponseErrorMessageExt for reqwest::Response { + async fn error_from_body(mut self) -> Result { + let status = self.status(); + if !(status.is_client_error() || status.is_server_error()) { + return Ok(self); + } + + let url = self.url().to_owned(); + Err(match self.json::().await { + Ok(HttpErrorBody { msg }) => Error::ApiError(msg), + Err(_) => { + Error::ReceiveErrorBody(format!("Http error ({}) at {}.", status.as_u16(), url)) + } + }) + } +} + +impl Client { + pub fn new(mgmt_api_endpoint: String, jwt: Option<&str>) -> Self { + Self { + mgmt_api_endpoint, + authorization_header: jwt.map(|jwt| format!("Bearer {jwt}")), + client: reqwest::Client::new(), + } + } + + pub async fn list_tenants(&self) -> Result> { + let uri = format!("{}/v1/tenant", self.mgmt_api_endpoint); + let resp = self.get(&uri).await?; + resp.json().await.map_err(Error::ReceiveBody) + } + + pub async fn list_timelines( + &self, + tenant_id: TenantId, + ) -> Result> { + let uri = format!("{}/v1/tenant/{tenant_id}/timeline", self.mgmt_api_endpoint); + self.get(&uri) + .await? + .json() + .await + .map_err(Error::ReceiveBody) + } + + pub async fn timeline_info( + &self, + tenant_id: TenantId, + timeline_id: TimelineId, + ) -> Result { + let uri = format!( + "{}/v1/tenant/{tenant_id}/timeline/{timeline_id}", + self.mgmt_api_endpoint + ); + self.get(&uri) + .await? + .json() + .await + .map_err(Error::ReceiveBody) + } + + async fn get(&self, uri: U) -> Result { + self.request(Method::GET, uri, ()).await + } + + async fn request( + &self, + method: Method, + uri: U, + body: B, + ) -> Result { + let req = self.client.request(method, uri); + let req = if let Some(value) = &self.authorization_header { + req.header(reqwest::header::AUTHORIZATION, value) + } else { + req + }; + let res = req.json(&body).send().await.map_err(Error::ReceiveBody)?; + let response = res.error_from_body().await?; + Ok(response) + } + + pub async fn status(&self) -> Result<()> { + let uri = format!("{}/v1/status", self.mgmt_api_endpoint); + self.get(&uri).await?; + Ok(()) + } + + pub async fn tenant_create(&self, req: &TenantCreateRequest) -> Result { + let uri = format!("{}/v1/tenant", self.mgmt_api_endpoint); + self.request(Method::POST, &uri, req) + .await? + .json() + .await + .map_err(Error::ReceiveBody) + } + + pub async fn tenant_config(&self, req: &TenantConfigRequest) -> Result<()> { + let uri = format!("{}/v1/tenant/config", self.mgmt_api_endpoint); + self.request(Method::PUT, &uri, req).await?; + Ok(()) + } + + pub async fn location_config( + &self, + tenant_id: TenantId, + config: LocationConfig, + flush_ms: Option, + ) -> Result<()> { + let req_body = TenantLocationConfigRequest { tenant_id, config }; + let path = format!( + "{}/v1/tenant/{}/location_config", + self.mgmt_api_endpoint, tenant_id + ); + let path = if let Some(flush_ms) = flush_ms { + format!("{}?flush_ms={}", path, flush_ms.as_millis()) + } else { + path + }; + self.request(Method::PUT, &path, &req_body).await?; + Ok(()) + } + + pub async fn timeline_create( + &self, + tenant_id: TenantId, + req: &TimelineCreateRequest, + ) -> Result { + let uri = format!( + "{}/v1/tenant/{}/timeline", + self.mgmt_api_endpoint, tenant_id + ); + self.request(Method::POST, &uri, req) + .await? + .json() + .await + .map_err(Error::ReceiveBody) + } +}