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..4188b26da --- /dev/null +++ b/src/aap_eda/core/migrations/0075_rulebookprocess_activation_job_namespace.py @@ -0,0 +1,63 @@ +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 = [ + ("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), + ), + migrations.RunPython( + backfill_rulebook_process_activation_job_namespace, + migrations.RunPython.noop, + ), + ] 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..c86f0b3ef 100644 --- a/src/aap_eda/services/activation/activation_manager.py +++ b/src/aap_eda/services/activation/activation_manager.py @@ -39,6 +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, + get_service_account_namespace, +) from .status_manager import StatusManager, run_with_lock from .tee_system_logger import TeeSystemLogger @@ -78,6 +82,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. 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 + return + + namespace = instance.activation_job_namespace + if not namespace: + namespace = get_service_account_namespace() + instance.activation_job_namespace = namespace + instance.save(update_fields=["activation_job_namespace"]) + LOGGER.info( + "Process %s has no recorded Job namespace, " + "recording legacy namespace %s", + instance.id, + namespace, + ) + + 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 +222,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 +295,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 +340,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 +358,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 +379,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 +477,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. """ + user_msg = ( + "Activation is unresponsive. " + f"{check_type} check for ansible-rulebook timed out." + ) try: self._cleanup(raise_on_error=True) - finally: - user_msg = ( - "Activation is unresponsive. " - f"{check_type} check for ansible-rulebook timed out." - ) + except engine_exceptions.ContainerCleanupError as exc: + if isinstance(self.container_engine, KubernetesEngine): + msg = ( + f"Activation {self.db_instance.id} is unresponsive. " + f"Cleanup failed. Reason: {exc}" + ) + LOGGER.error(msg) + raise self._fail_instance(msg=user_msg, cleanup_already_done=True) self.set_status(ActivationStatus.FAILED, user_msg) + raise + + 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 +597,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 +618,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 +655,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 +689,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,18 +734,20 @@ 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) 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) @@ -705,7 +788,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() @@ -754,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: @@ -886,7 +976,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 +1015,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}" + ) + LOGGER.error(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 +1085,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 +1105,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 +1161,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 +1196,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..8effe29cc 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,85 @@ def get_k8s_client() -> Client: raise ContainerEngineInitError(str(e)) from e +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 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 + + return get_service_account_namespace(), False + + +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,15 +189,19 @@ 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}" 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}") @@ -570,14 +654,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 +698,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 +719,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 +791,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 +875,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 + self.namespace = namespace + self._namespace_is_override = is_override 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/core/migrations/test_0075_rulebook_process_activation_job_namespace.py b/tests/integration/core/migrations/test_0075_rulebook_process_activation_job_namespace.py new file mode 100644 index 000000000..dc8ada76d --- /dev/null +++ b/tests/integration/core/migrations/test_0075_rulebook_process_activation_job_namespace.py @@ -0,0 +1,81 @@ +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." + "0075_rulebookprocess_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", "0075") + + 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." + "0075_rulebookprocess_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", "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 52c485a6d..c661b9b6a 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,8 +248,9 @@ def test_engine_init_with_exception(init_kubernetes_data): with pytest.raises( ContainerEngineInitError, match=( - "Namespace file /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( @@ -731,7 +734,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 +746,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 +757,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 +800,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 +882,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 +1254,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 +1476,148 @@ 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_defers_access_check( + init_kubernetes_data, +): + """An override does not block lifecycle operations at construction.""" + 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_not_called() + + +@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_start_with_env_override_verifies_access( + init_kubernetes_data, mock_log_handler, default_organization +): + """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" + ) + with mock.patch.dict( + "os.environ", + {"EDA_CONTAINER_GROUP_DEFAULT_NAMESPACE": "forbidden-ns"}, + ): + 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 +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..43f258360 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,256 @@ 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.RUNNING + 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" + ) + 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: + 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" + ) + 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: + 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_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) + engine.namespace = "new-ns" + 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: + manager.monitor() + + assert isinstance(exc.value.__cause__, exceptions.ActivationStopError) + running_activation.refresh_from_db() + 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( + 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 +1363,36 @@ 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.STARTING + 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 +1412,230 @@ 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.django_db +@patch( + "aap_eda.services.activation.activation_manager." + "get_service_account_namespace", + return_value="eda-control-plane", +) +def test_start_replaces_legacy_job_in_control_plane_namespace( + get_control_plane_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" + get_control_plane_namespace.assert_called_once_with() + + +@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 +@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 cleanup uses and records the control-plane namespace.""" + k8s_engine = MagicMock(spec=KubernetesEngine) + 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, + container_engine=k8s_engine, + ) + manager._cleanup() + + 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 1a3cae33f..2dcbee71e 100644 --- a/tests/unit/test_activation_job_namespace.py +++ b/tests/unit/test_activation_job_namespace.py @@ -16,14 +16,25 @@ 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, + get_service_account_namespace, + 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 +45,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 +56,22 @@ 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_skips_probe_during_construction(): + client = mock.Mock() + engine = Engine( + activation_id="1", + resource_prefix="activation", + client=client, + ) + + assert engine.namespace == "eda-jobs" + client.batch_api.list_namespaced_job.assert_not_called() + + +@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 +86,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 +101,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 +122,157 @@ 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 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" + 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() + + +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() + + +# --- 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..74f74e4f8 --- /dev/null +++ b/tests/unit/test_namespace_lifecycle.py @@ -0,0 +1,138 @@ +# 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, patch + +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" + + @patch( + "aap_eda.services.activation.activation_manager." + "get_service_account_namespace", + return_value="control-plane-ns", + ) + def test_null_namespace_records_and_uses_legacy_namespace( + self, get_control_plane_namespace + ): + """A legacy process never uses the configured namespace as source.""" + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "target-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 == "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"] + ) + get_control_plane_namespace.assert_called_once_with() + + @patch( + "aap_eda.services.activation.activation_manager." + "get_service_account_namespace", + return_value="control-plane-ns", + ) + def test_empty_namespace_records_and_uses_legacy_namespace( + self, get_control_plane_namespace + ): + """An empty legacy value is treated the same as a null value.""" + engine = MagicMock(spec=KubernetesEngine) + engine.namespace = "target-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 == "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"] + ) + 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.""" + 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