From 7ef45c4cbfd580acbda36efd68ec095b9e1abfeb Mon Sep 17 00:00:00 2001 From: Thomas Achatz Date: Thu, 3 Sep 2026 13:46:49 +0200 Subject: [PATCH] Fix grand-central exposure convergence --- CHANGES.rst | 4 + crate/operator/exposure.py | 44 +++---- tests/test_grand_central_exposure.py | 174 +++++++++++++++++++++++++++ 3 files changed, 196 insertions(+), 26 deletions(-) diff --git a/CHANGES.rst b/CHANGES.rst index 6f488bf6..49368129 100644 --- a/CHANGES.rst +++ b/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) ------------------- diff --git a/crate/operator/exposure.py b/crate/operator/exposure.py index 71d5c1b0..9f41c491 100644 --- a/crate/operator/exposure.py +++ b/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, ) diff --git a/tests/test_grand_central_exposure.py b/tests/test_grand_central_exposure.py index 4b72ee69..cd585f2e 100644 --- a/tests/test_grand_central_exposure.py +++ b/tests/test_grand_central_exposure.py @@ -19,8 +19,16 @@ # 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, @@ -28,6 +36,11 @@ ) +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()