Skip to content

Use broader exception to ensure things are properly deleted #48

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
brucetony wants to merge 2 commits into from
Closed

Use broader exception to ensure things are properly deleted #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6f50d22
fix(deploy): use broader exception to ensure things are properly deleted
brucetony Jun 12, 2026
7f8c8ee
fix(deploy): add extra timeouts and cacthes during deploy
brucetony Jun 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 64 additions & 25 deletions src/k8s/kubernetes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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.

Expand All @@ -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]:
Expand Down
17 changes: 11 additions & 6 deletions src/resources/analysis/entity.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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,
Expand Down
1 change: 1 addition & 0 deletions src/status/constants.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
19 changes: 14 additions & 5 deletions src/status/status.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Comment on lines +213 to +221

@coderabbitai coderabbitai Bot Jun 12, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Handle None value for time_created more defensively.

The expression analysis.time_created or 0 uses 0 as a fallback if time_created is None. This would make elapsed = time.time() - 0 evaluate to the current Unix timestamp (~1.7 billion seconds), immediately triggering the timeout.

While the database creation code sets time_created=time.time(), so this should not happen in normal operation, a more defensive approach would explicitly check for None and either log a warning or treat it as a data integrity error.

🛡️ Proposed fix 
         db_status = analysis.status
         if db_status == AnalysisStatus.STARTING.value:
-            elapsed = time.time() - (analysis.time_created or 0)
+            if analysis.time_created is None:
+                logger.error(f"Analysis {analysis_id} missing time_created — treating as timed out")
+                elapsed = _STARTING_TIMEOUT + 1  # Force timeout path
+            else:
+                elapsed = time.time() - analysis.time_created
             if elapsed < _STARTING_TIMEOUT:  # 360s / 6min
                 int_status = AnalysisStatus.STARTING.value
             else:
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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
if db_status == AnalysisStatus.STARTING.value:
if analysis.time_created is None:
logger.error(f"Analysis {analysis_id} missing time_created — treating as timed out")
elapsed = _STARTING_TIMEOUT + 1 # Force timeout path
else:
elapsed = time.time() - analysis.time_created
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
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/status/status.py` around lines 213 - 221, When handling
AnalysisStatus.STARTING in src/status/status.py, explicitly check if
analysis.time_created is None before computing elapsed; if None, log a warning
including analysis_id about missing time_created (data integrity), then treat it
as stuck by setting db_status = AnalysisStatus.STARTED.value and int_status =
AnalysisStatus.FAILED.value (same as the timeout branch); otherwise compute
elapsed = time.time() - analysis.time_created and keep the existing
_STARTING_TIMEOUT logic.

# 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:
Expand Down