rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-01-03 19:42:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add option allowing running multiple endpoints on the same branch.

Browse Source 
This is used by safekeeper tests.
This commit is contained in:
Arseny Sher
2024-05-02 11:51:24 +03:00
committed by Arseny Sher
parent 0353a72a00
commit e6da7e29ed
3 changed files with 58 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
control_plane/src/bin/neon_local.rs
View File

@@ -837,6 +837,8 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re
.copied()
.unwrap_or(false);
let allow_multiple = sub_args.get_flag("allow-multiple");
let mode = match (lsn, hot_standby) {
(Some(lsn), false) => ComputeMode::Static(lsn),
(None, true) => ComputeMode::Replica,
@@ -854,7 +856,9 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re
_ => {}
}
cplane.check_conflicting_endpoints(mode, tenant_id, timeline_id)?;
if !allow_multiple {
cplane.check_conflicting_endpoints(mode, tenant_id, timeline_id)?;
}
cplane.new_endpoint(
&endpoint_id,
@@ -883,6 +887,8 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re
let remote_ext_config = sub_args.get_one::<String>("remote-ext-config");
let allow_multiple = sub_args.get_flag("allow-multiple");
// If --safekeepers argument is given, use only the listed safekeeper nodes.
let safekeepers =
if let Some(safekeepers_str) = sub_args.get_one::<String>("safekeepers") {
@@ -908,11 +914,13 @@ async fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Re
.cloned()
.unwrap_or_default();
cplane.check_conflicting_endpoints(
endpoint.mode,
endpoint.tenant_id,
endpoint.timeline_id,
)?;
if !allow_multiple {
cplane.check_conflicting_endpoints(
endpoint.mode,
endpoint.tenant_id,
endpoint.timeline_id,
)?;
}
let (pageservers, stripe_size) = if let Some(pageserver_id) = pageserver_id {
let conf = env.get_pageserver_conf(pageserver_id).unwrap();
@@ -1444,6 +1452,12 @@ fn cli() -> Command {
.help("If set, will create test user `user` and `neondb` database. Requires `update-catalog = true`")
.required(false);
let allow_multiple = Arg::new("allow-multiple")
.help("Allow multiple primary endpoints running on the same branch. Shouldn't be used normally, but useful for tests.")
.long("allow-multiple")
.action(ArgAction::SetTrue)
.required(false);
Command::new("Neon CLI")
.arg_required_else_help(true)
.version(GIT_VERSION)
@@ -1601,6 +1615,7 @@ fn cli() -> Command {
.arg(pg_version_arg.clone())
.arg(hot_standby_arg.clone())
.arg(update_catalog)
.arg(allow_multiple.clone())
)
.subcommand(Command::new("start")
.about("Start postgres.\n If the endpoint doesn't exist yet, it is created.")
@@ -1609,6 +1624,7 @@ fn cli() -> Command {
.arg(safekeepers_arg)
.arg(remote_ext_config_args)
.arg(create_test_user)
.arg(allow_multiple.clone())
)
.subcommand(Command::new("reconfigure")
.about("Reconfigure the endpoint")

22
test_runner/fixtures/neon_fixtures.py
View File

@@ -1801,6 +1801,7 @@ class NeonCli(AbstractNeonCli):
hot_standby: bool = False,
lsn: Optional[Lsn] = None,
pageserver_id: Optional[int] = None,
allow_multiple=False,
) -> "subprocess.CompletedProcess[str]":
args = [
"endpoint",
@@ -1824,6 +1825,8 @@ class NeonCli(AbstractNeonCli):
args.extend(["--hot-standby", "true"])
if pageserver_id is not None:
args.extend(["--pageserver-id", str(pageserver_id)])
if allow_multiple:
args.extend(["--allow-multiple"])
res = self.raw_cli(args)
res.check_returncode()
@@ -1835,6 +1838,7 @@ class NeonCli(AbstractNeonCli):
safekeepers: Optional[List[int]] = None,
remote_ext_config: Optional[str] = None,
pageserver_id: Optional[int] = None,
allow_multiple=False,
) -> "subprocess.CompletedProcess[str]":
args = [
"endpoint",
@@ -1849,6 +1853,8 @@ class NeonCli(AbstractNeonCli):
args.append(endpoint_id)
if pageserver_id is not None:
args.extend(["--pageserver-id", str(pageserver_id)])
if allow_multiple:
args.extend(["--allow-multiple"])
res = self.raw_cli(args)
res.check_returncode()
@@ -3299,6 +3305,7 @@ class Endpoint(PgProtocol):
lsn: Optional[Lsn] = None,
config_lines: Optional[List[str]] = None,
pageserver_id: Optional[int] = None,
allow_multiple: bool = False,
) -> "Endpoint":
"""
Create a new Postgres endpoint.
@@ -3321,6 +3328,7 @@ class Endpoint(PgProtocol):
pg_port=self.pg_port,
http_port=self.http_port,
pageserver_id=pageserver_id,
allow_multiple=allow_multiple,
)
path = Path("endpoints") / self.endpoint_id / "pgdata"
self.pgdata_dir = os.path.join(self.env.repo_dir, path)
@@ -3337,7 +3345,10 @@ class Endpoint(PgProtocol):
return self
def start(
self, remote_ext_config: Optional[str] = None, pageserver_id: Optional[int] = None
self,
remote_ext_config: Optional[str] = None,
pageserver_id: Optional[int] = None,
allow_multiple: bool = False,
) -> "Endpoint":
"""
Start the Postgres instance.
@@ -3353,6 +3364,7 @@ class Endpoint(PgProtocol):
safekeepers=self.active_safekeepers,
remote_ext_config=remote_ext_config,
pageserver_id=pageserver_id,
allow_multiple=allow_multiple,
)
self.running = True
@@ -3482,6 +3494,7 @@ class Endpoint(PgProtocol):
config_lines: Optional[List[str]] = None,
remote_ext_config: Optional[str] = None,
pageserver_id: Optional[int] = None,
allow_multiple=False,
) -> "Endpoint":
"""
Create an endpoint, apply config, and start Postgres.
@@ -3497,7 +3510,12 @@ class Endpoint(PgProtocol):
hot_standby=hot_standby,
lsn=lsn,
pageserver_id=pageserver_id,
).start(remote_ext_config=remote_ext_config, pageserver_id=pageserver_id)
allow_multiple=allow_multiple,
).start(
remote_ext_config=remote_ext_config,
pageserver_id=pageserver_id,
allow_multiple=allow_multiple,
)
log.info(f"Postgres startup took {time.time() - started_at} seconds")

20
test_runner/regress/test_wal_acceptor_async.py
View File

@@ -254,7 +254,9 @@ def test_restarts_frequent_checkpoints(neon_env_builder: NeonEnvBuilder):
)
def endpoint_create_start(env: NeonEnv, branch: str, pgdir_name: Optional[str]):
def endpoint_create_start(
env: NeonEnv, branch: str, pgdir_name: Optional[str], allow_multiple: bool = False
):
endpoint = Endpoint(
env,
tenant_id=env.initial_tenant,
@@ -268,14 +270,23 @@ def endpoint_create_start(env: NeonEnv, branch: str, pgdir_name: Optional[str]):
# embed current time in endpoint ID
endpoint_id = pgdir_name or f"ep-{time.time()}"
return endpoint.create_start(
branch_name=branch, endpoint_id=endpoint_id, config_lines=["log_statement=all"]
branch_name=branch,
endpoint_id=endpoint_id,
config_lines=["log_statement=all"],
allow_multiple=allow_multiple,
)
async def exec_compute_query(
env: NeonEnv, branch: str, query: str, pgdir_name: Optional[str] = None
env: NeonEnv,
branch: str,
query: str,
pgdir_name: Optional[str] = None,
allow_multiple: bool = False,
):
with endpoint_create_start(env, branch=branch, pgdir_name=pgdir_name) as endpoint:
with endpoint_create_start(
env, branch=branch, pgdir_name=pgdir_name, allow_multiple=allow_multiple
) as endpoint:
before_conn = time.time()
conn = await endpoint.connect_async()
res = await conn.fetch(query)
@@ -347,6 +358,7 @@ class BackgroundCompute(object):
self.branch,
f"INSERT INTO query_log(index, verify_key) VALUES ({self.index}, {verify_key}) RETURNING verify_key",
pgdir_name=f"bgcompute{self.index}_key{verify_key}",
allow_multiple=True,
)
log.info(f"result: {res}")
if len(res) != 1: