From 572d20dfc04ca0f53fa04b29a2978ec847ac804e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcos=20M=C3=A9ndez?= Date: Fri, 9 Oct 2026 05:06:23 +0000 Subject: [PATCH 1/2] fix: the new primary takes writes as soon as it carries the VIP (keel#118) After keel vip promote --old-primary-gone on a real pair, the new primary refused writes for 11.7 s while it carried the VIP. keel database follow promoted the database 7.5 s after the VIP landed. The cause is STOP SLAVE IO_THREAD: it returns only when the I/O thread ends. Measured on MariaDB 11.8 with the primary gone: a thread that tries to connect ends when the try times out (slave_net_timeout, 10 s; 2.96 s with 3 s); a connected thread with semi-synchronous replication tries to connect to the primary to end its dump thread (5.06 s, 0.07 s without semi-synchronous replication). A KILL of the thread does not shorten either. The status shows "Ending IO thread" at once: the thread receives nothing from the moment the stop marks it killed. keel.system.dbreadonly.promote now runs the stop in a client of its own. Once the stop is in the server's killing phase (its process list STATE is "Killing slave"; INFO is set at dispatch, before the stop takes its lock) or ended without an error, the SQL thread drains, the received position and the killing phase are checked again, and read_only goes off; STOP SLAVE and RESET SLAVE ALL follow while the server already takes writes, and the stop is waited for. The SQL thread is not stopped first: it has nothing more to apply, and STOP SLAVE SQL_THREAD waits behind the I/O thread's stop (4.7 s measured). Against a frozen primary on 11.8 the server took writes 0.35 s after the promotion started (0.26 s with the thread connecting). When the position moved after the drain, or cannot be read, the old order applies: the stop waited for, the drain again, then the promotion. Follow looks for the VIP on wg0 every 0.2 s, not 0.5 s. A planned promote takes the same path; its stop ends at once. Spec apply's SeedReplica and PromoteReplica take follow's lock, so apply, follow and watch never interleave. A stop client that failed after the server may have taken the statement starts the I/O thread again. Tests: the writes before the stop ends; a stop not in the killing phase after the drain takes the old order; anything received after the drain takes the old order; a stop the server does not show is waited for first; each failure. In mariadb / trixie the promote no longer holds the agent's lock, as on a machine, where keel-database-follow.path runs follow while the promote goes on; the crash case checks that the server takes writes within 3 s after the VIP lands, with no acknowledged row lost. The crash case kills A only once C's writes were acknowledged, and runs the promote again, as an operator does, when etcd had no leader to answer (A, cut off, may have been etcd's leader), at most three times, the attempts kept. --- debian/changelog | 19 +++ docs/replication.md | 27 +++++ keel/__init__.py | 2 +- keel/system/dbfollow.py | 14 +-- keel/system/dbreadonly.py | 201 ++++++++++++++++++++++++++++++-- keel/system/effects.py | 6 +- pyproject.toml | 2 +- tests/mariadb_netns.py | 29 ++++- tests/test_mariadb_netns.py | 6 + tests/test_system_dbfollow.py | 13 ++- tests/test_system_dbreadonly.py | 180 +++++++++++++++++++++++++++- 11 files changed, 461 insertions(+), 38 deletions(-) diff --git a/debian/changelog b/debian/changelog index a75ce20..24b4d39 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,22 @@ +keel (0.23.10) unstable; urgency=medium + + * After keel vip promote --old-primary-gone the new primary refused + writes for 11.7 s while it carried the VIP (keel#118). keel database + follow promoted the database 7.5 s after the VIP landed: STOP SLAVE + IO_THREAD waits until the I/O thread ends, which with a dead primary + takes up to 10 s (a try to connect) or 5 s (the semi-synchronous + kill of the dump thread). The thread receives nothing once the stop + marks it killed. The promotion now runs the stop in a client of its + own; once the stop is in the server's killing phase, the SQL thread + drains, the received position and the killing phase are checked + again, and read_only goes off; the I/O thread is + waited for and the primary forgotten after. When anything + arrived after the drain, the old order applies. A planned promote + takes the same path. Spec apply's replication actions take follow's + lock. Follow looks for the VIP every 0.2 s, not 0.5 s. + + -- Marcos Méndez Fri, 09 Oct 2026 19:00:00 +0000 + keel (0.23.9) unstable; urgency=medium * With etcd, two nodes could carry the VIP for less than a second diff --git a/docs/replication.md b/docs/replication.md index f246241..fe8664e 100644 --- a/docs/replication.md +++ b/docs/replication.md @@ -230,6 +230,33 @@ without the flag when the old primary does not answer: 0049, second round, point 3), then the database: drain the relay log, `STOP SLAVE`, `RESET SLAVE ALL`, `read_only = OFF`, `READ_ONLY ADMIN` back. The semi-synchronous roles need no flip (both sides enabled on both nodes). +When the old primary is gone, `STOP SLAVE IO_THREAD` takes seconds +(measured on 11.8): a thread that tries to connect ends only when the +try times out (`slave_net_timeout`, 10 s), and a connected thread with +semi-synchronous replication tries to connect to the primary to end its +dump thread (5 s). A `KILL` of the thread does not shorten it. On a real +pair that cost 7.5 s of refused writes while the new primary already +carried the VIP (keel#118). The thread receives nothing from the moment +the stop marks it killed, long before it ends; a thread that connects +again (`MASTER_CONNECT_RETRY=2`) after that ends without receiving. So +the stop runs in a client of its own; once the stop is in the server's +killing phase (its process list `STATE` is `Killing slave`; its `INFO` +is set earlier, when it is dispatched) or ended without an error, the +SQL thread drains, the received position is read again, the killing +phase is checked again, and `read_only = OFF`; `STOP SLAVE` and `RESET +SLAVE ALL` follow while the server already takes writes. The SQL thread +is not stopped first: it has nothing more to apply, and `STOP SLAVE +SQL_THREAD` waits behind the I/O thread's stop (4.7 s measured). On +11.8 against a frozen primary, the server took writes 0.35 s after the +promotion started (0.26 s with the thread connecting), against 5 s and +7 s before. A planned promote takes the same path; its stop ends at +once, since the primary answers. Follow, `keel database watch`'s follow +and spec apply's replication actions take one lock, +`/var/lib/keel/database/follow.lock`. When the position moved after the drain, or +cannot be read, the old order applies: the stop waited for, the SQL +thread started and drained again, then the promotion. +`keel-database-follow.path` starts follow when the claim is written, +and follow looks for the VIP on `wg0` every 0.2 s. **The database follows the VIP** (0020: everything that depends on the role follows the elected role). `keel-database-follow.path` watches diff --git a/keel/__init__.py b/keel/__init__.py index 981037b..61c46fb 100644 --- a/keel/__init__.py +++ b/keel/__init__.py @@ -12,7 +12,7 @@ # an image's tree read offline): debian/changelog's and pyproject.toml's, # which tests/test_version.py keeps in step. Installed, the metadata # dpkg put beside the package answers (keel#98). -FALLBACK_VERSION = "0.23.9" +FALLBACK_VERSION = "0.23.10" def version() -> str: diff --git a/keel/system/dbfollow.py b/keel/system/dbfollow.py index 65493b8..e80765e 100644 --- a/keel/system/dbfollow.py +++ b/keel/system/dbfollow.py @@ -44,7 +44,6 @@ (keel.system.dbmariadb.CLIENT): root has no READ_ONLY ADMIN on a replica. """ -import fcntl import json import os import subprocess @@ -73,7 +72,7 @@ ) DIVERGED = "var/lib/keel/database/diverged" -FOLLOW_LOCK = "var/lib/keel/database/follow.lock" +FOLLOW_LOCK = dbreadonly.REPLICATION_LOCK BACKUPS = "var/backups/keel/mariadb" DROPIN_MODE = 0o644 GTID_QUESTION = mariadb.QUIET_CLIENT + ( @@ -84,7 +83,8 @@ # on wg0, before it leaves the server read only: keel vip promote's own # wait for the controller to carry it (keel.mesh.vippromote.CARRY_WAIT) PROOF_WAIT = 15.0 -PROOF_POLL = 0.5 +# how often follow looks for the VIP on wg0 while it waits (keel#118) +PROOF_POLL = 0.2 NO_PAIR = ("this node declares no appliance.vip: there is no VIP for the" " database to follow (keel spec apply converges a declared role)") DIVERGED_TITLE = "the old primary diverged and stays read only" @@ -130,14 +130,8 @@ def follow(root: str, spec: str, confirmed: bool, found: Followed, wait: float = PROOF_WAIT, sleep=time.sleep) -> None: """One run at a time: keel-database-follow.service and keel database watch never interleave between the proof and SET read_only OFF""" - lock = os.path.join(root, FOLLOW_LOCK) - os.makedirs(os.path.dirname(lock), exist_ok=True) - fd = os.open(lock, os.O_RDWR | os.O_CREAT, 0o600) - try: - fcntl.flock(fd, fcntl.LOCK_EX) + with dbreadonly.replication_locked(root): _follow(root, spec, confirmed, found, runner, output, wait, sleep) - finally: - os.close(fd) def _follow(root: str, spec: str, confirmed: bool, found: Followed, diff --git a/keel/system/dbreadonly.py b/keel/system/dbreadonly.py index 60a8d47..53ea51f 100644 --- a/keel/system/dbreadonly.py +++ b/keel/system/dbreadonly.py @@ -31,9 +31,13 @@ keel.system.effects is the only caller. """ +import contextlib +import fcntl +import io import os import subprocess import time +from collections.abc import Iterator from keel.inspect.dbengines import grantee, own_account from keel.inspect.dbreading import field_lines @@ -50,6 +54,25 @@ STATUS_QUESTION = mariadb.CLIENT + ("--execute", "SHOW REPLICA STATUS\\G") NO_BINLOG = "SET SESSION sql_log_bin = 0;\n" POLL_SECONDS = 1.0 +# the lock follow and spec apply take around the replication's statements +REPLICATION_LOCK = "var/lib/keel/database/follow.lock" +IO_STOP = "STOP SLAVE IO_THREAD" +# subprocess.run as the module found it: a runner that is anything else +# (a test's) runs the stop to its end itself +REAL_RUN = subprocess.run +# the stop running in the server: its own statement, exactly +STOPPING_QUESTION = QUIET + ( + "SELECT STATE FROM information_schema.PROCESSLIST" + " WHERE INFO = 'STOP SLAVE IO_THREAD'", +) +# the stop's STATE while it kills the thread (sql/slave.cc stop_slave) +KILLING = "Killing slave" +# how long the stop may take to show in the server, and how often it is +# looked for; how long it may take to end (slave_net_timeout and the +# 5 s of the semi-synchronous kill, with room) +SHOWN_WAIT = 5.0 +SHOWN_POLL = 0.1 +STOP_WAIT = 60 NOT_REPLICATING = ( "this server replicates from nowhere, so there is nothing to promote." @@ -132,9 +155,37 @@ def unlock(root: str, runner=subprocess.run) -> str | None: def promote( root: str, action, runner=subprocess.run, clock=time.monotonic, - sleep=time.sleep, + sleep=time.sleep, spawn=None, ) -> str | None: - """Drain the SQL thread, then promote; None on success, else why not""" + """Stop receiving, drain, then take writes; None on success, else + why not + + STOP SLAVE IO_THREAD can take seconds when the primary is dead + (keel#118). Measured on 11.8: a thread that tries to connect ends + only when the try times out (slave_net_timeout, 10 s); a connected + thread with semi-synchronous replication tries to connect to the + primary to end its dump thread (5 s). So the stop runs in a client + of its own, and read_only goes off before it ends only when all of + this holds: + + - the stop is in the server's killing phase (its process list STATE + is "Killing slave"), or it ended without an error, both before + the drain and again after it: the server marks the I/O thread + killed in that phase, and a killed thread receives no more events + (it leaves its read loop, and one that connects ends before it + asks for the binary log); + - the SQL thread applied everything received (the drain); + - the received position read after the drain is the drained one. + + Otherwise the old order applies: the stop waited for, the drain + again, then the promotion. The guarantee is the old order's: read_only + goes off only once the I/O thread receives nothing and the SQL thread + applied all it received. The SQL thread is not stopped first: it has + nothing more to apply, and STOP SLAVE SQL_THREAD waits behind the I/O + thread's stop (4.7 s measured). The I/O thread is waited for and the + primary forgotten after. A planned promote takes the same path; its + stop ends at once, since the primary answers. + """ values, problem = _status(runner) if problem: return f"cannot read the replica status: {problem}. Nothing was" \ @@ -143,11 +194,139 @@ def promote( return NOT_REPLICATING if values.get("Slave_SQL_Running", "").lower() != "yes": return STOPPED.format(error=_error(values)) - problem = _send(runner, "STOP SLAVE IO_THREAD;\n") + stopping, problem = _stop_io(runner, spawn) if problem: return f"stopping the I/O thread failed ({problem}). Nothing was" \ " changed" - problem = _drain(runner, action.timeout, clock, sleep) + if not _stop_shown(runner, stopping, clock, sleep): + problem = _waited(stopping) + if problem: + # the server may have taken the statement before the client + # died: the I/O thread is started again either way + return f"stopping the I/O thread failed ({problem}). Nothing" \ + " was promoted" + _resume(runner) + problem, drained = _drain(runner, action.timeout, clock, sleep) + if problem: + _waited(stopping) + return problem + _resume(runner) + # the SQL thread is left running: it has nothing more to apply, and + # STOP SLAVE SQL_THREAD would wait behind the I/O thread's stop + after, problem = _status(runner) + if problem or _received(drained, after) or \ + not _killing(runner, stopping): + return _promote_in_order(runner, stopping, action.timeout, clock, + sleep) + writable = mariadb.set_read_only(False) + problem = _send(runner, writable.text) + if problem: + _waited(stopping) + return f"{writable.summary} failed: {problem}" + forget = mariadb.stop_replicating() + problem = _send(runner, forget.text) + _waited(stopping) + if problem: + return (f"the server takes writes, but {forget.summary} failed:" + f" {problem}. Run STOP SLAVE; RESET SLAVE ALL before its" + " next restart") + return None + + +@contextlib.contextmanager +def replication_locked(root: str) -> Iterator[None]: + """One writer of the replication's statements at a time: keel + database follow, keel database watch's follow and spec apply's + replication actions (keel#118)""" + path = os.path.join(root, REPLICATION_LOCK) + os.makedirs(os.path.dirname(path), exist_ok=True) + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + yield + finally: + os.close(fd) + + +class _Ran: + """A client that already ran to its end, as a process looks""" + + def __init__(self, done): + self.returncode = done.returncode + self.stderr = io.StringIO(done.stderr or "") + + def poll(self) -> int: + return self.returncode + + def wait(self, timeout=None) -> int: + return self.returncode + + +def _stop_io(runner, spawn): + """STOP SLAVE IO_THREAD in a client of its own, `spawn` (Popen when + the runner is subprocess.run); with another runner, run to its end + by it. (the process, why it could not start)""" + argv = list(QUIET) + [IO_STOP] + try: + if spawn is None and runner is not REAL_RUN: + return _Ran(runner(argv, capture_output=True, text=True, + check=False)), "" + return (spawn or subprocess.Popen)( + argv, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + text=True), "" + except OSError as e: + return None, f"cannot run {mariadb.CLIENT[0]}: {e.strerror}" + + +def _stop_shown(runner, stopping, clock, sleep) -> bool: + """Whether the stop reached the server's killing phase, waited for + up to SHOWN_WAIT""" + deadline = clock() + SHOWN_WAIT + while True: + if _killing(runner, stopping): + return True + if stopping.poll() is not None or clock() >= deadline: + return False + sleep(SHOWN_POLL) + + +def _killing(runner, stopping) -> bool: + """Whether the stop ended without an error, or is in the server's + killing phase: its process list STATE is KILLING (the statement's + INFO is set when it is dispatched, before the stop takes its lock, + so INFO alone says nothing)""" + done = stopping.poll() + if done is not None: + return done == 0 + out, problem = _ask(runner, STOPPING_QUESTION) + return not problem and KILLING in out.splitlines() + + +def _waited(stopping) -> str: + """The stop waited for; an empty string, or why it failed""" + try: + stopping.wait(STOP_WAIT) + except subprocess.TimeoutExpired: + return f"it did not end within {STOP_WAIT} s" + if stopping.returncode != 0: + said = stopping.stderr.read() if stopping.stderr else "" + return (said or "").strip() or f"exited {stopping.returncode}" + return "" + + +def _received(drained: dict[str, str], after: dict[str, str]) -> bool: + """Whether the I/O thread received anything after the drain""" + return any(drained.get(one) != after.get(one) + for one in ("Master_Log_File", "Read_Master_Log_Pos")) + + +def _promote_in_order(runner, stopping, timeout: int, clock, + sleep) -> str | None: + """The old order: the stop waited for, the SQL thread started and + drained again, then STOP SLAVE, RESET SLAVE ALL, read_only off""" + problem = _waited(stopping) + if problem: + return f"stopping the I/O thread failed ({problem}). Nothing was" \ + " promoted" + _resume(runner) + problem, _ = _drain(runner, timeout, clock, sleep) if problem: return problem + _resume(runner) statements = mariadb.promote() @@ -166,20 +345,22 @@ def _resume(runner) -> str: return RESUMED -def _drain(runner, timeout: int, clock, sleep) -> str: - """Wait until the SQL thread applied everything received; or why not""" +def _drain(runner, timeout: int, clock, + sleep) -> tuple[str, dict[str, str]]: + """Wait until the SQL thread applied everything received; ("", the + status then), or (why not, {})""" deadline = clock() + timeout while True: values, problem = _status(runner) if problem: return f"the replica status could not be read while draining" \ - f" ({problem}); nothing was promoted" + f" ({problem}); nothing was promoted", {} if values.get("Slave_SQL_Running", "").lower() != "yes": - return STOPPED_DRAINING.format(error=_error(values)) + return STOPPED_DRAINING.format(error=_error(values)), {} if _drained(values): - return "" + return "", values if clock() >= deadline: - return SLOW.format(timeout=timeout) + return SLOW.format(timeout=timeout), {} sleep(POLL_SECONDS) diff --git a/keel/system/effects.py b/keel/system/effects.py index 09a58ba..1185fa3 100644 --- a/keel/system/effects.py +++ b/keel/system/effects.py @@ -76,13 +76,15 @@ def apply(self, action: Change) -> str | None: return wgkeys.adopt(self.tree.path(action.conf), self.tree.path(action.path.lstrip("/"))) if isinstance(action, SeedReplica): - return dbseed.seed(self.tree.root, action) + with dbreadonly.replication_locked(self.tree.root): + return dbseed.seed(self.tree.root, action) if isinstance(action, LockReplica): return dbreadonly.lock(self.tree.root) if isinstance(action, UnlockAccounts): return dbreadonly.unlock(self.tree.root) if isinstance(action, PromoteReplica): - return dbreadonly.promote(self.tree.root, action) + with dbreadonly.replication_locked(self.tree.root): + return dbreadonly.promote(self.tree.root, action) if isinstance(action, EnsureDatabaseTls): return self.database_tls(action) if isinstance(action, FollowVip): diff --git a/pyproject.toml b/pyproject.toml index a3abbcb..bf84793 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "keel" -version = "0.23.9" +version = "0.23.10" authors = [{name = "KeelLinux maintainers"}] description = "Declarative management of one appliance instance" readme = "README.md" diff --git a/tests/mariadb_netns.py b/tests/mariadb_netns.py index 00bc9d4..8852e8d 100644 --- a/tests/mariadb_netns.py +++ b/tests/mariadb_netns.py @@ -268,7 +268,10 @@ def database_command(here, words: list[str]) -> dict | None: from keel import cli out = io.StringIO() started = time.monotonic() - with LOCK, contextlib.redirect_stdout(out), \ + # not under LOCK: on a machine keel-database-follow.path runs + # follow while the promote goes on (when the claim is written), + # and follow's own lock orders the two runs (keel#118) + with contextlib.redirect_stdout(out), \ contextlib.redirect_stderr(out): code = cli.main(["database", "promote", "--spec", path, "--root", root] + (["--old-primary-gone"] @@ -765,6 +768,22 @@ def acked_missing(writer: Writer, agent_send) -> list[str]: return sorted(acked - held)[:20] +def promote_gone(agent_send) -> dict: + """keel database promote --old-primary-gone on B, as an operator runs + it: run again when etcd had no leader to answer (A, cut off, may + have been etcd's leader: the election takes up to 5 s), at most + three times; the attempts are kept""" + attempts = [] + for _ in range(3): + found = agent_send("B", "dbpromote gone", 300) + attempts.append({"code": found.get("code"), + "took_s": found.get("took_s")}) + said = " ".join(found.get("out") or []) + if found.get("code") != 16 or "etcd did not answer" not in said: + break + return {**found, "attempts": attempts} + + def crashed(report: dict, pids: list[int], roots: list[str]) -> None: """keel#104: the primary A crashes (its server killed, its VIP helper gone, cut off), B is promoted at a newer epoch with A gone, @@ -790,7 +809,11 @@ def crashed(report: dict, pids: list[int], roots: list[str]) -> None: watched = NewPrimary(pids[1], roots[1]) writer.start() watched.start() - time.sleep(5) + # writes acknowledged before the kill, so the downtime has a start: + # a connection at the VIP over 250 ms takes 1.3 to 1.9 s + report["k108_first_ack_s"] = wait_until( + lambda: any(one["ok"] for one in writer.done), 60, 0.2) + time.sleep(3) killed_at = time.time() sh("systemctl", "kill", "--signal=SIGKILL", service_of(a_root), check=False) @@ -800,7 +823,7 @@ def crashed(report: dict, pids: list[int], roots: list[str]) -> None: sh("tc", "qdisc", "replace", "dev", "uplink", "root", "netem", "loss", "100%", pid=a_pid) promote_at = time.time() - report["k104_promote"] = agent_send("B", "dbpromote gone", 300) + report["k104_promote"] = promote_gone(agent_send) promoted_at = time.time() wait_until(lambda: watched.writable_at is not None, 60, 0.2) time.sleep(5) diff --git a/tests/test_mariadb_netns.py b/tests/test_mariadb_netns.py index bf09c39..3accb0d 100644 --- a/tests/test_mariadb_netns.py +++ b/tests/test_mariadb_netns.py @@ -210,6 +210,12 @@ def test_the_unplanned_write_downtime_and_no_acked_row_lost(self): said = " ".join(self.found["k104_promote"]["out"]) self.assertIn("no answer within 1.5 s", said) self.assertNotIn("timed out", said) + # follow does not lag the VIP (keel#118: 7.5 s on real nodes) + self.assertLess(found["writable_after_carried_s"], 3.0, found) + # with etcd the old primary's lease must expire before the claim + # (0049); after the VIP lands, the writes resume within 3 s + self.assertLess(found["first_ok_after_s"] - found["carried_s"], 3.0, + found) def test_no_two_nodes_ever_carry_the_vip(self): self.assertEqual(self.found["max_holders"], 1, diff --git a/tests/test_system_dbfollow.py b/tests/test_system_dbfollow.py index 3aa1e14..5c9d595 100644 --- a/tests/test_system_dbfollow.py +++ b/tests/test_system_dbfollow.py @@ -230,10 +230,12 @@ def test_a_replica_that_takes_the_vip_is_drained_and_promoted(self): found = self.follow(0, server) self.assertIsNone(found.problem, found.lines) self.assertIn("drained and promoted", " ".join(found.lines)) - self.assertTrue(any("STOP SLAVE IO_THREAD" in one - for one in server.sent)) - self.assertTrue(any("RESET SLAVE ALL" in one and "read_only = OFF" - in one for one in server.sent)) + self.assertIn("STOP SLAVE IO_THREAD", [one[-1] for one in + server.argv]) + # writable once drained, the primary forgotten after + sent = "".join(server.sent) + self.assertLess(sent.index("read_only = OFF"), + sent.index("RESET SLAVE ALL")) def test_the_privilege_goes_back_to_what_the_replica_took_it_from(self): self.pair_nodes() @@ -595,7 +597,8 @@ def test_follow_runs_under_its_lock(self): def flock(fd, how): taken.append(how) return real(fd, how) - with mock.patch.object(dbfollow.fcntl, "flock", side_effect=flock): + with mock.patch.object(dbfollow.dbreadonly.fcntl, "flock", + side_effect=flock): self.follow(0, Server()) self.assertEqual(taken, [fcntl.LOCK_EX]) self.assertTrue(os.path.exists(os.path.join( diff --git a/tests/test_system_dbreadonly.py b/tests/test_system_dbreadonly.py index 3122c48..9e3443f 100644 --- a/tests/test_system_dbreadonly.py +++ b/tests/test_system_dbreadonly.py @@ -30,10 +30,12 @@ def status(sql: str = "Yes", read: int = 900, executed: int = 900, read_file: str = "mariadb-bin.000002", executed_file: str = "mariadb-bin.000002", - error: str = "") -> str: + error: str = "", io: str | None = None) -> str: """SHOW REPLICA STATUS\\G as the server prints it, the fields read""" return ( "*************************** 1. row ***************************\n" + + (f" Slave_IO_Running: {io}\n" if io else "") + + f" Master_Log_File: {read_file}\n" f" Read_Master_Log_Pos: {read}\n" f" Relay_Master_Log_File: {executed_file}\n" @@ -71,6 +73,10 @@ def __call__(self, argv, **kwargs): @staticmethod def key(argv, stdin: str) -> str: asked = argv[-1] if "--execute" in argv else "" + if "PROCESSLIST" in asked: + return "stopping" + if asked == "STOP SLAVE IO_THREAD": + return "stop-io" if "READ_ONLY ADMIN" in asked: return "bypass" if "mysql.user" in asked: @@ -83,6 +89,10 @@ def key(argv, stdin: str) -> str: return "grant" if "RESET SLAVE ALL" in stdin: return "promote" + if "STOP SLAVE SQL_THREAD" in stdin: + return "sql-stop" + if "read_only = OFF" in stdin: + return "writable" if "START SLAVE IO_THREAD" in stdin: return "resume" if "STOP SLAVE IO_THREAD" in stdin: @@ -271,12 +281,167 @@ def promote(self, runner: Runner, clock: Clock | None = None): self.root, PromoteReplica(60), runner, clock, clock.sleep ) + def spawned(self, events: list[str], code: int = 0): + """STOP SLAVE IO_THREAD in a client of its own that is still + running (a primary that is gone) until it is waited for""" + test = self + + class Stopping: + returncode = None + stderr = __import__("io").StringIO("ERROR 1198 stuck") + + def __init__(self, argv, **kwargs): + test.assertEqual(argv[-1], "STOP SLAVE IO_THREAD") + events.append("stop-io started") + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + if self.returncode is None: + events.append("stop-io ended") + self.returncode = code + return self.returncode + return Stopping + + def recorded(self, runner: Runner, events: list[str]) -> Runner: + real = runner.__call__ + + def call(argv, **kwargs): + done = real(argv, **kwargs) + events.append(runner.calls[-1][0]) + return done + runner.__call__ = call + return runner + + def promote_spawned(self, answers: dict, code: int = 0): + events: list[str] = [] + runner = Runner(answers) + recording = self.recorded(runner, events) + clock = Clock() + problem = dbreadonly.promote( + self.root, PromoteReplica(60), recording.__call__, clock, + clock.sleep, self.spawned(events, code)) + return problem, events, runner + + def test_writes_do_not_wait_for_the_io_thread_to_end(self): + """keel#118: the stop waits seconds for a dead primary (10 s + connecting, 5 s for the semi-synchronous kill, measured on + 11.8), while the thread receives nothing from the moment the + server shows the stop. So the server takes writes before the + stop ends, and forgets the primary after""" + problem, events, _ = self.promote_spawned({ + "status": (0, status(io="Connecting"), ""), + "stopping": (0, "Killing slave\n", "")}) + self.assertIsNone(problem, events) + self.assertEqual(events[:3], ["status", "stop-io started", + "stopping"]) + self.assertNotIn("sql-stop", events) + self.assertLess(events.index("writable"), + events.index("stop-io ended")) + self.assertLess(events.index("writable"), events.index("promote")) + + def test_anything_received_after_the_drain_takes_the_old_order(self): + """The safety review of keel#118: a thread that connects again + (MASTER_CONNECT_RETRY 2) could receive; when the position moved + after the drain, the stop is waited for, the SQL thread drains + again, and only then does the server take writes""" + problem, events, runner = self.promote_spawned({ + "status": [(0, status(io="Yes", read=900, executed=900), ""), + (0, status(io="Yes", read=900, executed=900), ""), + (0, status(io="Yes", read=950, executed=900), ""), + (0, status(io="No", read=950, executed=950), "")], + "stopping": (0, "Killing slave\n", "")}) + self.assertIsNone(problem, events) + self.assertNotIn("writable", events) + self.assertLess(events.index("stop-io ended"), + events.index("promote")) + self.assertEqual(runner.sent("promote"), + "STOP SLAVE;\nRESET SLAVE ALL;\n" + "SET GLOBAL read_only = OFF;\n") + + def test_a_stop_the_server_does_not_show_is_waited_for_first(self): + problem, events, _ = self.promote_spawned({ + "status": (0, status(io="Connecting"), ""), + "stopping": (0, "starting\n", "")}) + self.assertIsNone(problem, events) + self.assertLess(events.index("stop-io ended"), + events.index("writable")) + + def test_a_stop_not_yet_killing_after_the_drain_takes_the_old_order( + self): + """The re-review of keel#118: INFO is set when the statement is + dispatched, before the stop takes its lock; only the killing + phase, before and after the drain, lets writes come first""" + problem, events, runner = self.promote_spawned({ + "status": (0, status(io="Connecting"), ""), + "stopping": [(0, "Killing slave\n", ""), + (0, "Waiting for the stop lock\n", "")]}) + self.assertIsNone(problem, events) + self.assertNotIn("writable", events) + self.assertLess(events.index("stop-io ended"), + events.index("promote")) + self.assertEqual(runner.sent("promote"), + "STOP SLAVE;\nRESET SLAVE ALL;\n" + "SET GLOBAL read_only = OFF;\n") + + def test_a_stop_that_fails_promotes_nothing_and_resumes(self): + problem, events, _ = self.promote_spawned({ + "status": (0, status(io="Connecting"), ""), + "stopping": (0, "starting\n", "")}, code=1) + self.assertIn("ERROR 1198 stuck", problem) + self.assertIn("Nothing was promoted", problem) + # the server may have taken the stop before its client died + self.assertIn("The I/O thread was started again", problem) + self.assertIn("resume", events) + self.assertNotIn("writable", events) + + def test_failures_after_the_stop_are_said(self): + for key, said in (("writable", "set read_only OFF"),): + with self.subTest(key=key): + problem, events, runner = self.promote_spawned({ + "status": (0, status(io="Connecting"), ""), + "stopping": (0, "Killing slave\n", ""), + key: (1, "", "ERROR 1290")}) + self.assertIn("ERROR 1290", problem) + self.assertIn(said, problem) + self.assertIn("stop-io ended", events) + self.assertNotIn("promote", events) + problem, events, _ = self.promote_spawned({ + "status": (0, status(io="Connecting"), ""), + "stopping": (0, "Killing slave\n", ""), + "promote": (1, "", "ERROR 2013")}) + self.assertIn("takes writes", problem) + self.assertIn("ERROR 2013", problem) + + def test_a_status_lost_after_the_sql_thread_stopped_takes_the_old_order( + self): + problem, events, runner = self.promote_spawned({ + "status": [(0, status(io="Connecting"), ""), + (0, status(io="Connecting"), ""), + (1, "", "ERROR 2013"), + (0, status(io="No"), "")], + "stopping": (0, "Killing slave\n", "")}) + self.assertIsNone(problem, events) + self.assertNotIn("writable", events) + self.assertIn("promote", events) + + def test_a_client_that_cannot_start_the_stop(self): + def broken(argv, **kwargs): + raise FileNotFoundError(2, "No such file or directory") + clock = Clock() + problem = dbreadonly.promote( + self.root, PromoteReplica(60), Runner({"status": ( + 0, status(), "")}), clock, clock.sleep, broken) + self.assertIn("cannot run runuser", problem) + def test_the_io_thread_stops_before_anything_else(self): runner = Runner({"status": (0, status(), "")}) self.assertIsNone(self.promote(runner)) keys = runner.keys() - self.assertLess(keys.index("stop-io"), keys.index("promote")) - self.assertEqual(runner.sent("stop-io"), "STOP SLAVE IO_THREAD;\n") + self.assertEqual(keys[1], "stop-io") + self.assertLess(keys.index("stop-io"), keys.index("writable")) + self.assertLess(keys.index("writable"), keys.index("promote")) def test_it_waits_for_the_sql_thread_to_apply_the_backlog(self): runner = Runner({"status": [ @@ -289,8 +454,9 @@ def test_it_waits_for_the_sql_thread_to_apply_the_backlog(self): self.assertIsNone(self.promote(runner, clock)) self.assertEqual(clock.slept, 2) keys = runner.keys() - self.assertEqual(keys.count("status"), 4) - self.assertGreater(keys.index("promote"), + # the drain's four, and one more after the SQL thread stopped + self.assertEqual(keys.count("status"), 5) + self.assertGreater(keys.index("writable"), len(keys) - 1 - keys[::-1].index("status")) def test_a_backlog_in_an_older_binary_log_file_is_not_drained(self): @@ -306,9 +472,11 @@ def test_a_backlog_in_an_older_binary_log_file_is_not_drained(self): def test_then_it_forgets_the_primary_and_takes_writes(self): runner = Runner({"status": (0, status(), "")}) self.promote(runner) + self.assertEqual(runner.sent("writable"), + "SET GLOBAL read_only = OFF;\n") self.assertEqual( runner.sent("promote"), - "STOP SLAVE;\nRESET SLAVE ALL;\nSET GLOBAL read_only = OFF;\n", + "STOP SLAVE;\nRESET SLAVE ALL;\n", ) def test_giving_the_privilege_back_is_not_its_business(self): From 0be60b6e06a3d6dc92cd965ae761d015ad27dcda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcos=20M=C3=A9ndez?= Date: Fri, 9 Oct 2026 05:26:31 +0000 Subject: [PATCH 2/2] test: the unplanned failover of a cloud simple MariaDB pair, measured at 250 ms The real pair of the rollouts was in cloud simple: the VIP moves by the members' channel alone, with no etcd lease. The mariadb / trixie job runs with etcd, where the old primary's lease (20 s, decision 0049) comes before every claim. tests/mariadb_simple_netns.py runs the mariadb job's nodes and links (250 ms +-25 ms, 2 % loss, measured first) with every spec in cloud simple and etcd disabled. A primary, B replica, connected and semi-synchronous; C writes one row at the VIP every 0.1 s. A's server is killed and A is cut off; exactly 2 s after, keel database promote --old-primary-gone on B. Without etcd the promote waits the members' channel's full timeout for each of its three asks of the dead old primary (the safety review of keel#108), so the writes at the VIP must come back within 15 s plus three times 10 s of the promote's start, with no acknowledged row lost. The new job is "mariadb-simple / trixie". --- .github/workflows/tests.yml | 24 +++++ tests/mariadb_simple_netns.py | 142 +++++++++++++++++++++++++++++ tests/test_mariadb_simple_netns.py | 123 +++++++++++++++++++++++++ 3 files changed, 289 insertions(+) create mode 100644 tests/mariadb_simple_netns.py create mode 100644 tests/test_mariadb_simple_netns.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 2f64fa3..9a32866 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -150,6 +150,30 @@ jobs: systemctl stop mariadb KEEL_MARIADB_NETNS=1 PYTHONPATH=. python3 -m pytest -q -s \ tests/test_mariadb_netns.py + # The unplanned failover of a cloud simple MariaDB pair (keel#108, + # keel#118) on the mariadb job's nodes and links, the specs in cloud + # simple with etcd disabled, so the VIP moves without a lease, as on + # the real pair: the primary killed and cut off, the operator's + # promote 2 s after, the writes at the VIP back within 15 s of it plus + # the three full timeouts it waits on the dead old primary, and + # no acknowledged row lost (tests/test_mariadb_simple_netns.py). The + # check is "mariadb-simple / trixie". + mariadb-simple: + permissions: + contents: read + uses: keel-linux/.github/.github/workflows/lxc-trixie.yml@main + with: + systemd: true + timeout: 35 + run: | + apt-get update -qq + apt-get install -y -qq --no-install-recommends python3-pytest \ + python3-yaml wireguard-tools iproute2 iputils-ping openssl \ + etcd-server etcd-client util-linux mariadb-server mariadb-client \ + zstd + systemctl stop mariadb + KEEL_MARIADB_SIMPLE_NETNS=1 PYTHONPATH=. python3 -m pytest -q -s \ + tests/test_mariadb_simple_netns.py # A change that ships must come with a changelog entry, or it cannot be # installed and never reaches an appliance. The rule and its tests are # bin/require-changelog of the .github repository. The check is diff --git a/tests/mariadb_simple_netns.py b/tests/mariadb_simple_netns.py new file mode 100644 index 0000000..1e8a149 --- /dev/null +++ b/tests/mariadb_simple_netns.py @@ -0,0 +1,142 @@ +# Copyright (c) 2026 KeelLinux maintainers +"""The MariaDB pair in cloud simple: the primary crashes, the operator +promotes the replica 2 s after (keel#108, keel#118) + +Run as root in a network namespace of its own by +tests/test_mariadb_simple_netns.py; never run by hand on a machine. The +nodes, links (250 ms ±25 ms, 2 % loss, measured first), servers and +agents are tests/mariadb_netns.py's, with every node's spec in cloud +simple and the etcd overlay disabled: the VIP moves by the members' +channel alone, without a lease (keel.mesh.vippromote.promoted), as on +the real pair of Keel-Linux/handbook#60 and #62. etcd still runs in the +namespaces, as the harness forms it, and no spec uses it. + +The case: A primary, B replica, connected and semi-synchronous; C writes +one row at the VIP every 0.1 s. A's server is killed and A is cut off; +exactly 2 s after the kill, `keel database promote --old-primary-gone` +on B. The result is one JSON line, `RESULT {...}`: the write downtime +from the promote's start, its parts, and the acknowledged rows lost. +""" + +import os +import subprocess +import sys +import time + +import mariadb_netns as base + +# the operator's reaction, fixed so the number can be compared +PROMOTE_AFTER = 2.0 + + +def spec_of(root: str, index: int, pairs) -> str: + """mariadb_netns.spec_of, in cloud simple with etcd disabled""" + import yaml + path = base.spec_of(root, index, pairs) + with open(path) as fob: + doc = yaml.safe_load(fob) + doc["installation"]["mode"] = "cloud_simple" + doc["overlays"] = {"etcd": "disabled"} + with open(path, "w") as fob: + yaml.safe_dump(doc, fob, sort_keys=False) + return path + + +def scenario(report: dict, pids: list[int], pairs, roots: list[str], + samples: list[dict]) -> None: + try: + play(report, pids, roots) + except Exception: # noqa: BLE001 - the report says where it ended + import traceback + report["trace"] = traceback.format_exc()[-3000:] + raise + + +def play(report: dict, pids: list[int], roots: list[str]) -> None: + import vip_netns + from vip_netns import agent_send, wait_until + for index in range(len(vip_netns.NAMES) - 1): + base.server(index, pids[index], roots[index]) + report["a_promote"] = agent_send("A", "promote") + report["a_follow"] = agent_send("A", "follow", 180) + for statement in ( + f"CREATE DATABASE {base.APP_DB}", + f"CREATE TABLE {base.APP_DB}.t (id INT AUTO_INCREMENT PRIMARY" + " KEY, v VARCHAR(64), at DOUBLE)", + f"CREATE USER '{base.APP_USER}'@'%' IDENTIFIED BY" + f" '{base.APP_PASSWORD}'", + f"GRANT ALL ON {base.APP_DB}.* TO '{base.APP_USER}'@'%'"): + found = agent_send("A", f"sql {statement}") + if found.get("code"): + report["setup_error"] = found + return + report["a_apply"] = agent_send("A", "apply", 600) + report["b_apply"] = agent_send("B", "apply", 900) + report["b_streaming_s"] = wait_until( + lambda: base.replica_status(agent_send, "B").get( + "Slave_IO_State", "").startswith( + "Waiting for master to send event"), 120, 1) + report["semisync_s"] = wait_until( + lambda: base.status_value(agent_send, "A", + "Rpl_semi_sync_master_clients") == "1" + and base.status_value(agent_send, "A", + "Rpl_semi_sync_master_status") == "ON", + 120, 1) + report["with_etcd"] = agent_send("B", "status") + writer = base.Writer(pids[2]) + watched = base.NewPrimary(pids[1], roots[1]) + writer.start() + watched.start() + # writes acknowledged before the kill, so the downtime has a start + report["first_ack_s"] = wait_until( + lambda: any(one["ok"] for one in writer.done), 60, 0.2) + time.sleep(3) + killed_at = time.time() + base.sh("systemctl", "kill", "--signal=SIGKILL", + base.service_of(roots[0]), check=False) + vip_netns.leg("to1", "100%") + base.sh("tc", "qdisc", "replace", "dev", "uplink", "root", "netem", + "loss", "100%", pid=pids[0]) + time.sleep(max(0.0, killed_at + PROMOTE_AFTER - time.time())) + promote_at = time.time() + report["promote"] = agent_send("B", "dbpromote gone", 300) + promoted_at = time.time() + wait_until(lambda: watched.writable_at is not None, 60, 0.2) + time.sleep(5) + writer.stop() + watched.stop() + report["downtime"] = base.downtime(writer, watched, killed_at, + promote_at, promoted_at) + report["acked_missing"] = base.acked_missing(writer, agent_send) + for process in vip_netns.AGENTS.values(): + try: + process.stdin.write("quit\n") + process.stdin.close() + except OSError: + pass + + +def main() -> None: + import vip_netns + vip_netns.spec_of = spec_of + vip_netns.__file__ = os.path.abspath(__file__) + try: + vip_netns.main(play=scenario) + finally: + for unit in base.UNITS: + subprocess.run(["systemctl", "stop", unit], capture_output=True, + check=False) + try: + os.remove(f"/run/systemd/system/{unit}.service") + except OSError: + pass + + +if __name__ == "__main__": + if sys.argv[1:2] == ["agent"]: + base.agent_main(sys.argv[2], sys.argv[3]) + elif sys.argv[1:2] == ["lead"]: + import vip_netns + vip_netns.move_leader(sys.argv[2]) + else: + main() diff --git a/tests/test_mariadb_simple_netns.py b/tests/test_mariadb_simple_netns.py new file mode 100644 index 0000000..c12b276 --- /dev/null +++ b/tests/test_mariadb_simple_netns.py @@ -0,0 +1,123 @@ +# Copyright (c) 2026 KeelLinux maintainers +"""The unplanned failover of a cloud simple MariaDB pair, measured + +tests/mariadb_simple_netns.py, as root in a network namespace of its own +(tests/wgtools.py says how), on tests/mariadb_netns.py's nodes with the +specs in cloud simple: no etcd lease, the VIP moved by the members' +channel, as on the real pair (keel#108, keel#118). The primary's server +is killed and the primary cut off; the operator's `keel database promote +--old-primary-gone` on the replica starts exactly 2 s after. The writes +at the VIP come back within 15 s of the promote's start plus the three +full timeouts the promote waits on the dead old primary without etcd, +and no row the application had acknowledged is lost. In CI a missing tool fails instead +of skipping; it runs in its own job, "mariadb-simple / trixie", which +sets KEEL_MARIADB_SIMPLE_NETNS. +""" + +import json +import os +import subprocess +import sys +import unittest + +import wgtools +from test_etcd_netns import etcd_dir +from test_mariadb_netns import SCRATCH + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +DRIVER = os.path.join(REPO, "tests", "mariadb_simple_netns.py") +# the target the rollout of 0.23.7 set (Keel-Linux/handbook#62), plus +# what the safety review of keel#108 keeps without etcd: nothing fences +# a live old primary there, so each of the promote's three asks of it +# (its epoch, the release, the announcement) waits the members' +# channel's full connect timeout +ASKS_OF_THE_OLD_PRIMARY = 3 +TARGET_S = 15.0 + ASKS_OF_THE_OLD_PRIMARY * 10.0 + + +class TestACloudSimplePairFailsOver(unittest.TestCase): + found: dict = {} + + def setUp(self): + if os.environ.get("CI") and \ + not os.environ.get("KEEL_MARIADB_SIMPLE_NETNS"): + self.skipTest("runs in the mariadb-simple / trixie job") + tools = wgtools.require(self) + etcd = etcd_dir() + if etcd is None: + if os.environ.get("CI"): + self.fail("etcd is not installed in CI") + self.skipTest("etcd not found (set KEEL_ETCD_DIR)") + if not os.path.exists("/usr/sbin/mariadbd"): + if os.environ.get("CI"): + self.fail("mariadb-server is not installed in CI") + self.skipTest("mariadb-server not installed") + if not os.path.exists("/sys/module/wireguard") and \ + not os.environ.get("CI"): + self.skipTest("the wireguard kernel module is not loaded") + if not self.found: + self.found.update(self.run_driver(f"{tools}:{etcd}")) + + def run_driver(self, path: str) -> dict: + prefix = wgtools.namespace_prefix() + if prefix is None: + if os.environ.get("CI"): + self.fail("no way to run wg-quick as root in a namespace") + self.skipTest("neither root, user namespaces nor sudo -n") + os.makedirs(SCRATCH, exist_ok=True) + os.chmod(SCRATCH, 0o1777) + done = subprocess.run( + prefix + ["env", + f"PATH={path}:/usr/sbin:{os.environ.get('PATH', '')}", + f"PYTHONPATH={REPO}:{REPO}/tests", f"TMPDIR={SCRATCH}", + sys.executable, DRIVER], + capture_output=True, text=True, check=False, timeout=1800) + for line in done.stdout.splitlines(): + if line.startswith("RESULT "): + print(line, file=sys.stderr) + return json.loads(line[len("RESULT "):]) + self.fail(f"no result from {DRIVER}:\n{done.stdout[-4000:]}\n" + f"{done.stderr[-4000:]}") + + def test_the_links_are_the_design_case_and_the_pair_is_up(self): + found = self.found + link = found["link"] + self.assertGreater(link["rtt_min_ms"], 220, link) + self.assertLess(link["rtt_median_ms"], 300, link) + self.assertGreater(link["loss"], 0.005, link) + self.assertNotIn("setup_error", found, found) + self.assertNotIn("error", found, found) + self.assertFalse(found["a_apply"]["failed"], found["a_apply"]) + self.assertFalse(found["b_apply"]["failed"], found["b_apply"]) + self.assertIsNotNone(found["b_streaming_s"], found) + self.assertIsNotNone(found["semisync_s"], found) + # no etcd: the VIP holds no lease + self.assertFalse(any("etcd lease" in line for line in + found["with_etcd"]["lines"]), found) + + def test_the_writes_come_back_within_15_s_of_the_promote(self): + found = self.found["downtime"] + print(f"cloud simple downtime: {found}", file=sys.stderr) + self.assertEqual(self.found["promote"]["code"], 0, + self.found["promote"]) + self.assertIsNotNone(found["first_ok_after_s"], found) + self.assertAlmostEqual(found["promote_started_s"], 2.0, delta=0.5) + after_promote = found["first_ok_after_s"] - found["promote_started_s"] + self.assertLess(after_promote, TARGET_S, found) + self.assertLess(found["writable_after_carried_s"], 3.0, found) + + def test_no_acknowledged_row_is_lost(self): + self.assertEqual(self.found["acked_missing"], [], + self.found["downtime"]) + self.assertGreater(self.found["downtime"]["acked"], 0) + + def test_without_etcd_the_old_primary_is_waited_for_in_full(self): + from keel.mesh import memberlink + self.assertEqual(memberlink.CONNECT_TIMEOUT, 10) + said = " ".join(self.found["promote"]["out"]) + self.assertNotIn("no answer within", said) + self.assertIn("timed out", said) + + +if __name__ == "__main__": + unittest.main()