diff --git a/control_plane/src/local_env.rs b/control_plane/src/local_env.rs index 787028eece..55d0b00496 100644 --- a/control_plane/src/local_env.rs +++ b/control_plane/src/local_env.rs @@ -64,6 +64,8 @@ pub struct LocalEnv { #[derive(Serialize, Deserialize, Clone, Debug)] #[serde(default)] pub struct PageServerConf { + // node id + pub id: ZNodeId, // Pageserver connection settings pub listen_pg_addr: String, pub listen_http_addr: String, @@ -78,6 +80,7 @@ pub struct PageServerConf { impl Default for PageServerConf { fn default() -> Self { Self { + id: ZNodeId(0), listen_pg_addr: String::new(), listen_http_addr: String::new(), auth_type: AuthType::Trust, diff --git a/control_plane/src/storage.rs b/control_plane/src/storage.rs index be594889ab..cd429e3f7a 100644 --- a/control_plane/src/storage.rs +++ b/control_plane/src/storage.rs @@ -103,6 +103,8 @@ impl PageServerNode { ) -> anyhow::Result<()> { let mut cmd = Command::new(self.env.pageserver_bin()?); + let id = format!("id={}", self.env.pageserver.id); + // FIXME: the paths should be shell-escaped to handle paths with spaces, quotas etc. let base_data_dir_param = self.env.base_data_dir.display().to_string(); let pg_distrib_dir_param = @@ -122,6 +124,7 @@ impl PageServerNode { args.extend(["-c", &authg_type_param]); args.extend(["-c", &listen_http_addr_param]); args.extend(["-c", &listen_pg_addr_param]); + args.extend(["-c", &id]); for config_override in config_overrides { args.extend(["-c", config_override]); diff --git a/pageserver/src/bin/pageserver.rs b/pageserver/src/bin/pageserver.rs index fb8baa28f6..d8d4033340 100644 --- a/pageserver/src/bin/pageserver.rs +++ b/pageserver/src/bin/pageserver.rs @@ -61,7 +61,7 @@ fn main() -> Result<()> { .number_of_values(1) .multiple_occurrences(true) .help("Additional configuration overrides of the ones from the toml config file (or new ones to add there). - Any option has to be a valid toml document, example: `-c \"foo='hey'\"` `-c \"foo={value=1}\"`"), + Any option has to be a valid toml document, example: `-c=\"foo='hey'\"` `-c=\"foo={value=1}\"`"), ) .get_matches(); @@ -115,7 +115,14 @@ fn main() -> Result<()> { option_line ) })?; + for (key, item) in doc.iter() { + if key == "id" { + anyhow::ensure!( + init, + "node id can only be set during pageserver init and cannot be overridden" + ); + } toml.insert(key, item.clone()); } } diff --git a/pageserver/src/config.rs b/pageserver/src/config.rs index 3d64d830b5..215af337d2 100644 --- a/pageserver/src/config.rs +++ b/pageserver/src/config.rs @@ -8,7 +8,7 @@ use anyhow::{bail, ensure, Context, Result}; use toml_edit; use toml_edit::{Document, Item}; use zenith_utils::postgres_backend::AuthType; -use zenith_utils::zid::{ZTenantId, ZTimelineId}; +use zenith_utils::zid::{ZNodeId, ZTenantId, ZTimelineId}; use std::convert::TryInto; use std::env; @@ -72,6 +72,10 @@ pub mod defaults { #[derive(Debug, Clone, PartialEq, Eq)] pub struct PageServerConf { + // Identifier of that particular pageserver so e g safekeepers + // can safely distinguish different pageservers + pub id: ZNodeId, + /// Example (default): 127.0.0.1:64000 pub listen_pg_addr: String, /// Example (default): 127.0.0.1:9898 @@ -106,6 +110,168 @@ pub struct PageServerConf { pub remote_storage_config: Option, } +// needed to simplify config construction +struct PageServerConfigBuilder { + listen_pg_addr: Option, + + listen_http_addr: Option, + + checkpoint_distance: Option, + checkpoint_period: Option, + + gc_horizon: Option, + gc_period: Option, + superuser: Option, + + page_cache_size: Option, + max_file_descriptors: Option, + + workdir: Option, + + pg_distrib_dir: Option, + + auth_type: Option, + + auth_validation_public_key_path: Option>, + remote_storage_config: Option>, + + id: Option, +} + +impl Default for PageServerConfigBuilder { + fn default() -> Self { + use defaults::*; + Self { + listen_pg_addr: Some(DEFAULT_PG_LISTEN_ADDR.to_string()), + listen_http_addr: Some(DEFAULT_HTTP_LISTEN_ADDR.to_string()), + checkpoint_distance: Some(DEFAULT_CHECKPOINT_DISTANCE), + checkpoint_period: Some( + humantime::parse_duration(DEFAULT_CHECKPOINT_PERIOD) + .expect("cannot parse default checkpoint period"), + ), + gc_horizon: Some(DEFAULT_GC_HORIZON), + gc_period: Some( + humantime::parse_duration(DEFAULT_GC_PERIOD) + .expect("cannot parse default gc period"), + ), + superuser: Some(DEFAULT_SUPERUSER.to_string()), + page_cache_size: Some(DEFAULT_PAGE_CACHE_SIZE), + max_file_descriptors: Some(DEFAULT_MAX_FILE_DESCRIPTORS), + workdir: Default::default(), + pg_distrib_dir: Some(PathBuf::new()), + auth_type: Some(AuthType::Trust), + auth_validation_public_key_path: Some(None), + remote_storage_config: Some(None), + id: Default::default(), + } + } +} + +impl PageServerConfigBuilder { + pub fn listen_pg_addr(&mut self, listen_pg_addr: String) { + self.listen_pg_addr = Some(listen_pg_addr) + } + + pub fn listen_http_addr(&mut self, listen_http_addr: String) { + self.listen_http_addr = Some(listen_http_addr) + } + + pub fn checkpoint_distance(&mut self, checkpoint_distance: u64) { + self.checkpoint_distance = Some(checkpoint_distance) + } + + pub fn checkpoint_period(&mut self, checkpoint_period: Duration) { + self.checkpoint_period = Some(checkpoint_period) + } + + pub fn gc_horizon(&mut self, gc_horizon: u64) { + self.gc_horizon = Some(gc_horizon) + } + + pub fn gc_period(&mut self, gc_period: Duration) { + self.gc_period = Some(gc_period) + } + + pub fn superuser(&mut self, superuser: String) { + self.superuser = Some(superuser) + } + + pub fn page_cache_size(&mut self, page_cache_size: usize) { + self.page_cache_size = Some(page_cache_size) + } + + pub fn max_file_descriptors(&mut self, max_file_descriptors: usize) { + self.max_file_descriptors = Some(max_file_descriptors) + } + + pub fn workdir(&mut self, workdir: PathBuf) { + self.workdir = Some(workdir) + } + + pub fn pg_distrib_dir(&mut self, pg_distrib_dir: PathBuf) { + self.pg_distrib_dir = Some(pg_distrib_dir) + } + + pub fn auth_type(&mut self, auth_type: AuthType) { + self.auth_type = Some(auth_type) + } + + pub fn auth_validation_public_key_path( + &mut self, + auth_validation_public_key_path: Option, + ) { + self.auth_validation_public_key_path = Some(auth_validation_public_key_path) + } + + pub fn remote_storage_config(&mut self, remote_storage_config: Option) { + self.remote_storage_config = Some(remote_storage_config) + } + + pub fn id(&mut self, node_id: ZNodeId) { + self.id = Some(node_id) + } + + pub fn build(self) -> Result { + Ok(PageServerConf { + listen_pg_addr: self + .listen_pg_addr + .ok_or(anyhow::anyhow!("missing listen_pg_addr"))?, + listen_http_addr: self + .listen_http_addr + .ok_or(anyhow::anyhow!("missing listen_http_addr"))?, + checkpoint_distance: self + .checkpoint_distance + .ok_or(anyhow::anyhow!("missing checkpoint_distance"))?, + checkpoint_period: self + .checkpoint_period + .ok_or(anyhow::anyhow!("missing checkpoint_period"))?, + gc_horizon: self + .gc_horizon + .ok_or(anyhow::anyhow!("missing gc_horizon"))?, + gc_period: self.gc_period.ok_or(anyhow::anyhow!("missing gc_period"))?, + superuser: self.superuser.ok_or(anyhow::anyhow!("missing superuser"))?, + page_cache_size: self + .page_cache_size + .ok_or(anyhow::anyhow!("missing page_cache_size"))?, + max_file_descriptors: self + .max_file_descriptors + .ok_or(anyhow::anyhow!("missing max_file_descriptors"))?, + workdir: self.workdir.ok_or(anyhow::anyhow!("missing workdir"))?, + pg_distrib_dir: self + .pg_distrib_dir + .ok_or(anyhow::anyhow!("missing pg_distrib_dir"))?, + auth_type: self.auth_type.ok_or(anyhow::anyhow!("missing auth_type"))?, + auth_validation_public_key_path: self + .auth_validation_public_key_path + .ok_or(anyhow::anyhow!("missing auth_validation_public_key_path"))?, + remote_storage_config: self + .remote_storage_config + .ok_or(anyhow::anyhow!("missing remote_storage_config"))?, + id: self.id.ok_or(anyhow::anyhow!("missing id"))?, + }) + } +} + /// External backup storage configuration, enough for creating a client for that storage. #[derive(Debug, Clone, PartialEq, Eq)] pub struct RemoteStorageConfig { @@ -221,57 +387,39 @@ impl PageServerConf { /// /// This leaves any options not present in the file in the built-in defaults. pub fn parse_and_validate(toml: &Document, workdir: &Path) -> Result { - use defaults::*; - - let mut conf = PageServerConf { - workdir: workdir.to_path_buf(), - - listen_pg_addr: DEFAULT_PG_LISTEN_ADDR.to_string(), - listen_http_addr: DEFAULT_HTTP_LISTEN_ADDR.to_string(), - checkpoint_distance: DEFAULT_CHECKPOINT_DISTANCE, - checkpoint_period: humantime::parse_duration(DEFAULT_CHECKPOINT_PERIOD)?, - gc_horizon: DEFAULT_GC_HORIZON, - gc_period: humantime::parse_duration(DEFAULT_GC_PERIOD)?, - page_cache_size: DEFAULT_PAGE_CACHE_SIZE, - max_file_descriptors: DEFAULT_MAX_FILE_DESCRIPTORS, - - pg_distrib_dir: PathBuf::new(), - auth_validation_public_key_path: None, - auth_type: AuthType::Trust, - - remote_storage_config: None, - - superuser: DEFAULT_SUPERUSER.to_string(), - }; + let mut builder = PageServerConfigBuilder::default(); + builder.workdir(workdir.to_owned()); for (key, item) in toml.iter() { match key { - "listen_pg_addr" => conf.listen_pg_addr = parse_toml_string(key, item)?, - "listen_http_addr" => conf.listen_http_addr = parse_toml_string(key, item)?, - "checkpoint_distance" => conf.checkpoint_distance = parse_toml_u64(key, item)?, - "checkpoint_period" => conf.checkpoint_period = parse_toml_duration(key, item)?, - "gc_horizon" => conf.gc_horizon = parse_toml_u64(key, item)?, - "gc_period" => conf.gc_period = parse_toml_duration(key, item)?, - "initial_superuser_name" => conf.superuser = parse_toml_string(key, item)?, - "page_cache_size" => conf.page_cache_size = parse_toml_u64(key, item)? as usize, + "listen_pg_addr" => builder.listen_pg_addr(parse_toml_string(key, item)?), + "listen_http_addr" => builder.listen_http_addr(parse_toml_string(key, item)?), + "checkpoint_distance" => builder.checkpoint_distance(parse_toml_u64(key, item)?), + "checkpoint_period" => builder.checkpoint_period(parse_toml_duration(key, item)?), + "gc_horizon" => builder.gc_horizon(parse_toml_u64(key, item)?), + "gc_period" => builder.gc_period(parse_toml_duration(key, item)?), + "initial_superuser_name" => builder.superuser(parse_toml_string(key, item)?), + "page_cache_size" => builder.page_cache_size(parse_toml_u64(key, item)? as usize), "max_file_descriptors" => { - conf.max_file_descriptors = parse_toml_u64(key, item)? as usize + builder.max_file_descriptors(parse_toml_u64(key, item)? as usize) } "pg_distrib_dir" => { - conf.pg_distrib_dir = PathBuf::from(parse_toml_string(key, item)?) + builder.pg_distrib_dir(PathBuf::from(parse_toml_string(key, item)?)) } - "auth_validation_public_key_path" => { - conf.auth_validation_public_key_path = - Some(PathBuf::from(parse_toml_string(key, item)?)) - } - "auth_type" => conf.auth_type = parse_toml_auth_type(key, item)?, + "auth_validation_public_key_path" => builder.auth_validation_public_key_path(Some( + PathBuf::from(parse_toml_string(key, item)?), + )), + "auth_type" => builder.auth_type(parse_toml_auth_type(key, item)?), "remote_storage" => { - conf.remote_storage_config = Some(Self::parse_remote_storage_config(item)?) + builder.remote_storage_config(Some(Self::parse_remote_storage_config(item)?)) } + "id" => builder.id(ZNodeId(parse_toml_u64(key, item)?)), _ => bail!("unrecognized pageserver option '{}'", key), } } + let mut conf = builder.build().context("invalid config")?; + if conf.auth_type == AuthType::ZenithJWT { let auth_validation_public_key_path = conf .auth_validation_public_key_path @@ -382,6 +530,7 @@ impl PageServerConf { #[cfg(test)] pub fn dummy_conf(repo_dir: PathBuf) -> Self { PageServerConf { + id: ZNodeId(0), checkpoint_distance: defaults::DEFAULT_CHECKPOINT_DISTANCE, checkpoint_period: Duration::from_secs(10), gc_horizon: defaults::DEFAULT_GC_HORIZON, @@ -461,15 +610,16 @@ max_file_descriptors = 333 # initial superuser role name to use when creating a new tenant initial_superuser_name = 'zzzz' +id = 10 - "#; +"#; #[test] fn parse_defaults() -> anyhow::Result<()> { let tempdir = tempdir()?; let (workdir, pg_distrib_dir) = prepare_fs(&tempdir)?; // we have to create dummy pathes to overcome the validation errors - let config_string = format!("pg_distrib_dir='{}'", pg_distrib_dir.display()); + let config_string = format!("pg_distrib_dir='{}'\nid=10", pg_distrib_dir.display()); let toml = config_string.parse()?; let parsed_config = @@ -480,6 +630,7 @@ initial_superuser_name = 'zzzz' assert_eq!( parsed_config, PageServerConf { + id: ZNodeId(10), listen_pg_addr: defaults::DEFAULT_PG_LISTEN_ADDR.to_string(), listen_http_addr: defaults::DEFAULT_HTTP_LISTEN_ADDR.to_string(), checkpoint_distance: defaults::DEFAULT_CHECKPOINT_DISTANCE, @@ -521,6 +672,7 @@ initial_superuser_name = 'zzzz' assert_eq!( parsed_config, PageServerConf { + id: ZNodeId(10), listen_pg_addr: "127.0.0.1:64000".to_string(), listen_http_addr: "127.0.0.1:9898".to_string(), checkpoint_distance: 111, diff --git a/pageserver/src/http/models.rs b/pageserver/src/http/models.rs index 6ce377c535..5d7398ef03 100644 --- a/pageserver/src/http/models.rs +++ b/pageserver/src/http/models.rs @@ -1,6 +1,7 @@ use serde::{Deserialize, Serialize}; use crate::ZTenantId; +use zenith_utils::zid::ZNodeId; #[derive(Serialize, Deserialize)] pub struct BranchCreateRequest { @@ -15,3 +16,8 @@ pub struct TenantCreateRequest { #[serde(with = "hex")] pub tenant_id: ZTenantId, } + +#[derive(Serialize)] +pub struct StatusResponse { + pub id: ZNodeId, +} diff --git a/pageserver/src/http/openapi_spec.yml b/pageserver/src/http/openapi_spec.yml index dcb81849e0..baf81fcf21 100644 --- a/pageserver/src/http/openapi_spec.yml +++ b/pageserver/src/http/openapi_spec.yml @@ -17,6 +17,11 @@ paths: application/json: schema: type: object + required: + - id + properties: + id: + type: integer /v1/timeline/{tenant_id}: parameters: - name: tenant_id diff --git a/pageserver/src/http/routes.rs b/pageserver/src/http/routes.rs index b13a45750e..4fc41d6e82 100644 --- a/pageserver/src/http/routes.rs +++ b/pageserver/src/http/routes.rs @@ -1,7 +1,6 @@ use std::sync::Arc; use anyhow::{Context, Result}; -use hyper::header; use hyper::StatusCode; use hyper::{Body, Request, Response, Uri}; use serde::Serialize; @@ -23,6 +22,7 @@ use zenith_utils::lsn::Lsn; use zenith_utils::zid::{opt_display_serde, ZTimelineId}; use super::models::BranchCreateRequest; +use super::models::StatusResponse; use super::models::TenantCreateRequest; use crate::branches::BranchInfo; use crate::repository::RepositoryTimeline; @@ -64,12 +64,12 @@ fn get_config(request: &Request) -> &'static PageServerConf { } // healthcheck handler -async fn status_handler(_: Request) -> Result, ApiError> { - Ok(Response::builder() - .status(StatusCode::OK) - .header(header::CONTENT_TYPE, "application/json") - .body(Body::from("{}")) - .map_err(ApiError::from_err)?) +async fn status_handler(request: Request) -> Result, ApiError> { + let config = get_config(&request); + Ok(json_response( + StatusCode::OK, + StatusResponse { id: config.id }, + )?) } async fn branch_create_handler(mut request: Request) -> Result, ApiError> { diff --git a/test_runner/README.md b/test_runner/README.md index 514c5f1e3a..a56c2df2c0 100644 --- a/test_runner/README.md +++ b/test_runner/README.md @@ -89,7 +89,7 @@ def test_foobar(zenith_env_builder: ZenithEnvBuilder): # Now create the environment. This initializes the repository, and starts # up the page server and the safekeepers - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() # Run the test ... diff --git a/test_runner/batch_others/test_auth.py b/test_runner/batch_others/test_auth.py index 7f86986e2e..ee1a09c917 100644 --- a/test_runner/batch_others/test_auth.py +++ b/test_runner/batch_others/test_auth.py @@ -8,7 +8,7 @@ import pytest def test_pageserver_auth(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.pageserver_auth_enabled = True - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() ps = env.pageserver @@ -51,7 +51,7 @@ def test_compute_auth_to_pageserver(zenith_env_builder: ZenithEnvBuilder, with_w zenith_env_builder.pageserver_auth_enabled = True if with_wal_acceptors: zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() branch = f"test_compute_auth_to_pageserver{with_wal_acceptors}" env.zenith_cli.create_branch(branch, "main") diff --git a/test_runner/batch_others/test_branch_behind.py b/test_runner/batch_others/test_branch_behind.py index 860db51c8a..509c46975e 100644 --- a/test_runner/batch_others/test_branch_behind.py +++ b/test_runner/batch_others/test_branch_behind.py @@ -19,7 +19,7 @@ def test_branch_behind(zenith_env_builder: ZenithEnvBuilder): # # See https://github.com/zenithdb/zenith/issues/1068 zenith_env_builder.num_safekeepers = 1 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() # Branch at the point where only 100 rows were inserted env.zenith_cli.create_branch("test_branch_behind", "main") diff --git a/test_runner/batch_others/test_next_xid.py b/test_runner/batch_others/test_next_xid.py index 625abc39d3..fd0f761409 100644 --- a/test_runner/batch_others/test_next_xid.py +++ b/test_runner/batch_others/test_next_xid.py @@ -11,7 +11,7 @@ from fixtures.log_helper import log def test_next_xid(zenith_env_builder: ZenithEnvBuilder): # One safekeeper is enough for this test. zenith_env_builder.num_safekeepers = 1 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() pg = env.postgres.create_start('main') diff --git a/test_runner/batch_others/test_pageserver_api.py b/test_runner/batch_others/test_pageserver_api.py index eccffc4d69..ba1f106c4b 100644 --- a/test_runner/batch_others/test_pageserver_api.py +++ b/test_runner/batch_others/test_pageserver_api.py @@ -1,8 +1,15 @@ -import json from uuid import uuid4, UUID -from fixtures.zenith_fixtures import ZenithEnv, ZenithEnvBuilder, ZenithPageserverHttpClient -from typing import cast -import pytest, psycopg2 +import pytest +from fixtures.zenith_fixtures import ZenithEnv, ZenithEnvBuilder, ZenithPageserverHttpClient, zenith_binpath + + +# test that we cannot override node id +def test_pageserver_init_node_id(zenith_env_builder: ZenithEnvBuilder): + env = zenith_env_builder.init() + with pytest.raises( + Exception, + match="node id can only be set during pageserver init and cannot be overridden"): + env.pageserver.start(overrides=['--pageserver-config-override=id=10']) def check_client(client: ZenithPageserverHttpClient, initial_tenant: UUID): @@ -41,7 +48,7 @@ def test_pageserver_http_api_client(zenith_simple_env: ZenithEnv): def test_pageserver_http_api_client_auth_enabled(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.pageserver_auth_enabled = True - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() management_token = env.auth_keys.generate_management_token() diff --git a/test_runner/batch_others/test_pageserver_catchup.py b/test_runner/batch_others/test_pageserver_catchup.py index 97dc0f3260..985d1a3af0 100644 --- a/test_runner/batch_others/test_pageserver_catchup.py +++ b/test_runner/batch_others/test_pageserver_catchup.py @@ -14,7 +14,7 @@ from fixtures.log_helper import log # and new compute node contains all data. def test_pageserver_catchup_while_compute_down(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_pageserver_catchup_while_compute_down", "main") pg = env.postgres.create_start('test_pageserver_catchup_while_compute_down') diff --git a/test_runner/batch_others/test_pageserver_restart.py b/test_runner/batch_others/test_pageserver_restart.py index 0cfc50f0ff..ec93c2cf5b 100644 --- a/test_runner/batch_others/test_pageserver_restart.py +++ b/test_runner/batch_others/test_pageserver_restart.py @@ -13,7 +13,7 @@ from fixtures.log_helper import log def test_pageserver_restart(zenith_env_builder: ZenithEnvBuilder): # One safekeeper is enough for this test. zenith_env_builder.num_safekeepers = 1 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_pageserver_restart", "main") pg = env.postgres.create_start('test_pageserver_restart') diff --git a/test_runner/batch_others/test_remote_storage.py b/test_runner/batch_others/test_remote_storage.py index fa6feaf412..61feb1a5bd 100644 --- a/test_runner/batch_others/test_remote_storage.py +++ b/test_runner/batch_others/test_remote_storage.py @@ -42,7 +42,7 @@ def test_remote_storage_backup_and_restore(zenith_env_builder: ZenithEnvBuilder, data_secret = 'very secret secret' ##### First start, insert secret data and upload it to the remote storage - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() pg = env.postgres.create_start() tenant_id = pg.safe_psql("show zenith.zenith_tenant")[0][0] diff --git a/test_runner/batch_others/test_restart_compute.py b/test_runner/batch_others/test_restart_compute.py index f7810be555..d4dd3fb9e2 100644 --- a/test_runner/batch_others/test_restart_compute.py +++ b/test_runner/batch_others/test_restart_compute.py @@ -13,7 +13,7 @@ def test_restart_compute(zenith_env_builder: ZenithEnvBuilder, with_wal_acceptor zenith_env_builder.pageserver_auth_enabled = True if with_wal_acceptors: zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_restart_compute", "main") diff --git a/test_runner/batch_others/test_tenant_relocation.py b/test_runner/batch_others/test_tenant_relocation.py index 5c6d78e730..acff3ef62c 100644 --- a/test_runner/batch_others/test_tenant_relocation.py +++ b/test_runner/batch_others/test_tenant_relocation.py @@ -122,7 +122,7 @@ def test_tenant_relocation(zenith_env_builder: ZenithEnvBuilder, zenith_env_builder.num_safekeepers = 1 zenith_env_builder.enable_local_fs_remote_storage() - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() # create folder for remote storage mock remote_storage_mock_path = env.repo_dir / 'local_fs_remote_storage' diff --git a/test_runner/batch_others/test_tenants.py b/test_runner/batch_others/test_tenants.py index 232c724870..b665ae9022 100644 --- a/test_runner/batch_others/test_tenants.py +++ b/test_runner/batch_others/test_tenants.py @@ -10,7 +10,7 @@ def test_tenants_normal_work(zenith_env_builder: ZenithEnvBuilder, with_wal_acce if with_wal_acceptors: zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() """Tests tenants with and without wal acceptors""" tenant_1 = env.create_tenant() tenant_2 = env.create_tenant() diff --git a/test_runner/batch_others/test_timeline_size.py b/test_runner/batch_others/test_timeline_size.py index b48f830528..2c31267922 100644 --- a/test_runner/batch_others/test_timeline_size.py +++ b/test_runner/batch_others/test_timeline_size.py @@ -67,7 +67,7 @@ def wait_for_pageserver_catchup(pgmain: Postgres, polling_interval=1, timeout=60 def test_timeline_size_quota(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.num_safekeepers = 1 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_timeline_size_quota", "main") client = env.pageserver.http_client() diff --git a/test_runner/batch_others/test_wal_acceptor.py b/test_runner/batch_others/test_wal_acceptor.py index 5685bd57e4..c375c9626a 100644 --- a/test_runner/batch_others/test_wal_acceptor.py +++ b/test_runner/batch_others/test_wal_acceptor.py @@ -22,7 +22,7 @@ from typing import List, Optional, Any # succeed and data is written def test_normal_work(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_wal_acceptors_normal_work", "main") @@ -51,7 +51,7 @@ class BranchMetrics: # against different timelines. def test_many_timelines(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() n_timelines = 3 @@ -181,7 +181,7 @@ def test_restarts(zenith_env_builder: ZenithEnvBuilder): n_acceptors = 3 zenith_env_builder.num_safekeepers = n_acceptors - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_wal_acceptors_restarts", "main") pg = env.postgres.create_start('test_wal_acceptors_restarts') @@ -218,7 +218,7 @@ def delayed_wal_acceptor_start(wa): # When majority of acceptors is offline, commits are expected to be frozen def test_unavailability(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.num_safekeepers = 2 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_wal_acceptors_unavailability", "main") pg = env.postgres.create_start('test_wal_acceptors_unavailability') @@ -289,7 +289,7 @@ def stop_value(): def test_race_conditions(zenith_env_builder: ZenithEnvBuilder, stop_value): zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_wal_acceptors_race_conditions", "main") pg = env.postgres.create_start('test_wal_acceptors_race_conditions') @@ -404,7 +404,7 @@ def test_sync_safekeepers(zenith_env_builder: ZenithEnvBuilder, # We don't really need the full environment for this test, just the # safekeepers would be enough. zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() timeline_id = uuid.uuid4() tenant_id = uuid.uuid4() @@ -454,7 +454,7 @@ def test_sync_safekeepers(zenith_env_builder: ZenithEnvBuilder, def test_timeline_status(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.num_safekeepers = 1 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_timeline_status", "main") pg = env.postgres.create_start('test_timeline_status') @@ -629,7 +629,7 @@ def test_replace_safekeeper(zenith_env_builder: ZenithEnvBuilder): log.info(f"Safekeeper {sk.id} status error: {e}") zenith_env_builder.num_safekeepers = 4 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_replace_safekeeper", "main") log.info("Use only first 3 safekeepers") diff --git a/test_runner/batch_others/test_wal_acceptor_async.py b/test_runner/batch_others/test_wal_acceptor_async.py index 1d2a186eb7..4b6a27f73d 100644 --- a/test_runner/batch_others/test_wal_acceptor_async.py +++ b/test_runner/batch_others/test_wal_acceptor_async.py @@ -200,7 +200,7 @@ async def run_restarts_under_load(pg: Postgres, acceptors: List[Safekeeper], n_w # restart acceptors one by one, while executing and validating bank transactions def test_restarts_under_load(zenith_env_builder: ZenithEnvBuilder): zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() env.zenith_cli.create_branch("test_wal_acceptors_restarts_under_load", "main") pg = env.postgres.create_start('test_wal_acceptors_restarts_under_load') diff --git a/test_runner/batch_others/test_zenith_cli.py b/test_runner/batch_others/test_zenith_cli.py index ce051dfd6e..f1897e4b6f 100644 --- a/test_runner/batch_others/test_zenith_cli.py +++ b/test_runner/batch_others/test_zenith_cli.py @@ -97,7 +97,7 @@ def test_cli_tenant_list(zenith_simple_env: ZenithEnv): def test_cli_ipv4_listeners(zenith_env_builder: ZenithEnvBuilder): # Start with single sk zenith_env_builder.num_safekeepers = 1 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() # Connect to sk port on v4 loopback res = requests.get(f'http://127.0.0.1:{env.safekeepers[0].port.http}/v1/status') @@ -114,7 +114,7 @@ def test_cli_ipv4_listeners(zenith_env_builder: ZenithEnvBuilder): def test_cli_start_stop(zenith_env_builder: ZenithEnvBuilder): # Start with single sk zenith_env_builder.num_safekeepers = 1 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() # Stop default ps/sk env.zenith_cli.pageserver_stop() diff --git a/test_runner/fixtures/zenith_fixtures.py b/test_runner/fixtures/zenith_fixtures.py index 6312f5d99d..0f2a1b0af6 100644 --- a/test_runner/fixtures/zenith_fixtures.py +++ b/test_runner/fixtures/zenith_fixtures.py @@ -27,7 +27,7 @@ from dataclasses import dataclass # Type-related stuff from psycopg2.extensions import connection as PgConnection -from typing import Any, Callable, Dict, Iterator, List, Optional, TypeVar, cast, Union, Tuple +from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, TypeVar, cast, Union, Tuple from typing_extensions import Literal import pytest @@ -425,6 +425,14 @@ class ZenithEnvBuilder: self.env = ZenithEnv(self) return self.env + def start(self): + self.env.start() + + def init_start(self) -> ZenithEnv: + env = self.init() + self.start() + return env + """ Sets up the pageserver to use the local fs at the `test_dir/local_fs_remote_storage` path. Errors, if the pageserver has some remote storage configuration already, unless `force_enable` is not set to `True`. @@ -540,6 +548,7 @@ class ZenithEnv: toml += textwrap.dedent(f""" [pageserver] + id=1 listen_pg_addr = 'localhost:{pageserver_port.pg}' listen_http_addr = 'localhost:{pageserver_port.http}' auth_type = '{pageserver_auth_type}' @@ -557,13 +566,13 @@ class ZenithEnv: http=self.port_distributor.get_port(), ) id = i # assign ids sequentially - toml += f""" -[[safekeepers]] -id = {id} -pg_port = {port.pg} -http_port = {port.http} -sync = false # Disable fsyncs to make the tests go faster - """ + toml += textwrap.dedent(f""" + [[safekeepers]] + id = {id} + pg_port = {port.pg} + http_port = {port.http} + sync = false # Disable fsyncs to make the tests go faster + """) safekeeper = Safekeeper(env=self, id=id, port=port) self.safekeepers.append(safekeeper) @@ -571,6 +580,7 @@ sync = false # Disable fsyncs to make the tests go faster self.zenith_cli.init(toml) + def start(self): # Start up the page server and all the safekeepers self.pageserver.start() @@ -611,7 +621,7 @@ def _shared_simple_env(request: Any, port_distributor) -> Iterator[ZenithEnv]: with ZenithEnvBuilder(Path(repo_dir), port_distributor) as builder: - env = builder.init() + env = builder.init_start() # For convenience in tests, create a branch from the freshly-initialized cluster. env.zenith_cli.create_branch("empty", "main") @@ -645,7 +655,7 @@ def zenith_env_builder(test_output_dir, port_distributor) -> Iterator[ZenithEnvB To use, define 'zenith_env_builder' fixture in your test to get access to the builder object. Set properties on it to describe the environment. Finally, initialize and start up the environment by calling - zenith_env_builder.init(). + zenith_env_builder.init_start(). After the initialization, you can launch compute nodes by calling the functions in the 'env.postgres' factory object, stop/start the @@ -831,8 +841,9 @@ class ZenithCli: return self.raw_cli(cmd) - def pageserver_start(self) -> 'subprocess.CompletedProcess[str]': - start_args = ['pageserver', 'start'] + def pageserver_start(self, overrides=()) -> 'subprocess.CompletedProcess[str]': + start_args = ['pageserver', 'start', *overrides] + append_pageserver_param_overrides(start_args, self.env.pageserver.remote_storage) return self.raw_cli(start_args) @@ -985,14 +996,14 @@ class ZenithPageserver(PgProtocol): self.service_port = port # do not shadow PgProtocol.port which is just int self.remote_storage = remote_storage - def start(self) -> 'ZenithPageserver': + def start(self, overrides=()) -> 'ZenithPageserver': """ Start the page server. Returns self. """ assert self.running == False - self.env.zenith_cli.pageserver_start() + self.env.zenith_cli.pageserver_start(overrides=overrides) self.running = True return self @@ -1395,10 +1406,12 @@ class Safekeeper: port: SafekeeperPort id: int auth_token: Optional[str] = None + running: bool = False def start(self) -> 'Safekeeper': + assert self.running == False self.env.zenith_cli.safekeeper_start(self.id) - + self.running = True # wait for wal acceptor start by checking its status started_at = time.time() while True: @@ -1418,6 +1431,7 @@ class Safekeeper: def stop(self, immediate=False) -> 'Safekeeper': log.info('Stopping safekeeper {}'.format(self.id)) self.env.zenith_cli.safekeeper_stop(self.id, immediate) + self.running = False return self def append_logical_message(self, diff --git a/test_runner/performance/test_bulk_tenant_create.py b/test_runner/performance/test_bulk_tenant_create.py index 6fd77f3020..0247385211 100644 --- a/test_runner/performance/test_bulk_tenant_create.py +++ b/test_runner/performance/test_bulk_tenant_create.py @@ -23,7 +23,7 @@ def test_bulk_tenant_create( """Measure tenant creation time (with and without wal acceptors)""" if use_wal_acceptors == 'with_wa': zenith_env_builder.num_safekeepers = 3 - env = zenith_env_builder.init() + env = zenith_env_builder.init_start() time_slices = [] diff --git a/zenith/src/main.rs b/zenith/src/main.rs index 904896f01e..5500d924ea 100644 --- a/zenith/src/main.rs +++ b/zenith/src/main.rs @@ -25,21 +25,24 @@ use pageserver::branches::BranchInfo; // Default id of a safekeeper node, if not specified on the command line. const DEFAULT_SAFEKEEPER_ID: ZNodeId = ZNodeId(1); +const DEFAULT_PAGESERVER_ID: ZNodeId = ZNodeId(1); fn default_conf() -> String { format!( r#" # Default built-in configuration, defined in main.rs [pageserver] +id = {pageserver_id} listen_pg_addr = '{pageserver_pg_addr}' listen_http_addr = '{pageserver_http_addr}' auth_type = '{pageserver_auth_type}' [[safekeepers]] -id = '{safekeeper_id}' +id = {safekeeper_id} pg_port = {safekeeper_pg_port} http_port = {safekeeper_http_port} "#, + pageserver_id = DEFAULT_PAGESERVER_ID, pageserver_pg_addr = DEFAULT_PAGESERVER_PG_ADDR, pageserver_http_addr = DEFAULT_PAGESERVER_HTTP_ADDR, pageserver_auth_type = AuthType::Trust,