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

Generated by: LCOV version 2.1-beta