LCOV - code coverage report
Current view: top level - control_plane/src - pageserver.rs (source / functions) Coverage Total Hit
Test: 1b0a6a0c05cee5a7de360813c8034804e105ce1c.info Lines: 0.0 % 481 0
Test Date: 2025-03-12 00:01:28 Functions: 0.0 % 53 0

            Line data    Source code
       1              : //! Code to manage pageservers
       2              : //!
       3              : //! In the local test environment, the data for each pageserver is stored in
       4              : //!
       5              : //! ```text
       6              : //!   .neon/pageserver_<pageserver_id>
       7              : //! ```
       8              : //!
       9              : use std::collections::HashMap;
      10              : use std::io;
      11              : use std::io::Write;
      12              : use std::num::NonZeroU64;
      13              : use std::path::PathBuf;
      14              : use std::str::FromStr;
      15              : use std::time::Duration;
      16              : 
      17              : use anyhow::{Context, bail};
      18              : use camino::Utf8PathBuf;
      19              : use pageserver_api::models::{self, TenantInfo, TimelineInfo};
      20              : use pageserver_api::shard::TenantShardId;
      21              : use pageserver_client::mgmt_api;
      22              : use postgres_backend::AuthType;
      23              : use postgres_connection::{PgConnectionConfig, parse_host_port};
      24              : use reqwest::Certificate;
      25              : use utils::auth::{Claims, Scope};
      26              : use utils::id::{NodeId, TenantId, TimelineId};
      27              : use utils::lsn::Lsn;
      28              : 
      29              : use crate::background_process;
      30              : use crate::local_env::{LocalEnv, NeonLocalInitPageserverConf, PageServerConf};
      31              : 
      32              : /// Directory within .neon which will be used by default for LocalFs remote storage.
      33              : pub const PAGESERVER_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/pageserver";
      34              : 
      35              : //
      36              : // Control routines for pageserver.
      37              : //
      38              : // Used in CLI and tests.
      39              : //
      40              : #[derive(Debug)]
      41              : pub struct PageServerNode {
      42              :     pub pg_connection_config: PgConnectionConfig,
      43              :     pub conf: PageServerConf,
      44              :     pub env: LocalEnv,
      45              :     pub http_client: mgmt_api::Client,
      46              : }
      47              : 
      48              : impl PageServerNode {
      49            0 :     pub fn from_env(env: &LocalEnv, conf: &PageServerConf) -> PageServerNode {
      50            0 :         let (host, port) =
      51            0 :             parse_host_port(&conf.listen_pg_addr).expect("Unable to parse listen_pg_addr");
      52            0 :         let port = port.unwrap_or(5432);
      53            0 : 
      54            0 :         let ssl_ca_cert = env.ssl_ca_cert_path().map(|ssl_ca_file| {
      55            0 :             let buf = std::fs::read(ssl_ca_file).expect("SSL root CA file should exist");
      56            0 :             Certificate::from_pem(&buf).expect("CA certificate should be valid")
      57            0 :         });
      58              : 
      59            0 :         let endpoint = if env.storage_controller.use_https_pageserver_api {
      60            0 :             format!(
      61            0 :                 "https://{}",
      62            0 :                 conf.listen_https_addr.as_ref().expect(
      63            0 :                     "listen https address should be specified if use_https_pageserver_api is on"
      64            0 :                 )
      65            0 :             )
      66              :         } else {
      67            0 :             format!("http://{}", conf.listen_http_addr)
      68              :         };
      69              : 
      70              :         Self {
      71            0 :             pg_connection_config: PgConnectionConfig::new_host_port(host, port),
      72            0 :             conf: conf.clone(),
      73            0 :             env: env.clone(),
      74            0 :             http_client: mgmt_api::Client::new(
      75            0 :                 endpoint,
      76            0 :                 {
      77            0 :                     match conf.http_auth_type {
      78            0 :                         AuthType::Trust => None,
      79            0 :                         AuthType::NeonJWT => Some(
      80            0 :                             env.generate_auth_token(&Claims::new(None, Scope::PageServerApi))
      81            0 :                                 .unwrap(),
      82            0 :                         ),
      83              :                     }
      84              :                 }
      85            0 :                 .as_deref(),
      86            0 :                 ssl_ca_cert,
      87            0 :             )
      88            0 :             .expect("Client constructs with no errors"),
      89            0 :         }
      90            0 :     }
      91              : 
      92            0 :     fn pageserver_make_identity_toml(&self, node_id: NodeId) -> toml_edit::DocumentMut {
      93            0 :         toml_edit::DocumentMut::from_str(&format!("id={node_id}")).unwrap()
      94            0 :     }
      95              : 
      96            0 :     fn pageserver_init_make_toml(
      97            0 :         &self,
      98            0 :         conf: NeonLocalInitPageserverConf,
      99            0 :     ) -> anyhow::Result<toml_edit::DocumentMut> {
     100            0 :         assert_eq!(
     101            0 :             &PageServerConf::from(&conf),
     102            0 :             &self.conf,
     103            0 :             "during neon_local init, we derive the runtime state of ps conf (self.conf) from the --config flag fully"
     104              :         );
     105              : 
     106              :         // TODO(christian): instead of what we do here, create a pageserver_api::config::ConfigToml (PR #7656)
     107              : 
     108              :         // FIXME: the paths should be shell-escaped to handle paths with spaces, quotas etc.
     109            0 :         let pg_distrib_dir_param = format!(
     110            0 :             "pg_distrib_dir='{}'",
     111            0 :             self.env.pg_distrib_dir_raw().display()
     112            0 :         );
     113            0 : 
     114            0 :         let broker_endpoint_param = format!("broker_endpoint='{}'", self.env.broker.client_url());
     115            0 : 
     116            0 :         let mut overrides = vec![pg_distrib_dir_param, broker_endpoint_param];
     117            0 : 
     118            0 :         overrides.push(format!(
     119            0 :             "control_plane_api='{}'",
     120            0 :             self.env.control_plane_api.as_str()
     121            0 :         ));
     122              : 
     123              :         // Storage controller uses the same auth as pageserver: if JWT is enabled
     124              :         // for us, we will also need it to talk to them.
     125            0 :         if matches!(conf.http_auth_type, AuthType::NeonJWT) {
     126            0 :             let jwt_token = self
     127            0 :                 .env
     128            0 :                 .generate_auth_token(&Claims::new(None, Scope::GenerationsApi))
     129            0 :                 .unwrap();
     130            0 :             overrides.push(format!("control_plane_api_token='{}'", jwt_token));
     131            0 :         }
     132              : 
     133            0 :         if !conf.other.contains_key("remote_storage") {
     134            0 :             overrides.push(format!(
     135            0 :                 "remote_storage={{local_path='../{PAGESERVER_REMOTE_STORAGE_DIR}'}}"
     136            0 :             ));
     137            0 :         }
     138              : 
     139            0 :         if conf.http_auth_type != AuthType::Trust || conf.pg_auth_type != AuthType::Trust {
     140            0 :             // Keys are generated in the toplevel repo dir, pageservers' workdirs
     141            0 :             // are one level below that, so refer to keys with ../
     142            0 :             overrides.push("auth_validation_public_key_path='../auth_public_key.pem'".to_owned());
     143            0 :         }
     144              : 
     145              :         // Apply the user-provided overrides
     146            0 :         overrides.push({
     147            0 :             let mut doc =
     148            0 :                 toml_edit::ser::to_document(&conf).expect("we deserialized this from toml earlier");
     149            0 :             // `id` is written out to `identity.toml` instead of `pageserver.toml`
     150            0 :             doc.remove("id").expect("it's part of the struct");
     151            0 :             doc.to_string()
     152            0 :         });
     153            0 : 
     154            0 :         // Turn `overrides` into a toml document.
     155            0 :         // TODO: above code is legacy code, it should be refactored to use toml_edit directly.
     156            0 :         let mut config_toml = toml_edit::DocumentMut::new();
     157            0 :         for fragment_str in overrides {
     158            0 :             let fragment = toml_edit::DocumentMut::from_str(&fragment_str)
     159            0 :                 .expect("all fragments in `overrides` are valid toml documents, this function controls that");
     160            0 :             for (key, item) in fragment.iter() {
     161            0 :                 config_toml.insert(key, item.clone());
     162            0 :             }
     163              :         }
     164            0 :         Ok(config_toml)
     165            0 :     }
     166              : 
     167              :     /// Initializes a pageserver node by creating its config with the overrides provided.
     168            0 :     pub fn initialize(&self, conf: NeonLocalInitPageserverConf) -> anyhow::Result<()> {
     169            0 :         self.pageserver_init(conf)
     170            0 :             .with_context(|| format!("Failed to run init for pageserver node {}", self.conf.id))
     171            0 :     }
     172              : 
     173            0 :     pub fn repo_path(&self) -> PathBuf {
     174            0 :         self.env.pageserver_data_dir(self.conf.id)
     175            0 :     }
     176              : 
     177              :     /// The pid file is created by the pageserver process, with its pid stored inside.
     178              :     /// Other pageservers cannot lock the same file and overwrite it for as long as the current
     179              :     /// pageserver runs. (Unless someone removes the file manually; never do that!)
     180            0 :     fn pid_file(&self) -> Utf8PathBuf {
     181            0 :         Utf8PathBuf::from_path_buf(self.repo_path().join("pageserver.pid"))
     182            0 :             .expect("non-Unicode path")
     183            0 :     }
     184              : 
     185            0 :     pub async fn start(&self, retry_timeout: &Duration) -> anyhow::Result<()> {
     186            0 :         self.start_node(retry_timeout).await
     187            0 :     }
     188              : 
     189            0 :     fn pageserver_init(&self, conf: NeonLocalInitPageserverConf) -> anyhow::Result<()> {
     190            0 :         let datadir = self.repo_path();
     191            0 :         let node_id = self.conf.id;
     192            0 :         println!(
     193            0 :             "Initializing pageserver node {} at '{}' in {:?}",
     194            0 :             node_id,
     195            0 :             self.pg_connection_config.raw_address(),
     196            0 :             datadir
     197            0 :         );
     198            0 :         io::stdout().flush()?;
     199              : 
     200              :         // If the config file we got as a CLI argument includes the `availability_zone`
     201              :         // config, then use that to populate the `metadata.json` file for the pageserver.
     202              :         // In production the deployment orchestrator does this for us.
     203            0 :         let az_id = conf
     204            0 :             .other
     205            0 :             .get("availability_zone")
     206            0 :             .map(|toml| {
     207            0 :                 let az_str = toml.to_string();
     208            0 :                 // Trim the (") chars from the toml representation
     209            0 :                 if az_str.starts_with('"') && az_str.ends_with('"') {
     210            0 :                     az_str[1..az_str.len() - 1].to_string()
     211              :                 } else {
     212            0 :                     az_str
     213              :                 }
     214            0 :             })
     215            0 :             .unwrap_or("local".to_string());
     216              : 
     217            0 :         let config = self
     218            0 :             .pageserver_init_make_toml(conf)
     219            0 :             .context("make pageserver toml")?;
     220            0 :         let config_file_path = datadir.join("pageserver.toml");
     221            0 :         let mut config_file = std::fs::OpenOptions::new()
     222            0 :             .create_new(true)
     223            0 :             .write(true)
     224            0 :             .open(&config_file_path)
     225            0 :             .with_context(|| format!("open pageserver toml for write: {config_file_path:?}"))?;
     226            0 :         config_file
     227            0 :             .write_all(config.to_string().as_bytes())
     228            0 :             .context("write pageserver toml")?;
     229            0 :         drop(config_file);
     230            0 : 
     231            0 :         let identity_file_path = datadir.join("identity.toml");
     232            0 :         let mut identity_file = std::fs::OpenOptions::new()
     233            0 :             .create_new(true)
     234            0 :             .write(true)
     235            0 :             .open(identity_file_path)
     236            0 :             .with_context(|| format!("open identity toml for write: {config_file_path:?}"))?;
     237            0 :         let identity_toml = self.pageserver_make_identity_toml(node_id);
     238            0 :         identity_file
     239            0 :             .write_all(identity_toml.to_string().as_bytes())
     240            0 :             .context("write identity toml")?;
     241            0 :         drop(identity_toml);
     242            0 : 
     243            0 :         if self.env.generate_local_ssl_certs {
     244            0 :             self.env.generate_ssl_cert(
     245            0 :                 datadir.join("server.crt").as_path(),
     246            0 :                 datadir.join("server.key").as_path(),
     247            0 :             )?;
     248            0 :         }
     249              : 
     250              :         // TODO: invoke a TBD config-check command to validate that pageserver will start with the written config
     251              : 
     252              :         // Write metadata file, used by pageserver on startup to register itself with
     253              :         // the storage controller
     254            0 :         let metadata_path = datadir.join("metadata.json");
     255            0 : 
     256            0 :         let (_http_host, http_port) =
     257            0 :             parse_host_port(&self.conf.listen_http_addr).expect("Unable to parse listen_http_addr");
     258            0 :         let http_port = http_port.unwrap_or(9898);
     259              : 
     260            0 :         let https_port = match self.conf.listen_https_addr.as_ref() {
     261            0 :             Some(https_addr) => {
     262            0 :                 let (_https_host, https_port) =
     263            0 :                     parse_host_port(https_addr).expect("Unable to parse listen_https_addr");
     264            0 :                 Some(https_port.unwrap_or(9899))
     265              :             }
     266            0 :             None => None,
     267              :         };
     268              : 
     269              :         // Intentionally hand-craft JSON: this acts as an implicit format compat test
     270              :         // in case the pageserver-side structure is edited, and reflects the real life
     271              :         // situation: the metadata is written by some other script.
     272            0 :         std::fs::write(
     273            0 :             metadata_path,
     274            0 :             serde_json::to_vec(&pageserver_api::config::NodeMetadata {
     275            0 :                 postgres_host: "localhost".to_string(),
     276            0 :                 postgres_port: self.pg_connection_config.port(),
     277            0 :                 http_host: "localhost".to_string(),
     278            0 :                 http_port,
     279            0 :                 https_port,
     280            0 :                 other: HashMap::from([(
     281            0 :                     "availability_zone_id".to_string(),
     282            0 :                     serde_json::json!(az_id),
     283            0 :                 )]),
     284            0 :             })
     285            0 :             .unwrap(),
     286            0 :         )
     287            0 :         .expect("Failed to write metadata file");
     288            0 : 
     289            0 :         Ok(())
     290            0 :     }
     291              : 
     292            0 :     async fn start_node(&self, retry_timeout: &Duration) -> anyhow::Result<()> {
     293            0 :         // TODO: using a thread here because start_process() is not async but we need to call check_status()
     294            0 :         let datadir = self.repo_path();
     295            0 :         print!(
     296            0 :             "Starting pageserver node {} at '{}' in {:?}, retrying for {:?}",
     297            0 :             self.conf.id,
     298            0 :             self.pg_connection_config.raw_address(),
     299            0 :             datadir,
     300            0 :             retry_timeout
     301            0 :         );
     302            0 :         io::stdout().flush().context("flush stdout")?;
     303              : 
     304            0 :         let datadir_path_str = datadir.to_str().with_context(|| {
     305            0 :             format!(
     306            0 :                 "Cannot start pageserver node {} in path that has no string representation: {:?}",
     307            0 :                 self.conf.id, datadir,
     308            0 :             )
     309            0 :         })?;
     310            0 :         let args = vec!["-D", datadir_path_str];
     311            0 : 
     312            0 :         background_process::start_process(
     313            0 :             "pageserver",
     314            0 :             &datadir,
     315            0 :             &self.env.pageserver_bin(),
     316            0 :             args,
     317            0 :             self.pageserver_env_variables()?,
     318            0 :             background_process::InitialPidFile::Expect(self.pid_file()),
     319            0 :             retry_timeout,
     320            0 :             || async {
     321            0 :                 let st = self.check_status().await;
     322            0 :                 match st {
     323            0 :                     Ok(()) => Ok(true),
     324            0 :                     Err(mgmt_api::Error::ReceiveBody(_)) => Ok(false),
     325            0 :                     Err(e) => Err(anyhow::anyhow!("Failed to check node status: {e}")),
     326              :                 }
     327            0 :             },
     328            0 :         )
     329            0 :         .await?;
     330              : 
     331            0 :         Ok(())
     332            0 :     }
     333              : 
     334            0 :     fn pageserver_env_variables(&self) -> anyhow::Result<Vec<(String, String)>> {
     335            0 :         // FIXME: why is this tied to pageserver's auth type? Whether or not the safekeeper
     336            0 :         // needs a token, and how to generate that token, seems independent to whether
     337            0 :         // the pageserver requires a token in incoming requests.
     338            0 :         Ok(if self.conf.http_auth_type != AuthType::Trust {
     339              :             // Generate a token to connect from the pageserver to a safekeeper
     340            0 :             let token = self
     341            0 :                 .env
     342            0 :                 .generate_auth_token(&Claims::new(None, Scope::SafekeeperData))?;
     343            0 :             vec![("NEON_AUTH_TOKEN".to_owned(), token)]
     344              :         } else {
     345            0 :             Vec::new()
     346              :         })
     347            0 :     }
     348              : 
     349              :     ///
     350              :     /// Stop the server.
     351              :     ///
     352              :     /// If 'immediate' is true, we use SIGQUIT, killing the process immediately.
     353              :     /// Otherwise we use SIGTERM, triggering a clean shutdown
     354              :     ///
     355              :     /// If the server is not running, returns success
     356              :     ///
     357            0 :     pub fn stop(&self, immediate: bool) -> anyhow::Result<()> {
     358            0 :         background_process::stop_process(immediate, "pageserver", &self.pid_file())
     359            0 :     }
     360              : 
     361            0 :     pub async fn check_status(&self) -> mgmt_api::Result<()> {
     362            0 :         self.http_client.status().await
     363            0 :     }
     364              : 
     365            0 :     pub async fn tenant_list(&self) -> mgmt_api::Result<Vec<TenantInfo>> {
     366            0 :         self.http_client.list_tenants().await
     367            0 :     }
     368            0 :     pub fn parse_config(mut settings: HashMap<&str, &str>) -> anyhow::Result<models::TenantConfig> {
     369            0 :         let result = models::TenantConfig {
     370            0 :             checkpoint_distance: settings
     371            0 :                 .remove("checkpoint_distance")
     372            0 :                 .map(|x| x.parse::<u64>())
     373            0 :                 .transpose()
     374            0 :                 .context("Failed to parse 'checkpoint_distance' as an integer")?,
     375            0 :             checkpoint_timeout: settings
     376            0 :                 .remove("checkpoint_timeout")
     377            0 :                 .map(humantime::parse_duration)
     378            0 :                 .transpose()
     379            0 :                 .context("Failed to parse 'checkpoint_timeout' as duration")?,
     380            0 :             compaction_target_size: settings
     381            0 :                 .remove("compaction_target_size")
     382            0 :                 .map(|x| x.parse::<u64>())
     383            0 :                 .transpose()
     384            0 :                 .context("Failed to parse 'compaction_target_size' as an integer")?,
     385            0 :             compaction_period: settings
     386            0 :                 .remove("compaction_period")
     387            0 :                 .map(humantime::parse_duration)
     388            0 :                 .transpose()
     389            0 :                 .context("Failed to parse 'compaction_period' as duration")?,
     390            0 :             compaction_threshold: settings
     391            0 :                 .remove("compaction_threshold")
     392            0 :                 .map(|x| x.parse::<usize>())
     393            0 :                 .transpose()
     394            0 :                 .context("Failed to parse 'compaction_threshold' as an integer")?,
     395            0 :             compaction_upper_limit: settings
     396            0 :                 .remove("compaction_upper_limit")
     397            0 :                 .map(|x| x.parse::<usize>())
     398            0 :                 .transpose()
     399            0 :                 .context("Failed to parse 'compaction_upper_limit' as an integer")?,
     400            0 :             compaction_algorithm: settings
     401            0 :                 .remove("compaction_algorithm")
     402            0 :                 .map(serde_json::from_str)
     403            0 :                 .transpose()
     404            0 :                 .context("Failed to parse 'compaction_algorithm' json")?,
     405            0 :             compaction_l0_first: settings
     406            0 :                 .remove("compaction_l0_first")
     407            0 :                 .map(|x| x.parse::<bool>())
     408            0 :                 .transpose()
     409            0 :                 .context("Failed to parse 'compaction_l0_first' as a bool")?,
     410            0 :             compaction_l0_semaphore: settings
     411            0 :                 .remove("compaction_l0_semaphore")
     412            0 :                 .map(|x| x.parse::<bool>())
     413            0 :                 .transpose()
     414            0 :                 .context("Failed to parse 'compaction_l0_semaphore' as a bool")?,
     415            0 :             l0_flush_delay_threshold: settings
     416            0 :                 .remove("l0_flush_delay_threshold")
     417            0 :                 .map(|x| x.parse::<usize>())
     418            0 :                 .transpose()
     419            0 :                 .context("Failed to parse 'l0_flush_delay_threshold' as an integer")?,
     420            0 :             l0_flush_wait_upload: settings
     421            0 :                 .remove("l0_flush_wait_upload")
     422            0 :                 .map(|x| x.parse::<bool>())
     423            0 :                 .transpose()
     424            0 :                 .context("Failed to parse 'l0_flush_wait_upload' as a boolean")?,
     425            0 :             l0_flush_stall_threshold: settings
     426            0 :                 .remove("l0_flush_stall_threshold")
     427            0 :                 .map(|x| x.parse::<usize>())
     428            0 :                 .transpose()
     429            0 :                 .context("Failed to parse 'l0_flush_stall_threshold' as an integer")?,
     430            0 :             gc_horizon: settings
     431            0 :                 .remove("gc_horizon")
     432            0 :                 .map(|x| x.parse::<u64>())
     433            0 :                 .transpose()
     434            0 :                 .context("Failed to parse 'gc_horizon' as an integer")?,
     435            0 :             gc_period: settings.remove("gc_period")
     436            0 :                 .map(humantime::parse_duration)
     437            0 :                 .transpose()
     438            0 :                 .context("Failed to parse 'gc_period' as duration")?,
     439            0 :             image_creation_threshold: settings
     440            0 :                 .remove("image_creation_threshold")
     441            0 :                 .map(|x| x.parse::<usize>())
     442            0 :                 .transpose()
     443            0 :                 .context("Failed to parse 'image_creation_threshold' as non zero integer")?,
     444            0 :             image_layer_creation_check_threshold: settings
     445            0 :                 .remove("image_layer_creation_check_threshold")
     446            0 :                 .map(|x| x.parse::<u8>())
     447            0 :                 .transpose()
     448            0 :                 .context("Failed to parse 'image_creation_check_threshold' as integer")?,
     449            0 :             image_creation_preempt_threshold: settings
     450            0 :                 .remove("image_creation_preempt_threshold")
     451            0 :                 .map(|x| x.parse::<usize>())
     452            0 :                 .transpose()
     453            0 :                 .context("Failed to parse 'image_creation_preempt_threshold' as integer")?,
     454            0 :             pitr_interval: settings.remove("pitr_interval")
     455            0 :                 .map(humantime::parse_duration)
     456            0 :                 .transpose()
     457            0 :                 .context("Failed to parse 'pitr_interval' as duration")?,
     458            0 :             walreceiver_connect_timeout: settings
     459            0 :                 .remove("walreceiver_connect_timeout")
     460            0 :                 .map(humantime::parse_duration)
     461            0 :                 .transpose()
     462            0 :                 .context("Failed to parse 'walreceiver_connect_timeout' as duration")?,
     463            0 :             lagging_wal_timeout: settings
     464            0 :                 .remove("lagging_wal_timeout")
     465            0 :                 .map(humantime::parse_duration)
     466            0 :                 .transpose()
     467            0 :                 .context("Failed to parse 'lagging_wal_timeout' as duration")?,
     468            0 :             max_lsn_wal_lag: settings
     469            0 :                 .remove("max_lsn_wal_lag")
     470            0 :                 .map(|x| x.parse::<NonZeroU64>())
     471            0 :                 .transpose()
     472            0 :                 .context("Failed to parse 'max_lsn_wal_lag' as non zero integer")?,
     473            0 :             eviction_policy: settings
     474            0 :                 .remove("eviction_policy")
     475            0 :                 .map(serde_json::from_str)
     476            0 :                 .transpose()
     477            0 :                 .context("Failed to parse 'eviction_policy' json")?,
     478            0 :             min_resident_size_override: settings
     479            0 :                 .remove("min_resident_size_override")
     480            0 :                 .map(|x| x.parse::<u64>())
     481            0 :                 .transpose()
     482            0 :                 .context("Failed to parse 'min_resident_size_override' as integer")?,
     483            0 :             evictions_low_residence_duration_metric_threshold: settings
     484            0 :                 .remove("evictions_low_residence_duration_metric_threshold")
     485            0 :                 .map(humantime::parse_duration)
     486            0 :                 .transpose()
     487            0 :                 .context("Failed to parse 'evictions_low_residence_duration_metric_threshold' as duration")?,
     488            0 :             heatmap_period: settings
     489            0 :                 .remove("heatmap_period")
     490            0 :                 .map(humantime::parse_duration)
     491            0 :                 .transpose()
     492            0 :                 .context("Failed to parse 'heatmap_period' as duration")?,
     493            0 :             lazy_slru_download: settings
     494            0 :                 .remove("lazy_slru_download")
     495            0 :                 .map(|x| x.parse::<bool>())
     496            0 :                 .transpose()
     497            0 :                 .context("Failed to parse 'lazy_slru_download' as bool")?,
     498            0 :             timeline_get_throttle: settings
     499            0 :                 .remove("timeline_get_throttle")
     500            0 :                 .map(serde_json::from_str)
     501            0 :                 .transpose()
     502            0 :                 .context("parse `timeline_get_throttle` from json")?,
     503            0 :             lsn_lease_length: settings.remove("lsn_lease_length")
     504            0 :                 .map(humantime::parse_duration)
     505            0 :                 .transpose()
     506            0 :                 .context("Failed to parse 'lsn_lease_length' as duration")?,
     507            0 :             lsn_lease_length_for_ts: settings
     508            0 :                 .remove("lsn_lease_length_for_ts")
     509            0 :                 .map(humantime::parse_duration)
     510            0 :                 .transpose()
     511            0 :                 .context("Failed to parse 'lsn_lease_length_for_ts' as duration")?,
     512            0 :             timeline_offloading: settings
     513            0 :                 .remove("timeline_offloading")
     514            0 :                 .map(|x| x.parse::<bool>())
     515            0 :                 .transpose()
     516            0 :                 .context("Failed to parse 'timeline_offloading' as bool")?,
     517            0 :             wal_receiver_protocol_override: settings
     518            0 :                 .remove("wal_receiver_protocol_override")
     519            0 :                 .map(serde_json::from_str)
     520            0 :                 .transpose()
     521            0 :                 .context("parse `wal_receiver_protocol_override` from json")?,
     522            0 :             rel_size_v2_enabled: settings
     523            0 :                 .remove("rel_size_v2_enabled")
     524            0 :                 .map(|x| x.parse::<bool>())
     525            0 :                 .transpose()
     526            0 :                 .context("Failed to parse 'rel_size_v2_enabled' as bool")?,
     527            0 :             gc_compaction_enabled: settings
     528            0 :                 .remove("gc_compaction_enabled")
     529            0 :                 .map(|x| x.parse::<bool>())
     530            0 :                 .transpose()
     531            0 :                 .context("Failed to parse 'gc_compaction_enabled' as bool")?,
     532            0 :             gc_compaction_initial_threshold_kb: settings
     533            0 :                 .remove("gc_compaction_initial_threshold_kb")
     534            0 :                 .map(|x| x.parse::<u64>())
     535            0 :                 .transpose()
     536            0 :                 .context("Failed to parse 'gc_compaction_initial_threshold_kb' as integer")?,
     537            0 :             gc_compaction_ratio_percent: settings
     538            0 :                 .remove("gc_compaction_ratio_percent")
     539            0 :                 .map(|x| x.parse::<u64>())
     540            0 :                 .transpose()
     541            0 :                 .context("Failed to parse 'gc_compaction_ratio_percent' as integer")?,
     542              :         };
     543            0 :         if !settings.is_empty() {
     544            0 :             bail!("Unrecognized tenant settings: {settings:?}")
     545              :         } else {
     546            0 :             Ok(result)
     547              :         }
     548            0 :     }
     549              : 
     550            0 :     pub async fn tenant_config(
     551            0 :         &self,
     552            0 :         tenant_id: TenantId,
     553            0 :         settings: HashMap<&str, &str>,
     554            0 :     ) -> anyhow::Result<()> {
     555            0 :         let config = Self::parse_config(settings)?;
     556            0 :         self.http_client
     557            0 :             .set_tenant_config(&models::TenantConfigRequest { tenant_id, config })
     558            0 :             .await?;
     559              : 
     560            0 :         Ok(())
     561            0 :     }
     562              : 
     563            0 :     pub async fn timeline_list(
     564            0 :         &self,
     565            0 :         tenant_shard_id: &TenantShardId,
     566            0 :     ) -> anyhow::Result<Vec<TimelineInfo>> {
     567            0 :         Ok(self.http_client.list_timelines(*tenant_shard_id).await?)
     568            0 :     }
     569              : 
     570              :     /// Import a basebackup prepared using either:
     571              :     /// a) `pg_basebackup -F tar`, or
     572              :     /// b) The `fullbackup` pageserver endpoint
     573              :     ///
     574              :     /// # Arguments
     575              :     /// * `tenant_id` - tenant to import into. Created if not exists
     576              :     /// * `timeline_id` - id to assign to imported timeline
     577              :     /// * `base` - (start lsn of basebackup, path to `base.tar` file)
     578              :     /// * `pg_wal` - if there's any wal to import: (end lsn, path to `pg_wal.tar`)
     579            0 :     pub async fn timeline_import(
     580            0 :         &self,
     581            0 :         tenant_id: TenantId,
     582            0 :         timeline_id: TimelineId,
     583            0 :         base: (Lsn, PathBuf),
     584            0 :         pg_wal: Option<(Lsn, PathBuf)>,
     585            0 :         pg_version: u32,
     586            0 :     ) -> anyhow::Result<()> {
     587            0 :         // Init base reader
     588            0 :         let (start_lsn, base_tarfile_path) = base;
     589            0 :         let base_tarfile = tokio::fs::File::open(base_tarfile_path).await?;
     590            0 :         let base_tarfile =
     591            0 :             mgmt_api::ReqwestBody::wrap_stream(tokio_util::io::ReaderStream::new(base_tarfile));
     592              : 
     593              :         // Init wal reader if necessary
     594            0 :         let (end_lsn, wal_reader) = if let Some((end_lsn, wal_tarfile_path)) = pg_wal {
     595            0 :             let wal_tarfile = tokio::fs::File::open(wal_tarfile_path).await?;
     596            0 :             let wal_reader =
     597            0 :                 mgmt_api::ReqwestBody::wrap_stream(tokio_util::io::ReaderStream::new(wal_tarfile));
     598            0 :             (end_lsn, Some(wal_reader))
     599              :         } else {
     600            0 :             (start_lsn, None)
     601              :         };
     602              : 
     603              :         // Import base
     604            0 :         self.http_client
     605            0 :             .import_basebackup(
     606            0 :                 tenant_id,
     607            0 :                 timeline_id,
     608            0 :                 start_lsn,
     609            0 :                 end_lsn,
     610            0 :                 pg_version,
     611            0 :                 base_tarfile,
     612            0 :             )
     613            0 :             .await?;
     614              : 
     615              :         // Import wal if necessary
     616            0 :         if let Some(wal_reader) = wal_reader {
     617            0 :             self.http_client
     618            0 :                 .import_wal(tenant_id, timeline_id, start_lsn, end_lsn, wal_reader)
     619            0 :                 .await?;
     620            0 :         }
     621              : 
     622            0 :         Ok(())
     623            0 :     }
     624              : }
        

Generated by: LCOV version 2.1-beta