LCOV - code coverage report
Current view: top level - control_plane/src - local_env.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 0.0 % 606 0
Test Date: 2025-03-12 16:10:49 Functions: 0.0 % 142 0

            Line data    Source code
       1              : //! This module is responsible for locating and loading paths in a local setup.
       2              : //!
       3              : //! Now it also provides init method which acts like a stub for proper installation
       4              : //! script which will use local paths.
       5              : 
       6              : use std::collections::HashMap;
       7              : use std::net::{IpAddr, Ipv4Addr, SocketAddr};
       8              : use std::path::{Path, PathBuf};
       9              : use std::process::{Command, Stdio};
      10              : use std::time::Duration;
      11              : use std::{env, fs};
      12              : 
      13              : use anyhow::{Context, bail};
      14              : use clap::ValueEnum;
      15              : use postgres_backend::AuthType;
      16              : use reqwest::Url;
      17              : use serde::{Deserialize, Serialize};
      18              : use utils::auth::{Claims, encode_from_key_file};
      19              : use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
      20              : 
      21              : use crate::pageserver::{PAGESERVER_REMOTE_STORAGE_DIR, PageServerNode};
      22              : use crate::safekeeper::SafekeeperNode;
      23              : 
      24              : pub const DEFAULT_PG_VERSION: u32 = 16;
      25              : 
      26              : //
      27              : // This data structures represents neon_local CLI config
      28              : //
      29              : // It is deserialized from the .neon/config file, or the config file passed
      30              : // to 'neon_local init --config=<path>' option. See control_plane/simple.conf for
      31              : // an example.
      32              : //
      33              : #[derive(PartialEq, Eq, Clone, Debug)]
      34              : pub struct LocalEnv {
      35              :     // Base directory for all the nodes (the pageserver, safekeepers and
      36              :     // compute endpoints).
      37              :     //
      38              :     // This is not stored in the config file. Rather, this is the path where the
      39              :     // config file itself is. It is read from the NEON_REPO_DIR env variable which
      40              :     // must be an absolute path. If the env var is not set, $PWD/.neon is used.
      41              :     pub base_data_dir: PathBuf,
      42              : 
      43              :     // Path to postgres distribution. It's expected that "bin", "include",
      44              :     // "lib", "share" from postgres distribution are there. If at some point
      45              :     // in time we will be able to run against vanilla postgres we may split that
      46              :     // to four separate paths and match OS-specific installation layout.
      47              :     pub pg_distrib_dir: PathBuf,
      48              : 
      49              :     // Path to pageserver binary.
      50              :     pub neon_distrib_dir: PathBuf,
      51              : 
      52              :     // Default tenant ID to use with the 'neon_local' command line utility, when
      53              :     // --tenant_id is not explicitly specified.
      54              :     pub default_tenant_id: Option<TenantId>,
      55              : 
      56              :     // used to issue tokens during e.g pg start
      57              :     pub private_key_path: PathBuf,
      58              : 
      59              :     pub broker: NeonBroker,
      60              : 
      61              :     // Configuration for the storage controller (1 per neon_local environment)
      62              :     pub storage_controller: NeonStorageControllerConf,
      63              : 
      64              :     /// This Vec must always contain at least one pageserver
      65              :     /// Populdated by [`Self::load_config`] from the individual `pageserver.toml`s.
      66              :     /// NB: not used anymore except for informing users that they need to change their `.neon/config`.
      67              :     pub pageservers: Vec<PageServerConf>,
      68              : 
      69              :     pub safekeepers: Vec<SafekeeperConf>,
      70              : 
      71              :     // Control plane upcall API for pageserver: if None, we will not run storage_controller  If set, this will
      72              :     // be propagated into each pageserver's configuration.
      73              :     pub control_plane_api: Url,
      74              : 
      75              :     // Control plane upcall APIs for storage controller.  If set, this will be propagated into the
      76              :     // storage controller's configuration.
      77              :     pub control_plane_hooks_api: Option<Url>,
      78              : 
      79              :     /// Keep human-readable aliases in memory (and persist them to config), to hide ZId hex strings from the user.
      80              :     // A `HashMap<String, HashMap<TenantId, TimelineId>>` would be more appropriate here,
      81              :     // but deserialization into a generic toml object as `toml::Value::try_from` fails with an error.
      82              :     // https://toml.io/en/v1.0.0 does not contain a concept of "a table inside another table".
      83              :     pub branch_name_mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
      84              : 
      85              :     /// Flag to generate SSL certificates for components that need it.
      86              :     /// Also generates root CA certificate that is used to sign all other certificates.
      87              :     pub generate_local_ssl_certs: bool,
      88              : }
      89              : 
      90              : /// On-disk state stored in `.neon/config`.
      91            0 : #[derive(PartialEq, Eq, Clone, Debug, Default, Serialize, Deserialize)]
      92              : #[serde(default, deny_unknown_fields)]
      93              : pub struct OnDiskConfig {
      94              :     pub pg_distrib_dir: PathBuf,
      95              :     pub neon_distrib_dir: PathBuf,
      96              :     pub default_tenant_id: Option<TenantId>,
      97              :     pub private_key_path: PathBuf,
      98              :     pub broker: NeonBroker,
      99              :     pub storage_controller: NeonStorageControllerConf,
     100              :     #[serde(
     101              :         skip_serializing,
     102              :         deserialize_with = "fail_if_pageservers_field_specified"
     103              :     )]
     104              :     pub pageservers: Vec<PageServerConf>,
     105              :     pub safekeepers: Vec<SafekeeperConf>,
     106              :     pub control_plane_api: Option<Url>,
     107              :     pub control_plane_hooks_api: Option<Url>,
     108              :     pub control_plane_compute_hook_api: Option<Url>,
     109              :     branch_name_mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
     110              :     // Note: skip serializing because in compat tests old storage controller fails
     111              :     // to load new config file. May be removed after this field is in release branch.
     112              :     #[serde(skip_serializing_if = "std::ops::Not::not")]
     113              :     pub generate_local_ssl_certs: bool,
     114              : }
     115              : 
     116            0 : fn fail_if_pageservers_field_specified<'de, D>(_: D) -> Result<Vec<PageServerConf>, D::Error>
     117            0 : where
     118            0 :     D: serde::Deserializer<'de>,
     119            0 : {
     120            0 :     Err(serde::de::Error::custom(
     121            0 :         "The 'pageservers' field is no longer used; pageserver.toml is now authoritative; \
     122            0 :          Please remove the `pageservers` from your .neon/config.",
     123            0 :     ))
     124            0 : }
     125              : 
     126              : /// The description of the neon_local env to be initialized by `neon_local init --config`.
     127            0 : #[derive(Clone, Debug, Deserialize)]
     128              : #[serde(deny_unknown_fields)]
     129              : pub struct NeonLocalInitConf {
     130              :     // TODO: do we need this? Seems unused
     131              :     pub pg_distrib_dir: Option<PathBuf>,
     132              :     // TODO: do we need this? Seems unused
     133              :     pub neon_distrib_dir: Option<PathBuf>,
     134              :     pub default_tenant_id: TenantId,
     135              :     pub broker: NeonBroker,
     136              :     pub storage_controller: Option<NeonStorageControllerConf>,
     137              :     pub pageservers: Vec<NeonLocalInitPageserverConf>,
     138              :     pub safekeepers: Vec<SafekeeperConf>,
     139              :     pub control_plane_api: Option<Url>,
     140              :     pub control_plane_hooks_api: Option<Url>,
     141              :     pub generate_local_ssl_certs: bool,
     142              : }
     143              : 
     144              : /// Broker config for cluster internal communication.
     145            0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
     146              : #[serde(default)]
     147              : pub struct NeonBroker {
     148              :     /// Broker listen address for storage nodes coordination, e.g. '127.0.0.1:50051'.
     149              :     pub listen_addr: SocketAddr,
     150              : }
     151              : 
     152              : /// Broker config for cluster internal communication.
     153            0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
     154              : #[serde(default)]
     155              : pub struct NeonStorageControllerConf {
     156              :     /// Heartbeat timeout before marking a node offline
     157              :     #[serde(with = "humantime_serde")]
     158              :     pub max_offline: Duration,
     159              : 
     160              :     #[serde(with = "humantime_serde")]
     161              :     pub max_warming_up: Duration,
     162              : 
     163              :     pub start_as_candidate: bool,
     164              : 
     165              :     /// Database url used when running multiple storage controller instances
     166              :     pub database_url: Option<SocketAddr>,
     167              : 
     168              :     /// Threshold for auto-splitting a tenant into shards
     169              :     pub split_threshold: Option<u64>,
     170              : 
     171              :     pub max_secondary_lag_bytes: Option<u64>,
     172              : 
     173              :     #[serde(with = "humantime_serde")]
     174              :     pub heartbeat_interval: Duration,
     175              : 
     176              :     #[serde(with = "humantime_serde")]
     177              :     pub long_reconcile_threshold: Option<Duration>,
     178              : 
     179              :     #[serde(default)]
     180              :     pub use_https_pageserver_api: bool,
     181              : 
     182              :     pub timelines_onto_safekeepers: bool,
     183              : }
     184              : 
     185              : impl NeonStorageControllerConf {
     186              :     // Use a shorter pageserver unavailability interval than the default to speed up tests.
     187              :     const DEFAULT_MAX_OFFLINE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);
     188              : 
     189              :     const DEFAULT_MAX_WARMING_UP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);
     190              : 
     191              :     // Very tight heartbeat interval to speed up tests
     192              :     const DEFAULT_HEARTBEAT_INTERVAL: std::time::Duration = std::time::Duration::from_millis(1000);
     193              : }
     194              : 
     195              : impl Default for NeonStorageControllerConf {
     196            0 :     fn default() -> Self {
     197            0 :         Self {
     198            0 :             max_offline: Self::DEFAULT_MAX_OFFLINE_INTERVAL,
     199            0 :             max_warming_up: Self::DEFAULT_MAX_WARMING_UP_INTERVAL,
     200            0 :             start_as_candidate: false,
     201            0 :             database_url: None,
     202            0 :             split_threshold: None,
     203            0 :             max_secondary_lag_bytes: None,
     204            0 :             heartbeat_interval: Self::DEFAULT_HEARTBEAT_INTERVAL,
     205            0 :             long_reconcile_threshold: None,
     206            0 :             use_https_pageserver_api: false,
     207            0 :             timelines_onto_safekeepers: false,
     208            0 :         }
     209            0 :     }
     210              : }
     211              : 
     212              : // Dummy Default impl to satisfy Deserialize derive.
     213              : impl Default for NeonBroker {
     214            0 :     fn default() -> Self {
     215            0 :         NeonBroker {
     216            0 :             listen_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0),
     217            0 :         }
     218            0 :     }
     219              : }
     220              : 
     221              : impl NeonBroker {
     222            0 :     pub fn client_url(&self) -> Url {
     223            0 :         Url::parse(&format!("http://{}", self.listen_addr)).expect("failed to construct url")
     224            0 :     }
     225              : }
     226              : 
     227              : // neon_local needs to know this subset of pageserver configuration.
     228              : // For legacy reasons, this information is duplicated from `pageserver.toml` into `.neon/config`.
     229              : // It can get stale if `pageserver.toml` is changed.
     230              : // TODO(christian): don't store this at all in `.neon/config`, always load it from `pageserver.toml`
     231            0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
     232              : #[serde(default, deny_unknown_fields)]
     233              : pub struct PageServerConf {
     234              :     pub id: NodeId,
     235              :     pub listen_pg_addr: String,
     236              :     pub listen_http_addr: String,
     237              :     pub listen_https_addr: Option<String>,
     238              :     pub pg_auth_type: AuthType,
     239              :     pub http_auth_type: AuthType,
     240              :     pub no_sync: bool,
     241              : }
     242              : 
     243              : impl Default for PageServerConf {
     244            0 :     fn default() -> Self {
     245            0 :         Self {
     246            0 :             id: NodeId(0),
     247            0 :             listen_pg_addr: String::new(),
     248            0 :             listen_http_addr: String::new(),
     249            0 :             listen_https_addr: None,
     250            0 :             pg_auth_type: AuthType::Trust,
     251            0 :             http_auth_type: AuthType::Trust,
     252            0 :             no_sync: false,
     253            0 :         }
     254            0 :     }
     255              : }
     256              : 
     257              : /// The toml that can be passed to `neon_local init --config`.
     258              : /// This is a subset of the `pageserver.toml` configuration.
     259              : // TODO(christian): use pageserver_api::config::ConfigToml (PR #7656)
     260            0 : #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
     261              : pub struct NeonLocalInitPageserverConf {
     262              :     pub id: NodeId,
     263              :     pub listen_pg_addr: String,
     264              :     pub listen_http_addr: String,
     265              :     pub listen_https_addr: Option<String>,
     266              :     pub pg_auth_type: AuthType,
     267              :     pub http_auth_type: AuthType,
     268              :     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     269              :     pub no_sync: bool,
     270              :     #[serde(flatten)]
     271              :     pub other: HashMap<String, toml::Value>,
     272              : }
     273              : 
     274              : impl From<&NeonLocalInitPageserverConf> for PageServerConf {
     275            0 :     fn from(conf: &NeonLocalInitPageserverConf) -> Self {
     276            0 :         let NeonLocalInitPageserverConf {
     277            0 :             id,
     278            0 :             listen_pg_addr,
     279            0 :             listen_http_addr,
     280            0 :             listen_https_addr,
     281            0 :             pg_auth_type,
     282            0 :             http_auth_type,
     283            0 :             no_sync,
     284            0 :             other: _,
     285            0 :         } = conf;
     286            0 :         Self {
     287            0 :             id: *id,
     288            0 :             listen_pg_addr: listen_pg_addr.clone(),
     289            0 :             listen_http_addr: listen_http_addr.clone(),
     290            0 :             listen_https_addr: listen_https_addr.clone(),
     291            0 :             pg_auth_type: *pg_auth_type,
     292            0 :             http_auth_type: *http_auth_type,
     293            0 :             no_sync: *no_sync,
     294            0 :         }
     295            0 :     }
     296              : }
     297              : 
     298            0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
     299              : #[serde(default)]
     300              : pub struct SafekeeperConf {
     301              :     pub id: NodeId,
     302              :     pub pg_port: u16,
     303              :     pub pg_tenant_only_port: Option<u16>,
     304              :     pub http_port: u16,
     305              :     pub sync: bool,
     306              :     pub remote_storage: Option<String>,
     307              :     pub backup_threads: Option<u32>,
     308              :     pub auth_enabled: bool,
     309              :     pub listen_addr: Option<String>,
     310              : }
     311              : 
     312              : impl Default for SafekeeperConf {
     313            0 :     fn default() -> Self {
     314            0 :         Self {
     315            0 :             id: NodeId(0),
     316            0 :             pg_port: 0,
     317            0 :             pg_tenant_only_port: None,
     318            0 :             http_port: 0,
     319            0 :             sync: true,
     320            0 :             remote_storage: None,
     321            0 :             backup_threads: None,
     322            0 :             auth_enabled: false,
     323            0 :             listen_addr: None,
     324            0 :         }
     325            0 :     }
     326              : }
     327              : 
     328              : #[derive(Clone, Copy)]
     329              : pub enum InitForceMode {
     330              :     MustNotExist,
     331              :     EmptyDirOk,
     332              :     RemoveAllContents,
     333              : }
     334              : 
     335              : impl ValueEnum for InitForceMode {
     336            0 :     fn value_variants<'a>() -> &'a [Self] {
     337            0 :         &[
     338            0 :             Self::MustNotExist,
     339            0 :             Self::EmptyDirOk,
     340            0 :             Self::RemoveAllContents,
     341            0 :         ]
     342            0 :     }
     343              : 
     344            0 :     fn to_possible_value(&self) -> Option<clap::builder::PossibleValue> {
     345            0 :         Some(clap::builder::PossibleValue::new(match self {
     346            0 :             InitForceMode::MustNotExist => "must-not-exist",
     347            0 :             InitForceMode::EmptyDirOk => "empty-dir-ok",
     348            0 :             InitForceMode::RemoveAllContents => "remove-all-contents",
     349              :         }))
     350            0 :     }
     351              : }
     352              : 
     353              : impl SafekeeperConf {
     354              :     /// Compute is served by port on which only tenant scoped tokens allowed, if
     355              :     /// it is configured.
     356            0 :     pub fn get_compute_port(&self) -> u16 {
     357            0 :         self.pg_tenant_only_port.unwrap_or(self.pg_port)
     358            0 :     }
     359              : }
     360              : 
     361              : impl LocalEnv {
     362            0 :     pub fn pg_distrib_dir_raw(&self) -> PathBuf {
     363            0 :         self.pg_distrib_dir.clone()
     364            0 :     }
     365              : 
     366            0 :     pub fn pg_distrib_dir(&self, pg_version: u32) -> anyhow::Result<PathBuf> {
     367            0 :         let path = self.pg_distrib_dir.clone();
     368            0 : 
     369            0 :         #[allow(clippy::manual_range_patterns)]
     370            0 :         match pg_version {
     371            0 :             14 | 15 | 16 | 17 => Ok(path.join(format!("v{pg_version}"))),
     372            0 :             _ => bail!("Unsupported postgres version: {}", pg_version),
     373              :         }
     374            0 :     }
     375              : 
     376            0 :     pub fn pg_dir(&self, pg_version: u32, dir_name: &str) -> anyhow::Result<PathBuf> {
     377            0 :         Ok(self.pg_distrib_dir(pg_version)?.join(dir_name))
     378            0 :     }
     379              : 
     380            0 :     pub fn pg_bin_dir(&self, pg_version: u32) -> anyhow::Result<PathBuf> {
     381            0 :         self.pg_dir(pg_version, "bin")
     382            0 :     }
     383              : 
     384            0 :     pub fn pg_lib_dir(&self, pg_version: u32) -> anyhow::Result<PathBuf> {
     385            0 :         self.pg_dir(pg_version, "lib")
     386            0 :     }
     387              : 
     388            0 :     pub fn pageserver_bin(&self) -> PathBuf {
     389            0 :         self.neon_distrib_dir.join("pageserver")
     390            0 :     }
     391              : 
     392            0 :     pub fn storage_controller_bin(&self) -> PathBuf {
     393            0 :         // Irrespective of configuration, storage controller binary is always
     394            0 :         // run from the same location as neon_local.  This means that for compatibility
     395            0 :         // tests that run old pageserver/safekeeper, they still run latest storage controller.
     396            0 :         let neon_local_bin_dir = env::current_exe().unwrap().parent().unwrap().to_owned();
     397            0 :         neon_local_bin_dir.join("storage_controller")
     398            0 :     }
     399              : 
     400            0 :     pub fn safekeeper_bin(&self) -> PathBuf {
     401            0 :         self.neon_distrib_dir.join("safekeeper")
     402            0 :     }
     403              : 
     404            0 :     pub fn storage_broker_bin(&self) -> PathBuf {
     405            0 :         self.neon_distrib_dir.join("storage_broker")
     406            0 :     }
     407              : 
     408            0 :     pub fn endpoints_path(&self) -> PathBuf {
     409            0 :         self.base_data_dir.join("endpoints")
     410            0 :     }
     411              : 
     412            0 :     pub fn pageserver_data_dir(&self, pageserver_id: NodeId) -> PathBuf {
     413            0 :         self.base_data_dir
     414            0 :             .join(format!("pageserver_{pageserver_id}"))
     415            0 :     }
     416              : 
     417            0 :     pub fn safekeeper_data_dir(&self, data_dir_name: &str) -> PathBuf {
     418            0 :         self.base_data_dir.join("safekeepers").join(data_dir_name)
     419            0 :     }
     420              : 
     421            0 :     pub fn get_pageserver_conf(&self, id: NodeId) -> anyhow::Result<&PageServerConf> {
     422            0 :         if let Some(conf) = self.pageservers.iter().find(|node| node.id == id) {
     423            0 :             Ok(conf)
     424              :         } else {
     425            0 :             let have_ids = self
     426            0 :                 .pageservers
     427            0 :                 .iter()
     428            0 :                 .map(|node| format!("{}:{}", node.id, node.listen_http_addr))
     429            0 :                 .collect::<Vec<_>>();
     430            0 :             let joined = have_ids.join(",");
     431            0 :             bail!("could not find pageserver {id}, have ids {joined}")
     432              :         }
     433            0 :     }
     434              : 
     435            0 :     pub fn ssl_ca_cert_path(&self) -> Option<PathBuf> {
     436            0 :         if self.generate_local_ssl_certs {
     437            0 :             Some(self.base_data_dir.join("rootCA.crt"))
     438              :         } else {
     439            0 :             None
     440              :         }
     441            0 :     }
     442              : 
     443            0 :     pub fn ssl_ca_key_path(&self) -> Option<PathBuf> {
     444            0 :         if self.generate_local_ssl_certs {
     445            0 :             Some(self.base_data_dir.join("rootCA.key"))
     446              :         } else {
     447            0 :             None
     448              :         }
     449            0 :     }
     450              : 
     451            0 :     pub fn generate_ssl_ca_cert(&self) -> anyhow::Result<()> {
     452            0 :         let cert_path = self.ssl_ca_cert_path().unwrap();
     453            0 :         let key_path = self.ssl_ca_key_path().unwrap();
     454            0 :         if !fs::exists(cert_path.as_path())? {
     455            0 :             generate_ssl_ca_cert(cert_path.as_path(), key_path.as_path())?;
     456            0 :         }
     457            0 :         Ok(())
     458            0 :     }
     459              : 
     460            0 :     pub fn generate_ssl_cert(&self, cert_path: &Path, key_path: &Path) -> anyhow::Result<()> {
     461            0 :         self.generate_ssl_ca_cert()?;
     462            0 :         generate_ssl_cert(
     463            0 :             cert_path,
     464            0 :             key_path,
     465            0 :             self.ssl_ca_cert_path().unwrap().as_path(),
     466            0 :             self.ssl_ca_key_path().unwrap().as_path(),
     467            0 :         )
     468            0 :     }
     469              : 
     470              :     /// Inspect the base data directory and extract the instance id and instance directory path
     471              :     /// for all storage controller instances
     472            0 :     pub async fn storage_controller_instances(&self) -> std::io::Result<Vec<(u8, PathBuf)>> {
     473            0 :         let mut instances = Vec::default();
     474              : 
     475            0 :         let dir = std::fs::read_dir(self.base_data_dir.clone())?;
     476            0 :         for dentry in dir {
     477            0 :             let dentry = dentry?;
     478            0 :             let is_dir = dentry.metadata()?.is_dir();
     479            0 :             let filename = dentry.file_name().into_string().unwrap();
     480            0 :             let parsed_instance_id = match filename.strip_prefix("storage_controller_") {
     481            0 :                 Some(suffix) => suffix.parse::<u8>().ok(),
     482            0 :                 None => None,
     483              :             };
     484              : 
     485            0 :             let is_instance_dir = is_dir && parsed_instance_id.is_some();
     486              : 
     487            0 :             if !is_instance_dir {
     488            0 :                 continue;
     489            0 :             }
     490            0 : 
     491            0 :             instances.push((
     492            0 :                 parsed_instance_id.expect("Checked previously"),
     493            0 :                 dentry.path(),
     494            0 :             ));
     495              :         }
     496              : 
     497            0 :         Ok(instances)
     498            0 :     }
     499              : 
     500            0 :     pub fn register_branch_mapping(
     501            0 :         &mut self,
     502            0 :         branch_name: String,
     503            0 :         tenant_id: TenantId,
     504            0 :         timeline_id: TimelineId,
     505            0 :     ) -> anyhow::Result<()> {
     506            0 :         let existing_values = self
     507            0 :             .branch_name_mappings
     508            0 :             .entry(branch_name.clone())
     509            0 :             .or_default();
     510            0 : 
     511            0 :         let existing_ids = existing_values
     512            0 :             .iter()
     513            0 :             .find(|(existing_tenant_id, _)| existing_tenant_id == &tenant_id);
     514              : 
     515            0 :         if let Some((_, old_timeline_id)) = existing_ids {
     516            0 :             if old_timeline_id == &timeline_id {
     517            0 :                 Ok(())
     518              :             } else {
     519            0 :                 bail!(
     520            0 :                     "branch '{branch_name}' is already mapped to timeline {old_timeline_id}, cannot map to another timeline {timeline_id}"
     521            0 :                 );
     522              :             }
     523              :         } else {
     524            0 :             existing_values.push((tenant_id, timeline_id));
     525            0 :             Ok(())
     526              :         }
     527            0 :     }
     528              : 
     529            0 :     pub fn get_branch_timeline_id(
     530            0 :         &self,
     531            0 :         branch_name: &str,
     532            0 :         tenant_id: TenantId,
     533            0 :     ) -> Option<TimelineId> {
     534            0 :         self.branch_name_mappings
     535            0 :             .get(branch_name)?
     536            0 :             .iter()
     537            0 :             .find(|(mapped_tenant_id, _)| mapped_tenant_id == &tenant_id)
     538            0 :             .map(|&(_, timeline_id)| timeline_id)
     539            0 :     }
     540              : 
     541            0 :     pub fn timeline_name_mappings(&self) -> HashMap<TenantTimelineId, String> {
     542            0 :         self.branch_name_mappings
     543            0 :             .iter()
     544            0 :             .flat_map(|(name, tenant_timelines)| {
     545            0 :                 tenant_timelines.iter().map(|&(tenant_id, timeline_id)| {
     546            0 :                     (TenantTimelineId::new(tenant_id, timeline_id), name.clone())
     547            0 :                 })
     548            0 :             })
     549            0 :             .collect()
     550            0 :     }
     551              : 
     552              :     ///  Construct `Self` from on-disk state.
     553            0 :     pub fn load_config(repopath: &Path) -> anyhow::Result<Self> {
     554            0 :         if !repopath.exists() {
     555            0 :             bail!(
     556            0 :                 "Neon config is not found in {}. You need to run 'neon_local init' first",
     557            0 :                 repopath.to_str().unwrap()
     558            0 :             );
     559            0 :         }
     560              : 
     561              :         // TODO: check that it looks like a neon repository
     562              : 
     563              :         // load and parse file
     564            0 :         let config_file_contents = fs::read_to_string(repopath.join("config"))?;
     565            0 :         let on_disk_config: OnDiskConfig = toml::from_str(config_file_contents.as_str())?;
     566            0 :         let mut env = {
     567            0 :             let OnDiskConfig {
     568            0 :                 pg_distrib_dir,
     569            0 :                 neon_distrib_dir,
     570            0 :                 default_tenant_id,
     571            0 :                 private_key_path,
     572            0 :                 broker,
     573            0 :                 storage_controller,
     574            0 :                 pageservers,
     575            0 :                 safekeepers,
     576            0 :                 control_plane_api,
     577            0 :                 control_plane_hooks_api,
     578            0 :                 control_plane_compute_hook_api: _,
     579            0 :                 branch_name_mappings,
     580            0 :                 generate_local_ssl_certs,
     581            0 :             } = on_disk_config;
     582            0 :             LocalEnv {
     583            0 :                 base_data_dir: repopath.to_owned(),
     584            0 :                 pg_distrib_dir,
     585            0 :                 neon_distrib_dir,
     586            0 :                 default_tenant_id,
     587            0 :                 private_key_path,
     588            0 :                 broker,
     589            0 :                 storage_controller,
     590            0 :                 pageservers,
     591            0 :                 safekeepers,
     592            0 :                 control_plane_api: control_plane_api.unwrap(),
     593            0 :                 control_plane_hooks_api,
     594            0 :                 branch_name_mappings,
     595            0 :                 generate_local_ssl_certs,
     596            0 :             }
     597            0 :         };
     598            0 : 
     599            0 :         // The source of truth for pageserver configuration is the pageserver.toml.
     600            0 :         assert!(
     601            0 :             env.pageservers.is_empty(),
     602            0 :             "we ensure this during deserialization"
     603              :         );
     604            0 :         env.pageservers = {
     605            0 :             let iter = std::fs::read_dir(repopath).context("open dir")?;
     606            0 :             let mut pageservers = Vec::new();
     607            0 :             for res in iter {
     608            0 :                 let dentry = res?;
     609              :                 const PREFIX: &str = "pageserver_";
     610            0 :                 let dentry_name = dentry
     611            0 :                     .file_name()
     612            0 :                     .into_string()
     613            0 :                     .ok()
     614            0 :                     .with_context(|| format!("non-utf8 dentry: {:?}", dentry.path()))
     615            0 :                     .unwrap();
     616            0 :                 if !dentry_name.starts_with(PREFIX) {
     617            0 :                     continue;
     618            0 :                 }
     619            0 :                 if !dentry.file_type().context("determine file type")?.is_dir() {
     620            0 :                     anyhow::bail!("expected a directory, got {:?}", dentry.path());
     621            0 :                 }
     622            0 :                 let id = dentry_name[PREFIX.len()..]
     623            0 :                     .parse::<NodeId>()
     624            0 :                     .with_context(|| format!("parse id from {:?}", dentry.path()))?;
     625              :                 // TODO(christian): use pageserver_api::config::ConfigToml (PR #7656)
     626            0 :                 #[derive(serde::Serialize, serde::Deserialize)]
     627              :                 // (allow unknown fields, unlike PageServerConf)
     628              :                 struct PageserverConfigTomlSubset {
     629              :                     listen_pg_addr: String,
     630              :                     listen_http_addr: String,
     631              :                     listen_https_addr: Option<String>,
     632              :                     pg_auth_type: AuthType,
     633              :                     http_auth_type: AuthType,
     634              :                     #[serde(default)]
     635              :                     no_sync: bool,
     636              :                 }
     637            0 :                 let config_toml_path = dentry.path().join("pageserver.toml");
     638            0 :                 let config_toml: PageserverConfigTomlSubset = toml_edit::de::from_str(
     639            0 :                     &std::fs::read_to_string(&config_toml_path)
     640            0 :                         .with_context(|| format!("read {:?}", config_toml_path))?,
     641              :                 )
     642            0 :                 .context("parse pageserver.toml")?;
     643            0 :                 let identity_toml_path = dentry.path().join("identity.toml");
     644            0 :                 #[derive(serde::Serialize, serde::Deserialize)]
     645              :                 struct IdentityTomlSubset {
     646              :                     id: NodeId,
     647              :                 }
     648            0 :                 let identity_toml: IdentityTomlSubset = toml_edit::de::from_str(
     649            0 :                     &std::fs::read_to_string(&identity_toml_path)
     650            0 :                         .with_context(|| format!("read {:?}", identity_toml_path))?,
     651              :                 )
     652            0 :                 .context("parse identity.toml")?;
     653              :                 let PageserverConfigTomlSubset {
     654            0 :                     listen_pg_addr,
     655            0 :                     listen_http_addr,
     656            0 :                     listen_https_addr,
     657            0 :                     pg_auth_type,
     658            0 :                     http_auth_type,
     659            0 :                     no_sync,
     660            0 :                 } = config_toml;
     661            0 :                 let IdentityTomlSubset {
     662            0 :                     id: identity_toml_id,
     663            0 :                 } = identity_toml;
     664            0 :                 let conf = PageServerConf {
     665              :                     id: {
     666            0 :                         anyhow::ensure!(
     667            0 :                             identity_toml_id == id,
     668            0 :                             "id mismatch: identity.toml:id={identity_toml_id} pageserver_(.*) id={id}",
     669              :                         );
     670            0 :                         id
     671            0 :                     },
     672            0 :                     listen_pg_addr,
     673            0 :                     listen_http_addr,
     674            0 :                     listen_https_addr,
     675            0 :                     pg_auth_type,
     676            0 :                     http_auth_type,
     677            0 :                     no_sync,
     678            0 :                 };
     679            0 :                 pageservers.push(conf);
     680              :             }
     681            0 :             pageservers
     682            0 :         };
     683            0 : 
     684            0 :         Ok(env)
     685            0 :     }
     686              : 
     687            0 :     pub fn persist_config(&self) -> anyhow::Result<()> {
     688            0 :         Self::persist_config_impl(
     689            0 :             &self.base_data_dir,
     690            0 :             &OnDiskConfig {
     691            0 :                 pg_distrib_dir: self.pg_distrib_dir.clone(),
     692            0 :                 neon_distrib_dir: self.neon_distrib_dir.clone(),
     693            0 :                 default_tenant_id: self.default_tenant_id,
     694            0 :                 private_key_path: self.private_key_path.clone(),
     695            0 :                 broker: self.broker.clone(),
     696            0 :                 storage_controller: self.storage_controller.clone(),
     697            0 :                 pageservers: vec![], // it's skip_serializing anyway
     698            0 :                 safekeepers: self.safekeepers.clone(),
     699            0 :                 control_plane_api: Some(self.control_plane_api.clone()),
     700            0 :                 control_plane_hooks_api: self.control_plane_hooks_api.clone(),
     701            0 :                 control_plane_compute_hook_api: None,
     702            0 :                 branch_name_mappings: self.branch_name_mappings.clone(),
     703            0 :                 generate_local_ssl_certs: self.generate_local_ssl_certs,
     704            0 :             },
     705            0 :         )
     706            0 :     }
     707              : 
     708            0 :     pub fn persist_config_impl(base_path: &Path, config: &OnDiskConfig) -> anyhow::Result<()> {
     709            0 :         let conf_content = &toml::to_string_pretty(config)?;
     710            0 :         let target_config_path = base_path.join("config");
     711            0 :         fs::write(&target_config_path, conf_content).with_context(|| {
     712            0 :             format!(
     713            0 :                 "Failed to write config file into path '{}'",
     714            0 :                 target_config_path.display()
     715            0 :             )
     716            0 :         })
     717            0 :     }
     718              : 
     719              :     // this function is used only for testing purposes in CLI e g generate tokens during init
     720            0 :     pub fn generate_auth_token(&self, claims: &Claims) -> anyhow::Result<String> {
     721            0 :         let private_key_path = self.get_private_key_path();
     722            0 :         let key_data = fs::read(private_key_path)?;
     723            0 :         encode_from_key_file(claims, &key_data)
     724            0 :     }
     725              : 
     726            0 :     pub fn get_private_key_path(&self) -> PathBuf {
     727            0 :         if self.private_key_path.is_absolute() {
     728            0 :             self.private_key_path.to_path_buf()
     729              :         } else {
     730            0 :             self.base_data_dir.join(&self.private_key_path)
     731              :         }
     732            0 :     }
     733              : 
     734              :     /// Materialize the [`NeonLocalInitConf`] to disk. Called during [`neon_local init`].
     735            0 :     pub fn init(conf: NeonLocalInitConf, force: &InitForceMode) -> anyhow::Result<()> {
     736            0 :         let base_path = base_path();
     737            0 :         assert_ne!(base_path, Path::new(""));
     738            0 :         let base_path = &base_path;
     739            0 : 
     740            0 :         // create base_path dir
     741            0 :         if base_path.exists() {
     742            0 :             match force {
     743              :                 InitForceMode::MustNotExist => {
     744            0 :                     bail!(
     745            0 :                         "directory '{}' already exists. Perhaps already initialized?",
     746            0 :                         base_path.display()
     747            0 :                     );
     748              :                 }
     749              :                 InitForceMode::EmptyDirOk => {
     750            0 :                     if let Some(res) = std::fs::read_dir(base_path)?.next() {
     751            0 :                         res.context("check if directory is empty")?;
     752            0 :                         anyhow::bail!("directory not empty: {base_path:?}");
     753            0 :                     }
     754              :                 }
     755              :                 InitForceMode::RemoveAllContents => {
     756            0 :                     println!("removing all contents of '{}'", base_path.display());
     757              :                     // instead of directly calling `remove_dir_all`, we keep the original dir but removing
     758              :                     // all contents inside. This helps if the developer symbol links another directory (i.e.,
     759              :                     // S3 local SSD) to the `.neon` base directory.
     760            0 :                     for entry in std::fs::read_dir(base_path)? {
     761            0 :                         let entry = entry?;
     762            0 :                         let path = entry.path();
     763            0 :                         if path.is_dir() {
     764            0 :                             fs::remove_dir_all(&path)?;
     765              :                         } else {
     766            0 :                             fs::remove_file(&path)?;
     767              :                         }
     768              :                     }
     769              :                 }
     770              :             }
     771            0 :         }
     772            0 :         if !base_path.exists() {
     773            0 :             fs::create_dir(base_path)?;
     774            0 :         }
     775              : 
     776              :         let NeonLocalInitConf {
     777            0 :             pg_distrib_dir,
     778            0 :             neon_distrib_dir,
     779            0 :             default_tenant_id,
     780            0 :             broker,
     781            0 :             storage_controller,
     782            0 :             pageservers,
     783            0 :             safekeepers,
     784            0 :             control_plane_api,
     785            0 :             generate_local_ssl_certs,
     786            0 :             control_plane_hooks_api,
     787            0 :         } = conf;
     788            0 : 
     789            0 :         // Find postgres binaries.
     790            0 :         // Follow POSTGRES_DISTRIB_DIR if set, otherwise look in "pg_install".
     791            0 :         // Note that later in the code we assume, that distrib dirs follow the same pattern
     792            0 :         // for all postgres versions.
     793            0 :         let pg_distrib_dir = pg_distrib_dir.unwrap_or_else(|| {
     794            0 :             if let Some(postgres_bin) = env::var_os("POSTGRES_DISTRIB_DIR") {
     795            0 :                 postgres_bin.into()
     796              :             } else {
     797            0 :                 let cwd = env::current_dir().unwrap();
     798            0 :                 cwd.join("pg_install")
     799              :             }
     800            0 :         });
     801            0 : 
     802            0 :         // Find neon binaries.
     803            0 :         let neon_distrib_dir = neon_distrib_dir
     804            0 :             .unwrap_or_else(|| env::current_exe().unwrap().parent().unwrap().to_owned());
     805            0 : 
     806            0 :         // Generate keypair for JWT.
     807            0 :         //
     808            0 :         // The keypair is only needed if authentication is enabled in any of the
     809            0 :         // components. For convenience, we generate the keypair even if authentication
     810            0 :         // is not enabled, so that you can easily enable it after the initialization
     811            0 :         // step.
     812            0 :         generate_auth_keys(
     813            0 :             base_path.join("auth_private_key.pem").as_path(),
     814            0 :             base_path.join("auth_public_key.pem").as_path(),
     815            0 :         )
     816            0 :         .context("generate auth keys")?;
     817            0 :         let private_key_path = PathBuf::from("auth_private_key.pem");
     818            0 : 
     819            0 :         // create the runtime type because the remaining initialization code below needs
     820            0 :         // a LocalEnv instance op operation
     821            0 :         // TODO: refactor to avoid this, LocalEnv should only be constructed from on-disk state
     822            0 :         let env = LocalEnv {
     823            0 :             base_data_dir: base_path.clone(),
     824            0 :             pg_distrib_dir,
     825            0 :             neon_distrib_dir,
     826            0 :             default_tenant_id: Some(default_tenant_id),
     827            0 :             private_key_path,
     828            0 :             broker,
     829            0 :             storage_controller: storage_controller.unwrap_or_default(),
     830            0 :             pageservers: pageservers.iter().map(Into::into).collect(),
     831            0 :             safekeepers,
     832            0 :             control_plane_api: control_plane_api.unwrap(),
     833            0 :             control_plane_hooks_api,
     834            0 :             branch_name_mappings: Default::default(),
     835            0 :             generate_local_ssl_certs,
     836            0 :         };
     837            0 : 
     838            0 :         if generate_local_ssl_certs {
     839            0 :             env.generate_ssl_ca_cert()?;
     840            0 :         }
     841              : 
     842              :         // create endpoints dir
     843            0 :         fs::create_dir_all(env.endpoints_path())?;
     844              : 
     845              :         // create safekeeper dirs
     846            0 :         for safekeeper in &env.safekeepers {
     847            0 :             fs::create_dir_all(SafekeeperNode::datadir_path_by_id(&env, safekeeper.id))?;
     848              :         }
     849              : 
     850              :         // initialize pageserver state
     851            0 :         for (i, ps) in pageservers.into_iter().enumerate() {
     852            0 :             let runtime_ps = &env.pageservers[i];
     853            0 :             assert_eq!(&PageServerConf::from(&ps), runtime_ps);
     854            0 :             fs::create_dir(env.pageserver_data_dir(ps.id))?;
     855            0 :             PageServerNode::from_env(&env, runtime_ps)
     856            0 :                 .initialize(ps)
     857            0 :                 .context("pageserver init failed")?;
     858              :         }
     859              : 
     860              :         // setup remote remote location for default LocalFs remote storage
     861            0 :         std::fs::create_dir_all(env.base_data_dir.join(PAGESERVER_REMOTE_STORAGE_DIR))?;
     862              : 
     863            0 :         env.persist_config()
     864            0 :     }
     865              : }
     866              : 
     867            0 : pub fn base_path() -> PathBuf {
     868            0 :     let path = match std::env::var_os("NEON_REPO_DIR") {
     869            0 :         Some(val) => {
     870            0 :             let path = PathBuf::from(val);
     871            0 :             if !path.is_absolute() {
     872              :                 // repeat the env var in the error because our default is always absolute
     873            0 :                 panic!("NEON_REPO_DIR must be an absolute path, got {path:?}");
     874            0 :             }
     875            0 :             path
     876              :         }
     877              :         None => {
     878            0 :             let pwd = std::env::current_dir()
     879            0 :                 // technically this can fail but it's quite unlikeley
     880            0 :                 .expect("determine current directory");
     881            0 :             let pwd_abs = pwd.canonicalize().expect("canonicalize current directory");
     882            0 :             pwd_abs.join(".neon")
     883              :         }
     884              :     };
     885            0 :     assert!(path.is_absolute());
     886            0 :     path
     887            0 : }
     888              : 
     889              : /// Generate a public/private key pair for JWT authentication
     890            0 : fn generate_auth_keys(private_key_path: &Path, public_key_path: &Path) -> anyhow::Result<()> {
     891              :     // Generate the key pair
     892              :     //
     893              :     // openssl genpkey -algorithm ed25519 -out auth_private_key.pem
     894            0 :     let keygen_output = Command::new("openssl")
     895            0 :         .arg("genpkey")
     896            0 :         .args(["-algorithm", "ed25519"])
     897            0 :         .args(["-out", private_key_path.to_str().unwrap()])
     898            0 :         .stdout(Stdio::null())
     899            0 :         .output()
     900            0 :         .context("failed to generate auth private key")?;
     901            0 :     if !keygen_output.status.success() {
     902            0 :         bail!(
     903            0 :             "openssl failed: '{}'",
     904            0 :             String::from_utf8_lossy(&keygen_output.stderr)
     905            0 :         );
     906            0 :     }
     907              :     // Extract the public key from the private key file
     908              :     //
     909              :     // openssl pkey -in auth_private_key.pem -pubout -out auth_public_key.pem
     910            0 :     let keygen_output = Command::new("openssl")
     911            0 :         .arg("pkey")
     912            0 :         .args(["-in", private_key_path.to_str().unwrap()])
     913            0 :         .arg("-pubout")
     914            0 :         .args(["-out", public_key_path.to_str().unwrap()])
     915            0 :         .output()
     916            0 :         .context("failed to extract public key from private key")?;
     917            0 :     if !keygen_output.status.success() {
     918            0 :         bail!(
     919            0 :             "openssl failed: '{}'",
     920            0 :             String::from_utf8_lossy(&keygen_output.stderr)
     921            0 :         );
     922            0 :     }
     923            0 :     Ok(())
     924            0 : }
     925              : 
     926            0 : fn generate_ssl_ca_cert(cert_path: &Path, key_path: &Path) -> anyhow::Result<()> {
     927              :     // openssl req -x509 -newkey rsa:2048 -nodes -subj "/CN=Neon Local CA" -days 36500 \
     928              :     // -out rootCA.crt -keyout rootCA.key
     929            0 :     let keygen_output = Command::new("openssl")
     930            0 :         .args([
     931            0 :             "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "36500",
     932            0 :         ])
     933            0 :         .args(["-subj", "/CN=Neon Local CA"])
     934            0 :         .args(["-out", cert_path.to_str().unwrap()])
     935            0 :         .args(["-keyout", key_path.to_str().unwrap()])
     936            0 :         .output()
     937            0 :         .context("failed to generate CA certificate")?;
     938            0 :     if !keygen_output.status.success() {
     939            0 :         bail!(
     940            0 :             "openssl failed: '{}'",
     941            0 :             String::from_utf8_lossy(&keygen_output.stderr)
     942            0 :         );
     943            0 :     }
     944            0 :     Ok(())
     945            0 : }
     946              : 
     947            0 : fn generate_ssl_cert(
     948            0 :     cert_path: &Path,
     949            0 :     key_path: &Path,
     950            0 :     ca_cert_path: &Path,
     951            0 :     ca_key_path: &Path,
     952            0 : ) -> anyhow::Result<()> {
     953            0 :     // Generate Certificate Signing Request (CSR).
     954            0 :     let mut csr_path = cert_path.to_path_buf();
     955            0 :     csr_path.set_extension(".csr");
     956              : 
     957              :     // openssl req -new -nodes -newkey rsa:2048 -keyout server.key -out server.csr \
     958              :     // -subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
     959            0 :     let keygen_output = Command::new("openssl")
     960            0 :         .args(["req", "-new", "-nodes"])
     961            0 :         .args(["-newkey", "rsa:2048"])
     962            0 :         .args(["-subj", "/CN=localhost"])
     963            0 :         .args(["-addext", "subjectAltName=DNS:localhost,IP:127.0.0.1"])
     964            0 :         .args(["-keyout", key_path.to_str().unwrap()])
     965            0 :         .args(["-out", csr_path.to_str().unwrap()])
     966            0 :         .output()
     967            0 :         .context("failed to generate CSR")?;
     968            0 :     if !keygen_output.status.success() {
     969            0 :         bail!(
     970            0 :             "openssl failed: '{}'",
     971            0 :             String::from_utf8_lossy(&keygen_output.stderr)
     972            0 :         );
     973            0 :     }
     974              : 
     975              :     // Sign CSR with CA key.
     976              :     //
     977              :     // openssl x509 -req -in server.csr -CA rootCA.crt -CAkey rootCA.key -CAcreateserial \
     978              :     // -out server.crt -days 36500 -copy_extensions copyall
     979            0 :     let keygen_output = Command::new("openssl")
     980            0 :         .args(["x509", "-req"])
     981            0 :         .args(["-in", csr_path.to_str().unwrap()])
     982            0 :         .args(["-CA", ca_cert_path.to_str().unwrap()])
     983            0 :         .args(["-CAkey", ca_key_path.to_str().unwrap()])
     984            0 :         .arg("-CAcreateserial")
     985            0 :         .args(["-out", cert_path.to_str().unwrap()])
     986            0 :         .args(["-days", "36500"])
     987            0 :         .args(["-copy_extensions", "copyall"])
     988            0 :         .output()
     989            0 :         .context("failed to sign CSR")?;
     990            0 :     if !keygen_output.status.success() {
     991            0 :         bail!(
     992            0 :             "openssl failed: '{}'",
     993            0 :             String::from_utf8_lossy(&keygen_output.stderr)
     994            0 :         );
     995            0 :     }
     996            0 : 
     997            0 :     // Remove CSR file as it's not needed anymore.
     998            0 :     fs::remove_file(csr_path)?;
     999              : 
    1000            0 :     Ok(())
    1001            0 : }
        

Generated by: LCOV version 2.1-beta