Fix grand-central exposure convergence

CHANGES.rst

@@ -5,6 +5,10 @@ Changelog
 Unreleased
 ----------
 
+* Fixed grand-central routing not converging on a ``spec.cluster.exposure``
+  change while ``spec.grandCentral.exposure`` was unset, which could leave a
+  stale nginx Ingress behind.
+
 2.64.1 (2026-09-07)
 -------------------
 

crate/operator/exposure.py

@@ -531,23 +531,22 @@ async def migrate_grand_central_exposure(
     name: str,
     spec: kopf.Spec,
     meta: kopf.Meta,
-    old_use_traefik: bool,
-    new_use_traefik: bool,
+    use_traefik: bool,
     logger: logging.Logger,
 ) -> None:
     """
-    Migrate grand-central routing resources from one exposure to the other.
+    Reconcile grand-central routing resources to the given exposure.
 
-    Creates the routing resources for ``new_use_traefik`` and deletes the
-    resources belonging to the old exposure. Does nothing if grand-central is
-    not deployed for this cluster.
+    Creates the routing resources for ``use_traefik`` and deletes the resources
+    belonging to the *other* exposure. Does nothing if grand-central is not
+    deployed for this cluster.
 
     :param namespace: The Kubernetes namespace for the CrateDB cluster.
     :param name: The CrateDB custom resource name defining the CrateDB cluster.
     :param spec: The ``spec`` section of the CrateDB custom resource.
     :param meta: The ``metadata`` section of the CrateDB custom resource.
-    :param old_use_traefik: Whether grand-central was previously on Traefik.
-    :param new_use_traefik: Whether grand-central should now be on Traefik.
+    :param use_traefik: The target exposure - ``True`` for Traefik (HTTPRoute +
+        Middlewares), ``False`` for an nginx Ingress.
     :param logger: Logger for operation tracking.
     """
     gc_deployment = await read_grand_central_deployment(namespace, name)
@@ -561,13 +560,15 @@ async def migrate_grand_central_exposure(
         spec=spec,
         meta=meta,
         logger=logger,
-        use_traefik=new_use_traefik,
+        use_traefik=use_traefik,
     )
 
-    if old_use_traefik:
-        await delete_grand_central_traefik_resources(namespace, name, logger)
-    else:
+    # Always remove the resources for the OTHER exposure so the routing layer
+    # converges to `use_traefik` regardless of the previous (or fallback) state.
+    if use_traefik:
         await delete_grand_central_ingress(namespace, name, logger)
+    else:
+        await delete_grand_central_traefik_resources(namespace, name, logger)
 
 
 class CreateTraefikResourcesSubHandler(StateBasedSubHandler):
@@ -678,8 +679,7 @@ async def handle(
                 name,
                 spec,
                 meta=body["metadata"],
-                old_use_traefik=(old_exposure == "traefik"),
-                new_use_traefik=(new_exposure == "traefik"),
+                use_traefik=(new_exposure == "traefik"),
                 logger=logger,
             )
 
@@ -701,24 +701,16 @@ async def handle(
         logger: logging.Logger,
         **kwargs: Any,
     ):
-        old_use_traefik = grand_central_uses_traefik(old["spec"])
-        new_use_traefik = grand_central_uses_traefik(body["spec"])
-
-        if old_use_traefik == new_use_traefik:
-            logger.info("Grand-central exposure unchanged")
-            return
+        use_traefik = grand_central_uses_traefik(body["spec"])
 
-        logger.info(
-            "Changing grand-central exposure "
-            f"(traefik={old_use_traefik} -> traefik={new_use_traefik})"
-        )
+        # Always reconcile to the resolved exposure
+        logger.info(f"Reconciling grand-central exposure to traefik={use_traefik}")
 
         await migrate_grand_central_exposure(
             namespace,
             name,
             body["spec"],
             meta=body["metadata"],
-            old_use_traefik=old_use_traefik,
-            new_use_traefik=new_use_traefik,
+            use_traefik=use_traefik,
             logger=logger,
         )

tests/test_grand_central_exposure.py

@@ -19,15 +19,28 @@
 # with Crate these terms will supersede the license and you may use the
 # software solely pursuant to the terms of the relevant commercial agreement.
 
+from typing import Any, Optional
+from unittest import mock
+
 import pytest
 
+from crate.operator.exposure import (
+    ChangeExposureSubHandler,
+    ChangeGrandCentralExposureSubHandler,
+    migrate_grand_central_exposure,
+)
 from crate.operator.grand_central import (
     _grand_central_hostname,
     get_grand_central_exposure,
     grand_central_uses_traefik,
 )
 
 
+def _await_kwargs(m: Any) -> dict:
+    assert m.await_args is not None
+    return m.await_args.kwargs
+
+
 @pytest.mark.parametrize(
     "spec, expected_use_traefik",
     [
@@ -106,3 +119,164 @@ def test_grand_central_hostname_only_replaces_leading_label(
     cluster_name, external_dns, expected
 ):
     assert _grand_central_hostname(cluster_name, external_dns) == expected
+
+
+_MIGRATE_PATCHES = (
+    "crate.operator.exposure.read_grand_central_deployment",
+    "crate.operator.exposure.create_grand_central_exposure",
+    "crate.operator.exposure.delete_grand_central_ingress",
+    "crate.operator.exposure.delete_grand_central_traefik_resources",
+)
+
+
+async def test_migrate_to_traefik_creates_httproute_and_deletes_stale_ingress():
+    spec = {"cluster": {"name": "c", "externalDNS": "c.example.com"}}
+    with (
+        mock.patch(_MIGRATE_PATCHES[0]) as read_deploy,
+        mock.patch(_MIGRATE_PATCHES[1]) as create_exp,
+        mock.patch(_MIGRATE_PATCHES[2]) as del_ingress,
+        mock.patch(_MIGRATE_PATCHES[3]) as del_traefik,
+    ):
+        read_deploy.return_value = object()
+        await migrate_grand_central_exposure(
+            "ns", "c", spec, {}, use_traefik=True, logger=mock.MagicMock()
+        )
+
+    assert _await_kwargs(create_exp)["use_traefik"] is True
+    # Converge to Traefik: the stale nginx Ingress must be removed...
+    del_ingress.assert_awaited_once()
+    # ...and the Traefik resources must NOT be deleted.
+    del_traefik.assert_not_called()
+
+
+async def test_migrate_to_nginx_creates_ingress_and_deletes_traefik():
+    spec = {"cluster": {"name": "c", "externalDNS": "c.example.com"}}
+    with (
+        mock.patch(_MIGRATE_PATCHES[0]) as read_deploy,
+        mock.patch(_MIGRATE_PATCHES[1]) as create_exp,
+        mock.patch(_MIGRATE_PATCHES[2]) as del_ingress,
+        mock.patch(_MIGRATE_PATCHES[3]) as del_traefik,
+    ):
+        read_deploy.return_value = object()
+        await migrate_grand_central_exposure(
+            "ns", "c", spec, {}, use_traefik=False, logger=mock.MagicMock()
+        )
+
+    assert _await_kwargs(create_exp)["use_traefik"] is False
+    del_traefik.assert_awaited_once()
+    del_ingress.assert_not_called()
+
+
+async def test_migrate_noop_when_grand_central_not_deployed():
+    with (
+        mock.patch(_MIGRATE_PATCHES[0]) as read_deploy,
+        mock.patch(_MIGRATE_PATCHES[1]) as create_exp,
+        mock.patch(_MIGRATE_PATCHES[2]) as del_ingress,
+        mock.patch(_MIGRATE_PATCHES[3]) as del_traefik,
+    ):
+        read_deploy.return_value = None
+        await migrate_grand_central_exposure(
+            "ns",
+            "c",
+            {"cluster": {"name": "c"}},
+            {},
+            use_traefik=True,
+            logger=mock.MagicMock(),
+        )
+
+    create_exp.assert_not_called()
+    del_ingress.assert_not_called()
+    del_traefik.assert_not_called()
+
+
+async def test_change_gc_exposure_handler_reconciles_even_when_effective_unchanged(
+    faker,
+):
+    handler = ChangeGrandCentralExposureSubHandler(
+        faker.uuid4(), faker.domain_word(), faker.md5(), {}
+    )
+    old = {"spec": {"cluster": {"exposure": "traefik"}}, "metadata": {}}
+    body = {
+        "spec": {
+            "cluster": {"exposure": "traefik"},
+            "grandCentral": {"exposure": "traefik"},
+        },
+        "metadata": {},
+    }
+    assert grand_central_uses_traefik(old["spec"]) is True
+    assert grand_central_uses_traefik(body["spec"]) is True
+
+    with mock.patch(
+        "crate.operator.exposure.migrate_grand_central_exposure"
+    ) as migrate:
+        await handler.handle(
+            namespace="ns",
+            name="c",
+            body=body,
+            old=old,
+            logger=mock.MagicMock(),
+        )
+
+    migrate.assert_awaited_once()
+    assert _await_kwargs(migrate)["use_traefik"] is True
+
+
+class _AsyncCM:
+    async def __aenter__(self):
+        return mock.MagicMock()
+
+    async def __aexit__(self, *args):
+        return False
+
+
+def _cluster_exposure_body(exposure: str, grand_central: Optional[dict] = None) -> dict:
+    spec = {
+        "cluster": {
+            "name": "c",
+            "exposure": exposure,
+            "externalDNS": "example.aks1.eastus2.azure.cratedb-dev.net.",
+        },
+    }
+    if grand_central is not None:
+        spec["grandCentral"] = grand_central
+    return {"spec": spec, "metadata": {"name": "c"}}
+
+
+async def _run_change_exposure(body: dict, old: dict):
+    handler = ChangeExposureSubHandler("ns", "c", "hash", {})
+    with (
+        mock.patch("crate.operator.exposure.GlobalApiClient", return_value=_AsyncCM()),
+        mock.patch("crate.operator.exposure.CoreV1Api"),
+        mock.patch("crate.operator.exposure.patch_service_exposure"),
+        mock.patch("crate.operator.exposure.create_traefik_resources"),
+        mock.patch("crate.operator.exposure.delete_traefik_resources"),
+        mock.patch("crate.operator.exposure.get_owner_references", return_value=[]),
+        mock.patch("crate.operator.exposure.migrate_grand_central_exposure") as migrate,
+    ):
+        await handler.handle(
+            namespace="ns",
+            name="c",
+            body=body,
+            old=old,
+            logger=mock.MagicMock(),
+        )
+    return migrate
+
+
+async def test_change_cluster_exposure_converges_gc_when_gc_exposure_unset():
+    body = _cluster_exposure_body("traefik")
+    old = _cluster_exposure_body("loadbalancer")
+
+    migrate = await _run_change_exposure(body, old)
+
+    migrate.assert_awaited_once()
+    assert _await_kwargs(migrate)["use_traefik"] is True
+
+
+async def test_change_cluster_exposure_skips_gc_when_gc_exposure_explicit():
+    body = _cluster_exposure_body("traefik", grand_central={"exposure": "nginx"})
+    old = _cluster_exposure_body("loadbalancer", grand_central={"exposure": "nginx"})
+
+    migrate = await _run_change_exposure(body, old)
+
+    migrate.assert_not_awaited()