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