diff --git a/control_plane/src/attachment_service.rs b/control_plane/src/attachment_service.rs index aec57e75b3..4295659739 100644 --- a/control_plane/src/attachment_service.rs +++ b/control_plane/src/attachment_service.rs @@ -1,10 +1,14 @@ use crate::{background_process, local_env::LocalEnv}; use anyhow::anyhow; use camino::Utf8PathBuf; -use pageserver_api::shard::TenantShardId; +use hyper::StatusCode; +use pageserver_api::{ + models::{TenantCreateRequest, TimelineCreateRequest}, + shard::TenantShardId, +}; use serde::{Deserialize, Serialize}; use std::{path::PathBuf, process::Child}; -use utils::id::NodeId; +use utils::id::{NodeId, TenantId}; pub struct AttachmentService { env: LocalEnv, @@ -36,6 +40,25 @@ pub struct InspectResponse { pub attachment: Option<(u32, NodeId)>, } +#[derive(Serialize, Deserialize)] +pub struct TenantCreateResponseShard { + pub node_id: NodeId, + pub generation: u32, +} + +#[derive(Serialize, Deserialize)] +pub struct TenantCreateResponse { + pub shards: Vec, +} + +#[derive(Serialize, Deserialize)] +pub struct NodeRegisterRequest { + pub node_id: NodeId, + + pub listen_http_addr: String, + pub listen_http_port: u16, +} + impl AttachmentService { pub fn from_env(env: &LocalEnv) -> Self { let path = env.base_data_dir.join("attachments.json"); @@ -91,8 +114,6 @@ impl AttachmentService { tenant_shard_id: TenantShardId, pageserver_id: NodeId, ) -> anyhow::Result> { - use hyper::StatusCode; - let url = self .env .control_plane_api @@ -115,7 +136,10 @@ impl AttachmentService { Ok(response.gen) } - pub async fn inspect(&self, tenant_id: TenantShardId) -> anyhow::Result> { + pub async fn inspect( + &self, + tenant_shard_id: TenantShardId, + ) -> anyhow::Result> { use hyper::StatusCode; let url = self @@ -136,4 +160,42 @@ impl AttachmentService { let response = response.json::().await?; Ok(response.attachment) } + + pub fn tenant_create(&self, req: TenantCreateRequest) -> anyhow::Result<()> { + let url = self + .env + .control_plane_api + .clone() + .unwrap() + .join("tenant") + .unwrap(); + + let response = self.client.post(url).json(&req).send()?; + if response.status() != StatusCode::OK { + return Err(anyhow!("Unexpected status {}", response.status())); + } + + Ok(()) + } + + pub fn tenant_timeline_create( + &self, + tenant_id: TenantId, + req: TimelineCreateRequest, + ) -> anyhow::Result<()> { + let url = self + .env + .control_plane_api + .clone() + .unwrap() + .join(&format!("tenant/{tenant_id}/timeline")) + .unwrap(); + + let response = self.client.post(url).json(&req).send()?; + if response.status() != StatusCode::OK { + return Err(anyhow!("Unexpected status {}", response.status())); + } + + Ok(()) + } } diff --git a/control_plane/src/bin/attachment_service.rs b/control_plane/src/bin/attachment_service.rs index 63f33112d5..30377d970b 100644 --- a/control_plane/src/bin/attachment_service.rs +++ b/control_plane/src/bin/attachment_service.rs @@ -6,14 +6,21 @@ /// use anyhow::anyhow; use clap::Parser; -use hyper::StatusCode; use hyper::{Body, Request, Response}; -use pageserver_api::shard::TenantShardId; +use hyper::{Method, StatusCode}; +use pageserver_api::models::{ + LocationConfig, LocationConfigMode, TenantConfig, TenantCreateRequest, + TenantLocationConfigRequest, TimelineCreateRequest, +}; +use pageserver_api::shard::{ShardCount, ShardIdentity, ShardNumber, TenantShardId}; +use reqwest::Client; use serde::{Deserialize, Serialize}; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; use std::sync::Arc; use utils::http::endpoint::request_span; +use utils::http::request::parse_request_param; +use utils::id::TenantId; use utils::logging::{self, LogFormat}; use utils::signals::{ShutdownSignals, Signal}; @@ -34,7 +41,8 @@ use pageserver_api::control_api::{ }; use control_plane::attachment_service::{ - AttachHookRequest, AttachHookResponse, InspectRequest, InspectResponse, + AttachHookRequest, AttachHookResponse, InspectRequest, InspectResponse, NodeRegisterRequest, + TenantCreateResponse, TenantCreateResponseShard, }; #[derive(Parser)] @@ -50,15 +58,60 @@ struct Cli { path: PathBuf, } -// The persistent state of each Tenant -#[derive(Serialize, Deserialize, Clone)] +/// Our latest knowledge of how this tenant is configured in the outside world. +/// +/// Meaning: +/// * No instance of this type exists for a node: we are certain that we have nothing configured on that +/// node for this shard. +/// * Instance exists with conf==None: we *might* have some state on that node, but we don't know +/// what it is (e.g. we failed partway through configuring it) +/// * Instance exists with conf==Some: this tells us what we last successfully configured on this node, +/// and that configuration will still be present unless something external interfered. +#[derive(Serialize, Deserialize)] +struct ObservedStateLocation { + /// If None, it means we do not know the status of this shard's location on this node, but + /// we know that we might have some state on this node. + conf: Option, +} + +#[derive(Serialize, Deserialize, Default)] +struct ObservedState { + locations: HashMap, +} + +#[derive(Serialize, Deserialize)] struct TenantState { + tenant_shard_id: TenantShardId, + + shard: ShardIdentity, + // Currently attached pageserver pageserver: Option, // Latest generation number: next time we attach, increment this // and use the incremented number when attaching generation: u32, + + observed: ObservedState, + + config: TenantConfig, +} + +#[derive(Serialize, Deserialize, Clone)] +struct NodeState { + id: NodeId, + + listen_http_addr: String, + listen_http_port: u16, +} + +impl NodeState { + fn base_url(&self) -> String { + format!( + "http://{}:{}/v1", + self.listen_http_addr, self.listen_http_port + ) + } } // Top level state available to all HTTP handlers @@ -66,6 +119,8 @@ struct TenantState { struct PersistentState { tenants: BTreeMap, + pageservers: HashMap, + #[serde(skip)] path: PathBuf, } @@ -99,6 +154,7 @@ impl PersistentState { tracing::info!("Will create state file at {}", path.display()); Self { tenants: BTreeMap::new(), + pageservers: HashMap::new(), path: path.to_owned(), } } @@ -131,6 +187,126 @@ fn get_state(request: &Request) -> &State { .as_ref() } +impl TenantState { + async fn location_config( + &self, + node: &NodeState, + config: LocationConfig, + ) -> anyhow::Result<()> { + let configure_request = TenantLocationConfigRequest { + tenant_shard_id: self.tenant_shard_id, + config, + }; + + let client = Client::new(); + let response = client + .request( + Method::PUT, + format!( + "{}/tenant/{}/location_config", + node.base_url(), + self.tenant_shard_id + ), + ) + .json(&configure_request) + .send() + .await?; + response.error_for_status()?; + + Ok(()) + } + + async fn timeline_create( + &self, + node: &NodeState, + req: &TimelineCreateRequest, + ) -> anyhow::Result<()> { + let client = Client::new(); + let response = client + .request( + Method::POST, + format!( + "{}/tenant/{}/timeline", + node.base_url(), + self.tenant_shard_id + ), + ) + .json(req) + .send() + .await?; + response.error_for_status()?; + + Ok(()) + } + + fn schedule(&mut self, scheduler: &mut Scheduler) -> Result<(), ScheduleError> { + if self.pageserver.is_some() { + return Ok(()); + } + + self.pageserver = Some(scheduler.schedule_shard()?); + + Ok(()) + } + + async fn reconcile( + &mut self, + pageservers: &HashMap, + ) -> Result<(), ReconcileError> { + let wanted_conf = LocationConfig { + mode: LocationConfigMode::AttachedSingle, + generation: Some(self.generation), + secondary_conf: None, + shard_number: self.shard.number.0, + shard_count: self.shard.count.0, + shard_stripe_size: self.shard.stripe_size.0, + tenant_conf: self.config.clone(), + }; + + match self.pageserver { + Some(node_id) => { + match self.observed.locations.get(&node_id) { + Some(conf) if conf.conf.as_ref() == Some(&wanted_conf) => { + // Nothing to do + tracing::info!("Observed configuration already correct.") + } + Some(_) | None => { + // If there is no observed configuration, or if its value does not equal our intent, then we must call out to the pageserver. + tracing::info!("Observed configuration requires update."); + let node = pageservers + .get(&node_id) + .expect("Pageserver may not be removed while referenced"); + self.location_config(&node, wanted_conf).await?; + } + } + } + None => { + // Detach everything + for (node_id, _old_state) in &self.observed.locations { + let node = pageservers + .get(node_id) + .expect("Pageserver may not be removed while referenced"); + self.location_config( + &node, + LocationConfig { + mode: LocationConfigMode::Detached, + generation: None, + secondary_conf: None, + shard_number: self.shard.number.0, + shard_count: self.shard.count.0, + shard_stripe_size: self.shard.stripe_size.0, + tenant_conf: self.config.clone(), + }, + ) + .await?; + } + } + } + + Ok(()) + } +} + /// Pageserver calls into this on startup, to learn which tenants it should attach async fn handle_re_attach(mut req: Request) -> Result, ApiError> { let reattach_req = json_request::(&mut req).await?; @@ -198,8 +374,12 @@ async fn handle_attach_hook(mut req: Request) -> Result, Ap .tenants .entry(attach_req.tenant_shard_id) .or_insert_with(|| TenantState { + tenant_shard_id: attach_req.tenant_shard_id, pageserver: attach_req.node_id, generation: 0, + shard: ShardIdentity::unsharded(), + observed: ObservedState::default(), + config: TenantConfig::default(), }); if let Some(attaching_pageserver) = attach_req.node_id.as_ref() { @@ -257,21 +437,239 @@ async fn handle_inspect(mut req: Request) -> Result, ApiErr ) } +/// Scenarios in which we cannot find a suitable location for a tenant shard +#[derive(thiserror::Error, Debug)] +enum ScheduleError { + #[error("No pageservers found")] + NoPageservers, +} + +impl From for ApiError { + fn from(value: ScheduleError) -> Self { + ApiError::Conflict(format!("Scheduling error: {}", value)) + } +} + +#[derive(thiserror::Error, Debug)] +enum ReconcileError { + #[error(transparent)] + Other(#[from] anyhow::Error), +} + +impl From for ApiError { + fn from(value: ReconcileError) -> Self { + ApiError::Conflict(format!("Reconciliation error: {}", value)) + } +} + +struct Scheduler { + tenant_counts: HashMap, +} + +impl Scheduler { + fn new(persistent_state: &PersistentState) -> Self { + let mut tenant_counts = HashMap::new(); + for (node_id, _) in &persistent_state.pageservers { + tenant_counts.insert(*node_id, 0); + } + + for (_id, tenant) in &persistent_state.tenants { + if let Some(ps) = tenant.pageserver { + let entry = tenant_counts.entry(ps).or_insert(0); + *entry += 1; + } + } + Self { tenant_counts } + } + + fn schedule_shard(&mut self) -> Result { + if self.tenant_counts.is_empty() { + return Err(ScheduleError::NoPageservers); + } + + let mut tenant_counts: Vec<(NodeId, usize)> = + self.tenant_counts.iter().map(|(k, v)| (*k, *v)).collect(); + tenant_counts.sort_by_key(|i| i.1); + let node_id = tenant_counts.last().unwrap().0; + *self.tenant_counts.get_mut(&node_id).unwrap() += 1; + Ok(node_id) + } +} + async fn handle_tenant_create(mut req: Request) -> Result, ApiError> { - let inspect_req = json_request::(&mut req).await?; + let create_req = json_request::(&mut req).await?; let state = get_state(&req).inner.clone(); - let locked = state.write().await; - let tenant_state = locked.tenants.get(&inspect_req.tenant_shard_id); + let mut locked = state.write().await; + + tracing::info!( + "Creating tenant {}, have {} pageservers", + create_req.new_tenant_id, + locked.pageservers.len() + ); + + // This service expects to handle sharding itself: it is an error to try and directly create + // a particular shard here. + let tenant_id = if create_req.new_tenant_id.shard_count > ShardCount(1) { + return Err(ApiError::BadRequest(anyhow::anyhow!( + "Attempted to create a specific shard, this API is for creating the whole tenant" + ))); + } else { + create_req.new_tenant_id.tenant_id + }; + + // Shard count 0 is valid: it means create a single shard (ShardCount(0) means "unsharded") + let literal_shard_count = if create_req.shard_parameters.is_unsharded() { + 1 + } else { + create_req.shard_parameters.count.0 + }; + + let mut response_shards = Vec::new(); + + let mut scheduler = Scheduler::new(&locked); + + for i in 0..literal_shard_count { + let shard_number = ShardNumber(i); + + let tenant_shard_id = TenantShardId { + tenant_id, + shard_number, + shard_count: create_req.shard_parameters.count, + }; + + use std::collections::btree_map::Entry; + match locked.tenants.entry(tenant_shard_id) { + Entry::Occupied(mut entry) => { + tracing::info!("Tenant shard {tenant_shard_id} already exists while creating"); + if entry.get_mut().pageserver.is_none() { + entry.get_mut().pageserver = Some(scheduler.schedule_shard().map_err(|e| { + ApiError::Conflict(format!( + "Failed to schedule shard {tenant_shard_id}: {e}" + )) + })?); + } + + response_shards.push(TenantCreateResponseShard { + node_id: entry + .get() + .pageserver + .expect("We just set pageserver if it was None"), + generation: entry.get().generation, + }); + + continue; + } + Entry::Vacant(entry) => { + let state = TenantState { + tenant_shard_id, + pageserver: Some(scheduler.schedule_shard().map_err(|e| { + ApiError::Conflict(format!( + "Failed to schedule shard {tenant_shard_id}: {e}" + )) + })?), + generation: create_req.generation.unwrap_or(1), + shard: ShardIdentity::from_params(shard_number, &create_req.shard_parameters), + observed: ObservedState::default(), + config: create_req.config.clone(), + }; + response_shards.push(TenantCreateResponseShard { + node_id: state + .pageserver + .expect("We just set pageserver if it was None"), + generation: state.generation, + }); + entry.insert(state) + } + }; + } + + // Take a snapshot of pageservers + let pageservers = locked.pageservers.clone(); + + for (tenant_shard_id, shard) in locked + .tenants + .range_mut(TenantShardId::tenant_range(tenant_id)) + { + shard.reconcile(&pageservers).await.map_err(|e| { + ApiError::Conflict(format!( + "Failed to reconcile tenant shard {}: {}", + tenant_shard_id, e + )) + })?; + } json_response( StatusCode::OK, - InspectResponse { - attachment: tenant_state.and_then(|s| s.pageserver.map(|ps| (s.generation, ps))), + TenantCreateResponse { + shards: response_shards, }, ) } +async fn handle_tenant_timeline_create(mut req: Request) -> Result, ApiError> { + let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?; + let create_req = json_request::(&mut req).await?; + + let state = get_state(&req).inner.clone(); + let mut locked = state.write().await; + + tracing::info!( + "Creating timeline {}/{}, have {} pageservers", + tenant_id, + create_req.new_timeline_id, + locked.pageservers.len() + ); + + let mut scheduler = Scheduler::new(&locked); + + // Take a snapshot of pageservers + let pageservers = locked.pageservers.clone(); + + for (_tenant_shard_id, shard) in locked + .tenants + .range_mut(TenantShardId::tenant_range(tenant_id)) + { + shard.schedule(&mut scheduler)?; + shard.reconcile(&pageservers).await?; + + let node_id = shard.pageserver.expect("We just scheduled successfully"); + let node = pageservers + .get(&node_id) + .expect("Pageservers may not be deleted while referenced"); + + shard + .timeline_create(node, &create_req) + .await + .map_err(|e| ApiError::Conflict(format!("Failed to create timeline: {e}")))?; + } + + json_response(StatusCode::OK, ()) +} + +async fn handle_node_register(mut req: Request) -> Result, ApiError> { + let register_req = json_request::(&mut req).await?; + let state = get_state(&req).inner.clone(); + let mut locked = state.write().await; + + locked.pageservers.insert( + register_req.node_id, + NodeState { + id: register_req.node_id, + listen_http_addr: register_req.listen_http_addr, + listen_http_port: register_req.listen_http_port, + }, + ); + + tracing::info!( + "Registered pageserver {}, now have {} pageservers", + register_req.node_id, + locked.pageservers.len() + ); + + json_response(StatusCode::OK, ()) +} + fn make_router(persistent_state: PersistentState) -> RouterBuilder { endpoint::make_router() .data(Arc::new(State::new(persistent_state))) @@ -279,8 +677,10 @@ fn make_router(persistent_state: PersistentState) -> RouterBuilder("pg-version") .copied() .context("Failed to parse postgres version from the argument string")?; - let timeline_info = pageserver - .timeline_create( - tenant_id, + // FIXME: passing None for ancestor_start_lsn is not kosher in a sharded world: we can't have + // different shards picking different start lsns. Maybe we have to teach attachment service + // to let shard 0 branch first and then propagate the chosen LSN to other shards. + + attachment_service.tenant_timeline_create( + tenant_id, + TimelineCreateRequest { 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; + ancestor_timeline_id: None, + ancestor_start_lsn: None, + existing_initdb_timeline_id: None, + pg_version: Some(pg_version), + }, + )?; env.register_branch_mapping( DEFAULT_BRANCH_NAME.to_string(), @@ -467,9 +469,7 @@ async fn handle_tenant( new_timeline_id, )?; - println!( - "Created an initial timeline '{new_timeline_id}' at Lsn {last_record_lsn} for tenant: {tenant_id}", - ); + println!("Created an initial timeline '{new_timeline_id}' for tenant: {tenant_id}",); if create_match.get_flag("set-default") { println!("Setting tenant {tenant_id} as a default one"); diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs index c2d71d1eb2..1956bcb196 100644 --- a/control_plane/src/pageserver.rs +++ b/control_plane/src/pageserver.rs @@ -1,5 +1,6 @@ //! Code to manage pageservers //! +//! //! In the local test environment, the pageserver stores its data directly in //! //! .neon/ @@ -371,6 +372,20 @@ impl PageServerNode { .context("Failed to parse 'gc_feedback' as bool")?, heatmap_period: settings.remove("heatmap_period").map(|x| x.to_string()), }; + if !settings.is_empty() { + bail!("Unrecognized tenant settings: {settings:?}") + } else { + Ok(result) + } + } + + pub fn tenant_create( + &self, + new_tenant_id: TenantId, + generation: Option, + settings: HashMap<&str, &str>, + ) -> anyhow::Result { + let config = Self::parse_config(settings)?; let request = models::TenantCreateRequest { new_tenant_id: TenantShardId::unsharded(new_tenant_id), diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs index dfb43b6b8c..c5f8e403a0 100644 --- a/libs/pageserver_api/src/models.rs +++ b/libs/pageserver_api/src/models.rs @@ -202,7 +202,7 @@ pub struct ShardParameters { } impl ShardParameters { - fn is_unsharded(&self) -> bool { + pub fn is_unsharded(&self) -> bool { self.count == ShardCount(0) } } @@ -251,7 +251,7 @@ impl std::ops::Deref for TenantCreateRequest { /// An alternative representation of `pageserver::tenant::TenantConf` with /// simpler types. -#[derive(Serialize, Deserialize, Debug, Default)] +#[derive(Serialize, Deserialize, Debug, Default, PartialEq, Eq, Clone)] pub struct TenantConfig { pub checkpoint_distance: Option, pub checkpoint_timeout: Option, @@ -280,7 +280,7 @@ pub struct TenantConfig { /// A flattened analog of a `pagesever::tenant::LocationMode`, which /// lists out all possible states (and the virtual "Detached" state) /// in a flat form rather than using rust-style enums. -#[derive(Serialize, Deserialize, Debug)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] pub enum LocationConfigMode { AttachedSingle, AttachedMulti, @@ -289,14 +289,14 @@ pub enum LocationConfigMode { Detached, } -#[derive(Serialize, Deserialize, Debug)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] pub struct LocationConfigSecondary { pub warm: bool, } /// An alternative representation of `pageserver::tenant::LocationConf`, /// for use in external-facing APIs. -#[derive(Serialize, Deserialize, Debug)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] pub struct LocationConfig { pub mode: LocationConfigMode, /// If attaching, in what generation? diff --git a/libs/pageserver_api/src/shard.rs b/libs/pageserver_api/src/shard.rs index cf9f75ebad..75cb319f21 100644 --- a/libs/pageserver_api/src/shard.rs +++ b/libs/pageserver_api/src/shard.rs @@ -336,7 +336,7 @@ const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8); pub struct ShardIdentity { pub number: ShardNumber, pub count: ShardCount, - stripe_size: ShardStripeSize, + pub stripe_size: ShardStripeSize, layout: ShardLayout, }