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/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/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_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_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() 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):