From 1e542b31871214fa4a3079290b157d6fd32b4465 Mon Sep 17 00:00:00 2001 From: John Spray Date: Tue, 12 Dec 2023 15:23:13 +0000 Subject: [PATCH] neon_local: always get endpoint pageserver from attachment service --- control_plane/src/bin/neon_local.rs | 45 +++++++++++++++++---------- control_plane/src/endpoint.rs | 39 +++-------------------- control_plane/src/tenant_migration.rs | 2 +- 3 files changed, 35 insertions(+), 51 deletions(-) diff --git a/control_plane/src/bin/neon_local.rs b/control_plane/src/bin/neon_local.rs index 0e8857b5ac..f1ebd0922f 100644 --- a/control_plane/src/bin/neon_local.rs +++ b/control_plane/src/bin/neon_local.rs @@ -33,6 +33,7 @@ use std::path::PathBuf; use std::process::exit; use std::str::FromStr; use storage_broker::DEFAULT_LISTEN_ADDR as DEFAULT_BROKER_ADDR; +use url::Host; use utils::{ auth::{Claims, Scope}, id::{NodeId, TenantId, TenantTimelineId, TimelineId}, @@ -671,7 +672,6 @@ async fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::Local None, pg_version, ComputeMode::Primary, - DEFAULT_PAGESERVER_ID, )?; println!("Done"); } @@ -830,13 +830,6 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re .copied() .unwrap_or(false); - let pageserver_id = - if let Some(id_str) = sub_args.get_one::("endpoint-pageserver-id") { - NodeId(id_str.parse().context("while parsing pageserver id")?) - } else { - DEFAULT_PAGESERVER_ID - }; - let mode = match (lsn, hot_standby) { (Some(lsn), false) => ComputeMode::Static(lsn), (None, true) => ComputeMode::Replica, @@ -864,7 +857,6 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re http_port, pg_version, mode, - pageserver_id, )?; } "start" => { @@ -912,7 +904,13 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re .tenant_locate(endpoint.tenant_id)? .shards .into_iter() - .map(|shard| (shard.listen_pg_addr, shard.listen_pg_port)) + .map(|shard| { + ( + Host::parse(&shard.listen_pg_addr) + .expect("Attachment service reported bad hostname"), + shard.listen_pg_port, + ) + }) .collect::>(); let ps_conf = env.get_pageserver_conf(pageserver_id)?; @@ -937,15 +935,30 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re .endpoints .get(endpoint_id.as_str()) .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?; - let pageserver_id = + let pageservers = if let Some(id_str) = sub_args.get_one::("endpoint-pageserver-id") { - Some(NodeId( - id_str.parse().context("while parsing pageserver id")?, - )) + let ps_id = NodeId(id_str.parse().context("while parsing pageserver id")?); + let pageserver = PageServerNode::from_env(env, env.get_pageserver_conf(ps_id)?); + vec![( + pageserver.pg_connection_config.host().clone(), + pageserver.pg_connection_config.port(), + )] } else { - None + let attachment_service = AttachmentService::from_env(env); + attachment_service + .tenant_locate(endpoint.tenant_id)? + .shards + .into_iter() + .map(|shard| { + ( + Host::parse(&shard.listen_pg_addr) + .expect("Attachment service reported malformed host"), + shard.listen_pg_port, + ) + }) + .collect::>() }; - endpoint.reconfigure(pageserver_id).await?; + endpoint.reconfigure(pageservers).await?; } "stop" => { let endpoint_id = sub_args diff --git a/control_plane/src/endpoint.rs b/control_plane/src/endpoint.rs index f9645be575..49005429f6 100644 --- a/control_plane/src/endpoint.rs +++ b/control_plane/src/endpoint.rs @@ -47,10 +47,10 @@ use std::time::Duration; use anyhow::{anyhow, bail, Context, Result}; use compute_api::spec::RemoteExtSpec; use serde::{Deserialize, Serialize}; +use url::Host; use utils::id::{NodeId, TenantId, TimelineId}; use crate::local_env::LocalEnv; -use crate::pageserver::PageServerNode; use crate::postgresql_conf::PostgresConf; use compute_api::responses::{ComputeState, ComputeStatus}; @@ -67,7 +67,6 @@ pub struct EndpointConf { http_port: u16, pg_version: u32, skip_pg_catalog_updates: bool, - pageserver_id: NodeId, } // @@ -119,19 +118,14 @@ impl ComputeControlPlane { http_port: Option, pg_version: u32, mode: ComputeMode, - pageserver_id: NodeId, ) -> Result> { let pg_port = pg_port.unwrap_or_else(|| self.get_port()); let http_port = http_port.unwrap_or_else(|| self.get_port() + 1); - let pageserver = - PageServerNode::from_env(&self.env, self.env.get_pageserver_conf(pageserver_id)?); - let ep = Arc::new(Endpoint { endpoint_id: endpoint_id.to_owned(), pg_address: SocketAddr::new("127.0.0.1".parse().unwrap(), pg_port), http_address: SocketAddr::new("127.0.0.1".parse().unwrap(), http_port), env: self.env.clone(), - pageserver, timeline_id, mode, tenant_id, @@ -157,7 +151,6 @@ impl ComputeControlPlane { pg_port, pg_version, skip_pg_catalog_updates: true, - pageserver_id, })?, )?; std::fs::write( @@ -216,7 +209,6 @@ pub struct Endpoint { // These are not part of the endpoint as such, but the environment // the endpoint runs in. pub env: LocalEnv, - pageserver: PageServerNode, // Optimizations skip_pg_catalog_updates: bool, @@ -239,15 +231,11 @@ impl Endpoint { let conf: EndpointConf = serde_json::from_slice(&std::fs::read(entry.path().join("endpoint.json"))?)?; - let pageserver = - PageServerNode::from_env(env, env.get_pageserver_conf(conf.pageserver_id)?); - Ok(Endpoint { pg_address: SocketAddr::new("127.0.0.1".parse().unwrap(), conf.pg_port), http_address: SocketAddr::new("127.0.0.1".parse().unwrap(), conf.http_port), endpoint_id, env: env.clone(), - pageserver, timeline_id: conf.timeline_id, mode: conf.mode, tenant_id: conf.tenant_id, @@ -464,7 +452,7 @@ impl Endpoint { } } - fn build_pageserver_connstr(pageservers: &[(String, u16)]) -> String { + fn build_pageserver_connstr(pageservers: &[(Host, u16)]) -> String { pageservers .iter() .map(|(host, port)| format!("postgresql://no_user@{host}:{port}")) @@ -476,7 +464,7 @@ impl Endpoint { &self, auth_token: &Option, safekeepers: Vec, - pageservers: Vec<(String, u16)>, + pageservers: Vec<(Host, u16)>, remote_ext_config: Option<&String>, ) -> Result<()> { if self.status() == "running" { @@ -663,7 +651,7 @@ impl Endpoint { } } - pub async fn reconfigure(&self, pageserver_id: Option) -> Result<()> { + pub async fn reconfigure(&self, pageservers: Vec<(Host, u16)>) -> Result<()> { let mut spec: ComputeSpec = { let spec_path = self.endpoint_path().join("spec.json"); let file = std::fs::File::open(spec_path)?; @@ -673,24 +661,7 @@ impl Endpoint { let postgresql_conf = self.read_postgresql_conf()?; spec.cluster.postgresql_conf = Some(postgresql_conf); - if let Some(pageserver_id) = pageserver_id { - let endpoint_config_path = self.endpoint_path().join("endpoint.json"); - let mut endpoint_conf: EndpointConf = { - let file = std::fs::File::open(&endpoint_config_path)?; - serde_json::from_reader(file)? - }; - endpoint_conf.pageserver_id = pageserver_id; - std::fs::write( - endpoint_config_path, - serde_json::to_string_pretty(&endpoint_conf)?, - )?; - - let pageserver = - PageServerNode::from_env(&self.env, self.env.get_pageserver_conf(pageserver_id)?); - let ps_http_conf = &pageserver.pg_connection_config; - let (host, port) = (ps_http_conf.host(), ps_http_conf.port()); - spec.pageserver_connstring = Some(format!("postgresql://no_user@{host}:{port}")); - } + spec.pageserver_connstring = Some(Self::build_pageserver_connstr(&pageservers)); let client = reqwest::Client::new(); let response = client diff --git a/control_plane/src/tenant_migration.rs b/control_plane/src/tenant_migration.rs index 9aa15b1cb2..b58fe00e2f 100644 --- a/control_plane/src/tenant_migration.rs +++ b/control_plane/src/tenant_migration.rs @@ -157,7 +157,7 @@ pub async fn migrate_tenant( "🔁 Reconfiguring endpoint {} to use pageserver {}", endpoint_name, dest_ps.conf.id ); - endpoint.reconfigure(Some(dest_ps.conf.id)).await?; + endpoint.reconfigure(vec![]).await?; } }