neon_local: use attachment service for tenant creation

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<TenantCreateResponseShard>,
+}
+
+#[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<Option<u32>> {
-        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<Option<(u32, NodeId)>> {
+    pub async fn inspect(
+        &self,
+        tenant_shard_id: TenantShardId,
+    ) -> anyhow::Result<Option<(u32, NodeId)>> {
         use hyper::StatusCode;
 
         let url = self
@@ -136,4 +160,42 @@ impl AttachmentService {
         let response = response.json::<InspectResponse>().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(())
+    }
 }

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<LocationConfig>,
+}
+
+#[derive(Serialize, Deserialize, Default)]
+struct ObservedState {
+    locations: HashMap<NodeId, ObservedStateLocation>,
+}
+
+#[derive(Serialize, Deserialize)]
 struct TenantState {
+    tenant_shard_id: TenantShardId,
+
+    shard: ShardIdentity,
+
     // Currently attached pageserver
     pageserver: Option<NodeId>,
 
     // 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<TenantShardId, TenantState>,
 
+    pageservers: HashMap<NodeId, NodeState>,
+
     #[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<Body>) -> &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<NodeId, NodeState>,
+    ) -> 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<Body>) -> Result<Response<Body>, ApiError> {
     let reattach_req = json_request::<ReAttachRequest>(&mut req).await?;
@@ -198,8 +374,12 @@ async fn handle_attach_hook(mut req: Request<Body>) -> Result<Response<Body>, 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<Body>) -> Result<Response<Body>, 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<ScheduleError> 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<ReconcileError> for ApiError {
+    fn from(value: ReconcileError) -> Self {
+        ApiError::Conflict(format!("Reconciliation error: {}", value))
+    }
+}
+
+struct Scheduler {
+    tenant_counts: HashMap<NodeId, usize>,
+}
+
+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<NodeId, ScheduleError> {
+        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<Body>) -> Result<Response<Body>, ApiError> {
-    let inspect_req = json_request::<InspectRequest>(&mut req).await?;
+    let create_req = json_request::<TenantCreateRequest>(&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<Body>) -> Result<Response<Body>, ApiError> {
+    let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
+    let create_req = json_request::<TimelineCreateRequest>(&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<Body>) -> Result<Response<Body>, ApiError> {
+    let register_req = json_request::<NodeRegisterRequest>(&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<hyper::Body, ApiError> {
     endpoint::make_router()
         .data(Arc::new(State::new(persistent_state)))
@@ -279,8 +677,10 @@ fn make_router(persistent_state: PersistentState) -> RouterBuilder<hyper::Body,
         .post("/validate", |r| request_span(r, handle_validate))
         .post("/attach-hook", |r| request_span(r, handle_attach_hook))
         .post("/inspect", |r| request_span(r, handle_inspect))
-        .post("/tenant/:tenant_id", |r| {
-            request_span(r, handle_tenant_create)
+        .post("/node", |r| request_span(r, handle_node_register))
+        .post("/tenant", |r| request_span(r, handle_tenant_create))
+        .post("/tenant/:tenant_id/timeline", |r| {
+            request_span(r, handle_tenant_timeline_create)
         })
 }
 

control_plane/src/bin/neon_local.rs

@@ -15,7 +15,9 @@ use control_plane::pageserver::{PageServerNode, PAGESERVER_REMOTE_STORAGE_DIR};
 use control_plane::safekeeper::SafekeeperNode;
 use control_plane::tenant_migration::migrate_tenant;
 use control_plane::{broker, local_env};
-use pageserver_api::models::TimelineInfo;
+use pageserver_api::models::{
+    ShardParameters, TenantCreateRequest, TimelineCreateRequest, TimelineInfo,
+};
 use pageserver_api::shard::TenantShardId;
 use pageserver_api::{
     DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
@@ -419,47 +421,47 @@ async fn handle_tenant(
                 .map(|vals| vals.flat_map(|c| c.split_once(':')).collect())
                 .unwrap_or_default();
 
+            let tenant_conf = PageServerNode::parse_config(tenant_conf)?;
+
             // If tenant ID was not specified, generate one
             let tenant_id = parse_tenant_id(create_match)?.unwrap_or_else(TenantId::generate);
 
-            // TODO(sharding): create multiple shards if requested
-            let tenant_shard_id = TenantShardId::unsharded(tenant_id);
-
-            let generation = if env.control_plane_api.is_some() {
-                // 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_shard_id, pageserver.conf.id)
-                    .await?
-            } else {
-                None
-            };
-
-            pageserver
-                .tenant_create(tenant_id, generation, tenant_conf)
-                .await?;
+            // 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.tenant_create(TenantCreateRequest {
+                // Note that ::unsharded here isn't actually because the tenant is unsharded, its because the
+                // attachment service expecfs a shard-naive tenant_id in this attribute, and the TenantCreateRequest
+                // type is used both in attachment service (for creating tenants) and in pageserver (for creating shards)
+                new_tenant_id: TenantShardId::unsharded(tenant_id),
+                generation: None,
+                shard_parameters: ShardParameters::default(),
+                config: tenant_conf,
+            })?;
             println!("tenant {tenant_id} successfully created on the pageserver");
 
             // Create an initial timeline for the new tenant
-            let new_timeline_id = parse_timeline_id(create_match)?;
+            let new_timeline_id =
+                parse_timeline_id(create_match)?.unwrap_or(TimelineId::generate());
             let pg_version = create_match
                 .get_one::<u32>("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");

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<u32>,
+        settings: HashMap<&str, &str>,
+    ) -> anyhow::Result<TenantId> {
+        let config = Self::parse_config(settings)?;
 
         let request = models::TenantCreateRequest {
             new_tenant_id: TenantShardId::unsharded(new_tenant_id),

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<u64>,
     pub checkpoint_timeout: Option<String>,
@@ -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?

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,
 }