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