diff --git a/src/k8s/kubernetes.py b/src/k8s/kubernetes.py index 5aec805..994969c 100644 --- a/src/k8s/kubernetes.py +++ b/src/k8s/kubernetes.py @@ -18,6 +18,8 @@ 'analysis': [8000], 'service': [80]} +_IP_WAIT_TIMEOUT = 300 # seconds to wait for a pod IP before giving up + def create_harbor_secret(host_address: str, user: str, @@ -141,30 +143,10 @@ def delete_deployment(deployment_name: str, namespace: str = 'default') -> None: logger.action(f"Deleting deployment {deployment_name} in namespace {namespace} at {time.strftime('%Y-%m-%d %H:%M:%S')}") app_client = client.AppsV1Api() for name in [deployment_name, f'nginx-{deployment_name}']: - try: - app_client.delete_namespaced_deployment(async_req=False, name=name, namespace=namespace) - _delete_service(name, namespace) - except client.exceptions.ApiException as e: - if e.reason == 'Not Found': - logger.warning(f"Could not find {name} for deletion") - else: - logger.error(f"Unknown error when attempting to delete {name} (reason={e.reason})") - network_client = client.NetworkingV1Api() - try: - network_client.delete_namespaced_network_policy(name=f'nginx-to-{deployment_name}-policy', namespace=namespace) - except client.exceptions.ApiException as e: - if e.reason == 'Not Found': - logger.error(f"Could not find nginx-to-{deployment_name}-policy for deletion") - else: - logger.error(f"Unknown error when attempting to delete nginx-to-{deployment_name}-policy (reason={e.reason})") - core_client = client.CoreV1Api() - try: - core_client.delete_namespaced_config_map(name=f"nginx-{deployment_name}-config", namespace=namespace) - except client.exceptions.ApiException as e: - if e.reason == 'Not Found': - logger.error(f"Could not find {deployment_name}-config for deletion") - else: - logger.error(f"Unknown error when attempting to delete {deployment_name}-config (reason={e.reason})") + _delete_k8s_deployment(app_client, name, namespace) + _delete_service(name, namespace) + _delete_network_policy(f'nginx-to-{deployment_name}-policy', namespace) + _delete_config_map(f'nginx-{deployment_name}-config', namespace) def get_analysis_logs(deployment_names: dict[str, str], @@ -373,7 +355,10 @@ def _create_nginx_config_map(analysis_name: str, 'component=flame-message-broker', namespace=namespace)[0] message_broker_pod = None + _mb_wait_start = time.time() while message_broker_pod is None: + if time.time() - _mb_wait_start > _IP_WAIT_TIMEOUT: + raise TimeoutError(f"Timed out waiting for message broker pod IP (>{_IP_WAIT_TIMEOUT}s)") message_broker_pod = core_client.read_namespaced_pod(name=message_broker_pod_name, namespace=namespace) if message_broker_pod is not None: @@ -392,7 +377,11 @@ def _create_nginx_config_map(analysis_name: str, 'component=flame-po', namespace=namespace)[0] pod_orchestration_pod = None + _po_wait_start = time.time() while pod_orchestration_pod is None: + if time.time() - _po_wait_start > _IP_WAIT_TIMEOUT: + raise TimeoutError(f"Timed out waiting for pod orchestration pod IP (>{_IP_WAIT_TIMEOUT}s)") + try: pod_orchestration_pod = core_client.read_namespaced_pod(name=pod_orchestration_name, namespace=namespace) @@ -404,7 +393,11 @@ def _create_nginx_config_map(analysis_name: str, # await and get analysis pod ip analysis_ip = None + _analysis_wait_start = time.time() while analysis_ip is None: + if time.time() - _analysis_wait_start > _IP_WAIT_TIMEOUT: + raise TimeoutError(f"Timed out waiting for analysis pod IP for {analysis_name} (>{_IP_WAIT_TIMEOUT}s)") + pod_list_object = core_client.list_namespaced_pod(label_selector=f"app={analysis_name}", watch=False, namespace=namespace) @@ -624,6 +617,18 @@ def _create_analysis_network_policy(analysis_name: str, nginx_name: str, namespa network_client.create_namespaced_network_policy(namespace=namespace, body=network_body) +def _delete_k8s_deployment(app_client: client.AppsV1Api, name: str, namespace: str) -> None: + try: + app_client.delete_namespaced_deployment(async_req=False, name=name, namespace=namespace) + except client.exceptions.ApiException as e: + if e.reason == 'Not Found': + logger.warning(f"Could not find deployment {name} for deletion") + else: + logger.error(f"Unknown error when attempting to delete deployment {name} (reason={e.reason})") + except Exception as e: + logger.error(f"Unexpected error when attempting to delete deployment {name}: {e}") + + def _delete_service(name: str, namespace: str = 'default') -> None: """Delete a Kubernetes service by name. @@ -632,7 +637,41 @@ def _delete_service(name: str, namespace: str = 'default') -> None: namespace: Namespace the service lives in. """ core_client = client.CoreV1Api() - core_client.delete_namespaced_service(async_req=False, name=name, namespace=namespace) + try: + core_client.delete_namespaced_service(async_req=False, name=name, namespace=namespace) + except client.exceptions.ApiException as e: + if e.reason == 'Not Found': + logger.warning(f"Could not find service {name} for deletion") + else: + logger.error(f"Unknown error when attempting to delete service {name} (reason={e.reason})") + except Exception as e: + logger.error(f"Unexpected error when attempting to delete service {name}: {e}") + + +def _delete_network_policy(name: str, namespace: str) -> None: + network_client = client.NetworkingV1Api() + try: + network_client.delete_namespaced_network_policy(name=name, namespace=namespace) + except client.exceptions.ApiException as e: + if e.reason == 'Not Found': + logger.warning(f"Could not find network policy {name} for deletion") + else: + logger.error(f"Unknown error when attempting to delete network policy {name} (reason={e.reason})") + except Exception as e: + logger.error(f"Unexpected error when attempting to delete network policy {name}: {e}") + + +def _delete_config_map(name: str, namespace: str) -> None: + core_client = client.CoreV1Api() + try: + core_client.delete_namespaced_config_map(name=name, namespace=namespace) + except client.exceptions.ApiException as e: + if e.reason == 'Not Found': + logger.warning(f"Could not find config map {name} for deletion") + else: + logger.error(f"Unknown error when attempting to delete config map {name} (reason={e.reason})") + except Exception as e: + logger.error(f"Unexpected error when attempting to delete config map {name}: {e}") def _get_logs(name: str, pod_ids: Optional[list[str]] = None, namespace: str = 'default') -> list[str]: diff --git a/src/resources/analysis/entity.py b/src/resources/analysis/entity.py index dc9c875..444cff3 100644 --- a/src/resources/analysis/entity.py +++ b/src/resources/analysis/entity.py @@ -48,7 +48,6 @@ def start(self, database: Database, namespace: str = 'default') -> None: database: Database wrapper used to persist the new deployment. namespace: Namespace the Kubernetes resources are created in. """ - self.status = AnalysisStatus.STARTED.value self.deployment_name = "analysis-" + self.analysis_id + "-" + str(self.restart_counter) self.tokens = create_analysis_tokens(kong_token=self.kong_token, analysis_id=self.analysis_id) self.analysis_config = self.tokens @@ -60,7 +59,7 @@ def start(self, database: Database, namespace: str = 'default') -> None: deployment_name=self.deployment_name, project_id=self.project_id, pod_ids=None, - status=self.status, + status=AnalysisStatus.STARTING.value, log=self.log, registry_url=self.registry_url, image_url=self.image_url, @@ -70,12 +69,18 @@ def start(self, database: Database, namespace: str = 'default') -> None: kong_token=self.kong_token, restart_counter=self.restart_counter, progress=self.progress) - self.pod_ids = create_analysis_deployment(name=self.deployment_name, - image=self.image_url, - env=self.analysis_config, - namespace=namespace) + try: + self.pod_ids = create_analysis_deployment(name=self.deployment_name, + image=self.image_url, + env=self.analysis_config, + namespace=namespace) + except Exception: + database.update_deployment_status(self.deployment_name, AnalysisStatus.FAILED.value) + raise + self.status = AnalysisStatus.STARTED.value database.update_deployment(self.deployment_name, pod_ids=json.dumps(self.pod_ids)) + database.update_deployment_status(self.deployment_name, self.status) def stop(self, database: Database, diff --git a/src/status/constants.py b/src/status/constants.py index 5a389a9..cd89951 100644 --- a/src/status/constants.py +++ b/src/status/constants.py @@ -3,6 +3,7 @@ _INTERNAL_STATUS_TIMEOUT = 10 # Time in seconds to wait for internal status response +_STARTING_TIMEOUT = 360 # Seconds before a STARTING analysis is considered stuck _MAX_RESTARTS = 10 # Maximum number of restarts for a stuck analysis diff --git a/src/status/status.py b/src/status/status.py index 65615e4..8bfce77 100644 --- a/src/status/status.py +++ b/src/status/status.py @@ -14,7 +14,7 @@ delete_analysis, stream_logs, clean_up_the_rest) -from src.status.constants import AnalysisStatus, _MAX_RESTARTS, _INTERNAL_STATUS_TIMEOUT +from src.status.constants import AnalysisStatus, _MAX_RESTARTS, _INTERNAL_STATUS_TIMEOUT, _STARTING_TIMEOUT from src.utils.hub_client import (init_hub_client, get_node_id_by_client, get_node_analysis_id, @@ -210,16 +210,25 @@ def _get_analysis_status(analysis_id: str, database: Database) -> Optional[dict[ analysis = database.get_latest_deployment(analysis_id) if analysis is not None: db_status = analysis.status + if db_status == AnalysisStatus.STARTING.value: + elapsed = time.time() - (analysis.time_created or 0) + if elapsed < _STARTING_TIMEOUT: # 360s / 6min + int_status = AnalysisStatus.STARTING.value + else: + # Safety net: creation took too long treat as STARTED+FAILED to trigger unstuck + logger.warning(f"Analysis {analysis_id} stuck in STARTING for {elapsed:.0f}s — triggering unstuck") + db_status = AnalysisStatus.STARTED.value + int_status = AnalysisStatus.FAILED.value # Make the Finished status final, the internal status is not checked anymore, # because the analysis will already be deleted - if db_status == AnalysisStatus.EXECUTED.value: + elif db_status == AnalysisStatus.EXECUTED.value: int_status = AnalysisStatus.EXECUTED.value else: int_status = _get_internal_deployment_status(analysis.deployment_name, analysis_id) - status_action = _decide_status_action(analysis.status, int_status) - logger.status_loop(f"Calculating action: db_status={analysis.status}, int_status={int_status} -> status_action={status_action}") + status_action = _decide_status_action(db_status, int_status) + logger.status_loop(f"Calculating action: db_status={db_status}, int_status={int_status} -> status_action={status_action}") return {'analysis_id': analysis_id, - 'db_status': analysis.status, + 'db_status': db_status, 'int_status': int_status, 'status_action': status_action} else: