From 5d490babf8cebccc61f7f02d79d62f26bfba80a7 Mon Sep 17 00:00:00 2001 From: Dmitry Rodionov Date: Tue, 22 Feb 2022 13:52:01 +0300 Subject: [PATCH] add node id to pageserver This adds node id parameter to pageserver configuration. Also I use a simple builder to construct pageserver config struct to avoid setting node id to some temporary invalid value. Some of the changes in test fixtures are needed to split init and start operations for envrionment. --- control_plane/src/local_env.rs | 3 + control_plane/src/storage.rs | 3 + pageserver/src/bin/pageserver.rs | 9 +- pageserver/src/config.rs | 234 +++++++++++++++--- pageserver/src/http/models.rs | 6 + pageserver/src/http/openapi_spec.yml | 5 + pageserver/src/http/routes.rs | 14 +- test_runner/README.md | 2 +- test_runner/batch_others/test_auth.py | 4 +- .../batch_others/test_branch_behind.py | 2 +- test_runner/batch_others/test_next_xid.py | 2 +- .../batch_others/test_pageserver_api.py | 17 +- .../batch_others/test_pageserver_catchup.py | 2 +- .../batch_others/test_pageserver_restart.py | 2 +- .../batch_others/test_remote_storage.py | 2 +- .../batch_others/test_restart_compute.py | 2 +- .../batch_others/test_tenant_relocation.py | 2 +- test_runner/batch_others/test_tenants.py | 2 +- .../batch_others/test_timeline_size.py | 2 +- test_runner/batch_others/test_wal_acceptor.py | 16 +- .../batch_others/test_wal_acceptor_async.py | 2 +- test_runner/batch_others/test_zenith_cli.py | 4 +- test_runner/fixtures/zenith_fixtures.py | 44 ++-- .../performance/test_bulk_tenant_create.py | 2 +- zenith/src/main.rs | 5 +- 25 files changed, 294 insertions(+), 94 deletions(-) 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,