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