Line data Source code
1 : //!
2 : //! `neon_local` is an executable that can be used to create a local
3 : //! Neon environment, for testing purposes. The local environment is
4 : //! quite different from the cloud environment with Kubernetes, but it
5 : //! easier to work with locally. The python tests in `test_runner`
6 : //! rely on `neon_local` to set up the environment for each test.
7 : //!
8 : use anyhow::{anyhow, bail, Context, Result};
9 : use clap::{value_parser, Arg, ArgAction, ArgMatches, Command, ValueEnum};
10 : use compute_api::spec::ComputeMode;
11 : use control_plane::attachment_service::{
12 : AttachmentService, NodeAvailability, NodeConfigureRequest, NodeSchedulingPolicy,
13 : };
14 : use control_plane::endpoint::ComputeControlPlane;
15 : use control_plane::local_env::{InitForceMode, LocalEnv};
16 : use control_plane::pageserver::{PageServerNode, PAGESERVER_REMOTE_STORAGE_DIR};
17 : use control_plane::safekeeper::SafekeeperNode;
18 : use control_plane::{broker, local_env};
19 : use pageserver_api::models::{
20 : ShardParameters, TenantCreateRequest, TimelineCreateRequest, TimelineInfo,
21 : };
22 : use pageserver_api::shard::{ShardCount, ShardStripeSize, TenantShardId};
23 : use pageserver_api::{
24 : DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
25 : DEFAULT_PG_LISTEN_PORT as DEFAULT_PAGESERVER_PG_PORT,
26 : };
27 : use postgres_backend::AuthType;
28 : use postgres_connection::parse_host_port;
29 : use safekeeper_api::{
30 : DEFAULT_HTTP_LISTEN_PORT as DEFAULT_SAFEKEEPER_HTTP_PORT,
31 : DEFAULT_PG_LISTEN_PORT as DEFAULT_SAFEKEEPER_PG_PORT,
32 : };
33 : use std::collections::{BTreeSet, HashMap};
34 : use std::path::PathBuf;
35 : use std::process::exit;
36 : use std::str::FromStr;
37 : use storage_broker::DEFAULT_LISTEN_ADDR as DEFAULT_BROKER_ADDR;
38 : use url::Host;
39 : use utils::{
40 : auth::{Claims, Scope},
41 : id::{NodeId, TenantId, TenantTimelineId, TimelineId},
42 : lsn::Lsn,
43 : project_git_version,
44 : };
45 :
46 : // Default id of a safekeeper node, if not specified on the command line.
47 : const DEFAULT_SAFEKEEPER_ID: NodeId = NodeId(1);
48 : const DEFAULT_PAGESERVER_ID: NodeId = NodeId(1);
49 : const DEFAULT_BRANCH_NAME: &str = "main";
50 : project_git_version!(GIT_VERSION);
51 :
52 : const DEFAULT_PG_VERSION: &str = "15";
53 :
54 : const DEFAULT_PAGESERVER_CONTROL_PLANE_API: &str = "http://127.0.0.1:1234/upcall/v1/";
55 :
56 0 : fn default_conf(num_pageservers: u16) -> String {
57 0 : let mut template = format!(
58 0 : r#"
59 0 : # Default built-in configuration, defined in main.rs
60 0 : control_plane_api = '{DEFAULT_PAGESERVER_CONTROL_PLANE_API}'
61 0 :
62 0 : [broker]
63 0 : listen_addr = '{DEFAULT_BROKER_ADDR}'
64 0 :
65 0 : [[safekeepers]]
66 0 : id = {DEFAULT_SAFEKEEPER_ID}
67 0 : pg_port = {DEFAULT_SAFEKEEPER_PG_PORT}
68 0 : http_port = {DEFAULT_SAFEKEEPER_HTTP_PORT}
69 0 :
70 0 : "#,
71 0 : );
72 :
73 0 : for i in 0..num_pageservers {
74 0 : let pageserver_id = NodeId(DEFAULT_PAGESERVER_ID.0 + i as u64);
75 0 : let pg_port = DEFAULT_PAGESERVER_PG_PORT + i;
76 0 : let http_port = DEFAULT_PAGESERVER_HTTP_PORT + i;
77 0 :
78 0 : template += &format!(
79 0 : r#"
80 0 : [[pageservers]]
81 0 : id = {pageserver_id}
82 0 : listen_pg_addr = '127.0.0.1:{pg_port}'
83 0 : listen_http_addr = '127.0.0.1:{http_port}'
84 0 : pg_auth_type = '{trust_auth}'
85 0 : http_auth_type = '{trust_auth}'
86 0 : "#,
87 0 : trust_auth = AuthType::Trust,
88 0 : )
89 : }
90 :
91 0 : template
92 0 : }
93 :
94 : ///
95 : /// Timelines tree element used as a value in the HashMap.
96 : ///
97 : struct TimelineTreeEl {
98 : /// `TimelineInfo` received from the `pageserver` via the `timeline_list` http API call.
99 : pub info: TimelineInfo,
100 : /// Name, recovered from neon config mappings
101 : pub name: Option<String>,
102 : /// Holds all direct children of this timeline referenced using `timeline_id`.
103 : pub children: BTreeSet<TimelineId>,
104 : }
105 :
106 : // Main entry point for the 'neon_local' CLI utility
107 : //
108 : // This utility helps to manage neon installation. That includes following:
109 : // * Management of local postgres installations running on top of the
110 : // pageserver.
111 : // * Providing CLI api to the pageserver
112 : // * TODO: export/import to/from usual postgres
113 6164 : fn main() -> Result<()> {
114 6164 : let matches = cli().get_matches();
115 :
116 6164 : let (sub_name, sub_args) = match matches.subcommand() {
117 6164 : Some(subcommand_data) => subcommand_data,
118 0 : None => bail!("no subcommand provided"),
119 : };
120 :
121 : // Check for 'neon init' command first.
122 6164 : let subcommand_result = if sub_name == "init" {
123 357 : handle_init(sub_args).map(Some)
124 : } else {
125 : // all other commands need an existing config
126 5807 : let mut env = LocalEnv::load_config().context("Error loading config")?;
127 5807 : let original_env = env.clone();
128 5807 :
129 5807 : let rt = tokio::runtime::Builder::new_current_thread()
130 5807 : .enable_all()
131 5807 : .build()
132 5807 : .unwrap();
133 :
134 5807 : let subcommand_result = match sub_name {
135 5807 : "tenant" => rt.block_on(handle_tenant(sub_args, &mut env)),
136 5322 : "timeline" => rt.block_on(handle_timeline(sub_args, &mut env)),
137 4960 : "start" => rt.block_on(handle_start_all(sub_args, &env)),
138 4957 : "stop" => rt.block_on(handle_stop_all(sub_args, &env)),
139 4954 : "pageserver" => rt.block_on(handle_pageserver(sub_args, &env)),
140 3709 : "attachment_service" => rt.block_on(handle_attachment_service(sub_args, &env)),
141 2981 : "safekeeper" => rt.block_on(handle_safekeeper(sub_args, &env)),
142 1924 : "endpoint" => rt.block_on(handle_endpoint(sub_args, &env)),
143 1 : "mappings" => handle_mappings(sub_args, &mut env),
144 0 : "pg" => bail!("'pg' subcommand has been renamed to 'endpoint'"),
145 0 : _ => bail!("unexpected subcommand {sub_name}"),
146 : };
147 :
148 5807 : if original_env != env {
149 796 : subcommand_result.map(|()| Some(env))
150 : } else {
151 5011 : subcommand_result.map(|()| None)
152 : }
153 : };
154 :
155 6117 : match subcommand_result {
156 1153 : Ok(Some(updated_env)) => updated_env.persist_config(&updated_env.base_data_dir)?,
157 4964 : Ok(None) => (),
158 47 : Err(e) => {
159 47 : eprintln!("command failed: {e:?}");
160 47 : exit(1);
161 : }
162 : }
163 6117 : Ok(())
164 6117 : }
165 :
166 : ///
167 : /// Prints timelines list as a tree-like structure.
168 : ///
169 19 : fn print_timelines_tree(
170 19 : timelines: Vec<TimelineInfo>,
171 19 : mut timeline_name_mappings: HashMap<TenantTimelineId, String>,
172 19 : ) -> Result<()> {
173 19 : let mut timelines_hash = timelines
174 19 : .iter()
175 34 : .map(|t| {
176 34 : (
177 34 : t.timeline_id,
178 34 : TimelineTreeEl {
179 34 : info: t.clone(),
180 34 : children: BTreeSet::new(),
181 34 : name: timeline_name_mappings
182 34 : .remove(&TenantTimelineId::new(t.tenant_id.tenant_id, t.timeline_id)),
183 34 : },
184 34 : )
185 34 : })
186 19 : .collect::<HashMap<_, _>>();
187 :
188 : // Memorize all direct children of each timeline.
189 34 : for timeline in timelines.iter() {
190 34 : if let Some(ancestor_timeline_id) = timeline.ancestor_timeline_id {
191 15 : timelines_hash
192 15 : .get_mut(&ancestor_timeline_id)
193 15 : .context("missing timeline info in the HashMap")?
194 : .children
195 15 : .insert(timeline.timeline_id);
196 19 : }
197 : }
198 :
199 34 : for timeline in timelines_hash.values() {
200 : // Start with root local timelines (no ancestors) first.
201 34 : if timeline.info.ancestor_timeline_id.is_none() {
202 19 : print_timeline(0, &Vec::from([true]), timeline, &timelines_hash)?;
203 15 : }
204 : }
205 :
206 19 : Ok(())
207 19 : }
208 :
209 : ///
210 : /// Recursively prints timeline info with all its children.
211 : ///
212 34 : fn print_timeline(
213 34 : nesting_level: usize,
214 34 : is_last: &[bool],
215 34 : timeline: &TimelineTreeEl,
216 34 : timelines: &HashMap<TimelineId, TimelineTreeEl>,
217 34 : ) -> Result<()> {
218 34 : if nesting_level > 0 {
219 15 : let ancestor_lsn = match timeline.info.ancestor_lsn {
220 15 : Some(lsn) => lsn.to_string(),
221 0 : None => "Unknown Lsn".to_string(),
222 : };
223 :
224 15 : let mut br_sym = "┣━";
225 15 :
226 15 : // Draw each nesting padding with proper style
227 15 : // depending on whether its timeline ended or not.
228 15 : if nesting_level > 1 {
229 3 : for l in &is_last[1..is_last.len() - 1] {
230 3 : if *l {
231 0 : print!(" ");
232 3 : } else {
233 3 : print!("┃ ");
234 3 : }
235 : }
236 12 : }
237 :
238 : // We are the last in this sub-timeline
239 15 : if *is_last.last().unwrap() {
240 10 : br_sym = "┗━";
241 10 : }
242 :
243 15 : print!("{} @{}: ", br_sym, ancestor_lsn);
244 19 : }
245 :
246 : // Finally print a timeline id and name with new line
247 34 : println!(
248 34 : "{} [{}]",
249 34 : timeline.name.as_deref().unwrap_or("_no_name_"),
250 34 : timeline.info.timeline_id
251 34 : );
252 34 :
253 34 : let len = timeline.children.len();
254 34 : let mut i: usize = 0;
255 34 : let mut is_last_new = Vec::from(is_last);
256 34 : is_last_new.push(false);
257 :
258 49 : for child in &timeline.children {
259 15 : i += 1;
260 15 :
261 15 : // Mark that the last padding is the end of the timeline
262 15 : if i == len {
263 10 : if let Some(last) = is_last_new.last_mut() {
264 10 : *last = true;
265 10 : }
266 5 : }
267 :
268 : print_timeline(
269 15 : nesting_level + 1,
270 15 : &is_last_new,
271 15 : timelines
272 15 : .get(child)
273 15 : .context("missing timeline info in the HashMap")?,
274 15 : timelines,
275 0 : )?;
276 : }
277 :
278 34 : Ok(())
279 34 : }
280 :
281 : /// Returns a map of timeline IDs to timeline_id@lsn strings.
282 : /// Connects to the pageserver to query this information.
283 0 : async fn get_timeline_infos(
284 0 : env: &local_env::LocalEnv,
285 0 : tenant_shard_id: &TenantShardId,
286 0 : ) -> Result<HashMap<TimelineId, TimelineInfo>> {
287 0 : Ok(get_default_pageserver(env)
288 0 : .timeline_list(tenant_shard_id)
289 0 : .await?
290 0 : .into_iter()
291 0 : .map(|timeline_info| (timeline_info.timeline_id, timeline_info))
292 0 : .collect())
293 0 : }
294 :
295 : // Helper function to parse --tenant_id option, or get the default from config file
296 900 : fn get_tenant_id(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow::Result<TenantId> {
297 900 : if let Some(tenant_id_from_arguments) = parse_tenant_id(sub_match).transpose() {
298 900 : tenant_id_from_arguments
299 0 : } else if let Some(default_id) = env.default_tenant_id {
300 0 : Ok(default_id)
301 : } else {
302 0 : anyhow::bail!("No tenant id. Use --tenant-id, or set a default tenant");
303 : }
304 900 : }
305 :
306 : // Helper function to parse --tenant_id option, for commands that accept a shard suffix
307 23 : fn get_tenant_shard_id(
308 23 : sub_match: &ArgMatches,
309 23 : env: &local_env::LocalEnv,
310 23 : ) -> anyhow::Result<TenantShardId> {
311 23 : if let Some(tenant_id_from_arguments) = parse_tenant_shard_id(sub_match).transpose() {
312 23 : tenant_id_from_arguments
313 0 : } else if let Some(default_id) = env.default_tenant_id {
314 0 : Ok(TenantShardId::unsharded(default_id))
315 : } else {
316 0 : anyhow::bail!("No tenant shard id. Use --tenant-id, or set a default tenant");
317 : }
318 23 : }
319 :
320 1360 : fn parse_tenant_id(sub_match: &ArgMatches) -> anyhow::Result<Option<TenantId>> {
321 1360 : sub_match
322 1360 : .get_one::<String>("tenant-id")
323 1360 : .map(|tenant_id| TenantId::from_str(tenant_id))
324 1360 : .transpose()
325 1360 : .context("Failed to parse tenant id from the argument string")
326 1360 : }
327 :
328 23 : fn parse_tenant_shard_id(sub_match: &ArgMatches) -> anyhow::Result<Option<TenantShardId>> {
329 23 : sub_match
330 23 : .get_one::<String>("tenant-id")
331 23 : .map(|id_str| TenantShardId::from_str(id_str))
332 23 : .transpose()
333 23 : .context("Failed to parse tenant shard id from the argument string")
334 23 : }
335 :
336 556 : fn parse_timeline_id(sub_match: &ArgMatches) -> anyhow::Result<Option<TimelineId>> {
337 556 : sub_match
338 556 : .get_one::<String>("timeline-id")
339 556 : .map(|timeline_id| TimelineId::from_str(timeline_id))
340 556 : .transpose()
341 556 : .context("Failed to parse timeline id from the argument string")
342 556 : }
343 :
344 357 : fn handle_init(init_match: &ArgMatches) -> anyhow::Result<LocalEnv> {
345 357 : let num_pageservers = init_match
346 357 : .get_one::<u16>("num-pageservers")
347 357 : .expect("num-pageservers arg has a default");
348 : // Create config file
349 357 : let toml_file: String = if let Some(config_path) = init_match.get_one::<PathBuf>("config") {
350 : // load and parse the file
351 357 : std::fs::read_to_string(config_path).with_context(|| {
352 0 : format!(
353 0 : "Could not read configuration file '{}'",
354 0 : config_path.display()
355 0 : )
356 357 : })?
357 : } else {
358 : // Built-in default config
359 0 : default_conf(*num_pageservers)
360 : };
361 :
362 357 : let pg_version = init_match
363 357 : .get_one::<u32>("pg-version")
364 357 : .copied()
365 357 : .context("Failed to parse postgres version from the argument string")?;
366 :
367 357 : let mut env =
368 357 : LocalEnv::parse_config(&toml_file).context("Failed to create neon configuration")?;
369 357 : let force = init_match.get_one("force").expect("we set a default value");
370 357 : env.init(pg_version, force)
371 357 : .context("Failed to initialize neon repository")?;
372 :
373 : // Create remote storage location for default LocalFs remote storage
374 357 : std::fs::create_dir_all(env.base_data_dir.join(PAGESERVER_REMOTE_STORAGE_DIR))?;
375 :
376 : // Initialize pageserver, create initial tenant and timeline.
377 757 : for ps_conf in &env.pageservers {
378 400 : PageServerNode::from_env(&env, ps_conf)
379 400 : .initialize(&pageserver_config_overrides(init_match))
380 400 : .unwrap_or_else(|e| {
381 0 : eprintln!("pageserver init failed: {e:?}");
382 0 : exit(1);
383 400 : });
384 400 : }
385 :
386 357 : Ok(env)
387 357 : }
388 :
389 : /// The default pageserver is the one where CLI tenant/timeline operations are sent by default.
390 : /// For typical interactive use, one would just run with a single pageserver. Scenarios with
391 : /// tenant/timeline placement across multiple pageservers are managed by python test code rather
392 : /// than this CLI.
393 847 : fn get_default_pageserver(env: &local_env::LocalEnv) -> PageServerNode {
394 847 : let ps_conf = env
395 847 : .pageservers
396 847 : .first()
397 847 : .expect("Config is validated to contain at least one pageserver");
398 847 : PageServerNode::from_env(env, ps_conf)
399 847 : }
400 :
401 1026 : fn pageserver_config_overrides(init_match: &ArgMatches) -> Vec<&str> {
402 1026 : init_match
403 1026 : .get_many::<String>("pageserver-config-override")
404 1026 : .into_iter()
405 1026 : .flatten()
406 1026 : .map(String::as_str)
407 1026 : .collect()
408 1026 : }
409 :
410 485 : async fn handle_tenant(
411 485 : tenant_match: &ArgMatches,
412 485 : env: &mut local_env::LocalEnv,
413 485 : ) -> anyhow::Result<()> {
414 485 : let pageserver = get_default_pageserver(env);
415 485 : match tenant_match.subcommand() {
416 485 : Some(("list", _)) => {
417 24 : for t in pageserver.tenant_list().await? {
418 11 : println!("{} {:?}", t.id, t.state);
419 11 : }
420 : }
421 479 : Some(("create", create_match)) => {
422 461 : let tenant_conf: HashMap<_, _> = create_match
423 461 : .get_many::<String>("config")
424 461 : .map(|vals: clap::parser::ValuesRef<'_, String>| {
425 736 : vals.flat_map(|c| c.split_once(':')).collect()
426 461 : })
427 461 : .unwrap_or_default();
428 461 :
429 461 : let shard_count: u8 = create_match
430 461 : .get_one::<u8>("shard-count")
431 461 : .cloned()
432 461 : .unwrap_or(0);
433 461 :
434 461 : let shard_stripe_size: Option<u32> =
435 461 : create_match.get_one::<u32>("shard-stripe-size").cloned();
436 :
437 461 : let tenant_conf = PageServerNode::parse_config(tenant_conf)?;
438 :
439 : // If tenant ID was not specified, generate one
440 460 : let tenant_id = parse_tenant_id(create_match)?.unwrap_or_else(TenantId::generate);
441 460 :
442 460 : // We must register the tenant with the attachment service, so
443 460 : // that when the pageserver restarts, it will be re-attached.
444 460 : let attachment_service = AttachmentService::from_env(env);
445 460 : attachment_service
446 460 : .tenant_create(TenantCreateRequest {
447 460 : // Note that ::unsharded here isn't actually because the tenant is unsharded, its because the
448 460 : // attachment service expecfs a shard-naive tenant_id in this attribute, and the TenantCreateRequest
449 460 : // type is used both in attachment service (for creating tenants) and in pageserver (for creating shards)
450 460 : new_tenant_id: TenantShardId::unsharded(tenant_id),
451 460 : generation: None,
452 460 : shard_parameters: ShardParameters {
453 460 : count: ShardCount(shard_count),
454 460 : stripe_size: shard_stripe_size
455 460 : .map(ShardStripeSize)
456 460 : .unwrap_or(ShardParameters::DEFAULT_STRIPE_SIZE),
457 460 : },
458 460 : config: tenant_conf,
459 460 : })
460 1380 : .await?;
461 459 : println!("tenant {tenant_id} successfully created on the pageserver");
462 :
463 : // Create an initial timeline for the new tenant
464 459 : let new_timeline_id =
465 459 : parse_timeline_id(create_match)?.unwrap_or(TimelineId::generate());
466 459 : let pg_version = create_match
467 459 : .get_one::<u32>("pg-version")
468 459 : .copied()
469 459 : .context("Failed to parse postgres version from the argument string")?;
470 :
471 : // FIXME: passing None for ancestor_start_lsn is not kosher in a sharded world: we can't have
472 : // different shards picking different start lsns. Maybe we have to teach attachment service
473 : // to let shard 0 branch first and then propagate the chosen LSN to other shards.
474 459 : attachment_service
475 459 : .tenant_timeline_create(
476 459 : tenant_id,
477 459 : TimelineCreateRequest {
478 459 : new_timeline_id,
479 459 : ancestor_timeline_id: None,
480 459 : ancestor_start_lsn: None,
481 459 : existing_initdb_timeline_id: None,
482 459 : pg_version: Some(pg_version),
483 459 : },
484 459 : )
485 459 : .await?;
486 :
487 459 : env.register_branch_mapping(
488 459 : DEFAULT_BRANCH_NAME.to_string(),
489 459 : tenant_id,
490 459 : new_timeline_id,
491 459 : )?;
492 :
493 459 : println!("Created an initial timeline '{new_timeline_id}' for tenant: {tenant_id}",);
494 459 :
495 459 : if create_match.get_flag("set-default") {
496 1 : println!("Setting tenant {tenant_id} as a default one");
497 1 : env.default_tenant_id = Some(tenant_id);
498 458 : }
499 : }
500 18 : Some(("set-default", set_default_match)) => {
501 0 : let tenant_id =
502 0 : parse_tenant_id(set_default_match)?.context("No tenant id specified")?;
503 0 : println!("Setting tenant {tenant_id} as a default one");
504 0 : env.default_tenant_id = Some(tenant_id);
505 : }
506 18 : Some(("config", create_match)) => {
507 14 : let tenant_id = get_tenant_id(create_match, env)?;
508 14 : let tenant_conf: HashMap<_, _> = create_match
509 14 : .get_many::<String>("config")
510 47 : .map(|vals| vals.flat_map(|c| c.split_once(':')).collect())
511 14 : .unwrap_or_default();
512 14 :
513 14 : pageserver
514 14 : .tenant_config(tenant_id, tenant_conf)
515 56 : .await
516 14 : .with_context(|| format!("Tenant config failed for tenant with id {tenant_id}"))?;
517 14 : println!("tenant {tenant_id} successfully configured on the pageserver");
518 : }
519 4 : Some(("migrate", matches)) => {
520 4 : let tenant_shard_id = get_tenant_shard_id(matches, env)?;
521 4 : let new_pageserver = get_pageserver(env, matches)?;
522 4 : let new_pageserver_id = new_pageserver.conf.id;
523 4 :
524 4 : let attachment_service = AttachmentService::from_env(env);
525 4 : attachment_service
526 4 : .tenant_migrate(tenant_shard_id, new_pageserver_id)
527 12 : .await?;
528 :
529 4 : println!("tenant {tenant_shard_id} migrated to {}", new_pageserver_id);
530 : }
531 0 : Some(("status", matches)) => {
532 0 : let tenant_id = get_tenant_id(matches, env)?;
533 :
534 0 : let mut shard_table = comfy_table::Table::new();
535 0 : shard_table.set_header(["Shard", "Pageserver", "Physical Size"]);
536 0 :
537 0 : let mut tenant_synthetic_size = None;
538 0 :
539 0 : let attachment_service = AttachmentService::from_env(env);
540 0 : for shard in attachment_service.tenant_locate(tenant_id).await?.shards {
541 0 : let pageserver =
542 0 : PageServerNode::from_env(env, env.get_pageserver_conf(shard.node_id)?);
543 :
544 0 : let size = pageserver
545 0 : .http_client
546 0 : .tenant_details(shard.shard_id)
547 0 : .await?
548 : .tenant_info
549 : .current_physical_size
550 0 : .unwrap();
551 0 :
552 0 : shard_table.add_row([
553 0 : format!("{}", shard.shard_id.shard_slug()),
554 0 : format!("{}", shard.node_id.0),
555 0 : format!("{} MiB", size / (1024 * 1024)),
556 0 : ]);
557 0 :
558 0 : if shard.shard_id.is_zero() {
559 : tenant_synthetic_size =
560 0 : Some(pageserver.tenant_synthetic_size(shard.shard_id).await?);
561 0 : }
562 : }
563 :
564 0 : let Some(synthetic_size) = tenant_synthetic_size else {
565 0 : bail!("Shard 0 not found")
566 : };
567 :
568 0 : let mut tenant_table = comfy_table::Table::new();
569 0 : tenant_table.add_row(["Tenant ID".to_string(), tenant_id.to_string()]);
570 0 : tenant_table.add_row([
571 0 : "Synthetic size".to_string(),
572 0 : format!("{} MiB", synthetic_size.size.unwrap_or(0) / (1024 * 1024)),
573 0 : ]);
574 0 :
575 0 : println!("{tenant_table}");
576 0 : println!("{shard_table}");
577 : }
578 0 : Some(("shard-split", matches)) => {
579 0 : let tenant_id = get_tenant_id(matches, env)?;
580 0 : let shard_count: u8 = matches.get_one::<u8>("shard-count").cloned().unwrap_or(0);
581 0 :
582 0 : let attachment_service = AttachmentService::from_env(env);
583 0 : let result = attachment_service
584 0 : .tenant_split(tenant_id, shard_count)
585 0 : .await?;
586 0 : println!(
587 0 : "Split tenant {} into shards {}",
588 0 : tenant_id,
589 0 : result
590 0 : .new_shards
591 0 : .iter()
592 0 : .map(|s| format!("{:?}", s))
593 0 : .collect::<Vec<_>>()
594 0 : .join(",")
595 0 : );
596 : }
597 :
598 0 : Some((sub_name, _)) => bail!("Unexpected tenant subcommand '{}'", sub_name),
599 0 : None => bail!("no tenant subcommand provided"),
600 : }
601 483 : Ok(())
602 485 : }
603 :
604 362 : async fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) -> Result<()> {
605 362 : let pageserver = get_default_pageserver(env);
606 362 :
607 362 : match timeline_match.subcommand() {
608 362 : Some(("list", list_match)) => {
609 : // TODO(sharding): this command shouldn't have to specify a shard ID: we should ask the attachment service
610 : // where shard 0 is attached, and query there.
611 19 : let tenant_shard_id = get_tenant_shard_id(list_match, env)?;
612 76 : let timelines = pageserver.timeline_list(&tenant_shard_id).await?;
613 19 : print_timelines_tree(timelines, env.timeline_name_mappings())?;
614 : }
615 343 : Some(("create", create_match)) => {
616 91 : let tenant_id = get_tenant_id(create_match, env)?;
617 91 : let new_branch_name = create_match
618 91 : .get_one::<String>("branch-name")
619 91 : .ok_or_else(|| anyhow!("No branch name provided"))?;
620 :
621 91 : let pg_version = create_match
622 91 : .get_one::<u32>("pg-version")
623 91 : .copied()
624 91 : .context("Failed to parse postgres version from the argument string")?;
625 :
626 91 : let new_timeline_id_opt = parse_timeline_id(create_match)?;
627 91 : let new_timeline_id = new_timeline_id_opt.unwrap_or(TimelineId::generate());
628 91 :
629 91 : let attachment_service = AttachmentService::from_env(env);
630 91 : let create_req = TimelineCreateRequest {
631 91 : new_timeline_id,
632 91 : ancestor_timeline_id: None,
633 91 : existing_initdb_timeline_id: None,
634 91 : ancestor_start_lsn: None,
635 91 : pg_version: Some(pg_version),
636 91 : };
637 91 : let timeline_info = attachment_service
638 91 : .tenant_timeline_create(tenant_id, create_req)
639 273 : .await?;
640 :
641 91 : let last_record_lsn = timeline_info.last_record_lsn;
642 91 : env.register_branch_mapping(new_branch_name.to_string(), tenant_id, new_timeline_id)?;
643 :
644 91 : println!(
645 91 : "Created timeline '{}' at Lsn {last_record_lsn} for tenant: {tenant_id}",
646 91 : timeline_info.timeline_id
647 91 : );
648 : }
649 252 : Some(("import", import_match)) => {
650 6 : let tenant_id = get_tenant_id(import_match, env)?;
651 6 : let timeline_id = parse_timeline_id(import_match)?.expect("No timeline id provided");
652 6 : let name = import_match
653 6 : .get_one::<String>("node-name")
654 6 : .ok_or_else(|| anyhow!("No node name provided"))?;
655 :
656 : // Parse base inputs
657 6 : let base_tarfile = import_match
658 6 : .get_one::<PathBuf>("base-tarfile")
659 6 : .ok_or_else(|| anyhow!("No base-tarfile provided"))?
660 6 : .to_owned();
661 6 : let base_lsn = Lsn::from_str(
662 6 : import_match
663 6 : .get_one::<String>("base-lsn")
664 6 : .ok_or_else(|| anyhow!("No base-lsn provided"))?,
665 0 : )?;
666 6 : let base = (base_lsn, base_tarfile);
667 6 :
668 6 : // Parse pg_wal inputs
669 6 : let wal_tarfile = import_match.get_one::<PathBuf>("wal-tarfile").cloned();
670 6 : let end_lsn = import_match
671 6 : .get_one::<String>("end-lsn")
672 6 : .map(|s| Lsn::from_str(s).unwrap());
673 6 : // TODO validate both or none are provided
674 6 : let pg_wal = end_lsn.zip(wal_tarfile);
675 :
676 6 : let pg_version = import_match
677 6 : .get_one::<u32>("pg-version")
678 6 : .copied()
679 6 : .context("Failed to parse postgres version from the argument string")?;
680 :
681 6 : let mut cplane = ComputeControlPlane::load(env.clone())?;
682 6 : println!("Importing timeline into pageserver ...");
683 6 : pageserver
684 6 : .timeline_import(tenant_id, timeline_id, base, pg_wal, pg_version)
685 62740 : .await?;
686 3 : env.register_branch_mapping(name.to_string(), tenant_id, timeline_id)?;
687 :
688 3 : println!("Creating endpoint for imported timeline ...");
689 3 : cplane.new_endpoint(
690 3 : name,
691 3 : tenant_id,
692 3 : timeline_id,
693 3 : None,
694 3 : None,
695 3 : pg_version,
696 3 : ComputeMode::Primary,
697 3 : )?;
698 3 : println!("Done");
699 : }
700 246 : Some(("branch", branch_match)) => {
701 246 : let tenant_id = get_tenant_id(branch_match, env)?;
702 246 : let new_branch_name = branch_match
703 246 : .get_one::<String>("branch-name")
704 246 : .ok_or_else(|| anyhow!("No branch name provided"))?;
705 246 : let ancestor_branch_name = branch_match
706 246 : .get_one::<String>("ancestor-branch-name")
707 246 : .map(|s| s.as_str())
708 246 : .unwrap_or(DEFAULT_BRANCH_NAME);
709 246 : let ancestor_timeline_id = env
710 246 : .get_branch_timeline_id(ancestor_branch_name, tenant_id)
711 246 : .ok_or_else(|| {
712 0 : anyhow!("Found no timeline id for branch name '{ancestor_branch_name}'")
713 246 : })?;
714 :
715 246 : let start_lsn = branch_match
716 246 : .get_one::<String>("ancestor-start-lsn")
717 246 : .map(|lsn_str| Lsn::from_str(lsn_str))
718 246 : .transpose()
719 246 : .context("Failed to parse ancestor start Lsn from the request")?;
720 246 : let new_timeline_id = TimelineId::generate();
721 246 : let attachment_service = AttachmentService::from_env(env);
722 246 : let create_req = TimelineCreateRequest {
723 246 : new_timeline_id,
724 246 : ancestor_timeline_id: Some(ancestor_timeline_id),
725 246 : existing_initdb_timeline_id: None,
726 246 : ancestor_start_lsn: start_lsn,
727 246 : pg_version: None,
728 246 : };
729 246 : let timeline_info = attachment_service
730 246 : .tenant_timeline_create(tenant_id, create_req)
731 738 : .await?;
732 :
733 242 : let last_record_lsn = timeline_info.last_record_lsn;
734 242 :
735 242 : env.register_branch_mapping(new_branch_name.to_string(), tenant_id, new_timeline_id)?;
736 :
737 242 : println!(
738 242 : "Created timeline '{}' at Lsn {last_record_lsn} for tenant: {tenant_id}. Ancestor timeline: '{ancestor_branch_name}'",
739 242 : timeline_info.timeline_id
740 242 : );
741 : }
742 0 : Some((sub_name, _)) => bail!("Unexpected tenant subcommand '{sub_name}'"),
743 0 : None => bail!("no tenant subcommand provided"),
744 : }
745 :
746 355 : Ok(())
747 362 : }
748 :
749 1923 : async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
750 1923 : let (sub_name, sub_args) = match ep_match.subcommand() {
751 1923 : Some(ep_subcommand_data) => ep_subcommand_data,
752 0 : None => bail!("no endpoint subcommand provided"),
753 : };
754 1923 : let mut cplane = ComputeControlPlane::load(env.clone())?;
755 :
756 1923 : match sub_name {
757 1923 : "list" => {
758 : // TODO(sharding): this command shouldn't have to specify a shard ID: we should ask the attachment service
759 : // where shard 0 is attached, and query there.
760 0 : let tenant_shard_id = get_tenant_shard_id(sub_args, env)?;
761 0 : let timeline_infos = get_timeline_infos(env, &tenant_shard_id)
762 0 : .await
763 0 : .unwrap_or_else(|e| {
764 0 : eprintln!("Failed to load timeline info: {}", e);
765 0 : HashMap::new()
766 0 : });
767 0 :
768 0 : let timeline_name_mappings = env.timeline_name_mappings();
769 0 :
770 0 : let mut table = comfy_table::Table::new();
771 0 :
772 0 : table.load_preset(comfy_table::presets::NOTHING);
773 0 :
774 0 : table.set_header([
775 0 : "ENDPOINT",
776 0 : "ADDRESS",
777 0 : "TIMELINE",
778 0 : "BRANCH NAME",
779 0 : "LSN",
780 0 : "STATUS",
781 0 : ]);
782 :
783 0 : for (endpoint_id, endpoint) in cplane
784 0 : .endpoints
785 0 : .iter()
786 0 : .filter(|(_, endpoint)| endpoint.tenant_id == tenant_shard_id.tenant_id)
787 0 : {
788 0 : let lsn_str = match endpoint.mode {
789 0 : ComputeMode::Static(lsn) => {
790 0 : // -> read-only endpoint
791 0 : // Use the node's LSN.
792 0 : lsn.to_string()
793 : }
794 : _ => {
795 : // -> primary endpoint or hot replica
796 : // Use the LSN at the end of the timeline.
797 0 : timeline_infos
798 0 : .get(&endpoint.timeline_id)
799 0 : .map(|bi| bi.last_record_lsn.to_string())
800 0 : .unwrap_or_else(|| "?".to_string())
801 : }
802 : };
803 :
804 0 : let branch_name = timeline_name_mappings
805 0 : .get(&TenantTimelineId::new(
806 0 : tenant_shard_id.tenant_id,
807 0 : endpoint.timeline_id,
808 0 : ))
809 0 : .map(|name| name.as_str())
810 0 : .unwrap_or("?");
811 0 :
812 0 : table.add_row([
813 0 : endpoint_id.as_str(),
814 0 : &endpoint.pg_address.to_string(),
815 0 : &endpoint.timeline_id.to_string(),
816 0 : branch_name,
817 0 : lsn_str.as_str(),
818 0 : &format!("{}", endpoint.status()),
819 0 : ]);
820 : }
821 :
822 0 : println!("{table}");
823 : }
824 1923 : "create" => {
825 543 : let tenant_id = get_tenant_id(sub_args, env)?;
826 543 : let branch_name = sub_args
827 543 : .get_one::<String>("branch-name")
828 543 : .map(|s| s.as_str())
829 543 : .unwrap_or(DEFAULT_BRANCH_NAME);
830 543 : let endpoint_id = sub_args
831 543 : .get_one::<String>("endpoint_id")
832 543 : .map(String::to_string)
833 543 : .unwrap_or_else(|| format!("ep-{branch_name}"));
834 :
835 543 : let lsn = sub_args
836 543 : .get_one::<String>("lsn")
837 543 : .map(|lsn_str| Lsn::from_str(lsn_str))
838 543 : .transpose()
839 543 : .context("Failed to parse Lsn from the request")?;
840 543 : let timeline_id = env
841 543 : .get_branch_timeline_id(branch_name, tenant_id)
842 543 : .ok_or_else(|| anyhow!("Found no timeline id for branch name '{branch_name}'"))?;
843 :
844 543 : let pg_port: Option<u16> = sub_args.get_one::<u16>("pg-port").copied();
845 543 : let http_port: Option<u16> = sub_args.get_one::<u16>("http-port").copied();
846 543 : let pg_version = sub_args
847 543 : .get_one::<u32>("pg-version")
848 543 : .copied()
849 543 : .context("Failed to parse postgres version from the argument string")?;
850 :
851 543 : let hot_standby = sub_args
852 543 : .get_one::<bool>("hot-standby")
853 543 : .copied()
854 543 : .unwrap_or(false);
855 :
856 543 : let mode = match (lsn, hot_standby) {
857 49 : (Some(lsn), false) => ComputeMode::Static(lsn),
858 2 : (None, true) => ComputeMode::Replica,
859 492 : (None, false) => ComputeMode::Primary,
860 0 : (Some(_), true) => anyhow::bail!("cannot specify both lsn and hot-standby"),
861 : };
862 :
863 543 : match (mode, hot_standby) {
864 : (ComputeMode::Static(_), true) => {
865 0 : bail!("Cannot start a node in hot standby mode when it is already configured as a static replica")
866 : }
867 : (ComputeMode::Primary, true) => {
868 0 : bail!("Cannot start a node as a hot standby replica, it is already configured as primary node")
869 : }
870 543 : _ => {}
871 543 : }
872 543 :
873 543 : cplane.check_conflicting_endpoints(mode, tenant_id, timeline_id)?;
874 :
875 515 : cplane.new_endpoint(
876 515 : &endpoint_id,
877 515 : tenant_id,
878 515 : timeline_id,
879 515 : pg_port,
880 515 : http_port,
881 515 : pg_version,
882 515 : mode,
883 515 : )?;
884 : }
885 1380 : "start" => {
886 583 : let endpoint_id = sub_args
887 583 : .get_one::<String>("endpoint_id")
888 583 : .ok_or_else(|| anyhow!("No endpoint ID was provided to start"))?;
889 :
890 583 : let pageserver_id =
891 583 : if let Some(id_str) = sub_args.get_one::<String>("endpoint-pageserver-id") {
892 : Some(NodeId(
893 22 : id_str.parse().context("while parsing pageserver id")?,
894 : ))
895 : } else {
896 561 : None
897 : };
898 :
899 583 : let remote_ext_config = sub_args.get_one::<String>("remote-ext-config");
900 :
901 : // If --safekeepers argument is given, use only the listed safekeeper nodes.
902 583 : let safekeepers =
903 583 : if let Some(safekeepers_str) = sub_args.get_one::<String>("safekeepers") {
904 578 : let mut safekeepers: Vec<NodeId> = Vec::new();
905 725 : for sk_id in safekeepers_str.split(',').map(str::trim) {
906 725 : let sk_id = NodeId(u64::from_str(sk_id).map_err(|_| {
907 0 : anyhow!("invalid node ID \"{sk_id}\" in --safekeepers list")
908 725 : })?);
909 725 : safekeepers.push(sk_id);
910 : }
911 578 : safekeepers
912 : } else {
913 5 : env.safekeepers.iter().map(|sk| sk.id).collect()
914 : };
915 :
916 583 : let endpoint = cplane
917 583 : .endpoints
918 583 : .get(endpoint_id.as_str())
919 583 : .ok_or_else(|| anyhow::anyhow!("endpoint {endpoint_id} not found"))?;
920 :
921 583 : cplane.check_conflicting_endpoints(
922 583 : endpoint.mode,
923 583 : endpoint.tenant_id,
924 583 : endpoint.timeline_id,
925 583 : )?;
926 :
927 582 : let (pageservers, stripe_size) = if let Some(pageserver_id) = pageserver_id {
928 22 : let conf = env.get_pageserver_conf(pageserver_id).unwrap();
929 22 : let parsed = parse_host_port(&conf.listen_pg_addr).expect("Bad config");
930 22 : (
931 22 : vec![(parsed.0, parsed.1.unwrap_or(5432))],
932 22 : // If caller is telling us what pageserver to use, this is not a tenant which is
933 22 : // full managed by attachment service, therefore not sharded.
934 22 : ShardParameters::DEFAULT_STRIPE_SIZE,
935 22 : )
936 : } else {
937 : // Look up the currently attached location of the tenant, and its striping metadata,
938 : // to pass these on to postgres.
939 560 : let attachment_service = AttachmentService::from_env(env);
940 1680 : let locate_result = attachment_service.tenant_locate(endpoint.tenant_id).await?;
941 560 : let pageservers = locate_result
942 560 : .shards
943 560 : .into_iter()
944 575 : .map(|shard| {
945 575 : (
946 575 : Host::parse(&shard.listen_pg_addr)
947 575 : .expect("Attachment service reported bad hostname"),
948 575 : shard.listen_pg_port,
949 575 : )
950 575 : })
951 560 : .collect::<Vec<_>>();
952 560 : let stripe_size = locate_result.shard_params.stripe_size;
953 560 :
954 560 : (pageservers, stripe_size)
955 : };
956 582 : assert!(!pageservers.is_empty());
957 :
958 582 : let ps_conf = env.get_pageserver_conf(DEFAULT_PAGESERVER_ID)?;
959 582 : let auth_token = if matches!(ps_conf.pg_auth_type, AuthType::NeonJWT) {
960 15 : let claims = Claims::new(Some(endpoint.tenant_id), Scope::Tenant);
961 15 :
962 15 : Some(env.generate_auth_token(&claims)?)
963 : } else {
964 567 : None
965 : };
966 :
967 582 : println!("Starting existing endpoint {endpoint_id}...");
968 582 : endpoint
969 582 : .start(
970 582 : &auth_token,
971 582 : safekeepers,
972 582 : pageservers,
973 582 : remote_ext_config,
974 582 : stripe_size.0 as usize,
975 582 : )
976 9431 : .await?;
977 : }
978 797 : "reconfigure" => {
979 225 : let endpoint_id = sub_args
980 225 : .get_one::<String>("endpoint_id")
981 225 : .ok_or_else(|| anyhow!("No endpoint ID provided to reconfigure"))?;
982 225 : let endpoint = cplane
983 225 : .endpoints
984 225 : .get(endpoint_id.as_str())
985 225 : .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?;
986 225 : let pageservers =
987 225 : if let Some(id_str) = sub_args.get_one::<String>("endpoint-pageserver-id") {
988 213 : let ps_id = NodeId(id_str.parse().context("while parsing pageserver id")?);
989 213 : let pageserver = PageServerNode::from_env(env, env.get_pageserver_conf(ps_id)?);
990 213 : vec![(
991 213 : pageserver.pg_connection_config.host().clone(),
992 213 : pageserver.pg_connection_config.port(),
993 213 : )]
994 : } else {
995 12 : let attachment_service = AttachmentService::from_env(env);
996 12 : attachment_service
997 12 : .tenant_locate(endpoint.tenant_id)
998 36 : .await?
999 : .shards
1000 12 : .into_iter()
1001 60 : .map(|shard| {
1002 60 : (
1003 60 : Host::parse(&shard.listen_pg_addr)
1004 60 : .expect("Attachment service reported malformed host"),
1005 60 : shard.listen_pg_port,
1006 60 : )
1007 60 : })
1008 12 : .collect::<Vec<_>>()
1009 : };
1010 675 : endpoint.reconfigure(pageservers).await?;
1011 : }
1012 572 : "stop" => {
1013 572 : let endpoint_id = sub_args
1014 572 : .get_one::<String>("endpoint_id")
1015 572 : .ok_or_else(|| anyhow!("No endpoint ID was provided to stop"))?;
1016 572 : let destroy = sub_args.get_flag("destroy");
1017 572 : let mode = sub_args.get_one::<String>("mode").expect("has a default");
1018 :
1019 572 : let endpoint = cplane
1020 572 : .endpoints
1021 572 : .get(endpoint_id.as_str())
1022 572 : .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?;
1023 572 : endpoint.stop(mode, destroy)?;
1024 : }
1025 :
1026 0 : _ => bail!("Unexpected endpoint subcommand '{sub_name}'"),
1027 : }
1028 :
1029 1886 : Ok(())
1030 1923 : }
1031 :
1032 1 : fn handle_mappings(sub_match: &ArgMatches, env: &mut local_env::LocalEnv) -> Result<()> {
1033 1 : let (sub_name, sub_args) = match sub_match.subcommand() {
1034 1 : Some(ep_subcommand_data) => ep_subcommand_data,
1035 0 : None => bail!("no mappings subcommand provided"),
1036 : };
1037 :
1038 1 : match sub_name {
1039 1 : "map" => {
1040 1 : let branch_name = sub_args
1041 1 : .get_one::<String>("branch-name")
1042 1 : .expect("branch-name argument missing");
1043 1 :
1044 1 : let tenant_id = sub_args
1045 1 : .get_one::<String>("tenant-id")
1046 1 : .map(|x| TenantId::from_str(x))
1047 1 : .expect("tenant-id argument missing")
1048 1 : .expect("malformed tenant-id arg");
1049 1 :
1050 1 : let timeline_id = sub_args
1051 1 : .get_one::<String>("timeline-id")
1052 1 : .map(|x| TimelineId::from_str(x))
1053 1 : .expect("timeline-id argument missing")
1054 1 : .expect("malformed timeline-id arg");
1055 1 :
1056 1 : env.register_branch_mapping(branch_name.to_owned(), tenant_id, timeline_id)?;
1057 :
1058 1 : Ok(())
1059 : }
1060 0 : other => unimplemented!("mappings subcommand {other}"),
1061 : }
1062 1 : }
1063 :
1064 1249 : fn get_pageserver(env: &local_env::LocalEnv, args: &ArgMatches) -> Result<PageServerNode> {
1065 1249 : let node_id = if let Some(id_str) = args.get_one::<String>("pageserver-id") {
1066 1249 : NodeId(id_str.parse().context("while parsing pageserver id")?)
1067 : } else {
1068 0 : DEFAULT_PAGESERVER_ID
1069 : };
1070 :
1071 : Ok(PageServerNode::from_env(
1072 1249 : env,
1073 1249 : env.get_pageserver_conf(node_id)?,
1074 : ))
1075 1249 : }
1076 :
1077 1245 : async fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
1078 1245 : match sub_match.subcommand() {
1079 1245 : Some(("start", subcommand_args)) => {
1080 622 : let register = subcommand_args.get_one::<bool>("register").unwrap_or(&true);
1081 622 : if let Err(e) = get_pageserver(env, subcommand_args)?
1082 622 : .start(&pageserver_config_overrides(subcommand_args), *register)
1083 5653 : .await
1084 : {
1085 0 : eprintln!("pageserver start failed: {e}");
1086 0 : exit(1);
1087 622 : }
1088 : }
1089 :
1090 623 : Some(("stop", subcommand_args)) => {
1091 623 : let immediate = subcommand_args
1092 623 : .get_one::<String>("stop-mode")
1093 623 : .map(|s| s.as_str())
1094 623 : == Some("immediate");
1095 :
1096 623 : if let Err(e) = get_pageserver(env, subcommand_args)?.stop(immediate) {
1097 0 : eprintln!("pageserver stop failed: {}", e);
1098 0 : exit(1);
1099 622 : }
1100 : }
1101 :
1102 0 : Some(("restart", subcommand_args)) => {
1103 0 : let pageserver = get_pageserver(env, subcommand_args)?;
1104 : //TODO what shutdown strategy should we use here?
1105 0 : if let Err(e) = pageserver.stop(false) {
1106 0 : eprintln!("pageserver stop failed: {}", e);
1107 0 : exit(1);
1108 0 : }
1109 :
1110 0 : if let Err(e) = pageserver
1111 0 : .start(&pageserver_config_overrides(subcommand_args), false)
1112 0 : .await
1113 : {
1114 0 : eprintln!("pageserver start failed: {e}");
1115 0 : exit(1);
1116 0 : }
1117 : }
1118 :
1119 0 : Some(("set-state", subcommand_args)) => {
1120 0 : let pageserver = get_pageserver(env, subcommand_args)?;
1121 0 : let scheduling = subcommand_args.get_one("scheduling");
1122 0 : let availability = subcommand_args.get_one("availability");
1123 0 :
1124 0 : let attachment_service = AttachmentService::from_env(env);
1125 0 : attachment_service
1126 0 : .node_configure(NodeConfigureRequest {
1127 0 : node_id: pageserver.conf.id,
1128 0 : scheduling: scheduling.cloned(),
1129 0 : availability: availability.cloned(),
1130 0 : })
1131 0 : .await?;
1132 : }
1133 :
1134 0 : Some(("status", subcommand_args)) => {
1135 0 : match get_pageserver(env, subcommand_args)?.check_status().await {
1136 0 : Ok(_) => println!("Page server is up and running"),
1137 0 : Err(err) => {
1138 0 : eprintln!("Page server is not available: {}", err);
1139 0 : exit(1);
1140 : }
1141 : }
1142 : }
1143 :
1144 0 : Some((sub_name, _)) => bail!("Unexpected pageserver subcommand '{}'", sub_name),
1145 0 : None => bail!("no pageserver subcommand provided"),
1146 : }
1147 1244 : Ok(())
1148 1245 : }
1149 :
1150 728 : async fn handle_attachment_service(
1151 728 : sub_match: &ArgMatches,
1152 728 : env: &local_env::LocalEnv,
1153 728 : ) -> Result<()> {
1154 728 : let svc = AttachmentService::from_env(env);
1155 728 : match sub_match.subcommand() {
1156 728 : Some(("start", _start_match)) => {
1157 4770 : if let Err(e) = svc.start().await {
1158 0 : eprintln!("start failed: {e}");
1159 0 : exit(1);
1160 363 : }
1161 : }
1162 :
1163 365 : Some(("stop", stop_match)) => {
1164 365 : let immediate = stop_match
1165 365 : .get_one::<String>("stop-mode")
1166 365 : .map(|s| s.as_str())
1167 365 : == Some("immediate");
1168 :
1169 734 : if let Err(e) = svc.stop(immediate).await {
1170 0 : eprintln!("stop failed: {}", e);
1171 0 : exit(1);
1172 365 : }
1173 : }
1174 0 : Some((sub_name, _)) => bail!("Unexpected attachment_service subcommand '{}'", sub_name),
1175 0 : None => bail!("no attachment_service subcommand provided"),
1176 : }
1177 728 : Ok(())
1178 728 : }
1179 :
1180 1057 : fn get_safekeeper(env: &local_env::LocalEnv, id: NodeId) -> Result<SafekeeperNode> {
1181 1392 : if let Some(node) = env.safekeepers.iter().find(|node| node.id == id) {
1182 1057 : Ok(SafekeeperNode::from_env(env, node))
1183 : } else {
1184 0 : bail!("could not find safekeeper {id}")
1185 : }
1186 1057 : }
1187 :
1188 : // Get list of options to append to safekeeper command invocation.
1189 509 : fn safekeeper_extra_opts(init_match: &ArgMatches) -> Vec<String> {
1190 509 : init_match
1191 509 : .get_many::<String>("safekeeper-extra-opt")
1192 509 : .into_iter()
1193 509 : .flatten()
1194 509 : .map(|s| s.to_owned())
1195 509 : .collect()
1196 509 : }
1197 :
1198 1057 : async fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
1199 1057 : let (sub_name, sub_args) = match sub_match.subcommand() {
1200 1057 : Some(safekeeper_command_data) => safekeeper_command_data,
1201 0 : None => bail!("no safekeeper subcommand provided"),
1202 : };
1203 :
1204 : // All the commands take an optional safekeeper name argument
1205 1057 : let sk_id = if let Some(id_str) = sub_args.get_one::<String>("id") {
1206 1039 : NodeId(id_str.parse().context("while parsing safekeeper id")?)
1207 : } else {
1208 18 : DEFAULT_SAFEKEEPER_ID
1209 : };
1210 1057 : let safekeeper = get_safekeeper(env, sk_id)?;
1211 :
1212 1057 : match sub_name {
1213 1057 : "start" => {
1214 509 : let extra_opts = safekeeper_extra_opts(sub_args);
1215 :
1216 2060 : if let Err(e) = safekeeper.start(extra_opts).await {
1217 0 : eprintln!("safekeeper start failed: {}", e);
1218 0 : exit(1);
1219 509 : }
1220 : }
1221 :
1222 548 : "stop" => {
1223 548 : let immediate =
1224 548 : sub_args.get_one::<String>("stop-mode").map(|s| s.as_str()) == Some("immediate");
1225 :
1226 548 : if let Err(e) = safekeeper.stop(immediate) {
1227 0 : eprintln!("safekeeper stop failed: {}", e);
1228 0 : exit(1);
1229 548 : }
1230 : }
1231 :
1232 0 : "restart" => {
1233 0 : let immediate =
1234 0 : sub_args.get_one::<String>("stop-mode").map(|s| s.as_str()) == Some("immediate");
1235 :
1236 0 : if let Err(e) = safekeeper.stop(immediate) {
1237 0 : eprintln!("safekeeper stop failed: {}", e);
1238 0 : exit(1);
1239 0 : }
1240 0 :
1241 0 : let extra_opts = safekeeper_extra_opts(sub_args);
1242 0 : if let Err(e) = safekeeper.start(extra_opts).await {
1243 0 : eprintln!("safekeeper start failed: {}", e);
1244 0 : exit(1);
1245 0 : }
1246 : }
1247 :
1248 : _ => {
1249 0 : bail!("Unexpected safekeeper subcommand '{}'", sub_name)
1250 : }
1251 : }
1252 1057 : Ok(())
1253 1057 : }
1254 :
1255 3 : async fn handle_start_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> anyhow::Result<()> {
1256 3 : // Endpoints are not started automatically
1257 3 :
1258 10 : broker::start_broker_process(env).await?;
1259 :
1260 : // Only start the attachment service if the pageserver is configured to need it
1261 3 : if env.control_plane_api.is_some() {
1262 3 : let attachment_service = AttachmentService::from_env(env);
1263 38 : if let Err(e) = attachment_service.start().await {
1264 0 : eprintln!("attachment_service start failed: {:#}", e);
1265 0 : try_stop_all(env, true).await;
1266 0 : exit(1);
1267 3 : }
1268 0 : }
1269 :
1270 7 : for ps_conf in &env.pageservers {
1271 4 : let pageserver = PageServerNode::from_env(env, ps_conf);
1272 4 : if let Err(e) = pageserver
1273 4 : .start(&pageserver_config_overrides(sub_match), true)
1274 36 : .await
1275 : {
1276 0 : eprintln!("pageserver {} start failed: {:#}", ps_conf.id, e);
1277 0 : try_stop_all(env, true).await;
1278 0 : exit(1);
1279 4 : }
1280 : }
1281 :
1282 4 : for node in env.safekeepers.iter() {
1283 4 : let safekeeper = SafekeeperNode::from_env(env, node);
1284 16 : if let Err(e) = safekeeper.start(vec![]).await {
1285 0 : eprintln!("safekeeper {} start failed: {:#}", safekeeper.id, e);
1286 0 : try_stop_all(env, false).await;
1287 0 : exit(1);
1288 4 : }
1289 : }
1290 3 : Ok(())
1291 3 : }
1292 :
1293 3 : async fn handle_stop_all(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
1294 3 : let immediate =
1295 3 : sub_match.get_one::<String>("stop-mode").map(|s| s.as_str()) == Some("immediate");
1296 3 :
1297 6 : try_stop_all(env, immediate).await;
1298 :
1299 3 : Ok(())
1300 3 : }
1301 :
1302 3 : async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
1303 3 : // Stop all endpoints
1304 3 : match ComputeControlPlane::load(env.clone()) {
1305 3 : Ok(cplane) => {
1306 5 : for (_k, node) in cplane.endpoints {
1307 2 : if let Err(e) = node.stop(if immediate { "immediate" } else { "fast " }, false) {
1308 2 : eprintln!("postgres stop failed: {e:#}");
1309 2 : }
1310 : }
1311 : }
1312 0 : Err(e) => {
1313 0 : eprintln!("postgres stop failed, could not restore control plane data from env: {e:#}")
1314 : }
1315 : }
1316 :
1317 7 : for ps_conf in &env.pageservers {
1318 4 : let pageserver = PageServerNode::from_env(env, ps_conf);
1319 4 : if let Err(e) = pageserver.stop(immediate) {
1320 0 : eprintln!("pageserver {} stop failed: {:#}", ps_conf.id, e);
1321 4 : }
1322 : }
1323 :
1324 4 : for node in env.safekeepers.iter() {
1325 4 : let safekeeper = SafekeeperNode::from_env(env, node);
1326 4 : if let Err(e) = safekeeper.stop(immediate) {
1327 0 : eprintln!("safekeeper {} stop failed: {:#}", safekeeper.id, e);
1328 4 : }
1329 : }
1330 :
1331 3 : if let Err(e) = broker::stop_broker_process(env) {
1332 0 : eprintln!("neon broker stop failed: {e:#}");
1333 3 : }
1334 :
1335 3 : if env.control_plane_api.is_some() {
1336 3 : let attachment_service = AttachmentService::from_env(env);
1337 6 : if let Err(e) = attachment_service.stop(immediate).await {
1338 0 : eprintln!("attachment service stop failed: {e:#}");
1339 3 : }
1340 0 : }
1341 3 : }
1342 :
1343 6166 : fn cli() -> Command {
1344 6166 : let branch_name_arg = Arg::new("branch-name")
1345 6166 : .long("branch-name")
1346 6166 : .help("Name of the branch to be created or used as an alias for other services")
1347 6166 : .required(false);
1348 6166 :
1349 6166 : let endpoint_id_arg = Arg::new("endpoint_id")
1350 6166 : .help("Postgres endpoint id")
1351 6166 : .required(false);
1352 6166 :
1353 6166 : let safekeeper_id_arg = Arg::new("id").help("safekeeper id").required(false);
1354 6166 :
1355 6166 : // --id, when using a pageserver command
1356 6166 : let pageserver_id_arg = Arg::new("pageserver-id")
1357 6166 : .long("id")
1358 6166 : .global(true)
1359 6166 : .help("pageserver id")
1360 6166 : .required(false);
1361 6166 : // --pageserver-id when using a non-pageserver command
1362 6166 : let endpoint_pageserver_id_arg = Arg::new("endpoint-pageserver-id")
1363 6166 : .long("pageserver-id")
1364 6166 : .required(false);
1365 6166 :
1366 6166 : let safekeeper_extra_opt_arg = Arg::new("safekeeper-extra-opt")
1367 6166 : .short('e')
1368 6166 : .long("safekeeper-extra-opt")
1369 6166 : .num_args(1)
1370 6166 : .action(ArgAction::Append)
1371 6166 : .help("Additional safekeeper invocation options, e.g. -e=--http-auth-public-key-path=foo")
1372 6166 : .required(false);
1373 6166 :
1374 6166 : let tenant_id_arg = Arg::new("tenant-id")
1375 6166 : .long("tenant-id")
1376 6166 : .help("Tenant id. Represented as a hexadecimal string 32 symbols length")
1377 6166 : .required(false);
1378 6166 :
1379 6166 : let timeline_id_arg = Arg::new("timeline-id")
1380 6166 : .long("timeline-id")
1381 6166 : .help("Timeline id. Represented as a hexadecimal string 32 symbols length")
1382 6166 : .required(false);
1383 6166 :
1384 6166 : let pg_version_arg = Arg::new("pg-version")
1385 6166 : .long("pg-version")
1386 6166 : .help("Postgres version to use for the initial tenant")
1387 6166 : .required(false)
1388 6166 : .value_parser(value_parser!(u32))
1389 6166 : .default_value(DEFAULT_PG_VERSION);
1390 6166 :
1391 6166 : let pg_port_arg = Arg::new("pg-port")
1392 6166 : .long("pg-port")
1393 6166 : .required(false)
1394 6166 : .value_parser(value_parser!(u16))
1395 6166 : .value_name("pg-port");
1396 6166 :
1397 6166 : let http_port_arg = Arg::new("http-port")
1398 6166 : .long("http-port")
1399 6166 : .required(false)
1400 6166 : .value_parser(value_parser!(u16))
1401 6166 : .value_name("http-port");
1402 6166 :
1403 6166 : let safekeepers_arg = Arg::new("safekeepers")
1404 6166 : .long("safekeepers")
1405 6166 : .required(false)
1406 6166 : .value_name("safekeepers");
1407 6166 :
1408 6166 : let stop_mode_arg = Arg::new("stop-mode")
1409 6166 : .short('m')
1410 6166 : .value_parser(["fast", "immediate"])
1411 6166 : .default_value("fast")
1412 6166 : .help("If 'immediate', don't flush repository data at shutdown")
1413 6166 : .required(false)
1414 6166 : .value_name("stop-mode");
1415 6166 :
1416 6166 : let pageserver_config_args = Arg::new("pageserver-config-override")
1417 6166 : .long("pageserver-config-override")
1418 6166 : .num_args(1)
1419 6166 : .action(ArgAction::Append)
1420 6166 : .help("Additional pageserver's configuration options or overrides, refer to pageserver's 'config-override' CLI parameter docs for more")
1421 6166 : .required(false);
1422 6166 :
1423 6166 : let remote_ext_config_args = Arg::new("remote-ext-config")
1424 6166 : .long("remote-ext-config")
1425 6166 : .num_args(1)
1426 6166 : .help("Configure the remote extensions storage proxy gateway to request for extensions.")
1427 6166 : .required(false);
1428 6166 :
1429 6166 : let lsn_arg = Arg::new("lsn")
1430 6166 : .long("lsn")
1431 6166 : .help("Specify Lsn on the timeline to start from. By default, end of the timeline would be used.")
1432 6166 : .required(false);
1433 6166 :
1434 6166 : let hot_standby_arg = Arg::new("hot-standby")
1435 6166 : .value_parser(value_parser!(bool))
1436 6166 : .long("hot-standby")
1437 6166 : .help("If set, the node will be a hot replica on the specified timeline")
1438 6166 : .required(false);
1439 6166 :
1440 6166 : let force_arg = Arg::new("force")
1441 6166 : .value_parser(value_parser!(InitForceMode))
1442 6166 : .long("force")
1443 6166 : .default_value(
1444 6166 : InitForceMode::MustNotExist
1445 6166 : .to_possible_value()
1446 6166 : .unwrap()
1447 6166 : .get_name()
1448 6166 : .to_owned(),
1449 6166 : )
1450 6166 : .help("Force initialization even if the repository is not empty")
1451 6166 : .required(false);
1452 6166 :
1453 6166 : let num_pageservers_arg = Arg::new("num-pageservers")
1454 6166 : .value_parser(value_parser!(u16))
1455 6166 : .long("num-pageservers")
1456 6166 : .help("How many pageservers to create (default 1)")
1457 6166 : .required(false)
1458 6166 : .default_value("1");
1459 6166 :
1460 6166 : Command::new("Neon CLI")
1461 6166 : .arg_required_else_help(true)
1462 6166 : .version(GIT_VERSION)
1463 6166 : .subcommand(
1464 6166 : Command::new("init")
1465 6166 : .about("Initialize a new Neon repository, preparing configs for services to start with")
1466 6166 : .arg(pageserver_config_args.clone())
1467 6166 : .arg(num_pageservers_arg.clone())
1468 6166 : .arg(
1469 6166 : Arg::new("config")
1470 6166 : .long("config")
1471 6166 : .required(false)
1472 6166 : .value_parser(value_parser!(PathBuf))
1473 6166 : .value_name("config"),
1474 6166 : )
1475 6166 : .arg(pg_version_arg.clone())
1476 6166 : .arg(force_arg)
1477 6166 : )
1478 6166 : .subcommand(
1479 6166 : Command::new("timeline")
1480 6166 : .about("Manage timelines")
1481 6166 : .subcommand(Command::new("list")
1482 6166 : .about("List all timelines, available to this pageserver")
1483 6166 : .arg(tenant_id_arg.clone()))
1484 6166 : .subcommand(Command::new("branch")
1485 6166 : .about("Create a new timeline, using another timeline as a base, copying its data")
1486 6166 : .arg(tenant_id_arg.clone())
1487 6166 : .arg(branch_name_arg.clone())
1488 6166 : .arg(Arg::new("ancestor-branch-name").long("ancestor-branch-name")
1489 6166 : .help("Use last Lsn of another timeline (and its data) as base when creating the new timeline. The timeline gets resolved by its branch name.").required(false))
1490 6166 : .arg(Arg::new("ancestor-start-lsn").long("ancestor-start-lsn")
1491 6166 : .help("When using another timeline as base, use a specific Lsn in it instead of the latest one").required(false)))
1492 6166 : .subcommand(Command::new("create")
1493 6166 : .about("Create a new blank timeline")
1494 6166 : .arg(tenant_id_arg.clone())
1495 6166 : .arg(timeline_id_arg.clone())
1496 6166 : .arg(branch_name_arg.clone())
1497 6166 : .arg(pg_version_arg.clone())
1498 6166 : )
1499 6166 : .subcommand(Command::new("import")
1500 6166 : .about("Import timeline from basebackup directory")
1501 6166 : .arg(tenant_id_arg.clone())
1502 6166 : .arg(timeline_id_arg.clone())
1503 6166 : .arg(Arg::new("node-name").long("node-name")
1504 6166 : .help("Name to assign to the imported timeline"))
1505 6166 : .arg(Arg::new("base-tarfile")
1506 6166 : .long("base-tarfile")
1507 6166 : .value_parser(value_parser!(PathBuf))
1508 6166 : .help("Basebackup tarfile to import")
1509 6166 : )
1510 6166 : .arg(Arg::new("base-lsn").long("base-lsn")
1511 6166 : .help("Lsn the basebackup starts at"))
1512 6166 : .arg(Arg::new("wal-tarfile")
1513 6166 : .long("wal-tarfile")
1514 6166 : .value_parser(value_parser!(PathBuf))
1515 6166 : .help("Wal to add after base")
1516 6166 : )
1517 6166 : .arg(Arg::new("end-lsn").long("end-lsn")
1518 6166 : .help("Lsn the basebackup ends at"))
1519 6166 : .arg(pg_version_arg.clone())
1520 6166 : )
1521 6166 : ).subcommand(
1522 6166 : Command::new("tenant")
1523 6166 : .arg_required_else_help(true)
1524 6166 : .about("Manage tenants")
1525 6166 : .subcommand(Command::new("list"))
1526 6166 : .subcommand(Command::new("create")
1527 6166 : .arg(tenant_id_arg.clone())
1528 6166 : .arg(timeline_id_arg.clone().help("Use a specific timeline id when creating a tenant and its initial timeline"))
1529 6166 : .arg(Arg::new("config").short('c').num_args(1).action(ArgAction::Append).required(false))
1530 6166 : .arg(pg_version_arg.clone())
1531 6166 : .arg(Arg::new("set-default").long("set-default").action(ArgAction::SetTrue).required(false)
1532 6166 : .help("Use this tenant in future CLI commands where tenant_id is needed, but not specified"))
1533 6166 : .arg(Arg::new("shard-count").value_parser(value_parser!(u8)).long("shard-count").action(ArgAction::Set).help("Number of shards in the new tenant (default 1)"))
1534 6166 : .arg(Arg::new("shard-stripe-size").value_parser(value_parser!(u32)).long("shard-stripe-size").action(ArgAction::Set).help("Sharding stripe size in pages"))
1535 6166 : )
1536 6166 : .subcommand(Command::new("set-default").arg(tenant_id_arg.clone().required(true))
1537 6166 : .about("Set a particular tenant as default in future CLI commands where tenant_id is needed, but not specified"))
1538 6166 : .subcommand(Command::new("config")
1539 6166 : .arg(tenant_id_arg.clone())
1540 6166 : .arg(Arg::new("config").short('c').num_args(1).action(ArgAction::Append).required(false)))
1541 6166 : .subcommand(Command::new("migrate")
1542 6166 : .about("Migrate a tenant from one pageserver to another")
1543 6166 : .arg(tenant_id_arg.clone())
1544 6166 : .arg(pageserver_id_arg.clone()))
1545 6166 : .subcommand(Command::new("status")
1546 6166 : .about("Human readable summary of the tenant's shards and attachment locations")
1547 6166 : .arg(tenant_id_arg.clone()))
1548 6166 : .subcommand(Command::new("shard-split")
1549 6166 : .about("Increase the number of shards in the tenant")
1550 6166 : .arg(tenant_id_arg.clone())
1551 6166 : .arg(Arg::new("shard-count").value_parser(value_parser!(u8)).long("shard-count").action(ArgAction::Set).help("Number of shards in the new tenant (default 1)"))
1552 6166 : )
1553 6166 : )
1554 6166 : .subcommand(
1555 6166 : Command::new("pageserver")
1556 6166 : .arg_required_else_help(true)
1557 6166 : .about("Manage pageserver")
1558 6166 : .arg(pageserver_id_arg)
1559 6166 : .subcommand(Command::new("status"))
1560 6166 : .subcommand(Command::new("start")
1561 6166 : .about("Start local pageserver")
1562 6166 : .arg(pageserver_config_args.clone()).arg(Arg::new("register")
1563 6166 : .long("register")
1564 6166 : .default_value("true").required(false)
1565 6166 : .value_parser(value_parser!(bool))
1566 6166 : .value_name("register"))
1567 6166 : )
1568 6166 : .subcommand(Command::new("stop")
1569 6166 : .about("Stop local pageserver")
1570 6166 : .arg(stop_mode_arg.clone())
1571 6166 : )
1572 6166 : .subcommand(Command::new("restart")
1573 6166 : .about("Restart local pageserver")
1574 6166 : .arg(pageserver_config_args.clone())
1575 6166 : )
1576 6166 : .subcommand(Command::new("set-state")
1577 6166 : .arg(Arg::new("availability").value_parser(value_parser!(NodeAvailability)).long("availability").action(ArgAction::Set).help("Availability state: offline,active"))
1578 6166 : .arg(Arg::new("scheduling").value_parser(value_parser!(NodeSchedulingPolicy)).long("scheduling").action(ArgAction::Set).help("Scheduling state: draining,pause,filling,active"))
1579 6166 : .about("Set scheduling or availability state of pageserver node")
1580 6166 : .arg(pageserver_config_args.clone())
1581 6166 : )
1582 6166 : )
1583 6166 : .subcommand(
1584 6166 : Command::new("attachment_service")
1585 6166 : .arg_required_else_help(true)
1586 6166 : .about("Manage attachment_service")
1587 6166 : .subcommand(Command::new("start").about("Start local pageserver").arg(pageserver_config_args.clone()))
1588 6166 : .subcommand(Command::new("stop").about("Stop local pageserver")
1589 6166 : .arg(stop_mode_arg.clone()))
1590 6166 : )
1591 6166 : .subcommand(
1592 6166 : Command::new("safekeeper")
1593 6166 : .arg_required_else_help(true)
1594 6166 : .about("Manage safekeepers")
1595 6166 : .subcommand(Command::new("start")
1596 6166 : .about("Start local safekeeper")
1597 6166 : .arg(safekeeper_id_arg.clone())
1598 6166 : .arg(safekeeper_extra_opt_arg.clone())
1599 6166 : )
1600 6166 : .subcommand(Command::new("stop")
1601 6166 : .about("Stop local safekeeper")
1602 6166 : .arg(safekeeper_id_arg.clone())
1603 6166 : .arg(stop_mode_arg.clone())
1604 6166 : )
1605 6166 : .subcommand(Command::new("restart")
1606 6166 : .about("Restart local safekeeper")
1607 6166 : .arg(safekeeper_id_arg)
1608 6166 : .arg(stop_mode_arg.clone())
1609 6166 : .arg(safekeeper_extra_opt_arg)
1610 6166 : )
1611 6166 : )
1612 6166 : .subcommand(
1613 6166 : Command::new("endpoint")
1614 6166 : .arg_required_else_help(true)
1615 6166 : .about("Manage postgres instances")
1616 6166 : .subcommand(Command::new("list").arg(tenant_id_arg.clone()))
1617 6166 : .subcommand(Command::new("create")
1618 6166 : .about("Create a compute endpoint")
1619 6166 : .arg(endpoint_id_arg.clone())
1620 6166 : .arg(branch_name_arg.clone())
1621 6166 : .arg(tenant_id_arg.clone())
1622 6166 : .arg(lsn_arg.clone())
1623 6166 : .arg(pg_port_arg.clone())
1624 6166 : .arg(http_port_arg.clone())
1625 6166 : .arg(endpoint_pageserver_id_arg.clone())
1626 6166 : .arg(
1627 6166 : Arg::new("config-only")
1628 6166 : .help("Don't do basebackup, create endpoint directory with only config files")
1629 6166 : .long("config-only")
1630 6166 : .required(false))
1631 6166 : .arg(pg_version_arg.clone())
1632 6166 : .arg(hot_standby_arg.clone())
1633 6166 : )
1634 6166 : .subcommand(Command::new("start")
1635 6166 : .about("Start postgres.\n If the endpoint doesn't exist yet, it is created.")
1636 6166 : .arg(endpoint_id_arg.clone())
1637 6166 : .arg(endpoint_pageserver_id_arg.clone())
1638 6166 : .arg(safekeepers_arg)
1639 6166 : .arg(remote_ext_config_args)
1640 6166 : )
1641 6166 : .subcommand(Command::new("reconfigure")
1642 6166 : .about("Reconfigure the endpoint")
1643 6166 : .arg(endpoint_pageserver_id_arg)
1644 6166 : .arg(endpoint_id_arg.clone())
1645 6166 : .arg(tenant_id_arg.clone())
1646 6166 : )
1647 6166 : .subcommand(
1648 6166 : Command::new("stop")
1649 6166 : .arg(endpoint_id_arg)
1650 6166 : .arg(
1651 6166 : Arg::new("destroy")
1652 6166 : .help("Also delete data directory (now optional, should be default in future)")
1653 6166 : .long("destroy")
1654 6166 : .action(ArgAction::SetTrue)
1655 6166 : .required(false)
1656 6166 : )
1657 6166 : .arg(
1658 6166 : Arg::new("mode")
1659 6166 : .help("Postgres shutdown mode, passed to \"pg_ctl -m <mode>\"")
1660 6166 : .long("mode")
1661 6166 : .action(ArgAction::Set)
1662 6166 : .required(false)
1663 6166 : .value_parser(["smart", "fast", "immediate"])
1664 6166 : .default_value("fast")
1665 6166 : )
1666 6166 : )
1667 6166 :
1668 6166 : )
1669 6166 : .subcommand(
1670 6166 : Command::new("mappings")
1671 6166 : .arg_required_else_help(true)
1672 6166 : .about("Manage neon_local branch name mappings")
1673 6166 : .subcommand(
1674 6166 : Command::new("map")
1675 6166 : .about("Create new mapping which cannot exist already")
1676 6166 : .arg(branch_name_arg.clone())
1677 6166 : .arg(tenant_id_arg.clone())
1678 6166 : .arg(timeline_id_arg.clone())
1679 6166 : )
1680 6166 : )
1681 6166 : // Obsolete old name for 'endpoint'. We now just print an error if it's used.
1682 6166 : .subcommand(
1683 6166 : Command::new("pg")
1684 6166 : .hide(true)
1685 6166 : .arg(Arg::new("ignore-rest").allow_hyphen_values(true).num_args(0..).required(false))
1686 6166 : .trailing_var_arg(true)
1687 6166 : )
1688 6166 : .subcommand(
1689 6166 : Command::new("start")
1690 6166 : .about("Start page server and safekeepers")
1691 6166 : .arg(pageserver_config_args)
1692 6166 : )
1693 6166 : .subcommand(
1694 6166 : Command::new("stop")
1695 6166 : .about("Stop page server and safekeepers")
1696 6166 : .arg(stop_mode_arg)
1697 6166 : )
1698 6166 : }
1699 :
1700 2 : #[test]
1701 2 : fn verify_cli() {
1702 2 : cli().debug_assert();
1703 2 : }
|
