From 17a21e6d1ec894944b3b4bb6feec56b20f1cd984 Mon Sep 17 00:00:00 2001 From: wfealdel Date: Tue, 20 Oct 2026 17:39:00 -0400 Subject: [PATCH 1/4] [AAP-98100] feat: namespace-aware activation Job lifecycle Persist the Kubernetes namespace where each activation Job is created on RulebookProcess.activation_job_namespace so that lifecycle operations (monitor, stop, cleanup, logs) target the correct namespace even when the namespace env var changes at runtime. Model and migration: - Add activation_job_namespace nullable TextField to RulebookProcess (migration 0075, based on 0074). - Add ACTIVATION_TERMINAL_STATUSES frozenset to enums. Engine (kubernetes.py): - Extract shared helpers: resolve_activation_namespace(), verify_namespace_accessible(), make_job_name(). - resolve_activation_namespace() reads EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE, falls back to deprecated EDA_ACTIVATION_JOB_NAMESPACE with warning, then to the SA namespace file. - Refactor _set_namespace() to use shared helpers; reject inaccessible overrides on 4xx and retry transient server errors with backoff. - Log environment-selected namespace overrides at DEBUG level. - Verify Job and Pod deletion during cleanup. Skip the deletion watch when no Pods remain, while retaining post-deletion verification. Activation manager: - Use _instance_engine to target each process's recorded namespace, retaining the engine default for legacy processes with a null value. - Record the engine namespace on new processes. - Log each successfully started Kubernetes Job with activation name/ID, new instance ID and namespace; include the previous instance ID and its recorded namespace when available. Remove unscoped engine INFO messages. - Map Kubernetes cleanup failures to lifecycle errors, retain the Job reference when deletion is uncertain, and block unsafe replacement or restart. For unresponsive activations, report ERROR without clearing the process; retain FAILED handling for other engines. - Avoid repeating cleanup after a successful terminal-Job cleanup. Tests: - Cover namespace resolution and recording, isolated deprecated-env fallback, namespace-probe retries, deletion-watch behavior, Job and Pod deletion verification, and the analytics CSV contract. - Cover cleanup failure and recovery, unresponsive Kubernetes cleanup, both FAILED restart-policy branches, and one-time Job cleanup. - Cover first starts, prior recorded or unknown namespaces, and failed starts without a misleading success log. Co-Authored-By: Claude Opus 4.6 (1M context) Assisted-by: OpenCode / GPT-6 Sol (OpenAI) --- src/aap_eda/core/enums.py | 10 + ...ulebookprocess_activation_job_namespace.py | 15 + src/aap_eda/core/models/rulebook_process.py | 1 + .../services/activation/activation_manager.py | 193 ++++++-- .../services/activation/engine/kubernetes.py | 162 +++++-- .../analytics/test_analytics_collectors.py | 1 + .../activation/engine/test_kubernetes.py | 235 +++++++++- .../services/activation/test_manager.py | 438 ++++++++++++++++++ tests/unit/test_activation_job_namespace.py | 243 +++++++++- tests/unit/test_namespace_lifecycle.py | 114 +++++ 10 files changed, 1327 insertions(+), 85 deletions(-) create mode 100644 src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py create mode 100644 tests/unit/test_namespace_lifecycle.py diff --git a/src/aap_eda/core/enums.py b/src/aap_eda/core/enums.py index aa2e3a007..7e6ed37db 100644 --- a/src/aap_eda/core/enums.py +++ b/src/aap_eda/core/enums.py @@ -87,6 +87,16 @@ class ActivationStatus(DjangoStrEnum): WORKERS_OFFLINE = "workers offline" +ACTIVATION_TERMINAL_STATUSES = frozenset( + { + ActivationStatus.STOPPED, + ActivationStatus.COMPLETED, + ActivationStatus.FAILED, + ActivationStatus.ERROR, + } +) + + # TODO: Deprecated, will be removed in future version, use # DefaultCredentialType instead class CredentialType(DjangoStrEnum): diff --git a/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py b/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py new file mode 100644 index 000000000..22911ea98 --- /dev/null +++ b/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py @@ -0,0 +1,15 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0074_awxtoken_ck_awx_token_name_not_empty_and_more"), + ] + + operations = [ + migrations.AddField( + model_name="rulebookprocess", + name="activation_job_namespace", + field=models.TextField(null=True), + ), + ] diff --git a/src/aap_eda/core/models/rulebook_process.py b/src/aap_eda/core/models/rulebook_process.py index f76199461..6c43bf690 100644 --- a/src/aap_eda/core/models/rulebook_process.py +++ b/src/aap_eda/core/models/rulebook_process.py @@ -68,6 +68,7 @@ class RulebookProcess(BaseOrgModel): ended_at = models.DateTimeField(null=True) # TODO: rename field to pod_id activation_pod_id = models.TextField(null=True) + activation_job_namespace = models.TextField(null=True) status_message = models.TextField(null=True, default=None) log_read_at = models.DateTimeField(null=True) diff --git a/src/aap_eda/services/activation/activation_manager.py b/src/aap_eda/services/activation/activation_manager.py index 495c2926f..994c5231c 100644 --- a/src/aap_eda/services/activation/activation_manager.py +++ b/src/aap_eda/services/activation/activation_manager.py @@ -39,6 +39,7 @@ from .drools_cleanup import drools_cleanup from .engine.common import ContainerableInvalidError, ContainerEngine from .engine.factory import new_container_engine +from .engine.kubernetes import Engine as KubernetesEngine from .status_manager import StatusManager, run_with_lock from .tee_system_logger import TeeSystemLogger @@ -78,6 +79,38 @@ def __init__( self.container_logger_class = container_logger_class + @contextlib.contextmanager + def _instance_engine(self, instance): + """Use the Job's persisted namespace for operations on that Job. + + When the process has a recorded namespace, temporarily swap the + engine's namespace for the duration of the block. When it is + None (legacy process created before this field existed), use the + engine's current namespace unchanged — that is where pre-feature + Jobs were created. + """ + if not isinstance(self.container_engine, KubernetesEngine): + yield + return + + namespace = instance.activation_job_namespace + if not namespace: + LOGGER.info( + "Process %s has no recorded Job namespace, " + "using engine default %s", + instance.id, + self.container_engine.namespace, + ) + yield + return + + configured_namespace = self.container_engine.namespace + self.container_engine.namespace = namespace + try: + yield + finally: + self.container_engine.namespace = configured_namespace + def _update_started_time(self) -> None: """Update latest instance's started_at to now.""" self.latest_instance.started_at = timezone.now() @@ -186,6 +219,11 @@ def _start_activation_instance(self): # the previous instance should be previously managed. # This is a safety check. self._check_non_finalized_instances() + previous_instance = ( + self.latest_instance + if isinstance(self.container_engine, KubernetesEngine) + else None + ) # create a new instance LOGGER.info( @@ -254,6 +292,23 @@ def _start_activation_instance(self): self._set_activation_pod_id(pod_id=container_id) self._update_started_time() + if isinstance(self.container_engine, KubernetesEngine): + msg = ( + f"Activation {self.db_instance.id} " + f"(name={self.db_instance.name!r}), " + f"instance {self.latest_instance.id}: Job started in " + f"namespace {self.latest_instance.activation_job_namespace!r}" + ) + if ( + previous_instance + and previous_instance.activation_job_namespace + ): + msg += ( + f"; previous instance {previous_instance.id} recorded " + "Job namespace " + f"{previous_instance.activation_job_namespace!r}" + ) + log_handler.write(f"{msg}.", flush=True) # update logs LOGGER.info( @@ -282,15 +337,16 @@ def _cleanup(self, raise_on_error=False): log_handler = self.container_logger_class(latest_instance.id) try: - self.container_engine.cleanup( - latest_instance.activation_pod_id, - log_handler, - ) + with self._instance_engine(latest_instance): + self.container_engine.cleanup( + latest_instance.activation_pod_id, + log_handler, + ) - # We don't have identified cases where we want to stop the workflow - # because of a cleanup error. - # We are just logging the error and continue. - except engine_exceptions.ContainerCleanupError as exc: + # Kubernetes replacements must not proceed if the previous Job could + # still exist; preserve the Job reference when cleanup is uncertain. + # Other engines retain their existing best-effort cleanup behavior. + except engine_exceptions.ContainerEngineError as exc: pod_id = latest_instance.activation_pod_id msg = ( f"Activation {self.db_instance.id} failed to cleanup its " @@ -299,8 +355,10 @@ def _cleanup(self, raise_on_error=False): ) LOGGER.error(msg) log_handler.write(msg, flush=True) - if raise_on_error: - raise + if raise_on_error or isinstance( + self.container_engine, KubernetesEngine + ): + raise engine_exceptions.ContainerCleanupError(msg) from exc return def _is_in_status(self, status: ActivationStatus) -> bool: @@ -318,7 +376,10 @@ def _is_in_status(self, status: ActivationStatus) -> bool: latest_instance = self.latest_instance if pod_id: - with contextlib.suppress(engine_exceptions.ContainerNotFoundError): + with ( + contextlib.suppress(engine_exceptions.ContainerNotFoundError), + self._instance_engine(latest_instance), + ): container_status = self.container_engine.get_status( container_id=latest_instance.activation_pod_id, ) @@ -413,20 +474,29 @@ def _unresponsive_policy(self, check_type: str): def _handle_unresponsive_activation(self, check_type: str): """Clean up and fail an unresponsive activation. - Attempts cleanup of the container, then marks the activation - as FAILED regardless of cleanup outcome. If cleanup raises - ContainerCleanupError, the exception propagates to the caller - after the activation is marked FAILED. + Leave the process intact if Kubernetes Job cleanup fails so it can + be retried. Other engines retain FAILED handling on cleanup errors. """ + kubernetes_cleanup_failed = False try: self._cleanup(raise_on_error=True) + except engine_exceptions.ContainerCleanupError as exc: + if isinstance(self.container_engine, KubernetesEngine): + kubernetes_cleanup_failed = True + msg = ( + f"Activation {self.db_instance.id} is unresponsive. " + f"Cleanup failed. Reason: {exc}" + ) + self._error_activation(msg) + raise finally: - user_msg = ( - "Activation is unresponsive. " - f"{check_type} check for ansible-rulebook timed out." - ) - self._fail_instance(msg=user_msg, cleanup_already_done=True) - self.set_status(ActivationStatus.FAILED, user_msg) + if not kubernetes_cleanup_failed: + user_msg = ( + "Activation is unresponsive. " + f"{check_type} check for ansible-rulebook timed out." + ) + self._fail_instance(msg=user_msg, cleanup_already_done=True) + self.set_status(ActivationStatus.FAILED, user_msg) def _missing_container_policy(self): LOGGER.info( @@ -524,7 +594,9 @@ def _completed_policy(self, container_msg: str): user_msg, ) - def _failed_policy(self, container_msg: str): + def _failed_policy( + self, container_msg: str, cleanup_already_done: bool = False + ): container_logger = self.container_logger_class(self.latest_instance.id) """Apply the failed restart policy.""" LOGGER.info( @@ -543,7 +615,9 @@ def _failed_policy(self, container_msg: str): user_msg = f"{container_msg} {user_msg}" container_logger.write(user_msg, flush=True) try: - self._fail_instance(user_msg) + self._fail_instance( + user_msg, cleanup_already_done=cleanup_already_done + ) self.set_status( ActivationStatus.FAILED, user_msg, @@ -578,7 +652,9 @@ def _failed_policy(self, container_msg: str): user_msg = f"{container_msg} {user_msg}" container_logger.write(user_msg, flush=True) try: - self._fail_instance(user_msg) + self._fail_instance( + user_msg, cleanup_already_done=cleanup_already_done + ) self.set_status( ActivationStatus.FAILED, user_msg, @@ -610,7 +686,9 @@ def _failed_policy(self, container_msg: str): user_msg = f"{container_msg} {user_msg}" container_logger.write(user_msg, flush=True) try: - self._fail_instance(user_msg) + self._fail_instance( + user_msg, cleanup_already_done=cleanup_already_done + ) self.set_status( ActivationStatus.FAILED, user_msg, @@ -653,12 +731,15 @@ def _fail_instance( self._set_activation_pod_id(pod_id=None) self._increase_failure_count() - def _error_instance(self, msg: tp.Optional[str] = None): + def _error_instance( + self, msg: tp.Optional[str] = None, cleanup_already_done=False + ): """Error the latest activation instance.""" kwargs = {} if msg: kwargs["status_message"] = msg - self._cleanup() + if not cleanup_already_done: + self._cleanup() self.set_latest_instance_status(ActivationStatus.ERROR, **kwargs) self._set_activation_pod_id(pod_id=None) @@ -705,7 +786,15 @@ def start(self, is_restart: bool = False): "were found, recreating.", ) - self._start_activation_instance() + try: + self._start_activation_instance() + except engine_exceptions.ContainerCleanupError as exc: + msg = ( + f"Activation {self.db_instance.id} failed to start because " + f"cleanup failed. Reason: {exc}" + ) + self._error_activation(msg) + raise exceptions.ActivationStartError(msg) from exc if is_restart: self._increase_restart_count() @@ -886,7 +975,10 @@ def monitor(self): "Activation is disabled. " "Applicable stop policy will be applied.", ) - self.stop() + try: + self.stop() + except exceptions.ActivationStopError as exc: + raise exceptions.ActivationMonitorError(str(exc)) from exc return if self.db_instance.status not in [ @@ -922,13 +1014,28 @@ def monitor(self): self.update_logs() + if container_status.status in ( + ActivationStatus.COMPLETED, + ActivationStatus.FAILED, + ActivationStatus.ERROR, + ): + try: + self._cleanup() + except engine_exceptions.ContainerCleanupError as exc: + msg = ( + f"Monitor operation: activation {self.db_instance.id} " + f"failed to cleanup its Job. Reason: {exc}" + ) + self._error_activation(msg) + raise exceptions.ActivationMonitorError(msg) from exc + if container_status.status == ActivationStatus.COMPLETED: # RC == 0 - self._cleanup() self._completed_policy(container_status.message) return if container_status.status == ActivationStatus.FAILED: # RC != 0 - self._cleanup() - self._failed_policy(container_status.message) + self._failed_policy( + container_status.message, cleanup_already_done=True + ) return # Detect unresponsive activation instance @@ -977,8 +1084,7 @@ def monitor(self): f"Container is in an error state. {container_status.message}" ) LOGGER.error(msg) - self._cleanup() - self._error_instance(msg) + self._error_instance(msg, cleanup_already_done=True) self._error_activation(msg) raise exceptions.ActivationMonitorError(msg) @@ -998,10 +1104,12 @@ def _get_container_status(self): exit early. """ try: - return ( - self.container_engine.get_status( + with self._instance_engine(self.latest_instance): + container_status = self.container_engine.get_status( container_id=self.latest_instance.activation_pod_id, - ), + ) + return ( + container_status, False, ) except engine_exceptions.ContainerNotFoundError: @@ -1052,10 +1160,11 @@ def update_logs(self): return try: - self.container_engine.update_logs( - container_id=self.latest_instance.activation_pod_id, - log_handler=log_handler, - ) + with self._instance_engine(self.latest_instance): + self.container_engine.update_logs( + container_id=self.latest_instance.activation_pod_id, + log_handler=log_handler, + ) # We don't have identified cases where we want to stop the workflow # because of a log error. @@ -1086,6 +1195,8 @@ def _create_activation_instance(self): "organization": self.db_instance.organization, } args[f"{self.db_instance_type}"] = self.db_instance + if isinstance(self.container_engine, KubernetesEngine): + args["activation_job_namespace"] = self.container_engine.namespace try: rulebook_process = models.RulebookProcess.objects.create(**args) except IntegrityError as exc: diff --git a/src/aap_eda/services/activation/engine/kubernetes.py b/src/aap_eda/services/activation/engine/kubernetes.py index fc5133f97..cc59dec4a 100644 --- a/src/aap_eda/services/activation/engine/kubernetes.py +++ b/src/aap_eda/services/activation/engine/kubernetes.py @@ -64,6 +64,7 @@ IMAGE_PULL_ERROR, ] POD_DELETE_TIMEOUT = 60 +JOB_DELETE_VERIFY_TIMEOUT = 15 @dataclass @@ -88,6 +89,81 @@ def get_k8s_client() -> Client: raise ContainerEngineInitError(str(e)) from e +SA_NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace" + + +def verify_namespace_accessible(batch_api, namespace: str) -> None: + """Probe a namespace, retrying transient errors but failing on 4xx.""" + for attempt in range(1, K8S_API_RETRIES + 1): + try: + batch_api.list_namespaced_job(namespace=namespace, limit=1) + return + except ApiException as exc: + if isinstance(exc.status, int) and 400 <= exc.status < 500: + raise ContainerEngineInitError( + f"Namespace {namespace!r} is not accessible " + f"(HTTP {exc.status}). Fix the namespace configuration " + f"or RBAC before proceeding." + ) from exc + if exc.status not in K8S_API_TRANSIENT_STATUS_CODES: + raise ContainerEngineInitError( + f"Cannot verify namespace {namespace!r}: {exc}" + ) from exc + LOGGER.warning( + "Transient K8s API error (HTTP %s) on attempt %d/%d " + "for namespace %s: %s", + exc.status, + attempt, + K8S_API_RETRIES, + namespace, + exc.reason, + ) + if attempt == K8S_API_RETRIES: + raise ContainerEngineInitError( + f"Cannot verify namespace {namespace!r} after " + f"{K8S_API_RETRIES} retries: {exc}" + ) from exc + time.sleep(K8S_API_RETRY_BACKOFF * attempt) + + +def resolve_activation_namespace() -> tuple[str, bool]: + """Resolve the activation Job namespace from env var or SA file. + + Returns (namespace, is_override) where is_override is True when + the namespace came from an env-var override. + """ + ns_override = os.environ.get( + "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE", "" + ).strip() + if ns_override: + return ns_override, True + + old_override = os.environ.get("EDA_ACTIVATION_JOB_NAMESPACE", "").strip() + if old_override: + LOGGER.warning( + "EDA_ACTIVATION_JOB_NAMESPACE is deprecated; " + "use EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE instead." + ) + return old_override, True + + try: + with open(SA_NAMESPACE_FILE, encoding="utf-8") as f: + return f.read().strip(), False + except FileNotFoundError as e: + raise ContainerEngineInitError( + "Cannot determine activation namespace: " + "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE is not set and " + f"{SA_NAMESPACE_FILE} does not exist." + ) from e + + +def make_job_name( + resource_prefix: str, parent_id: int, process_id: int +) -> str: + """Build the deterministic Job name for an activation process.""" + return f"{resource_prefix}-job-{parent_id}-{process_id}" + + class Engine(ContainerEngine): def __init__( self, @@ -109,9 +185,10 @@ def __init__( def start(self, request: ContainerRequest, log_handler: LogHandler) -> str: # TODO : Should this be compatible with the previous version # Previous Version - self.job_name = ( - f"{self.resource_prefix}-job-{request.process_parent_id}" - f"-{request.rulebook_process_id}" + self.job_name = make_job_name( + self.resource_prefix, + request.process_parent_id, + request.rulebook_process_id, ) self.pod_name = ( f"{self.resource_prefix}-pod-{request.process_parent_id}" @@ -570,14 +647,43 @@ def _delete_job(self, log_handler: LogHandler) -> None: ) if not (activation_job.items and activation_job.items[0].metadata): + self._verify_job_deleted() LOGGER.info(f"Job for {self.job_name} has been removed") return activation_job_name = activation_job.items[0].metadata.name self._delete_job_resource(activation_job_name, log_handler) + self._verify_job_deleted() + + def _verify_job_deleted(self) -> None: + selector = f"job-name={self.job_name}" + deadline = time.monotonic() + JOB_DELETE_VERIFY_TIMEOUT + while True: + jobs = self._call_k8s_api( + self.client.batch_api.list_namespaced_job, + namespace=self.namespace, + label_selector=selector, + error_cls=ContainerCleanupError, + description=f"verify job deletion {self.job_name}", + ) + pods = self._call_k8s_api( + self.client.core_api.list_namespaced_pod, + namespace=self.namespace, + label_selector=selector, + error_cls=ContainerCleanupError, + description=f"verify pod deletion {self.job_name}", + ) + if not jobs.items and not pods.items: + return + if time.monotonic() >= deadline: + raise ContainerCleanupError( + f"Job or Pod {self.job_name} still exists in " + f"{self.namespace}" + ) + time.sleep(1) - def _get_resource_version(self) -> str: - """Return the current resourceVersion for this job's pods.""" + def _get_resource_version(self) -> str | None: + """Return a watch position only when this Job still has Pods.""" pod_list = self._call_k8s_api( self.client.core_api.list_namespaced_pod, namespace=self.namespace, @@ -585,7 +691,9 @@ def _get_resource_version(self) -> str: error_cls=ContainerCleanupError, description=f"List pods for {self.job_name}", ) - return pod_list.metadata.resource_version + if pod_list.items: + return pod_list.metadata.resource_version + return None def _delete_job_resource( self, job_name: str, log_handler: LogHandler @@ -604,9 +712,10 @@ def _delete_job_resource( if result.status == "Failure": raise ContainerCleanupError(f"{result}") - self._watch_pod_deletion( - log_handler, resource_version=resource_version - ) + if resource_version is not None: + self._watch_pod_deletion( + log_handler, resource_version=resource_version + ) def _watch_for_deleted_event( self, @@ -675,9 +784,11 @@ def _watch_pod_deletion( last_exc = None for attempt in range(1, K8S_API_RETRIES + 1): watcher = watch.Watch() - if attempt > 1: - resource_version = self._get_resource_version() try: + if attempt > 1: + resource_version = self._get_resource_version() + if resource_version is None: + return if self._watch_for_deleted_event( watcher, resource_version, log_handler ): @@ -757,29 +868,14 @@ def _wait_for_pod_to_start(self, _log_handler: LogHandler) -> None: ) from last_exc def _set_namespace(self) -> None: - ns_override = os.environ.get("EDA_ACTIVATION_JOB_NAMESPACE", "") - if ns_override.strip(): - self.namespace = ns_override.strip() - LOGGER.info( - "Using activation job namespace override: %s", - self.namespace, + namespace, is_override = resolve_activation_namespace() + if is_override: + LOGGER.debug( + "Activation Job namespace override from environment: %s", + namespace, ) - return - - namespace_file = ( - "/var/run/secrets/kubernetes.io/serviceaccount/namespace" - ) - try: - with open( - namespace_file, mode="r", encoding="utf-8" - ) as namespace_ref: - self.namespace = namespace_ref.read() - - LOGGER.info(f"Namespace is {self.namespace}") - except FileNotFoundError as e: - message = f"Namespace file {namespace_file} does not exist." - LOGGER.error(message) - raise ContainerEngineInitError(message) from e + verify_namespace_accessible(self.client.batch_api, namespace) + self.namespace = namespace def _create_secret( self, diff --git a/tests/integration/analytics/test_analytics_collectors.py b/tests/integration/analytics/test_analytics_collectors.py index 24f540715..a93b9e735 100644 --- a/tests/integration/analytics/test_analytics_collectors.py +++ b/tests/integration/analytics/test_analytics_collectors.py @@ -864,6 +864,7 @@ def test_rulebook_processes_table_collector( "updated_at", "ended_at", "activation_pod_id", + "activation_job_namespace", "status_message", "log_read_at", ] diff --git a/tests/integration/services/activation/engine/test_kubernetes.py b/tests/integration/services/activation/engine/test_kubernetes.py index 52c485a6d..d6298be8c 100644 --- a/tests/integration/services/activation/engine/test_kubernetes.py +++ b/tests/integration/services/activation/engine/test_kubernetes.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os from unittest import mock import pytest @@ -35,6 +36,7 @@ IMAGE_PULL_BACK_OFF, IMAGE_PULL_ERROR, INVALID_IMAGE_NAME, + JOB_DELETE_VERIFY_TIMEOUT, K8S_API_RETRIES, Engine, get_k8s_client, @@ -246,7 +248,9 @@ def test_engine_init_with_exception(init_kubernetes_data): with pytest.raises( ContainerEngineInitError, match=( - "Namespace file /var/run/secrets/kubernetes.io/serviceaccount" + "Cannot determine activation namespace: " + "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE is not set and " + "/var/run/secrets/kubernetes.io/serviceaccount" "/namespace does not exist." ), ): @@ -731,7 +735,9 @@ def test_delete_job(mock_watch, init_kubernetes_data, kubernetes_engine): job_mock = mock.Mock() job_mock.metadata.name = job_name - with mock.patch.object(engine.client, "batch_api") as batch_api_mock: + with mock.patch.object( + engine.client, "batch_api" + ) as batch_api_mock, mock.patch.object(engine, "_verify_job_deleted"): batch_api_mock.list_namespaced_job.return_value.items = [job_mock] mock_watch.return_value.stream.return_value = [{"type": "DELETED"}] with mock.patch.object( @@ -741,7 +747,9 @@ def test_delete_job(mock_watch, init_kubernetes_data, kubernetes_engine): batch_api_mock.delete_namespaced_job.assert_called_once() - with mock.patch.object(engine.client, "batch_api") as batch_api_mock: + with mock.patch.object( + engine.client, "batch_api" + ) as batch_api_mock, mock.patch.object(engine, "_verify_job_deleted"): batch_api_mock.list_namespaced_job.return_value.items = None engine._delete_job(log_handler) @@ -750,7 +758,9 @@ def test_delete_job(mock_watch, init_kubernetes_data, kubernetes_engine): ) event = {"type": "DELETED"} - with mock.patch.object(engine.client, "batch_api") as batch_api_mock: + with mock.patch.object( + engine.client, "batch_api" + ) as batch_api_mock, mock.patch.object(engine, "_verify_job_deleted"): batch_api_mock.list_namespaced_job.return_value.items = [job_mock] mock_watch.return_value.stream.return_value = [event] with mock.patch.object( @@ -791,6 +801,60 @@ def raise_api_error(*args, **kwargs): engine._delete_job(log_handler) +@pytest.mark.django_db +def test_get_resource_version_without_pods(kubernetes_engine): + engine = kubernetes_engine + engine.job_name = "test-job" + engine.client.core_api.list_namespaced_pod.return_value.items = [] + + assert engine._get_resource_version() is None + + +@pytest.mark.django_db +def test_verify_job_deleted_waits_for_resources(kubernetes_engine): + engine = kubernetes_engine + engine.job_name = "test-job" + engine.client.batch_api.list_namespaced_job.side_effect = [ + mock.Mock(items=[mock.Mock()]), + mock.Mock(items=[]), + ] + engine.client.core_api.list_namespaced_pod.return_value.items = [] + + with mock.patch( + "aap_eda.services.activation.engine.kubernetes.time.sleep" + ) as sleep: + engine._verify_job_deleted() + + sleep.assert_called_once_with(1) + assert engine.client.batch_api.list_namespaced_job.call_count == 2 + + +@pytest.mark.parametrize("remaining_resource", ["job", "pod"]) +@pytest.mark.django_db +def test_verify_job_deleted_times_out_on_remaining_resource( + kubernetes_engine, remaining_resource +): + engine = kubernetes_engine + engine.job_name = "test-job" + engine.client.batch_api.list_namespaced_job.return_value.items = ( + [mock.Mock()] if remaining_resource == "job" else [] + ) + engine.client.core_api.list_namespaced_pod.return_value.items = ( + [mock.Mock()] if remaining_resource == "pod" else [] + ) + + with mock.patch( + "aap_eda.services.activation.engine.kubernetes.time.monotonic", + side_effect=[0, JOB_DELETE_VERIFY_TIMEOUT], + ), mock.patch( + "aap_eda.services.activation.engine.kubernetes.time.sleep" + ) as sleep: + with pytest.raises(ContainerCleanupError, match="still exists"): + engine._verify_job_deleted() + + sleep.assert_not_called() + + @pytest.mark.django_db def test_cleanup_orig(init_kubernetes_data, kubernetes_engine): engine = kubernetes_engine @@ -819,7 +883,9 @@ def test_pod_cleanup_exception_handling( job_mock = mock.Mock() job_mock.metadata.name = job_name - with mock.patch.object(engine.client, "batch_api") as batch_api_mock: + with mock.patch.object( + engine.client, "batch_api" + ) as batch_api_mock, mock.patch.object(engine, "_verify_job_deleted"): batch_api_mock.list_namespaced_job.return_value.items = [job_mock] # Test 404 handling @@ -1189,6 +1255,61 @@ def test_watch_pod_deletion_retries_on_transient( assert mock_watch.return_value.stop.call_count == K8S_API_RETRIES +@mock.patch("aap_eda.services.activation.engine.kubernetes.watch.Watch") +@pytest.mark.django_db +def test_delete_job_verifies_when_pods_disappear_before_watch_retry( + mock_watch, init_kubernetes_data, kubernetes_engine +): + """No retry watch is needed, but Job and Pod deletion is verified.""" + engine = kubernetes_engine + engine.job_name = "test-job" + log_handler = DBLogger(init_kubernetes_data.activation_instance.id) + job = mock.Mock(metadata=mock.Mock(name="test-job")) + engine.client.batch_api.list_namespaced_job.side_effect = [ + mock.Mock(items=[job]), + mock.Mock(items=[]), + ] + engine.client.core_api.list_namespaced_pod.return_value.items = [] + mock_watch.return_value.stream.return_value = iter([]) + + with mock.patch.object( + engine, "_get_resource_version", side_effect=["1", None] + ) as get_version: + engine._delete_job(log_handler) + + assert get_version.call_count == 2 + assert mock_watch.return_value.stream.call_count == 1 + assert mock_watch.return_value.stop.call_count == 2 + assert engine.client.batch_api.list_namespaced_job.call_count == 2 + engine.client.core_api.list_namespaced_pod.assert_called_once() + + +@mock.patch("aap_eda.services.activation.engine.kubernetes.watch.Watch") +@pytest.mark.django_db +def test_watch_pod_deletion_retries_with_available_resource_version( + mock_watch, init_kubernetes_data, kubernetes_engine +): + engine = kubernetes_engine + engine.job_name = "test-job" + log_handler = DBLogger(init_kubernetes_data.activation_instance.id) + mock_watch.return_value.stream.side_effect = [ + iter([]), + iter([{"type": "DELETED"}]), + ] + + with mock.patch.object(engine, "_get_resource_version", return_value="2"): + engine._watch_pod_deletion(log_handler, resource_version="1") + + assert mock_watch.return_value.stream.call_count == 2 + assert ( + mock_watch.return_value.stream.call_args_list[1].kwargs[ + "resource_version" + ] + == "2" + ) + assert mock_watch.return_value.stop.call_count == 2 + + @mock.patch("aap_eda.services.activation.engine.kubernetes.watch.Watch") @mock.patch("aap_eda.services.activation.engine.kubernetes.time.sleep") @pytest.mark.django_db @@ -1356,3 +1477,107 @@ def test_process_pod_start_event_unrecognized(kubernetes_engine): ) event = {"object": pod} assert engine._process_pod_start_event(event) is False + + +# --- Namespace-aware engine tests --- + + +@pytest.mark.django_db +def test_set_namespace_with_env_override_verifies_access( + init_kubernetes_data, +): + """When env var is set, _set_namespace probes the target namespace.""" + mock_client = mock.Mock() + mock_client.batch_api.list_namespaced_job.return_value = None + + with mock.patch.dict( + "os.environ", + {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "eda-activations"}, + ): + engine = Engine( + activation_id=str(init_kubernetes_data.activation.id), + resource_prefix="activation", + client=mock_client, + ) + assert engine.namespace == "eda-activations" + mock_client.batch_api.list_namespaced_job.assert_called_once_with( + namespace="eda-activations", + limit=1, + ) + + +@pytest.mark.django_db +def test_set_namespace_without_env_skips_verification(init_kubernetes_data): + """Without env override, _set_namespace does not probe K8s.""" + mock_client = mock.Mock() + env = { + k: v + for k, v in os.environ.items() + if k != "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE" + } + with mock.patch.dict("os.environ", env, clear=True): + with mock.patch("builtins.open", mock.mock_open(read_data="aap-eda")): + engine = Engine( + activation_id=str(init_kubernetes_data.activation.id), + resource_prefix="activation", + client=mock_client, + ) + assert engine.namespace == "aap-eda" + mock_client.batch_api.list_namespaced_job.assert_not_called() + + +@pytest.mark.django_db +def test_set_namespace_env_override_inaccessible_raises( + init_kubernetes_data, +): + """Inaccessible override namespace raises during init.""" + mock_client = mock.Mock() + mock_client.batch_api.list_namespaced_job.side_effect = ApiException( + status=403, reason="Forbidden" + ) + with mock.patch.dict( + "os.environ", + {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "forbidden-ns"}, + ): + with pytest.raises(ContainerEngineInitError, match="not accessible"): + Engine( + activation_id=str(init_kubernetes_data.activation.id), + resource_prefix="activation", + client=mock_client, + ) + + +@pytest.mark.django_db +def test_start_uses_make_job_name( + init_kubernetes_data, mock_log_handler, default_organization +): + """Engine.start() builds deterministic Job name.""" + mock_client = mock.Mock() + env = { + k: v + for k, v in os.environ.items() + if k != "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE" + } + with mock.patch.dict("os.environ", env, clear=True): + with mock.patch("builtins.open", mock.mock_open(read_data="aap-eda")): + engine = Engine( + activation_id=str(init_kubernetes_data.activation.id), + resource_prefix="activation", + client=mock_client, + ) + + request = get_request( + init_kubernetes_data, + "tester", + default_organization, + ) + + with mock.patch.object(engine, "_create_job"), mock.patch.object( + engine, "_wait_for_pod_to_start" + ), mock.patch.object(engine, "_create_service"): + engine.start(request, mock_log_handler) + + act_id = init_kubernetes_data.activation.id + proc_id = init_kubernetes_data.activation_instance.id + expected_name = f"activation-job-{act_id}-{proc_id}" + assert engine.job_name == expected_name diff --git a/tests/integration/services/activation/test_manager.py b/tests/integration/services/activation/test_manager.py index 322f0d944..fe7ba6329 100644 --- a/tests/integration/services/activation/test_manager.py +++ b/tests/integration/services/activation/test_manager.py @@ -35,6 +35,9 @@ ContainerEngine, ContainerRequest, ) +from aap_eda.services.activation.engine.kubernetes import ( + Engine as KubernetesEngine, +) from aap_eda.services.activation.status_manager import StatusManager @@ -367,6 +370,7 @@ def test_start_first_run( activation_instance=basic_activation.latest_instance ).all() assert not any("The activation was edited at" in log.log for log in logs) + assert not any("Job started in namespace" in log.log for log in logs) @pytest.mark.django_db @@ -456,6 +460,234 @@ def test_monitor_container_not_found( assert running_activation.status == enums.ActivationStatus.FAILED +@pytest.mark.parametrize( + "container_status", + [ + enums.ActivationStatus.COMPLETED, + enums.ActivationStatus.FAILED, + enums.ActivationStatus.ERROR, + ], +) +@pytest.mark.django_db +def test_monitor_cleanup_failure_preserves_job_for_retry( + running_activation: models.Activation, + container_status: enums.ActivationStatus, +): + """Do not apply terminal policies when Job deletion is uncertain.""" + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "new-ns" + engine.get_status.return_value = MagicMock( + status=container_status, message="container exited" + ) + engine.cleanup.side_effect = engine_exceptions.ContainerCleanupError( + "cleanup failed" + ) + process = running_activation.latest_instance + process.activation_job_namespace = "old-ns" + process.save(update_fields=["activation_job_namespace"]) + manager = ActivationManager(running_activation, container_engine=engine) + + with patch(SYSTEM_RESTART_PATH) as restart_mock: + with pytest.raises(exceptions.ActivationMonitorError) as exc: + manager.monitor() + + assert isinstance( + exc.value.__cause__, engine_exceptions.ContainerCleanupError + ) + running_activation.refresh_from_db() + process.refresh_from_db() + assert running_activation.status == enums.ActivationStatus.ERROR + assert "cleanup failed" in running_activation.status_message + assert process.status == enums.ActivationStatus.RUNNING + assert process.activation_pod_id == "test-pod-id" + assert engine.namespace == "new-ns" + engine.cleanup.assert_called_once() + restart_mock.assert_not_called() + + +@pytest.mark.parametrize( + "container_status", + [ + enums.ActivationStatus.COMPLETED, + enums.ActivationStatus.FAILED, + enums.ActivationStatus.ERROR, + ], +) +@pytest.mark.django_db +def test_monitor_terminal_status_after_cleanup( + running_activation: models.Activation, + container_status: enums.ActivationStatus, +): + """Successful cleanup still applies the corresponding terminal policy.""" + running_activation.restart_policy = enums.RestartPolicy.NEVER + running_activation.save(update_fields=["restart_policy"]) + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "new-ns" + engine.get_status.return_value = MagicMock( + status=container_status, message="container exited" + ) + manager = ActivationManager(running_activation, container_engine=engine) + + if container_status == enums.ActivationStatus.ERROR: + with pytest.raises(exceptions.ActivationMonitorError): + manager.monitor() + else: + manager.monitor() + + running_activation.refresh_from_db() + assert running_activation.status == container_status + assert running_activation.latest_instance.status == container_status + engine.cleanup.assert_called_once() + + +@pytest.mark.parametrize( + ("failure_count", "should_restart"), + [ + (0, True), + (django_settings.ACTIVATION_MAX_RESTARTS_ON_FAILURE, False), + ], +) +@pytest.mark.django_db +def test_monitor_failed_job_cleans_up_once_with_restart_policy( + running_activation: models.Activation, + failure_count: int, + should_restart: bool, +): + """Failed Jobs are not cleaned up again when applying restart policy.""" + running_activation.restart_policy = enums.RestartPolicy.ON_FAILURE + running_activation.failure_count = failure_count + running_activation.save(update_fields=["restart_policy", "failure_count"]) + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "new-ns" + engine.get_status.return_value = MagicMock( + status=enums.ActivationStatus.FAILED, message="container exited" + ) + manager = ActivationManager(running_activation, container_engine=engine) + + with patch(SYSTEM_RESTART_PATH) as restart_mock: + manager.monitor() + + running_activation.refresh_from_db() + process = running_activation.latest_instance + process.refresh_from_db() + assert running_activation.status == enums.ActivationStatus.FAILED + assert process.status == enums.ActivationStatus.FAILED + assert process.activation_pod_id is None + assert running_activation.failure_count == failure_count + 1 + engine.cleanup.assert_called_once() + if should_restart: + restart_delay = django_settings.ACTIVATION_RESTART_SECONDS_ON_FAILURE + restart_mock.assert_called_once_with( + enums.ProcessParentType.ACTIVATION, + running_activation.id, + delay_seconds=restart_delay, + ) + else: + restart_mock.assert_not_called() + + +@pytest.mark.django_db +def test_error_instance_cleans_up_job_when_not_already_done( + running_activation: models.Activation, +): + """Errors with a live Job clean up in the recorded namespace.""" + process = running_activation.latest_instance + process.activation_job_namespace = "old-ns" + process.save(update_fields=["activation_job_namespace"]) + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "new-ns" + namespaces = [] + engine.cleanup.side_effect = lambda *_: namespaces.append(engine.namespace) + manager = ActivationManager(running_activation, container_engine=engine) + + manager._error_instance("engine error") + + process.refresh_from_db() + assert process.status == enums.ActivationStatus.ERROR + assert process.activation_pod_id is None + assert namespaces == ["old-ns"] + assert engine.namespace == "new-ns" + engine.cleanup.assert_called_once() + assert engine.cleanup.call_args.args[0] == "test-pod-id" + + +@pytest.mark.django_db +def test_monitor_disabled_cleanup_failure_maps_to_monitor_error( + running_activation: models.Activation, +): + running_activation.is_enabled = False + running_activation.save(update_fields=["is_enabled"]) + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "new-ns" + engine.cleanup.side_effect = engine_exceptions.ContainerCleanupError( + "cleanup failed" + ) + manager = ActivationManager(running_activation, container_engine=engine) + + with pytest.raises(exceptions.ActivationMonitorError) as exc: + manager.monitor() + + assert isinstance(exc.value.__cause__, exceptions.ActivationStopError) + running_activation.refresh_from_db() + assert running_activation.status == enums.ActivationStatus.ERROR + assert running_activation.latest_instance.status == ( + enums.ActivationStatus.STOPPING + ) + assert ( + running_activation.latest_instance.activation_pod_id == "test-pod-id" + ) + + +@pytest.mark.django_db +def test_start_cleanup_failure_blocks_replacement( + running_activation: models.Activation, + preseed_credential_types, +): + running_activation.status = enums.ActivationStatus.STOPPED + running_activation.save(update_fields=["status"]) + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "new-ns" + engine.get_status.return_value = MagicMock( + status=enums.ActivationStatus.STOPPED + ) + engine.cleanup.side_effect = engine_exceptions.ContainerCleanupError( + "cleanup failed" + ) + process = running_activation.latest_instance + process.activation_job_namespace = "old-ns" + process.save(update_fields=["activation_job_namespace"]) + manager = ActivationManager(running_activation, container_engine=engine) + + with pytest.raises(exceptions.ActivationStartError) as exc: + manager.start() + + assert isinstance( + exc.value.__cause__, engine_exceptions.ContainerCleanupError + ) + running_activation.refresh_from_db() + process.refresh_from_db() + assert running_activation.status == enums.ActivationStatus.ERROR + assert process.status == enums.ActivationStatus.RUNNING + assert process.activation_pod_id == "test-pod-id" + assert ( + models.RulebookProcess.objects.filter( + activation=running_activation + ).count() + == 1 + ) + assert engine.namespace == "new-ns" + engine.start.assert_not_called() + + engine.cleanup.side_effect = None + engine.start.return_value = "replacement-pod-id" + manager.start() + + process.refresh_from_db() + assert process.status == enums.ActivationStatus.STOPPED + assert running_activation.latest_instance.id != process.id + engine.start.assert_called_once() + + @pytest.mark.django_db def test_start_restart( running_activation: models.Activation, @@ -1109,6 +1341,37 @@ def test_handle_unresponsive_activation_reraises_cleanup_error( assert activation.status == enums.ActivationStatus.FAILED +@pytest.mark.django_db +def test_handle_unresponsive_kubernetes_cleanup_failure_preserves_job( + starting_activation: models.Activation, +): + """Failed Job cleanup keeps the process available for a later retry.""" + activation = starting_activation + process = activation.latest_instance + process.activation_job_namespace = "old-ns" + process.save(update_fields=["activation_job_namespace"]) + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "new-ns" + engine.cleanup.side_effect = engine_exceptions.ContainerCleanupError( + "cleanup failed" + ) + manager = ActivationManager(activation, container_engine=engine) + + with pytest.raises(engine_exceptions.ContainerCleanupError): + manager._handle_unresponsive_activation(check_type="Readiness") + + activation.refresh_from_db() + process.refresh_from_db() + assert activation.status == enums.ActivationStatus.ERROR + assert "cleanup failed" in activation.status_message + assert process.status == enums.ActivationStatus.STARTING + assert process.activation_pod_id == "test-pod-id" + assert process.activation_job_namespace == "old-ns" + assert engine.namespace == "new-ns" + engine.cleanup.assert_called_once() + assert engine.cleanup.call_args.args[0] == "test-pod-id" + + @pytest.mark.django_db def test_handle_unresponsive_activation_fails_instance_on_success( starting_activation: models.Activation, @@ -1128,3 +1391,178 @@ def test_handle_unresponsive_activation_fails_instance_on_success( activation.refresh_from_db() assert activation.status == enums.ActivationStatus.FAILED assert "Liveness" in activation.status_message + + +# --- Namespace-aware lifecycle tests --- + + +@pytest.mark.django_db +def test_create_activation_instance_records_k8s_namespace( + basic_activation: models.Activation, + settings: SettingsWrapper, +): + """When engine is KubernetesEngine, namespace is recorded on process.""" + override_settings = { + "WEBSOCKET_BASE_URL": "ws://localhost:8000", + "WEBSOCKET_SSL_VERIFY": "no", + "RULEBOOK_LIVENESS_CHECK_SECONDS": 73, + } + apply_settings(settings, **override_settings) + + k8s_engine = MagicMock(spec=KubernetesEngine) + k8s_engine.namespace = "eda-activations" + + manager = ActivationManager( + basic_activation, + container_engine=k8s_engine, + ) + manager._create_activation_instance() + + process = basic_activation.latest_instance + assert process is not None + assert process.activation_job_namespace == "eda-activations" + + +@pytest.mark.parametrize( + ("has_previous_instance", "previous_namespace"), + [(False, None), (True, None), (True, "aap-eda")], +) +@pytest.mark.django_db +def test_start_logs_k8s_namespace_with_activation_and_instance( + basic_activation: models.Activation, + preseed_credential_types, + has_previous_instance: bool, + previous_namespace: str | None, +): + """Record the new Job's namespace and any known previous namespace.""" + k8s_engine = MagicMock(spec=KubernetesEngine) + k8s_engine.namespace = "eda-activations" + k8s_engine.start.return_value = "activation-job" + previous_instance = None + if has_previous_instance: + previous_instance = models.RulebookProcess.objects.create( + name=basic_activation.name, + activation=basic_activation, + organization=basic_activation.organization, + status=enums.ActivationStatus.STOPPED, + activation_job_namespace=previous_namespace, + ) + + manager = ActivationManager(basic_activation, container_engine=k8s_engine) + manager.start() + + process = basic_activation.latest_instance + assert process.activation_job_namespace == "eda-activations" + logs = models.RulebookProcessLog.objects.filter( + activation_instance=process, + log__contains="Job started in namespace", + ) + expected_log = ( + f"Activation {basic_activation.id} " + f"(name={basic_activation.name!r}), " + f"instance {process.id}: Job started in namespace 'eda-activations'" + ) + if previous_namespace: + expected_log += ( + f"; previous instance {previous_instance.id} recorded " + "Job namespace 'aap-eda'" + ) + assert [log.log for log in logs] == [f"{expected_log}."] + + +@pytest.mark.django_db +def test_failed_k8s_start_does_not_log_job_started( + basic_activation: models.Activation, + preseed_credential_types, +): + k8s_engine = MagicMock(spec=KubernetesEngine) + k8s_engine.namespace = "eda-activations" + k8s_engine.start.side_effect = engine_exceptions.ContainerStartError( + "job creation failed" + ) + + manager = ActivationManager(basic_activation, container_engine=k8s_engine) + with pytest.raises(exceptions.ActivationStartError): + manager.start() + + assert not models.RulebookProcessLog.objects.filter( + activation_instance=basic_activation.latest_instance, + log__contains="Job started in namespace", + ).exists() + + +@pytest.mark.django_db +def test_create_activation_instance_null_namespace_non_k8s( + basic_activation: models.Activation, + container_engine_mock: MagicMock, + settings: SettingsWrapper, +): + """Non-K8s engine leaves activation_job_namespace as None.""" + override_settings = { + "WEBSOCKET_BASE_URL": "ws://localhost:8000", + "WEBSOCKET_SSL_VERIFY": "no", + "RULEBOOK_LIVENESS_CHECK_SECONDS": 73, + } + apply_settings(settings, **override_settings) + + manager = ActivationManager( + basic_activation, + container_engine=container_engine_mock, + ) + manager._create_activation_instance() + + process = basic_activation.latest_instance + assert process is not None + assert process.activation_job_namespace is None + + +@pytest.mark.django_db +def test_cleanup_uses_instance_engine_namespace( + running_activation: models.Activation, +): + """Cleanup targets the process's recorded namespace, not engine default.""" + k8s_engine = MagicMock(spec=KubernetesEngine) + k8s_engine.namespace = "new-ns" + + proc = running_activation.latest_instance + proc.activation_job_namespace = "old-ns" + proc.save(update_fields=["activation_job_namespace"]) + + namespace_during_cleanup = [] + + def capture_namespace(*args, **kwargs): + namespace_during_cleanup.append(k8s_engine.namespace) + + k8s_engine.cleanup.side_effect = capture_namespace + + manager = ActivationManager( + running_activation, + container_engine=k8s_engine, + ) + manager._cleanup() + + assert k8s_engine.namespace == "new-ns" + k8s_engine.cleanup.assert_called_once() + assert namespace_during_cleanup == ["old-ns"] + + +@pytest.mark.django_db +def test_cleanup_uses_engine_default_for_null_namespace( + running_activation: models.Activation, +): + """Legacy process (null namespace) uses engine default for cleanup.""" + k8s_engine = MagicMock(spec=KubernetesEngine) + k8s_engine.namespace = "current-ns" + + proc = running_activation.latest_instance + proc.activation_job_namespace = None + proc.save(update_fields=["activation_job_namespace"]) + + manager = ActivationManager( + running_activation, + container_engine=k8s_engine, + ) + manager._cleanup() + + assert k8s_engine.namespace == "current-ns" + k8s_engine.cleanup.assert_called_once() diff --git a/tests/unit/test_activation_job_namespace.py b/tests/unit/test_activation_job_namespace.py index 1a3cae33f..397a3ce5b 100644 --- a/tests/unit/test_activation_job_namespace.py +++ b/tests/unit/test_activation_job_namespace.py @@ -16,14 +16,24 @@ from unittest import mock import pytest +from kubernetes.client.rest import ApiException from aap_eda.services.activation.engine.exceptions import ( ContainerEngineInitError, ) -from aap_eda.services.activation.engine.kubernetes import Engine +from aap_eda.services.activation.engine.kubernetes import ( + K8S_API_RETRIES, + K8S_API_RETRY_BACKOFF, + Engine, + make_job_name, + resolve_activation_namespace, + verify_namespace_accessible, +) -@mock.patch.dict("os.environ", {"EDA_ACTIVATION_JOB_NAMESPACE": "eda-jobs"}) +@mock.patch.dict( + "os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "eda-jobs"} +) def test_set_namespace_env_override(): engine = Engine( activation_id="1", @@ -34,7 +44,7 @@ def test_set_namespace_env_override(): @mock.patch.dict( - "os.environ", {"EDA_ACTIVATION_JOB_NAMESPACE": " eda-jobs "} + "os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": " eda-jobs "} ) def test_set_namespace_env_override_strips_whitespace(): engine = Engine( @@ -45,7 +55,30 @@ def test_set_namespace_env_override_strips_whitespace(): assert engine.namespace == "eda-jobs" -@mock.patch.dict("os.environ", {"EDA_ACTIVATION_JOB_NAMESPACE": ""}) +@mock.patch.dict( + "os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "eda-jobs"} +) +def test_set_namespace_env_override_retries_transient_probe(): + client = mock.Mock() + client.batch_api.list_namespaced_job.side_effect = [ + ApiException(status=503, reason="Unavailable"), + None, + ] + + with mock.patch( + "aap_eda.services.activation.engine.kubernetes.time.sleep" + ): + engine = Engine( + activation_id="1", + resource_prefix="activation", + client=client, + ) + + assert engine.namespace == "eda-jobs" + assert client.batch_api.list_namespaced_job.call_count == 2 + + +@mock.patch.dict("os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": ""}) def test_set_namespace_empty_env_falls_back_to_file(): with mock.patch("builtins.open", mock.mock_open(read_data="aap-eda")): engine = Engine( @@ -60,7 +93,7 @@ def test_set_namespace_unset_env_falls_back_to_file(): env = { k: v for k, v in os.environ.items() - if k != "EDA_ACTIVATION_JOB_NAMESPACE" + if k != "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE" } with mock.patch.dict("os.environ", env, clear=True): with mock.patch( @@ -75,7 +108,9 @@ def test_set_namespace_unset_env_falls_back_to_file(): assert engine.namespace == "my-namespace" -@mock.patch.dict("os.environ", {"EDA_ACTIVATION_JOB_NAMESPACE": " "}) +@mock.patch.dict( + "os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": " "} +) def test_set_namespace_whitespace_only_env_falls_back_to_file(): with mock.patch("builtins.open", mock.mock_open(read_data="aap-eda")): engine = Engine( @@ -94,3 +129,199 @@ def test_set_namespace_no_env_no_file_raises(): resource_prefix="activation", client=mock.Mock(), ) + + +# --- verify_namespace_accessible tests --- + + +def test_verify_namespace_accessible_success(): + """No exception when list_namespaced_job succeeds.""" + batch_api = mock.Mock() + batch_api.list_namespaced_job.return_value = None + verify_namespace_accessible(batch_api, "test-ns") + batch_api.list_namespaced_job.assert_called_once_with( + namespace="test-ns", limit=1 + ) + + +def test_verify_namespace_accessible_4xx_raises(): + """4xx errors raise ContainerEngineInitError with namespace info.""" + batch_api = mock.Mock() + batch_api.list_namespaced_job.side_effect = ApiException( + status=403, reason="Forbidden" + ) + with pytest.raises(ContainerEngineInitError, match="not accessible"): + verify_namespace_accessible(batch_api, "forbidden-ns") + batch_api.list_namespaced_job.assert_called_once() + + +def test_verify_namespace_accessible_retries_5xx_then_succeeds(): + """Transient server failure during construction need not block init.""" + batch_api = mock.Mock() + batch_api.list_namespaced_job.side_effect = [ + ApiException(status=503, reason="Unavailable"), + None, + ] + with mock.patch( + "aap_eda.services.activation.engine.kubernetes.time.sleep" + ) as sleep: + verify_namespace_accessible(batch_api, "test-ns") + + assert batch_api.list_namespaced_job.call_count == 2 + sleep.assert_called_once_with(K8S_API_RETRY_BACKOFF) + + +def test_verify_namespace_accessible_5xx_raises_after_retries(): + """Persistent transient errors still fail closed after bounded retries.""" + batch_api = mock.Mock() + batch_api.list_namespaced_job.side_effect = ApiException( + status=500, reason="Internal" + ) + with mock.patch( + "aap_eda.services.activation.engine.kubernetes.time.sleep" + ) as sleep: + with pytest.raises(ContainerEngineInitError, match="Cannot verify"): + verify_namespace_accessible(batch_api, "error-ns") + + assert batch_api.list_namespaced_job.call_count == K8S_API_RETRIES + assert sleep.call_args_list == [ + mock.call(K8S_API_RETRY_BACKOFF * attempt) + for attempt in range(1, K8S_API_RETRIES) + ] + + +def test_verify_namespace_accessible_non_transient_5xx_raises_immediately(): + batch_api = mock.Mock() + batch_api.list_namespaced_job.side_effect = ApiException( + status=501, reason="Not Implemented" + ) + with mock.patch( + "aap_eda.services.activation.engine.kubernetes.time.sleep" + ) as sleep: + with pytest.raises(ContainerEngineInitError, match="Cannot verify"): + verify_namespace_accessible(batch_api, "error-ns") + + batch_api.list_namespaced_job.assert_called_once() + sleep.assert_not_called() + + +# --- resolve_activation_namespace tests --- + + +@mock.patch.dict( + "os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "override-ns"} +) +def test_resolve_activation_namespace_env_override(): + """Env var returns (namespace, True).""" + ns, is_override = resolve_activation_namespace() + assert ns == "override-ns" + assert is_override is True + + +def test_resolve_activation_namespace_sa_file(): + """SA file returns (namespace, False) when both overrides are unset.""" + env = { + k: v + for k, v in os.environ.items() + if k + not in { + "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE", + "EDA_ACTIVATION_JOB_NAMESPACE", + } + } + with mock.patch.dict("os.environ", env, clear=True): + with mock.patch("builtins.open", mock.mock_open(read_data="sa-ns")): + ns, is_override = resolve_activation_namespace() + assert ns == "sa-ns" + assert is_override is False + + +def test_resolve_activation_namespace_no_env_no_file(): + """No env and no file raises ContainerEngineInitError.""" + with mock.patch.dict("os.environ", {}, clear=True): + with pytest.raises(ContainerEngineInitError, match="Cannot determine"): + resolve_activation_namespace() + + +# --- deprecated env var fallback tests --- + + +@mock.patch.dict( + "os.environ", + {"EDA_ACTIVATION_JOB_NAMESPACE": "legacy-ns"}, + clear=True, +) +def test_resolve_deprecated_env_var_fallback(): + """Old env var works as fallback with is_override=True.""" + ns, is_override = resolve_activation_namespace() + assert ns == "legacy-ns" + assert is_override is True + + +@mock.patch.dict( + "os.environ", + { + "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "new-ns", + "EDA_ACTIVATION_JOB_NAMESPACE": "old-ns", + }, +) +def test_resolve_new_env_var_takes_precedence(): + """New env var takes precedence over old.""" + ns, is_override = resolve_activation_namespace() + assert ns == "new-ns" + assert is_override is True + + +@mock.patch.dict( + "os.environ", + {"EDA_ACTIVATION_JOB_NAMESPACE": "legacy-ns"}, + clear=True, +) +def test_set_namespace_deprecated_env_override(): + """Engine uses deprecated env var when new one is not set.""" + engine = Engine( + activation_id="1", + resource_prefix="activation", + client=mock.Mock(), + ) + assert engine.namespace == "legacy-ns" + + +# --- make_job_name tests --- + + +def test_make_job_name(): + assert make_job_name("activation", 42, 7) == "activation-job-42-7" + + +def test_make_job_name_different_prefix(): + assert make_job_name("event-stream", 1, 2) == "event-stream-job-1-2" + + +# --- ACTIVATION_TERMINAL_STATUSES tests --- + + +def test_activation_terminal_statuses_contains_expected(): + """All four terminal statuses are present.""" + from aap_eda.core.enums import ( + ACTIVATION_TERMINAL_STATUSES, + ActivationStatus, + ) + + expected = { + ActivationStatus.STOPPED, + ActivationStatus.COMPLETED, + ActivationStatus.FAILED, + ActivationStatus.ERROR, + } + assert ACTIVATION_TERMINAL_STATUSES == expected + + +def test_activation_terminal_statuses_excludes_running(): + from aap_eda.core.enums import ( + ACTIVATION_TERMINAL_STATUSES, + ActivationStatus, + ) + + assert ActivationStatus.RUNNING not in ACTIVATION_TERMINAL_STATUSES + assert ActivationStatus.STARTING not in ACTIVATION_TERMINAL_STATUSES diff --git a/tests/unit/test_namespace_lifecycle.py b/tests/unit/test_namespace_lifecycle.py new file mode 100644 index 000000000..c66721396 --- /dev/null +++ b/tests/unit/test_namespace_lifecycle.py @@ -0,0 +1,114 @@ +# Copyright 2026 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for namespace-aware lifecycle in ActivationManager.""" + +from unittest.mock import MagicMock, create_autospec + +import pytest + +from aap_eda.services.activation.activation_manager import ActivationManager +from aap_eda.services.activation.engine.common import ContainerEngine +from aap_eda.services.activation.engine.kubernetes import ( + Engine as KubernetesEngine, +) + + +class TestInstanceEngine: + """Tests for _instance_engine context manager.""" + + def _make_manager(self, activation, engine): + return ActivationManager(activation, container_engine=engine) + + def test_swaps_namespace_for_recorded_process(self): + """When process has a namespace, engine namespace is swapped.""" + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "current-ns" + + instance = MagicMock() + instance.activation_job_namespace = "old-ns" + instance.id = 1 + + activation = MagicMock() + manager = self._make_manager(activation, engine) + + with manager._instance_engine(instance): + assert engine.namespace == "old-ns" + assert engine.namespace == "current-ns" + + def test_restores_namespace_on_exception(self): + """Engine namespace restored even when block raises.""" + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "current-ns" + + instance = MagicMock() + instance.activation_job_namespace = "old-ns" + instance.id = 1 + + activation = MagicMock() + manager = self._make_manager(activation, engine) + + def raise_inside_instance_engine(): + with manager._instance_engine(instance): + assert engine.namespace == "old-ns" + raise ValueError("test error") + + with pytest.raises(ValueError, match="test error"): + raise_inside_instance_engine() + assert engine.namespace == "current-ns" + + def test_no_swap_for_null_namespace(self): + """Legacy process with null namespace uses engine default.""" + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "current-ns" + + instance = MagicMock() + instance.activation_job_namespace = None + instance.id = 2 + + activation = MagicMock() + manager = self._make_manager(activation, engine) + + with manager._instance_engine(instance): + assert engine.namespace == "current-ns" + assert engine.namespace == "current-ns" + + def test_no_swap_for_empty_string_namespace(self): + """Empty string namespace treated same as None.""" + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "current-ns" + + instance = MagicMock() + instance.activation_job_namespace = "" + instance.id = 3 + + activation = MagicMock() + manager = self._make_manager(activation, engine) + + with manager._instance_engine(instance): + assert engine.namespace == "current-ns" + assert engine.namespace == "current-ns" + + def test_non_k8s_engine_yields_without_swap(self): + """Non-Kubernetes engine yields without any namespace logic.""" + engine = create_autospec(ContainerEngine, instance=True) + + instance = MagicMock() + instance.activation_job_namespace = "some-ns" + + activation = MagicMock() + manager = self._make_manager(activation, engine) + + with manager._instance_engine(instance): + pass From fa900340e815e97c8d2b3100e5ec4d5b3dcab9e0 Mon Sep 17 00:00:00 2001 From: Kaio Oliveira Date: Thu, 22 Oct 2026 23:18:51 -0300 Subject: [PATCH 2/4] [AAP-98100] fix: preserve legacy activation namespaces Backfill activation_job_namespace for pre-feature Kubernetes RulebookProcess records from the legacy execution namespace. When a legacy row is still NULL at lifecycle time, persist and use the control-plane (or deprecated legacy override) namespace instead of the current configured default. This ensures a namespace change removes the old Job before creating a replacement in the target namespace. Co-authored-by: Codex --- ...lebook_process_activation_job_namespace.py | 61 +++++++++++++++++++ .../services/activation/activation_manager.py | 21 ++++--- .../services/activation/engine/kubernetes.py | 44 ++++++++++--- ...lebook_process_activation_job_namespace.py | 60 ++++++++++++++++++ .../services/activation/test_manager.py | 38 ++++++++++++ tests/unit/test_activation_job_namespace.py | 38 ++++++++++++ tests/unit/test_namespace_lifecycle.py | 48 +++++++++++---- 7 files changed, 280 insertions(+), 30 deletions(-) create mode 100644 src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py create mode 100644 tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py diff --git a/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py b/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py new file mode 100644 index 000000000..c97757e78 --- /dev/null +++ b/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py @@ -0,0 +1,61 @@ +import os + +from django.conf import settings +from django.db import migrations + +SA_NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace" + + +def legacy_activation_namespace() -> str: + """Return the namespace used by Kubernetes Jobs before this feature. + + The deprecated override is the only pre-feature setting that could place + activation Jobs outside the control-plane namespace. The new default + namespace setting is intentionally ignored because it may be configured + while this migration is running and applies only to new Jobs. + """ + old_override = os.environ.get("EDA_ACTIVATION_JOB_NAMESPACE", "").strip() + if old_override: + return old_override + + try: + with open(SA_NAMESPACE_FILE, encoding="utf-8") as namespace_file: + namespace = namespace_file.read().strip() + except FileNotFoundError as exc: + raise RuntimeError( + "Cannot backfill legacy activation Job namespaces because " + f"{SA_NAMESPACE_FILE} does not exist." + ) from exc + + if not namespace: + raise RuntimeError( + "Cannot backfill legacy activation Job namespaces because " + f"{SA_NAMESPACE_FILE} is empty." + ) + + return namespace + + +def backfill_rulebook_process_activation_job_namespace(apps, schema_editor): + """Record the legacy Kubernetes namespace for existing processes.""" + if settings.DEPLOYMENT_TYPE != "k8s": + return + + RulebookProcess = apps.get_model("core", "RulebookProcess") # noqa: N806 + db_alias = schema_editor.connection.alias + RulebookProcess.objects.using(db_alias).filter( + activation_job_namespace__isnull=True + ).update(activation_job_namespace=legacy_activation_namespace()) + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0075_rulebookprocess_activation_job_namespace"), + ] + + operations = [ + migrations.RunPython( + backfill_rulebook_process_activation_job_namespace, + migrations.RunPython.noop, + ), + ] diff --git a/src/aap_eda/services/activation/activation_manager.py b/src/aap_eda/services/activation/activation_manager.py index 994c5231c..f6ff64f5c 100644 --- a/src/aap_eda/services/activation/activation_manager.py +++ b/src/aap_eda/services/activation/activation_manager.py @@ -39,7 +39,10 @@ from .drools_cleanup import drools_cleanup from .engine.common import ContainerableInvalidError, ContainerEngine from .engine.factory import new_container_engine -from .engine.kubernetes import Engine as KubernetesEngine +from .engine.kubernetes import ( + Engine as KubernetesEngine, + resolve_legacy_activation_namespace, +) from .status_manager import StatusManager, run_with_lock from .tee_system_logger import TeeSystemLogger @@ -84,10 +87,9 @@ def _instance_engine(self, instance): """Use the Job's persisted namespace for operations on that Job. When the process has a recorded namespace, temporarily swap the - engine's namespace for the duration of the block. When it is - None (legacy process created before this field existed), use the - engine's current namespace unchanged — that is where pre-feature - Jobs were created. + engine's namespace for the duration of the block. A process without + one was created before this field existed, so persist and use the + legacy namespace rather than the current default for new Jobs. """ if not isinstance(self.container_engine, KubernetesEngine): yield @@ -95,14 +97,15 @@ def _instance_engine(self, instance): namespace = instance.activation_job_namespace if not namespace: + namespace = resolve_legacy_activation_namespace() + instance.activation_job_namespace = namespace + instance.save(update_fields=["activation_job_namespace"]) LOGGER.info( "Process %s has no recorded Job namespace, " - "using engine default %s", + "recording legacy namespace %s", instance.id, - self.container_engine.namespace, + namespace, ) - yield - return configured_namespace = self.container_engine.namespace self.container_engine.namespace = namespace diff --git a/src/aap_eda/services/activation/engine/kubernetes.py b/src/aap_eda/services/activation/engine/kubernetes.py index cc59dec4a..1ad4f7d55 100644 --- a/src/aap_eda/services/activation/engine/kubernetes.py +++ b/src/aap_eda/services/activation/engine/kubernetes.py @@ -92,6 +92,40 @@ def get_k8s_client() -> Client: SA_NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace" +def get_service_account_namespace() -> str: + """Return the namespace of the EDA control-plane ServiceAccount.""" + try: + with open(SA_NAMESPACE_FILE, encoding="utf-8") as f: + namespace = f.read().strip() + except FileNotFoundError as e: + raise ContainerEngineInitError( + f"Cannot determine control-plane namespace: " + f"{SA_NAMESPACE_FILE} does not exist." + ) from e + + if not namespace: + raise ContainerEngineInitError( + "Cannot determine control-plane namespace: " + f"{SA_NAMESPACE_FILE} is empty." + ) + + return namespace + + +def resolve_legacy_activation_namespace() -> str: + """Return the namespace used by activation Jobs before this feature. + + ``EDA_ACTIVATION_JOB_NAMESPACE`` was the legacy override. Do not use + ``EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE`` here: that setting selects the + destination for new Jobs and may have changed since a legacy Job started. + """ + old_override = os.environ.get("EDA_ACTIVATION_JOB_NAMESPACE", "").strip() + if old_override: + return old_override + + return get_service_account_namespace() + + def verify_namespace_accessible(batch_api, namespace: str) -> None: """Probe a namespace, retrying transient errors but failing on 4xx.""" for attempt in range(1, K8S_API_RETRIES + 1): @@ -146,15 +180,7 @@ def resolve_activation_namespace() -> tuple[str, bool]: ) return old_override, True - try: - with open(SA_NAMESPACE_FILE, encoding="utf-8") as f: - return f.read().strip(), False - except FileNotFoundError as e: - raise ContainerEngineInitError( - "Cannot determine activation namespace: " - "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE is not set and " - f"{SA_NAMESPACE_FILE} does not exist." - ) from e + return get_service_account_namespace(), False def make_job_name( diff --git a/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py b/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py new file mode 100644 index 000000000..387ccc1ea --- /dev/null +++ b/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py @@ -0,0 +1,60 @@ +from importlib import import_module +from unittest import mock + +import pytest +from django.conf import settings +from django.core.management import call_command +from django.db import connection +from django.db.migrations.executor import MigrationExecutor + +from aap_eda.core import models + + +def get_historical_model(app_label, model_name, migration): + """Get a model as it existed at a specific migration.""" + executor = MigrationExecutor(connection) + state = executor.loader.project_state((app_label, migration)) + return state.apps.get_model(app_label, model_name) + + +@pytest.fixture +def rollback_migration(): + """Rollback to before the namespace field exists and restore later.""" + call_command("migrate", "core", "0074") + yield + call_command("migrate") + + +@pytest.mark.django_db(transaction=True) +def test_migration_records_control_plane_namespace( + rollback_migration, default_organization, tmp_path +): + """Pre-feature Kubernetes processes retain their original namespace.""" + HistoricalRulebookProcess = get_historical_model( # noqa: N806 + "core", + "RulebookProcess", + "0074_awxtoken_ck_awx_token_name_not_empty_and_more", + ) + process = HistoricalRulebookProcess.objects.create( + name="legacy-process", + activation_pod_id="activation-job-1-1", + organization_id=default_organization.id, + ) + + namespace_file = tmp_path / "namespace" + namespace_file.write_text("eda-control-plane\n", encoding="utf-8") + migration_module = import_module( + "aap_eda.core.migrations." + "0076_backfill_rulebook_process_activation_job_namespace" + ) + + with ( + mock.patch.object(settings, "DEPLOYMENT_TYPE", "k8s"), + mock.patch.object( + migration_module, "SA_NAMESPACE_FILE", str(namespace_file) + ), + ): + call_command("migrate", "core", "0076") + + migrated_process = models.RulebookProcess.objects.get(pk=process.pk) + assert migrated_process.activation_job_namespace == "eda-control-plane" diff --git a/tests/integration/services/activation/test_manager.py b/tests/integration/services/activation/test_manager.py index fe7ba6329..746ae1d22 100644 --- a/tests/integration/services/activation/test_manager.py +++ b/tests/integration/services/activation/test_manager.py @@ -1423,6 +1423,44 @@ def test_create_activation_instance_records_k8s_namespace( assert process.activation_job_namespace == "eda-activations" +@pytest.mark.django_db +@patch( + "aap_eda.services.activation.activation_manager." + "resolve_legacy_activation_namespace", + return_value="eda-control-plane", +) +def test_start_replaces_legacy_job_in_control_plane_namespace( + resolve_legacy_namespace, + running_activation: models.Activation, + preseed_credential_types, +): + """Clean up a legacy Job before creating a target-namespace replacement.""" + legacy_process = running_activation.latest_instance + assert legacy_process.activation_job_namespace is None + + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "eda-activation-jobs" + cleanup_namespaces = [] + engine.cleanup.side_effect = lambda *_: cleanup_namespaces.append( + engine.namespace + ) + engine.start.return_value = "replacement-job" + manager = ActivationManager(running_activation, container_engine=engine) + + manager.start() + + legacy_process.refresh_from_db() + replacement_process = running_activation.latest_instance + assert cleanup_namespaces == ["eda-control-plane"] + assert legacy_process.activation_job_namespace == "eda-control-plane" + assert replacement_process.id != legacy_process.id + assert replacement_process.activation_job_namespace == ( + "eda-activation-jobs" + ) + assert engine.namespace == "eda-activation-jobs" + resolve_legacy_namespace.assert_called_once_with() + + @pytest.mark.parametrize( ("has_previous_instance", "previous_namespace"), [(False, None), (True, None), (True, "aap-eda")], diff --git a/tests/unit/test_activation_job_namespace.py b/tests/unit/test_activation_job_namespace.py index 397a3ce5b..b79d5825b 100644 --- a/tests/unit/test_activation_job_namespace.py +++ b/tests/unit/test_activation_job_namespace.py @@ -25,8 +25,10 @@ K8S_API_RETRIES, K8S_API_RETRY_BACKOFF, Engine, + get_service_account_namespace, make_job_name, resolve_activation_namespace, + resolve_legacy_activation_namespace, verify_namespace_accessible, ) @@ -243,6 +245,17 @@ def test_resolve_activation_namespace_no_env_no_file(): resolve_activation_namespace() +def test_get_service_account_namespace_strips_whitespace(): + with mock.patch("builtins.open", mock.mock_open(read_data=" eda-api\n")): + assert get_service_account_namespace() == "eda-api" + + +def test_get_service_account_namespace_empty_file_raises(): + with mock.patch("builtins.open", mock.mock_open(read_data=" \n")): + with pytest.raises(ContainerEngineInitError, match="is empty"): + get_service_account_namespace() + + # --- deprecated env var fallback tests --- @@ -258,6 +271,31 @@ def test_resolve_deprecated_env_var_fallback(): assert is_override is True +@mock.patch.dict( + "os.environ", + { + "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "target-ns", + "EDA_ACTIVATION_JOB_NAMESPACE": "old-ns", + }, + clear=True, +) +def test_resolve_legacy_namespace_ignores_new_default(): + """Legacy cleanup must not use a configured destination namespace.""" + assert resolve_legacy_activation_namespace() == "old-ns" + + +@mock.patch.dict( + "os.environ", + {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "target-ns"}, + clear=True, +) +def test_resolve_legacy_namespace_uses_service_account_namespace(): + with mock.patch( + "builtins.open", mock.mock_open(read_data="control-plane-ns") + ): + assert resolve_legacy_activation_namespace() == "control-plane-ns" + + @mock.patch.dict( "os.environ", { diff --git a/tests/unit/test_namespace_lifecycle.py b/tests/unit/test_namespace_lifecycle.py index c66721396..a24f362e1 100644 --- a/tests/unit/test_namespace_lifecycle.py +++ b/tests/unit/test_namespace_lifecycle.py @@ -14,7 +14,7 @@ """Tests for namespace-aware lifecycle in ActivationManager.""" -from unittest.mock import MagicMock, create_autospec +from unittest.mock import MagicMock, create_autospec, patch import pytest @@ -68,10 +68,17 @@ def raise_inside_instance_engine(): raise_inside_instance_engine() assert engine.namespace == "current-ns" - def test_no_swap_for_null_namespace(self): - """Legacy process with null namespace uses engine default.""" + @patch( + "aap_eda.services.activation.activation_manager." + "resolve_legacy_activation_namespace", + return_value="control-plane-ns", + ) + def test_null_namespace_records_and_uses_legacy_namespace( + self, resolve_legacy_namespace + ): + """A legacy process never uses the configured namespace as source.""" engine = MagicMock(spec=KubernetesEngine) - engine.namespace = "current-ns" + engine.namespace = "target-ns" instance = MagicMock() instance.activation_job_namespace = None @@ -81,13 +88,25 @@ def test_no_swap_for_null_namespace(self): manager = self._make_manager(activation, engine) with manager._instance_engine(instance): - assert engine.namespace == "current-ns" - assert engine.namespace == "current-ns" - - def test_no_swap_for_empty_string_namespace(self): - """Empty string namespace treated same as None.""" + assert engine.namespace == "control-plane-ns" + assert engine.namespace == "target-ns" + assert instance.activation_job_namespace == "control-plane-ns" + instance.save.assert_called_once_with( + update_fields=["activation_job_namespace"] + ) + resolve_legacy_namespace.assert_called_once_with() + + @patch( + "aap_eda.services.activation.activation_manager." + "resolve_legacy_activation_namespace", + return_value="control-plane-ns", + ) + def test_empty_namespace_records_and_uses_legacy_namespace( + self, resolve_legacy_namespace + ): + """An empty legacy value is treated the same as a null value.""" engine = MagicMock(spec=KubernetesEngine) - engine.namespace = "current-ns" + engine.namespace = "target-ns" instance = MagicMock() instance.activation_job_namespace = "" @@ -97,8 +116,13 @@ def test_no_swap_for_empty_string_namespace(self): manager = self._make_manager(activation, engine) with manager._instance_engine(instance): - assert engine.namespace == "current-ns" - assert engine.namespace == "current-ns" + assert engine.namespace == "control-plane-ns" + assert engine.namespace == "target-ns" + assert instance.activation_job_namespace == "control-plane-ns" + instance.save.assert_called_once_with( + update_fields=["activation_job_namespace"] + ) + resolve_legacy_namespace.assert_called_once_with() def test_non_k8s_engine_yields_without_swap(self): """Non-Kubernetes engine yields without any namespace logic.""" From 84ecd64cc18a693c45eee14f6049dd73ff3201ad Mon Sep 17 00:00:00 2001 From: Kaio Oliveira Date: Fri, 23 Oct 2026 00:14:29 -0300 Subject: [PATCH 3/4] [AAP-98100] fix: remove experimental namespace fallback Remove support for the upstream-only EDA_ACTIVATION_JOB_NAMESPACE setting. Persist the control-plane ServiceAccount namespace for activation Jobs that predate namespace-aware lifecycle handling. The configured container-group default remains the destination for new Jobs only. Keep Kubernetes processes monitorable when Job cleanup fails, preserving the recorded source namespace and Job reference for a later cleanup retry. This prevents an unsafe replacement Job from being created in a new namespace. --- ...lebook_process_activation_job_namespace.py | 21 ++--- .../services/activation/activation_manager.py | 28 +++---- .../services/activation/engine/kubernetes.py | 22 ----- ...lebook_process_activation_job_namespace.py | 21 +++++ .../activation/engine/test_kubernetes.py | 7 +- .../services/activation/test_manager.py | 40 ++++++--- tests/unit/test_activation_job_namespace.py | 83 +------------------ tests/unit/test_namespace_lifecycle.py | 12 +-- 8 files changed, 84 insertions(+), 150 deletions(-) diff --git a/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py b/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py index c97757e78..260caabdb 100644 --- a/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py +++ b/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py @@ -1,5 +1,3 @@ -import os - from django.conf import settings from django.db import migrations @@ -9,15 +7,10 @@ def legacy_activation_namespace() -> str: """Return the namespace used by Kubernetes Jobs before this feature. - The deprecated override is the only pre-feature setting that could place - activation Jobs outside the control-plane namespace. The new default - namespace setting is intentionally ignored because it may be configured - while this migration is running and applies only to new Jobs. + Before this feature, Kubernetes activation Jobs ran in the EDA + control-plane namespace. The configured default namespace is intentionally + ignored because it applies only to new Jobs. """ - old_override = os.environ.get("EDA_ACTIVATION_JOB_NAMESPACE", "").strip() - if old_override: - return old_override - try: with open(SA_NAMESPACE_FILE, encoding="utf-8") as namespace_file: namespace = namespace_file.read().strip() @@ -43,9 +36,13 @@ def backfill_rulebook_process_activation_job_namespace(apps, schema_editor): RulebookProcess = apps.get_model("core", "RulebookProcess") # noqa: N806 db_alias = schema_editor.connection.alias - RulebookProcess.objects.using(db_alias).filter( + processes = RulebookProcess.objects.using(db_alias).filter( activation_job_namespace__isnull=True - ).update(activation_job_namespace=legacy_activation_namespace()) + ) + if not processes.exists(): + return + + processes.update(activation_job_namespace=legacy_activation_namespace()) class Migration(migrations.Migration): diff --git a/src/aap_eda/services/activation/activation_manager.py b/src/aap_eda/services/activation/activation_manager.py index f6ff64f5c..e8d14a81c 100644 --- a/src/aap_eda/services/activation/activation_manager.py +++ b/src/aap_eda/services/activation/activation_manager.py @@ -41,7 +41,7 @@ from .engine.factory import new_container_engine from .engine.kubernetes import ( Engine as KubernetesEngine, - resolve_legacy_activation_namespace, + get_service_account_namespace, ) from .status_manager import StatusManager, run_with_lock from .tee_system_logger import TeeSystemLogger @@ -97,7 +97,7 @@ def _instance_engine(self, instance): namespace = instance.activation_job_namespace if not namespace: - namespace = resolve_legacy_activation_namespace() + namespace = get_service_account_namespace() instance.activation_job_namespace = namespace instance.save(update_fields=["activation_job_namespace"]) LOGGER.info( @@ -480,26 +480,26 @@ def _handle_unresponsive_activation(self, check_type: str): Leave the process intact if Kubernetes Job cleanup fails so it can be retried. Other engines retain FAILED handling on cleanup errors. """ - kubernetes_cleanup_failed = False + user_msg = ( + "Activation is unresponsive. " + f"{check_type} check for ansible-rulebook timed out." + ) try: self._cleanup(raise_on_error=True) except engine_exceptions.ContainerCleanupError as exc: if isinstance(self.container_engine, KubernetesEngine): - kubernetes_cleanup_failed = True msg = ( f"Activation {self.db_instance.id} is unresponsive. " f"Cleanup failed. Reason: {exc}" ) - self._error_activation(msg) + LOGGER.error(msg) + raise + self._fail_instance(msg=user_msg, cleanup_already_done=True) + self.set_status(ActivationStatus.FAILED, user_msg) raise - finally: - if not kubernetes_cleanup_failed: - user_msg = ( - "Activation is unresponsive. " - f"{check_type} check for ansible-rulebook timed out." - ) - self._fail_instance(msg=user_msg, cleanup_already_done=True) - self.set_status(ActivationStatus.FAILED, user_msg) + + self._fail_instance(msg=user_msg, cleanup_already_done=True) + self.set_status(ActivationStatus.FAILED, user_msg) def _missing_container_policy(self): LOGGER.info( @@ -1029,7 +1029,7 @@ def monitor(self): f"Monitor operation: activation {self.db_instance.id} " f"failed to cleanup its Job. Reason: {exc}" ) - self._error_activation(msg) + LOGGER.error(msg) raise exceptions.ActivationMonitorError(msg) from exc if container_status.status == ActivationStatus.COMPLETED: # RC == 0 diff --git a/src/aap_eda/services/activation/engine/kubernetes.py b/src/aap_eda/services/activation/engine/kubernetes.py index 1ad4f7d55..6f4391632 100644 --- a/src/aap_eda/services/activation/engine/kubernetes.py +++ b/src/aap_eda/services/activation/engine/kubernetes.py @@ -112,20 +112,6 @@ def get_service_account_namespace() -> str: return namespace -def resolve_legacy_activation_namespace() -> str: - """Return the namespace used by activation Jobs before this feature. - - ``EDA_ACTIVATION_JOB_NAMESPACE`` was the legacy override. Do not use - ``EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE`` here: that setting selects the - destination for new Jobs and may have changed since a legacy Job started. - """ - old_override = os.environ.get("EDA_ACTIVATION_JOB_NAMESPACE", "").strip() - if old_override: - return old_override - - return get_service_account_namespace() - - def verify_namespace_accessible(batch_api, namespace: str) -> None: """Probe a namespace, retrying transient errors but failing on 4xx.""" for attempt in range(1, K8S_API_RETRIES + 1): @@ -172,14 +158,6 @@ def resolve_activation_namespace() -> tuple[str, bool]: if ns_override: return ns_override, True - old_override = os.environ.get("EDA_ACTIVATION_JOB_NAMESPACE", "").strip() - if old_override: - LOGGER.warning( - "EDA_ACTIVATION_JOB_NAMESPACE is deprecated; " - "use EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE instead." - ) - return old_override, True - return get_service_account_namespace(), False diff --git a/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py b/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py index 387ccc1ea..7e71af167 100644 --- a/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py +++ b/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py @@ -58,3 +58,24 @@ def test_migration_records_control_plane_namespace( migrated_process = models.RulebookProcess.objects.get(pk=process.pk) assert migrated_process.activation_job_namespace == "eda-control-plane" + + +@pytest.mark.django_db(transaction=True) +def test_migration_skips_namespace_lookup_without_legacy_processes( + rollback_migration, +): + """An empty installation does not need the ServiceAccount namespace.""" + migration_module = import_module( + "aap_eda.core.migrations." + "0076_backfill_rulebook_process_activation_job_namespace" + ) + + with ( + mock.patch.object(settings, "DEPLOYMENT_TYPE", "k8s"), + mock.patch.object( + migration_module, "legacy_activation_namespace" + ) as legacy_activation_namespace, + ): + call_command("migrate", "core", "0076") + + legacy_activation_namespace.assert_not_called() diff --git a/tests/integration/services/activation/engine/test_kubernetes.py b/tests/integration/services/activation/engine/test_kubernetes.py index d6298be8c..d542c1c0b 100644 --- a/tests/integration/services/activation/engine/test_kubernetes.py +++ b/tests/integration/services/activation/engine/test_kubernetes.py @@ -248,10 +248,9 @@ def test_engine_init_with_exception(init_kubernetes_data): with pytest.raises( ContainerEngineInitError, match=( - "Cannot determine activation namespace: " - "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE is not set and " - "/var/run/secrets/kubernetes.io/serviceaccount" - "/namespace does not exist." + "Cannot determine control-plane namespace: " + "/var/run/secrets/kubernetes.io/serviceaccount/namespace " + "does not exist." ), ): Engine( diff --git a/tests/integration/services/activation/test_manager.py b/tests/integration/services/activation/test_manager.py index 746ae1d22..3cd41ef04 100644 --- a/tests/integration/services/activation/test_manager.py +++ b/tests/integration/services/activation/test_manager.py @@ -496,8 +496,7 @@ def test_monitor_cleanup_failure_preserves_job_for_retry( ) running_activation.refresh_from_db() process.refresh_from_db() - assert running_activation.status == enums.ActivationStatus.ERROR - assert "cleanup failed" in running_activation.status_message + assert running_activation.status == enums.ActivationStatus.RUNNING assert process.status == enums.ActivationStatus.RUNNING assert process.activation_pod_id == "test-pod-id" assert engine.namespace == "new-ns" @@ -526,6 +525,9 @@ def test_monitor_terminal_status_after_cleanup( engine.get_status.return_value = MagicMock( status=container_status, message="container exited" ) + process = running_activation.latest_instance + process.activation_job_namespace = "new-ns" + process.save(update_fields=["activation_job_namespace"]) manager = ActivationManager(running_activation, container_engine=engine) if container_status == enums.ActivationStatus.ERROR: @@ -562,6 +564,9 @@ def test_monitor_failed_job_cleans_up_once_with_restart_policy( engine.get_status.return_value = MagicMock( status=enums.ActivationStatus.FAILED, message="container exited" ) + process = running_activation.latest_instance + process.activation_job_namespace = "new-ns" + process.save(update_fields=["activation_job_namespace"]) manager = ActivationManager(running_activation, container_engine=engine) with patch(SYSTEM_RESTART_PATH) as restart_mock: @@ -1362,8 +1367,7 @@ def test_handle_unresponsive_kubernetes_cleanup_failure_preserves_job( activation.refresh_from_db() process.refresh_from_db() - assert activation.status == enums.ActivationStatus.ERROR - assert "cleanup failed" in activation.status_message + assert activation.status == enums.ActivationStatus.STARTING assert process.status == enums.ActivationStatus.STARTING assert process.activation_pod_id == "test-pod-id" assert process.activation_job_namespace == "old-ns" @@ -1426,11 +1430,11 @@ def test_create_activation_instance_records_k8s_namespace( @pytest.mark.django_db @patch( "aap_eda.services.activation.activation_manager." - "resolve_legacy_activation_namespace", + "get_service_account_namespace", return_value="eda-control-plane", ) def test_start_replaces_legacy_job_in_control_plane_namespace( - resolve_legacy_namespace, + get_control_plane_namespace, running_activation: models.Activation, preseed_credential_types, ): @@ -1458,7 +1462,7 @@ def test_start_replaces_legacy_job_in_control_plane_namespace( "eda-activation-jobs" ) assert engine.namespace == "eda-activation-jobs" - resolve_legacy_namespace.assert_called_once_with() + get_control_plane_namespace.assert_called_once_with() @pytest.mark.parametrize( @@ -1585,16 +1589,26 @@ def capture_namespace(*args, **kwargs): @pytest.mark.django_db -def test_cleanup_uses_engine_default_for_null_namespace( +@patch( + "aap_eda.services.activation.activation_manager." + "get_service_account_namespace", + return_value="control-plane-ns", +) +def test_cleanup_records_control_plane_namespace_for_legacy_process( + get_control_plane_namespace, running_activation: models.Activation, ): - """Legacy process (null namespace) uses engine default for cleanup.""" + """Legacy cleanup uses and records the control-plane namespace.""" k8s_engine = MagicMock(spec=KubernetesEngine) - k8s_engine.namespace = "current-ns" + k8s_engine.namespace = "target-ns" proc = running_activation.latest_instance proc.activation_job_namespace = None proc.save(update_fields=["activation_job_namespace"]) + namespaces = [] + k8s_engine.cleanup.side_effect = lambda *_: namespaces.append( + k8s_engine.namespace + ) manager = ActivationManager( running_activation, @@ -1602,5 +1616,9 @@ def test_cleanup_uses_engine_default_for_null_namespace( ) manager._cleanup() - assert k8s_engine.namespace == "current-ns" + proc.refresh_from_db() + assert k8s_engine.namespace == "target-ns" + assert namespaces == ["control-plane-ns"] + assert proc.activation_job_namespace == "control-plane-ns" + get_control_plane_namespace.assert_called_once_with() k8s_engine.cleanup.assert_called_once() diff --git a/tests/unit/test_activation_job_namespace.py b/tests/unit/test_activation_job_namespace.py index b79d5825b..84ec065cd 100644 --- a/tests/unit/test_activation_job_namespace.py +++ b/tests/unit/test_activation_job_namespace.py @@ -28,7 +28,6 @@ get_service_account_namespace, make_job_name, resolve_activation_namespace, - resolve_legacy_activation_namespace, verify_namespace_accessible, ) @@ -221,17 +220,8 @@ def test_resolve_activation_namespace_env_override(): def test_resolve_activation_namespace_sa_file(): - """SA file returns (namespace, False) when both overrides are unset.""" - env = { - k: v - for k, v in os.environ.items() - if k - not in { - "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE", - "EDA_ACTIVATION_JOB_NAMESPACE", - } - } - with mock.patch.dict("os.environ", env, clear=True): + """SA file returns (namespace, False) when no default is configured.""" + with mock.patch.dict("os.environ", {}, clear=True): with mock.patch("builtins.open", mock.mock_open(read_data="sa-ns")): ns, is_override = resolve_activation_namespace() assert ns == "sa-ns" @@ -256,75 +246,6 @@ def test_get_service_account_namespace_empty_file_raises(): get_service_account_namespace() -# --- deprecated env var fallback tests --- - - -@mock.patch.dict( - "os.environ", - {"EDA_ACTIVATION_JOB_NAMESPACE": "legacy-ns"}, - clear=True, -) -def test_resolve_deprecated_env_var_fallback(): - """Old env var works as fallback with is_override=True.""" - ns, is_override = resolve_activation_namespace() - assert ns == "legacy-ns" - assert is_override is True - - -@mock.patch.dict( - "os.environ", - { - "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "target-ns", - "EDA_ACTIVATION_JOB_NAMESPACE": "old-ns", - }, - clear=True, -) -def test_resolve_legacy_namespace_ignores_new_default(): - """Legacy cleanup must not use a configured destination namespace.""" - assert resolve_legacy_activation_namespace() == "old-ns" - - -@mock.patch.dict( - "os.environ", - {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "target-ns"}, - clear=True, -) -def test_resolve_legacy_namespace_uses_service_account_namespace(): - with mock.patch( - "builtins.open", mock.mock_open(read_data="control-plane-ns") - ): - assert resolve_legacy_activation_namespace() == "control-plane-ns" - - -@mock.patch.dict( - "os.environ", - { - "EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "new-ns", - "EDA_ACTIVATION_JOB_NAMESPACE": "old-ns", - }, -) -def test_resolve_new_env_var_takes_precedence(): - """New env var takes precedence over old.""" - ns, is_override = resolve_activation_namespace() - assert ns == "new-ns" - assert is_override is True - - -@mock.patch.dict( - "os.environ", - {"EDA_ACTIVATION_JOB_NAMESPACE": "legacy-ns"}, - clear=True, -) -def test_set_namespace_deprecated_env_override(): - """Engine uses deprecated env var when new one is not set.""" - engine = Engine( - activation_id="1", - resource_prefix="activation", - client=mock.Mock(), - ) - assert engine.namespace == "legacy-ns" - - # --- make_job_name tests --- diff --git a/tests/unit/test_namespace_lifecycle.py b/tests/unit/test_namespace_lifecycle.py index a24f362e1..74f74e4f8 100644 --- a/tests/unit/test_namespace_lifecycle.py +++ b/tests/unit/test_namespace_lifecycle.py @@ -70,11 +70,11 @@ def raise_inside_instance_engine(): @patch( "aap_eda.services.activation.activation_manager." - "resolve_legacy_activation_namespace", + "get_service_account_namespace", return_value="control-plane-ns", ) def test_null_namespace_records_and_uses_legacy_namespace( - self, resolve_legacy_namespace + self, get_control_plane_namespace ): """A legacy process never uses the configured namespace as source.""" engine = MagicMock(spec=KubernetesEngine) @@ -94,15 +94,15 @@ def test_null_namespace_records_and_uses_legacy_namespace( instance.save.assert_called_once_with( update_fields=["activation_job_namespace"] ) - resolve_legacy_namespace.assert_called_once_with() + get_control_plane_namespace.assert_called_once_with() @patch( "aap_eda.services.activation.activation_manager." - "resolve_legacy_activation_namespace", + "get_service_account_namespace", return_value="control-plane-ns", ) def test_empty_namespace_records_and_uses_legacy_namespace( - self, resolve_legacy_namespace + self, get_control_plane_namespace ): """An empty legacy value is treated the same as a null value.""" engine = MagicMock(spec=KubernetesEngine) @@ -122,7 +122,7 @@ def test_empty_namespace_records_and_uses_legacy_namespace( instance.save.assert_called_once_with( update_fields=["activation_job_namespace"] ) - resolve_legacy_namespace.assert_called_once_with() + get_control_plane_namespace.assert_called_once_with() def test_non_k8s_engine_yields_without_swap(self): """Non-Kubernetes engine yields without any namespace logic.""" From e04a90144e181c30fa7987a92393973f214f0b99 Mon Sep 17 00:00:00 2001 From: Kaio Oliveira Date: Fri, 23 Oct 2026 17:30:21 -0300 Subject: [PATCH 4/4] [AAP-98100] fix: preserve namespace lifecycle retries Move configured-namespace validation from engine construction to Job creation. Existing activation Jobs can then be monitored and cleaned up in their recorded namespace when the configured destination is inaccessible. Keep disabled activations and processes monitorable when cleanup fails, so the scheduler retries rather than leaving a surviving Job orphaned. Fold the namespace backfill into migration 0075 because neither migration has been released. Co-authored-by: Codex --- ...ulebookprocess_activation_job_namespace.py | 48 +++++++++++++ ...lebook_process_activation_job_namespace.py | 58 --------------- .../services/activation/activation_manager.py | 4 +- .../services/activation/engine/kubernetes.py | 5 +- ...ebook_process_activation_job_namespace.py} | 8 +-- .../activation/engine/test_kubernetes.py | 71 +++++++++++++++---- .../services/activation/test_manager.py | 25 +++++-- tests/unit/test_activation_job_namespace.py | 22 ++---- 8 files changed, 141 insertions(+), 100 deletions(-) delete mode 100644 src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py rename tests/integration/core/migrations/{test_0076_backfill_rulebook_process_activation_job_namespace.py => test_0075_rulebook_process_activation_job_namespace.py} (91%) diff --git a/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py b/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py index 22911ea98..4188b26da 100644 --- a/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py +++ b/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py @@ -1,5 +1,49 @@ +from django.conf import settings from django.db import migrations, models +SA_NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace" + + +def legacy_activation_namespace() -> str: + """Return the namespace used by Kubernetes Jobs before this feature. + + Before this feature, Kubernetes activation Jobs ran in the EDA + control-plane namespace. The configured default namespace is intentionally + ignored because it applies only to new Jobs. + """ + try: + with open(SA_NAMESPACE_FILE, encoding="utf-8") as namespace_file: + namespace = namespace_file.read().strip() + except FileNotFoundError as exc: + raise RuntimeError( + "Cannot backfill legacy activation Job namespaces because " + f"{SA_NAMESPACE_FILE} does not exist." + ) from exc + + if not namespace: + raise RuntimeError( + "Cannot backfill legacy activation Job namespaces because " + f"{SA_NAMESPACE_FILE} is empty." + ) + + return namespace + + +def backfill_rulebook_process_activation_job_namespace(apps, schema_editor): + """Record the legacy Kubernetes namespace for existing processes.""" + if settings.DEPLOYMENT_TYPE != "k8s": + return + + RulebookProcess = apps.get_model("core", "RulebookProcess") # noqa: N806 + db_alias = schema_editor.connection.alias + processes = RulebookProcess.objects.using(db_alias).filter( + activation_job_namespace__isnull=True + ) + if not processes.exists(): + return + + processes.update(activation_job_namespace=legacy_activation_namespace()) + class Migration(migrations.Migration): dependencies = [ @@ -12,4 +56,8 @@ class Migration(migrations.Migration): name="activation_job_namespace", field=models.TextField(null=True), ), + migrations.RunPython( + backfill_rulebook_process_activation_job_namespace, + migrations.RunPython.noop, + ), ] diff --git a/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py b/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py deleted file mode 100644 index 260caabdb..000000000 --- a/src/aap_eda/core/migrations/0076_backfill_rulebook_process_activation_job_namespace.py +++ /dev/null @@ -1,58 +0,0 @@ -from django.conf import settings -from django.db import migrations - -SA_NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace" - - -def legacy_activation_namespace() -> str: - """Return the namespace used by Kubernetes Jobs before this feature. - - Before this feature, Kubernetes activation Jobs ran in the EDA - control-plane namespace. The configured default namespace is intentionally - ignored because it applies only to new Jobs. - """ - try: - with open(SA_NAMESPACE_FILE, encoding="utf-8") as namespace_file: - namespace = namespace_file.read().strip() - except FileNotFoundError as exc: - raise RuntimeError( - "Cannot backfill legacy activation Job namespaces because " - f"{SA_NAMESPACE_FILE} does not exist." - ) from exc - - if not namespace: - raise RuntimeError( - "Cannot backfill legacy activation Job namespaces because " - f"{SA_NAMESPACE_FILE} is empty." - ) - - return namespace - - -def backfill_rulebook_process_activation_job_namespace(apps, schema_editor): - """Record the legacy Kubernetes namespace for existing processes.""" - if settings.DEPLOYMENT_TYPE != "k8s": - return - - RulebookProcess = apps.get_model("core", "RulebookProcess") # noqa: N806 - db_alias = schema_editor.connection.alias - processes = RulebookProcess.objects.using(db_alias).filter( - activation_job_namespace__isnull=True - ) - if not processes.exists(): - return - - processes.update(activation_job_namespace=legacy_activation_namespace()) - - -class Migration(migrations.Migration): - dependencies = [ - ("core", "0075_rulebookprocess_activation_job_namespace"), - ] - - operations = [ - migrations.RunPython( - backfill_rulebook_process_activation_job_namespace, - migrations.RunPython.noop, - ), - ] diff --git a/src/aap_eda/services/activation/activation_manager.py b/src/aap_eda/services/activation/activation_manager.py index e8d14a81c..c86f0b3ef 100644 --- a/src/aap_eda/services/activation/activation_manager.py +++ b/src/aap_eda/services/activation/activation_manager.py @@ -748,7 +748,6 @@ def _error_instance( def _stop_instance(self): """Stop the latest activation instance.""" - self.set_latest_instance_status(ActivationStatus.STOPPING) self._cleanup() self.set_latest_instance_status(ActivationStatus.STOPPED) self._set_activation_pod_id(pod_id=None) @@ -846,8 +845,7 @@ def stop(self): f"Activation {self.db_instance.id} failed to stop. " f"Reason: {exc}" ) - # Alex: Not sure if we want to change the status here. - self._error_activation(msg) + LOGGER.error(msg) raise exceptions.ActivationStopError(msg) from exc user_msg = "Stop requested by user." if self.db_instance.status != ActivationStatus.ERROR: diff --git a/src/aap_eda/services/activation/engine/kubernetes.py b/src/aap_eda/services/activation/engine/kubernetes.py index 6f4391632..8effe29cc 100644 --- a/src/aap_eda/services/activation/engine/kubernetes.py +++ b/src/aap_eda/services/activation/engine/kubernetes.py @@ -199,6 +199,9 @@ def start(self, request: ContainerRequest, log_handler: LogHandler) -> str: f"-{request.rulebook_process_id}" ) + if self._namespace_is_override: + verify_namespace_accessible(self.client.batch_api, self.namespace) + try: log_handler.write("Creating Job") log_handler.write(f"Log tracking id: {request.log_tracking_id}") @@ -878,8 +881,8 @@ def _set_namespace(self) -> None: "Activation Job namespace override from environment: %s", namespace, ) - verify_namespace_accessible(self.client.batch_api, namespace) self.namespace = namespace + self._namespace_is_override = is_override def _create_secret( self, diff --git a/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py b/tests/integration/core/migrations/test_0075_rulebook_process_activation_job_namespace.py similarity index 91% rename from tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py rename to tests/integration/core/migrations/test_0075_rulebook_process_activation_job_namespace.py index 7e71af167..dc8ada76d 100644 --- a/tests/integration/core/migrations/test_0076_backfill_rulebook_process_activation_job_namespace.py +++ b/tests/integration/core/migrations/test_0075_rulebook_process_activation_job_namespace.py @@ -45,7 +45,7 @@ def test_migration_records_control_plane_namespace( namespace_file.write_text("eda-control-plane\n", encoding="utf-8") migration_module = import_module( "aap_eda.core.migrations." - "0076_backfill_rulebook_process_activation_job_namespace" + "0075_rulebookprocess_activation_job_namespace" ) with ( @@ -54,7 +54,7 @@ def test_migration_records_control_plane_namespace( migration_module, "SA_NAMESPACE_FILE", str(namespace_file) ), ): - call_command("migrate", "core", "0076") + call_command("migrate", "core", "0075") migrated_process = models.RulebookProcess.objects.get(pk=process.pk) assert migrated_process.activation_job_namespace == "eda-control-plane" @@ -67,7 +67,7 @@ def test_migration_skips_namespace_lookup_without_legacy_processes( """An empty installation does not need the ServiceAccount namespace.""" migration_module = import_module( "aap_eda.core.migrations." - "0076_backfill_rulebook_process_activation_job_namespace" + "0075_rulebookprocess_activation_job_namespace" ) with ( @@ -76,6 +76,6 @@ def test_migration_skips_namespace_lookup_without_legacy_processes( migration_module, "legacy_activation_namespace" ) as legacy_activation_namespace, ): - call_command("migrate", "core", "0076") + call_command("migrate", "core", "0075") legacy_activation_namespace.assert_not_called() diff --git a/tests/integration/services/activation/engine/test_kubernetes.py b/tests/integration/services/activation/engine/test_kubernetes.py index d542c1c0b..c661b9b6a 100644 --- a/tests/integration/services/activation/engine/test_kubernetes.py +++ b/tests/integration/services/activation/engine/test_kubernetes.py @@ -1482,10 +1482,10 @@ def test_process_pod_start_event_unrecognized(kubernetes_engine): @pytest.mark.django_db -def test_set_namespace_with_env_override_verifies_access( +def test_set_namespace_with_env_override_defers_access_check( init_kubernetes_data, ): - """When env var is set, _set_namespace probes the target namespace.""" + """An override does not block lifecycle operations at construction.""" mock_client = mock.Mock() mock_client.batch_api.list_namespaced_job.return_value = None @@ -1499,10 +1499,7 @@ def test_set_namespace_with_env_override_verifies_access( client=mock_client, ) assert engine.namespace == "eda-activations" - mock_client.batch_api.list_namespaced_job.assert_called_once_with( - namespace="eda-activations", - limit=1, - ) + mock_client.batch_api.list_namespaced_job.assert_not_called() @pytest.mark.django_db @@ -1526,10 +1523,48 @@ def test_set_namespace_without_env_skips_verification(init_kubernetes_data): @pytest.mark.django_db -def test_set_namespace_env_override_inaccessible_raises( - init_kubernetes_data, +def test_start_with_env_override_verifies_access( + init_kubernetes_data, mock_log_handler, default_organization ): - """Inaccessible override namespace raises during init.""" + """Starting a new Job verifies the configured target namespace.""" + request = get_request( + init_kubernetes_data, + "tester", + default_organization, + ) + mock_client = mock.Mock() + mock_client.batch_api.list_namespaced_job.return_value = None + with mock.patch.dict( + "os.environ", + {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "eda-activations"}, + ): + engine = Engine( + activation_id=str(init_kubernetes_data.activation.id), + resource_prefix="activation", + client=mock_client, + ) + + with mock.patch.object(engine, "_create_job"), mock.patch.object( + engine, "_wait_for_pod_to_start" + ), mock.patch.object(engine, "_create_service"): + engine.start(request, mock_log_handler) + + mock_client.batch_api.list_namespaced_job.assert_called_once_with( + namespace="eda-activations", + limit=1, + ) + + +@pytest.mark.django_db +def test_start_with_inaccessible_env_override_raises( + init_kubernetes_data, mock_log_handler, default_organization +): + """An inaccessible configured target fails before Job creation.""" + request = get_request( + init_kubernetes_data, + "tester", + default_organization, + ) mock_client = mock.Mock() mock_client.batch_api.list_namespaced_job.side_effect = ApiException( status=403, reason="Forbidden" @@ -1538,12 +1573,18 @@ def test_set_namespace_env_override_inaccessible_raises( "os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "forbidden-ns"}, ): - with pytest.raises(ContainerEngineInitError, match="not accessible"): - Engine( - activation_id=str(init_kubernetes_data.activation.id), - resource_prefix="activation", - client=mock_client, - ) + engine = Engine( + activation_id=str(init_kubernetes_data.activation.id), + resource_prefix="activation", + client=mock_client, + ) + with mock.patch.object(engine, "_create_job") as create_job: + with pytest.raises( + ContainerEngineInitError, match="not accessible" + ): + engine.start(request, mock_log_handler) + + create_job.assert_not_called() @pytest.mark.django_db diff --git a/tests/integration/services/activation/test_manager.py b/tests/integration/services/activation/test_manager.py index 3cd41ef04..43f258360 100644 --- a/tests/integration/services/activation/test_manager.py +++ b/tests/integration/services/activation/test_manager.py @@ -617,9 +617,10 @@ def test_error_instance_cleans_up_job_when_not_already_done( @pytest.mark.django_db -def test_monitor_disabled_cleanup_failure_maps_to_monitor_error( +def test_monitor_disabled_cleanup_failure_retries_until_stopped( running_activation: models.Activation, ): + """Disabled activations retry cleanup until the Job is removed.""" running_activation.is_enabled = False running_activation.save(update_fields=["is_enabled"]) engine = MagicMock(spec=KubernetesEngine) @@ -627,6 +628,9 @@ def test_monitor_disabled_cleanup_failure_maps_to_monitor_error( engine.cleanup.side_effect = engine_exceptions.ContainerCleanupError( "cleanup failed" ) + process = running_activation.latest_instance + process.activation_job_namespace = "old-ns" + process.save(update_fields=["activation_job_namespace"]) manager = ActivationManager(running_activation, container_engine=engine) with pytest.raises(exceptions.ActivationMonitorError) as exc: @@ -634,14 +638,27 @@ def test_monitor_disabled_cleanup_failure_maps_to_monitor_error( assert isinstance(exc.value.__cause__, exceptions.ActivationStopError) running_activation.refresh_from_db() - assert running_activation.status == enums.ActivationStatus.ERROR - assert running_activation.latest_instance.status == ( - enums.ActivationStatus.STOPPING + assert running_activation.status == enums.ActivationStatus.RUNNING + assert ( + running_activation.latest_instance.status + == enums.ActivationStatus.RUNNING ) assert ( running_activation.latest_instance.activation_pod_id == "test-pod-id" ) + engine.cleanup.side_effect = None + manager.monitor() + + running_activation.refresh_from_db() + assert running_activation.status == enums.ActivationStatus.STOPPED + assert ( + running_activation.latest_instance.status + == enums.ActivationStatus.STOPPED + ) + assert running_activation.latest_instance.activation_pod_id is None + assert engine.cleanup.call_count == 2 + @pytest.mark.django_db def test_start_cleanup_failure_blocks_replacement( diff --git a/tests/unit/test_activation_job_namespace.py b/tests/unit/test_activation_job_namespace.py index 84ec065cd..2dcbee71e 100644 --- a/tests/unit/test_activation_job_namespace.py +++ b/tests/unit/test_activation_job_namespace.py @@ -59,24 +59,16 @@ def test_set_namespace_env_override_strips_whitespace(): @mock.patch.dict( "os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "eda-jobs"} ) -def test_set_namespace_env_override_retries_transient_probe(): +def test_set_namespace_env_override_skips_probe_during_construction(): client = mock.Mock() - client.batch_api.list_namespaced_job.side_effect = [ - ApiException(status=503, reason="Unavailable"), - None, - ] - - with mock.patch( - "aap_eda.services.activation.engine.kubernetes.time.sleep" - ): - engine = Engine( - activation_id="1", - resource_prefix="activation", - client=client, - ) + engine = Engine( + activation_id="1", + resource_prefix="activation", + client=client, + ) assert engine.namespace == "eda-jobs" - assert client.batch_api.list_namespaced_job.call_count == 2 + client.batch_api.list_namespaced_job.assert_not_called() @mock.patch.dict("os.environ", {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": ""})