ci: wait for mongod before initiating the replica set, and initiate again

devtools/catalog_checks.py

@@ -8,7 +8,9 @@
 check-postgres-catalog-lock, check-postgres-older-binary,
 check-cassandra-catalog-lock, check-cassandra-older-binary,
 check-cassandra-authorizer and the check-mongodb-* scripts import it; the
-MongoDB ones start their replica sets with `start_mongo`. Each
+MongoDB ones start their replica sets with `start_mongo`, except
+check-mongodb-auth, which starts mongod with access control itself and
+initiates the set with `initiate_replica_set`. Each
 check prints one line, `ok` or `FAIL` and what was checked, with the
 command's output indented under a FAIL; `finish` prints the count of failed
 checks or `all checks passed` and returns the script's exit code.
@@ -140,7 +142,8 @@ def docker(*args: str, timeout: float = 120) -> subprocess.CompletedProcess:
 
 
 def mongosh(container: str, port: int, script: str, timeout: float = 60) -> tuple[int, str]:
-    """Run `script` with mongosh in a container `start_mongo` started."""
+    """Run `script` with mongosh, without credentials, in `container`, on the
+    mongod listening on `port`."""
     done = docker("exec", container, "mongosh", "--quiet", "--port", str(port), "--eval", script, timeout=timeout)
     return done.returncode, done.stdout + done.stderr
 
@@ -149,7 +152,8 @@ def start_mongo(container: str, port: int, mongod_args: tuple[str, ...] = (), mo
     """Start mongo:7 as a one-node replica set named rs0 in `container`,
     listening on `port` inside the container and published on 127.0.0.1
     only, so the same connection string works from the host and from a
-    container that shares its network. Waits for the member to be primary.
+    container that shares its network. Initiates the set and waits for the
+    member to be primary with `initiate_replica_set`.
     `mongod_args` are added to mongod's command line, `mounts` are `-v`
     arguments; a set started with `--keyFile` initiates through the
     localhost exception."""
@@ -161,13 +165,52 @@ def start_mongo(container: str, port: int, mongod_args: tuple[str, ...] = (), mo
     )
     if started.returncode != 0:
         return report(False, f"start {container}", started.stderr)
-    for _ in range(60):
-        if mongosh(container, port, "db.runCommand({ping: 1}).ok")[1].strip().endswith("1"):
-            break
-        time.sleep(1)
-    mongosh(container, port, f"rs.initiate({{_id: 'rs0', members: [{{_id: 0, host: 'localhost:{port}'}}]}})")
-    for _ in range(60):
-        if mongosh(container, port, "db.hello().isWritablePrimary")[1].strip().endswith("true"):
+    return initiate_replica_set(container, port)
+
+
+def _last_line(output: str) -> str:
+    lines = output.strip().splitlines()
+    return lines[-1].strip() if lines else ""
+
+
+# What `db.hello()` says about the member: `primary` once it is the writable
+# primary, `initiated` while the set has a configuration but no primary yet,
+# and `uninitiated` before any `rs.initiate` has taken effect (no `setName`).
+_MEMBER_STATE = (
+    "const h = db.hello(); print(h.isWritablePrimary ? 'primary' : h.setName ? 'initiated' : 'uninitiated')"
+)
+
+
+def initiate_replica_set(container: str, port: int) -> bool:
+    """Initiate the one-node replica set rs0 on the mongod listening on
+    `port` in `container`, and wait for its member to be primary.
+
+    Each poll runs `_MEMBER_STATE` and reads only the last line of its
+    output, which has to be one of the three states exactly. While mongod
+    refuses connections the poll prints `MongoNetworkError: connect
+    ECONNREFUSED 127.0.0.1:<port>`, and the wait goes on. `rs.initiate` is
+    sent after each poll that prints `uninitiated`, so it goes only to a
+    mongod that answered, and an initiate that returned an error is sent
+    again. Every call runs without credentials: `hello` needs none, and a
+    set started with `--keyFile` initiates through the localhost exception.
+    A member that is not primary after 120 polls, with 1 s between them, is
+    reported with the last poll's output, the last initiate's output and
+    the end of the container's log."""
+    initiate = f"rs.initiate({{_id: 'rs0', members: [{{_id: 0, host: 'localhost:{port}'}}]}})"
+    last_poll = ""
+    last_initiate = "(no rs.initiate sent: no poll printed `uninitiated`)"
+    for _ in range(120):
+        last_poll = mongosh(container, port, _MEMBER_STATE)[1]
+        state = _last_line(last_poll)
+        if state == "primary":
             return True
+        if state == "uninitiated":
+            last_initiate = mongosh(container, port, initiate)[1]
         time.sleep(1)
-    return report(False, f"{container} did not become primary", docker("logs", container).stdout[-2000:])
+    return report(
+        False,
+        f"{container} did not become primary",
+        f"last db.hello() poll output:\n{last_poll.strip()}\n"
+        f"last rs.initiate output:\n{last_initiate.strip()}\n"
+        "end of the container's log:\n" + docker("logs", container).stdout[-2000:],
+    )

devtools/catalog_checks_test.py

@@ -3,12 +3,17 @@
 """Tests of devtools/catalog_checks.py, which the catalog check scripts
 report through: a check passes only when the command's exit code and its
 output both match, every failure is counted, and finish returns 1 after one.
+initiate_replica_set sends rs.initiate only to a mongod that answered with
+no configuration, sends it again until it takes effect, and reports a member
+that is never primary with the last poll's and the last initiate's output.
 
     python3 -m unittest discover -s devtools -p '*_test.py'
 """
 import contextlib
 import io
+import subprocess
 import unittest
+from unittest import mock
 
 import catalog_checks
 
@@ -65,5 +70,108 @@ def test_finish_returns_1_after_a_failure_and_0_without(self) -> None:
         self.assertIn("1 check(s) failed", text)
 
 
+class FakeMongod:
+    """mongosh calls against a mongod that refuses the calls whose numbers,
+    counted from 0, are in `refused`, on which the first `failing` initiates
+    take no effect, and whose member answers `initiated` to the first
+    `electing` polls after the initiate that took effect, then `primary`."""
+
+    def __init__(self, refused: range = range(0), failing: int = 0, electing: int = 0) -> None:
+        self.refused = refused
+        self.failing = failing
+        self.electing = electing
+        self.initiated = False
+        self.calls: list[tuple[str, bool]] = []
+
+    def __call__(self, container: str, port: int, script: str, timeout: float = 60) -> tuple[int, str]:
+        listening = len(self.calls) not in self.refused
+        kind = "initiate" if "rs.initiate" in script else "hello"
+        self.calls.append((kind, listening))
+        if not listening:
+            return 1, f"MongoNetworkError: connect ECONNREFUSED 127.0.0.1:{port}\n"
+        if kind == "initiate":
+            if self.failing:
+                self.failing -= 1
+                return 1, "MongoServerError: the initiate did not take effect\n"
+            self.initiated = True
+            return 0, "{ ok: 1 }\n"
+        if not self.initiated:
+            return 0, "uninitiated\n"
+        if self.electing:
+            self.electing -= 1
+            return 0, "initiated\n"
+        return 0, "primary\n"
+
+
+class InitiateReplicaSetTest(unittest.TestCase):
+    """initiate_replica_set on port 27071, whose refused-connection error
+    ends in 1: the ping check it replaces took that error for an answer
+    (#793)."""
+
+    def setUp(self) -> None:
+        catalog_checks.failures = 0
+
+    def initiate(self, mongod: FakeMongod) -> tuple[bool, str]:
+        printed = io.StringIO()
+        log = subprocess.CompletedProcess([], 0, "the container's log\n", "")
+        with (
+            mock.patch.object(catalog_checks, "mongosh", mongod),
+            mock.patch.object(catalog_checks, "docker", lambda *args, **kwargs: log),
+            mock.patch.object(catalog_checks.time, "sleep", lambda seconds: None),
+            contextlib.redirect_stdout(printed),
+        ):
+            ok = catalog_checks.initiate_replica_set("c", 27071)
+        return ok, printed.getvalue()
+
+    def test_no_initiate_is_sent_while_connections_are_refused(self) -> None:
+        mongod = FakeMongod(refused=range(3))
+        ok, printed = self.initiate(mongod)
+        self.assertTrue(ok, printed)
+        self.assertEqual(
+            mongod.calls, [("hello", False)] * 3 + [("hello", True), ("initiate", True), ("hello", True)]
+        )
+
+    def test_the_initiate_goes_only_to_a_member_without_a_configuration(self) -> None:
+        # The two polls after the initiate are refused, and the two after
+        # those find the configuration with no primary yet.
+        mongod = FakeMongod(refused=range(2, 4), electing=2)
+        ok, printed = self.initiate(mongod)
+        self.assertTrue(ok, printed)
+        self.assertEqual(
+            mongod.calls, [("hello", True), ("initiate", True)] + [("hello", False)] * 2 + [("hello", True)] * 3
+        )
+
+    def test_an_initiate_that_takes_no_effect_is_sent_again(self) -> None:
+        mongod = FakeMongod(failing=2)
+        ok, printed = self.initiate(mongod)
+        self.assertTrue(ok, printed)
+        self.assertEqual([kind for kind, _ in mongod.calls].count("initiate"), 3)
+        self.assertEqual(catalog_checks.failures, 0)
+
+    def test_a_member_never_primary_is_reported_with_the_last_poll_and_initiate(self) -> None:
+        ok, printed = self.initiate(FakeMongod(failing=1000))
+        self.assertFalse(ok)
+        self.assertEqual(catalog_checks.failures, 1)
+        self.assertIn("FAIL c did not become primary", printed)
+        self.assertIn("     | last db.hello() poll output:\n     | uninitiated\n", printed)
+        self.assertIn(
+            "     | last rs.initiate output:\n     | MongoServerError: the initiate did not take effect\n", printed
+        )
+        self.assertIn("     | the container's log", printed)
+
+    def test_a_mongod_that_never_answers_is_reported_with_no_initiate_sent(self) -> None:
+        mongod = FakeMongod(refused=range(1000))
+        ok, printed = self.initiate(mongod)
+        self.assertFalse(ok)
+        self.assertEqual(set(mongod.calls), {("hello", False)})
+        self.assertIn(
+            "     | last db.hello() poll output:\n     | MongoNetworkError: connect ECONNREFUSED 127.0.0.1:27071\n",
+            printed,
+        )
+        self.assertIn(
+            "     | last rs.initiate output:\n     | (no rs.initiate sent: no poll printed `uninitiated`)\n", printed
+        )
+
+
 if __name__ == "__main__":
     unittest.main()

devtools/check-mongodb-auth

@@ -21,9 +21,8 @@ import shutil
 import subprocess
 import sys
 import tempfile
-import time
 
-from catalog_checks import docker, expect_exit, finish, healthy, report, run, start, stop
+from catalog_checks import docker, expect_exit, finish, healthy, initiate_replica_set, report, run, start, stop
 
 HERE = os.path.dirname(os.path.realpath(__file__))
 SEED = os.path.join(HERE, "..", "crates", "storage-mongodb", "tests", "fixtures", "released", "seed.py")
@@ -47,17 +46,9 @@ def start_with_auth(container: str, port: int) -> bool:
     )
     if not report(started.returncode == 0, f"start {container} with access control", started.stderr):
         return False
-    for _ in range(60):
-        if shell(container, port, "db.runCommand({ping: 1}).ok", root=False)[1].strip().endswith("1"):
-            break
-        time.sleep(1)
     # The localhost exception allows the first user and the initiation.
-    shell(container, port, f"rs.initiate({{_id: 'rs0', members: [{{_id: 0, host: 'localhost:{port}'}}]}})",
-          root=False)
-    for _ in range(60):
-        if shell(container, port, "db.hello().isWritablePrimary", root=False)[1].strip().endswith("true"):
-            break
-        time.sleep(1)
+    if not initiate_replica_set(container, port):
+        return False
     code, out = shell(container, port, f"db.getSiblingDB('admin').createUser({{user: 'root', pwd: "
                                        f"'{ROOT_PASSWORD}', roles: ['root']}})", root=False)
     if not report(code == 0, "create the administrator", out):