Require stable cluster health before advancing a rolling restart

CHANGES.rst

@@ -5,6 +5,9 @@ Changelog
 Unreleased
 ----------
 
+* Fixed the operator restarting the next node during a rolling restart before
+  the cluster had actually recovered.
+
 2.63.0 (2026-08-12)
 -------------------
 

crate/operator/config.py

@@ -55,6 +55,17 @@ class Config:
     #: the pods are quick to start up.
     HEALTH_CHECK_RETRY_DELAY: Optional[int] = 30
 
+    #: Number of consecutive positive readings ``is_cluster_healthy`` requires
+    #: before it reports a cluster as healthy. A single reading can catch a
+    #: stale ``GREEN`` in the short window after a node rejoins but before the
+    #: master has recomputed cluster health, which lets a rolling restart
+    #: terminate the next node too early.
+    HEALTH_CHECK_STABILITY_CHECKS: Optional[int] = 3
+
+    #: Delay in seconds between the consecutive readings taken by
+    #: ``is_cluster_healthy``.
+    HEALTH_CHECK_STABILITY_DELAY: Optional[int] = 2
+
     #: When set, enable special handling for the defind cloud provider, e.g. on
     #: AWS pass the availability zone as a CrateDB node attribute.
     CLOUD_PROVIDER: Optional[CloudProvider] = None
@@ -221,6 +232,32 @@ def load(self):
                 f"'{bootstrap_timeout}'. Needs to be a positive integer."
             )
 
+        stability_checks = self.env(
+            "HEALTH_CHECK_STABILITY_CHECKS",
+            default=str(self.HEALTH_CHECK_STABILITY_CHECKS),
+        )
+        try:
+            self.HEALTH_CHECK_STABILITY_CHECKS = int(stability_checks)
+            if self.HEALTH_CHECK_STABILITY_CHECKS < 1:
+                raise ValueError()
+        except ValueError:
+            raise ConfigurationError(
+                f"Invalid {self._prefix}HEALTH_CHECK_STABILITY_CHECKS="
+                f"'{stability_checks}'. Needs to be an integer >= 1."
+            )
+
+        stability_delay = self.env(
+            "HEALTH_CHECK_STABILITY_DELAY",
+            default=str(self.HEALTH_CHECK_STABILITY_DELAY),
+        )
+        try:
+            self.HEALTH_CHECK_STABILITY_DELAY = int(stability_delay)
+        except ValueError:
+            raise ConfigurationError(
+                f"Invalid {self._prefix}HEALTH_CHECK_STABILITY_DELAY="
+                f"'{stability_delay}'. Needs to be a positive integer."
+            )
+
         cloud_provider = self.env("CLOUD_PROVIDER", default=self.CLOUD_PROVIDER)
         if cloud_provider is not None:
             try:

crate/operator/cratedb.py

@@ -19,6 +19,7 @@
 # with Crate these terms will supersede the license and you may use the
 # software solely pursuant to the terms of the relevant commercial agreement.
 
+import asyncio
 import functools
 import logging
 from typing import Dict, List, Optional, Tuple, Union
@@ -164,47 +165,108 @@ async def get_healthiness(cursor: Cursor) -> int:
     return row and row[0]
 
 
-async def is_cluster_healthy(
-    conn_factory, expected_nodes: int, logger: logging.Logger
+async def _read_cluster_health(
+    conn_factory,
+    expected_nodes: int,
+    logger: logging.Logger,
+    attempt: Optional[int] = None,
+    total: Optional[int] = None,
 ) -> bool:
     """
-    Check if a cluster is healthy.
+    Take a single reading of cluster health.
 
-    The function checks for the cluster health using the `sys.health
-    <https://crate.io/docs/crate/reference/en/latest/admin/system-information.html#health>`_
-    table and the expected number of nodes in the cluster three times.
-
-    :param conn_factory: A callable that allows the operator to connect
-        to the database. We regularly need to reconnect to ensure the
-        connection wasn't closed because it was opened to a CrateDB node that
-        was shut down since the connection was opened.
-    :param expected_nodes: The number of nodes that make up a healthy cluster.
+    Returns ``True`` only when the cluster has the expected number of nodes and
+    is not in a ``YELLOW``/``RED`` state.
     """
+    prefix = ""
+    if attempt is not None and total is not None:
+        prefix = f"Health check {attempt}/{total}: "
+
     # We need to establish a new connection because the peer of a
     # previous connection could have been shut down. And by
     # re-establishing a connection for _each_ polling we can assert
     # that the connection is open
     async with conn_factory() as conn:
         async with conn.cursor() as cursor:
-            logger.debug("Checking if cluster is healthy ...")
             try:
                 num_nodes = await get_number_of_nodes(cursor)
                 healthiness = await get_healthiness(cursor)
-                if num_nodes == expected_nodes and healthiness in {1, None}:
-                    logger.info("Cluster has expected number of nodes and is healthy")
-                    return True
-                else:
-                    logger.info(
-                        "Cluster has %d of %d nodes and is in %s state.",
-                        num_nodes,
-                        expected_nodes,
-                        HEALTHINESS.get(healthiness, "UNKNOWN"),
-                    )
-                    return False
             except ProgrammingError as e:
-                logger.warning("Failed to run health check query", exc_info=e)
+                logger.warning("%sFailed to run health check query", prefix, exc_info=e)
                 return False
 
+    # ``None`` means there are no tables at all (empty cluster), which is
+    # healthy. Any real severity other than GREEN (1) is not.
+    if healthiness is None:
+        state = "GREEN (no user tables)"
+    else:
+        state = HEALTHINESS.get(healthiness, f"severity {healthiness}")
+    logger.info(
+        "%sCluster has %d/%d nodes and is %s.",
+        prefix,
+        num_nodes,
+        expected_nodes,
+        state,
+    )
+
+    if num_nodes != expected_nodes:
+        return False
+    if healthiness not in {1, None}:
+        return False
+    return True
+
+
+async def is_cluster_healthy(
+    conn_factory,
+    expected_nodes: int,
+    logger: logging.Logger,
+    stability_checks: Optional[int] = None,
+    stability_delay: Optional[int] = None,
+) -> bool:
+    """
+    Check if a cluster is healthy.
+
+    The function checks the expected number of nodes and the cluster health from
+    the ``sys.health`` table, and requires that reading to hold across several
+    consecutive polls before reporting the cluster as healthy.
+
+    The consecutive-poll requirement exists because a single reading can catch
+    a stale ``GREEN`` in the short window after a node rejoins ``sys.nodes`` but
+    before the master has recomputed cluster health.
+
+    :param conn_factory: The connection factory to connect to CrateDB.
+    :param expected_nodes: The number of nodes that make up a healthy cluster.
+    :param stability_checks: Number of consecutive positive readings required.
+        Defaults to ``config.HEALTH_CHECK_STABILITY_CHECKS``.
+    :param stability_delay: Delay in seconds between the readings. Defaults to
+        ``config.HEALTH_CHECK_STABILITY_DELAY``.
+    """
+    checks = (
+        stability_checks
+        if stability_checks is not None
+        else config.HEALTH_CHECK_STABILITY_CHECKS
+    )
+    delay = (
+        stability_delay
+        if stability_delay is not None
+        else config.HEALTH_CHECK_STABILITY_DELAY
+    )
+    assert checks is not None and delay is not None
+
+    for attempt in range(checks):
+        if not await _read_cluster_health(
+            conn_factory, expected_nodes, logger, attempt=attempt + 1, total=checks
+        ):
+            return False
+        if attempt < checks - 1:
+            await asyncio.sleep(delay)
+
+    logger.info(
+        "Cluster has expected number of nodes and is healthy across %d checks.",
+        checks,
+    )
+    return True
+
 
 async def are_snapshots_in_progress(
     conn_factory, logger: logging.Logger

tests/test_cratedb.py

@@ -24,7 +24,12 @@
 import pytest
 from kubernetes_asyncio.client import CoreV1Api, CustomObjectsApi
 
-from crate.operator.cratedb import create_user, get_healthiness, get_number_of_nodes
+from crate.operator.cratedb import (
+    create_user,
+    get_healthiness,
+    get_number_of_nodes,
+    is_cluster_healthy,
+)
 from crate.operator.webhooks import (
     WebhookClusterHealthPayload,
     WebhookEvent,
@@ -197,6 +202,142 @@ async def test_get_healthiness(healthiness):
     cursor.fetchone.assert_awaited_once()
 
 
+class _FakeCursor:
+    def __init__(self, rows):
+        self._rows = list(rows)
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+    async def execute(self, sql, *args):
+        return None
+
+    async def fetchone(self):
+        return self._rows.pop(0)
+
+
+class _FakeConn:
+    def __init__(self, rows):
+        self._rows = rows
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+    def cursor(self):
+        return _FakeCursor(self._rows)
+
+
+class _conn_factory_for:
+    """
+    A ``conn_factory`` that returns one reading per connection.
+
+    ``readings`` is a list of ``(num_nodes, healthiness)`` tuples, one per poll
+    ``is_cluster_healthy`` performs. ``healthiness`` of ``None`` simulates an
+    empty ``sys.health`` (a cluster with no tables). ``calls`` records how many
+    connections were opened, so tests can assert that a failing reading
+    short-circuits the remaining polls.
+    """
+
+    def __init__(self, readings):
+        self._remaining = list(readings)
+        self.calls = 0
+
+    def __call__(self):
+        self.calls += 1
+        num_nodes, healthiness = self._remaining.pop(0)
+        rows = [
+            (num_nodes,),
+            (healthiness,) if healthiness is not None else None,
+        ]
+        return _FakeConn(rows)
+
+
+@pytest.mark.parametrize(
+    "readings, expected",
+    [
+        # Steady GREEN across every poll -> healthy.
+        ([(3, 1), (3, 1), (3, 1)], True),
+        # Empty cluster (no tables): no health rows -> healthy.
+        ([(3, None), (3, None), (3, None)], True),
+        # A GREEN cluster under ingest is still GREEN on every poll -> healthy.
+        # (Shards initializing/relocating from new partitions or rebalancing
+        # must NOT be treated as unhealthy; that would stall an upgrade.)
+        ([(3, 1), (3, 1), (3, 1)], True),
+        # Not all nodes back yet -> not healthy (short-circuits on first poll).
+        ([(2, 1)], False),
+        # YELLOW/RED -> not healthy.
+        ([(3, 2)], False),
+        ([(3, 3)], False),
+    ],
+)
+async def test_is_cluster_healthy_readings(readings, expected):
+    logger = mock.Mock()
+    conn_factory = _conn_factory_for(readings)
+    result = await is_cluster_healthy(
+        conn_factory,
+        expected_nodes=3,
+        logger=logger,
+        stability_checks=len(readings),
+        stability_delay=0,
+    )
+    assert result is expected
+
+
+async def test_is_cluster_healthy_requires_stable_readings():
+    logger = mock.Mock()
+    # First poll catches a stale GREEN, the second sees the cluster go YELLOW
+    # as the master recomputes health. A single snapshot used to pass here and
+    # let the next node be terminated too early; requiring consecutive readings
+    # must now report the cluster as not healthy.
+    conn_factory = _conn_factory_for([(3, 1), (3, 2)])
+    result = await is_cluster_healthy(
+        conn_factory,
+        expected_nodes=3,
+        logger=logger,
+        stability_checks=3,
+        stability_delay=0,
+    )
+    assert result is False
+
+
+async def test_is_cluster_healthy_recovers_after_flap():
+    logger = mock.Mock()
+    # A cluster that is briefly YELLOW then settles to a stable GREEN across the
+    # required number of consecutive polls is healthy.
+    conn_factory = _conn_factory_for([(3, 1), (3, 1), (3, 1)])
+    result = await is_cluster_healthy(
+        conn_factory,
+        expected_nodes=3,
+        logger=logger,
+        stability_checks=3,
+        stability_delay=0,
+    )
+    assert result is True
+    assert conn_factory.calls == 3
+
+
+async def test_is_cluster_healthy_stops_polling_on_first_failure():
+    logger = mock.Mock()
+    # A failing first reading must short-circuit without polling again, so only
+    # one connection is opened even though three checks were requested.
+    conn_factory = _conn_factory_for([(2, 1)])
+    result = await is_cluster_healthy(
+        conn_factory,
+        expected_nodes=3,
+        logger=logger,
+        stability_checks=3,
+        stability_delay=0,
+    )
+    assert result is False
+    assert conn_factory.calls == 1
+
+
 @pytest.mark.k8s
 @pytest.mark.asyncio
 @mock.patch("crate.operator.webhooks.webhook_client.send_notification")