Canary

[antidodo]

This pull request introduces several improvements and refactorings to the API, analysis lifecycle, and cleanup logic, with a focus on more robust handling of the FLAME Hub client, better logging, and improved cleanup of orphaned resources. The most important changes are grouped below by theme.

API and Hub Client Handling:

• Refactored the initialization and usage of the Hub client and node ID in PodOrchestrationAPI, introducing a _set_node_id_and_hub_client method with retry logic to ensure the client and node ID are set before performing log streaming and cleanup operations. This method is now called before relevant API actions to improve reliability. [1] [2] [3] [4] [5] [6] [7]

• Updated the API to use the new init_hub_client function instead of the old init_hub_client_with_client, and adjusted imports accordingly.

Analysis Lifecycle and Logging:

• In the Analysis entity, changed the creation and updating of pod_ids to occur after the analysis database entry is created, ensuring consistency between the database and deployment. Also, added logic to clean up message-broker subscriptions when stopping an analysis. [1] [2] [3]

• Updated AnalysisStoppedLog to use the AnalysisStatus.STOPPED.value for the status field, improving consistency.

Cleanup and Resource Management:

• Refactored the cleanup logic to accept the Hub client as an argument, enabling validation of known analysis IDs against the Hub before deleting orphaned (zombie) Kubernetes resources. Added a helper method _validate_analyses_with_hub to perform this validation and updated the cleanup routines to use it. [1] [2] [3] [4] [5]

Utility and Main Entrypoint Adjustments:

• Changed the order of environment and cluster configuration loading in src/main.py to ensure the logger and configs are initialized in the correct sequence. [1] [2] [3]

Other Minor Improvements:

• Updated utility imports and usages to match the new Hub client interface, and improved error logging in several places for better observability. [1] [2] [3]

These changes collectively improve the robustness of the API, ensure better synchronization with the Hub, and make resource cleanup safer and more reliable.

Summary by CodeRabbit

Release Notes

• Bug Fixes

  • Improved reliability of Hub client initialization with automatic retry logic
  • Enhanced resource cleanup to properly remove orphaned Kubernetes resources
  • Fixed Keycloak subscription cleanup on analysis termination
  • Better error handling for Hub connectivity failures

• Refactor

  • Optimized status reconciliation loop with improved error recovery
  • Enhanced failure detection and automatic status database updates
  • Streamlined environment configuration loading process

[coderabbitai]

Warning

Rate limit exceeded

@antidodo has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 55 minutes and 43 seconds before requesting another review.

You’ve run out of usage credits. Purchase more in the billing tab.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: f9c9044a-e37b-4578-8be4-ebf48b8a4a23

📥 Commits

Reviewing files that changed from the base of the PR and between 4c6af92 and d34319a.

📒 Files selected for processing (15)

• pyproject.toml
• src/api/api.py
• src/resources/utils.py
• src/status/status.py
• src/utils/hub_client.py
• src/utils/mb_client.py
• tests/conftest.py
• tests/test_api_api.py
• tests/test_main.py
• tests/test_resources_analysis_entity.py
• tests/test_resources_log_entity.py
• tests/test_resources_utils.py
• tests/test_status_status.py
• tests/test_utils_hub_client.py
• tests/test_utils_other.py

📝 Walkthrough

Walkthrough

Pod Orchestration refactors Hub client initialization from eager to lazy on-demand with automatic retries, introduces Hub-aware zombie cleanup validation, and updates analysis lifecycle to defer pod-ID persistence and delete message broker subscriptions on stop.

Changes

Hub Client Initialization and Cleanup Refactoring

Layer / File(s)	Summary
Hub Client API Contract Updates src/utils/hub_client.py	init_hub_client_with_client is renamed to init_hub_client. get_node_analysis_id becomes overloaded: when node_id_object_id is omitted, it returns a list of analysis-node IDs; when provided, it returns a single ID. get_analysis_node_statuses explicitly returns None when no nodes are found. init_hub_client_and_update_hub_status_with_client is renamed and updated to use init_hub_client.
Cleanup Utilities with Hub Validation src/utils/mb_client.py, src/utils/other.py, src/resources/utils.py	delete_subscription is added to issue HTTP DELETE requests to the message broker. resource_name_to_analysis is simplified to remove the max_r_split parameter. cleanup and clean_up_the_rest now accept hub_client, and clean_up_the_rest validates known analysis IDs against the hub using _validate_analyses_with_hub before deleting zombie resources; when hub_client is missing, validation is skipped with a warning.
Module-Level Bootstrap Initialization src/main.py	Kubernetes cluster configuration and .env loading are moved from main() to module import time, with module-level logger initialization.
API Endpoints with Lazy Hub Initialization src/api/api.py	PodOrchestrationAPI initializes node_id and hub_client to None and calls a new _set_node_id_and_hub_client retry helper during construction. Stop, stream_logs, and cleanup endpoints now call this helper with max_attempts=5 before forwarding logs; if node_id cannot be resolved, endpoints log warnings or raise HTTP 500 instead of proceeding.
Log Entity Status Standardization src/resources/log/entity.py	AnalysisStoppedLog now uses AnalysisStatus.STOPPED.value enum instead of the hardcoded "stopped" string.
Analysis Lifecycle Pod IDs and Subscription Cleanup src/resources/analysis/entity.py	Analysis.start now creates the database row with pod_ids=None first, then updates it after deployment creation. Analysis.stop calls delete_subscription with the Keycloak token after deletion and status updates.
Resource Creation/Destruction Hub Status Propagation src/resources/utils.py	create_analysis and stop_analysis now use init_hub_client_and_update_hub_status for Hub status synchronization.
Status Reconciliation with Zombie Cleanup and STOPPED Handling src/status/status.py	Adds clean_up_the_rest call during the reconciliation iteration. Explicitly handles STOPPED status from health checks instead of defaulting to FAILED. Broadens timeout exception handling to catch both TimeoutException and ConnectTimeout. Refactors status decision logging and includes STOPPED in terminal-status preferences when forwarding to the Hub.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• PrivateAIM/node-pod-orchestration#42: Introduces AnalysisStoppedLog and updates stop endpoint log streaming behavior via Hub client, overlapping with stop/stream_logs changes in this PR.
• PrivateAIM/node-pod-orchestration#43: Modifies src/api/api.py Hub client initialization and node-id resolution for Hub-dependent endpoints with similar client-based wiring changes.

Poem

A rabbit hops through Hub connections,
Retrying with patience through sections,
Pod IDs now deferred with care,
Zombies cleaned with Hub aware,
Status flowing everywhere. 🐰

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Title check	⚠️ Warning	The title 'Canary' is vague and does not convey any meaningful information about the changeset's primary modifications.	Replace with a specific, descriptive title that summarizes the main change, such as 'Add lazy Hub client initialization with retry logic and improve resource cleanup' or 'Refactor Hub client setup and enhance analysis lifecycle management'.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Docstring Coverage	✅ Passed	Docstring coverage is 91.67% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch canary

Tip

💬 Introducing Slack Agent: The best way for teams to turn conversations into code.

Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.

• Generate code and open pull requests
• Plan features and break down work
• Investigate incidents and troubleshoot customer tickets together
• Automate recurring tasks and respond to alerts with triggers
• Summarize progress and report instantly

Built for teams:

• Shared memory across your entire org—no repeating context
• Per-thread sandboxes to safely plan and execute work
• Governance built-in—scoped access, auditability, and budget controls

One agent for your entire SDLC. Right inside Slack.

👉 Get started

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 8

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (3)

src/resources/analysis/entity.py (1)

59-78: ⚠️ Potential issue | 🔴 Critical | ⚡ Quick win

Race window: pod_ids=None is persisted before deployment creation.

Inserting the row with pod_ids=None and updating it only after create_analysis_deployment returns leaves a window during which other code paths (e.g., read_db_analysis on line 117 of this file, which unconditionally does json.loads(analysis.pod_ids)) will raise TypeError: the JSON object must be str, bytes or bytearray, not NoneType.

This affects:

• The status reconciliation loop (status_loop in src/status/status.py) which reads the latest deployment on every tick.
• retrieve_history / retrieve_logs / stop_analysis if invoked while the deployment is being created.
• unstuck_analysis_deployments, which calls create_analysis (potentially before the prior row's pod_ids is populated).

Mitigations:

• Persist an empty JSON list (json.dumps([])) on the initial insert instead of None.
• And/or make read_db_analysis tolerant of None (json.loads(analysis.pod_ids) if analysis.pod_ids else []).

Proposed fix

         database.create_analysis(analysis_id=self.analysis_id,
                                  deployment_name=self.deployment_name,
                                  project_id=self.project_id,
-                                 pod_ids=None,
+                                 pod_ids=json.dumps([]),
                                  status=self.status,

And/or:

-                    pod_ids=json.loads(analysis.pod_ids),
+                    pod_ids=json.loads(analysis.pod_ids) if analysis.pod_ids else [],

🤖 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/resources/analysis/entity.py` around lines 59 - 78, The initial insert
persists pod_ids=None creating a race when read_db_analysis unconditionally does
json.loads(analysis.pod_ids); update the create path so
database.create_analysis(…) is called with pod_ids=json.dumps([]) instead of
None (the code around create_analysis and create_analysis_deployment and
subsequent database.update_deployment) and/or make read_db_analysis defensive by
parsing pod_ids with json.loads(analysis.pod_ids) if analysis.pod_ids else [] so
status_loop, retrieve_history, retrieve_logs, stop_analysis and
unstuck_analysis_deployments cannot encounter TypeError during deployment
creation.

src/utils/other.py (1)

28-41: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Brittle parsing may silently misclassify resources as zombies.

The new fixed-segment split assumes the resource name always follows analysis-{uuid}-{counter}[...]. If the input doesn't match (e.g., a resource is renamed, or analysis- is missing), this still returns something (an arbitrary substring), which is then compared against known_analysis_ids in clean_up_the_rest. A mismatch makes the resource look orphaned and triggers delete_k8s_resource — i.e., a parsing bug becomes silent data loss.

Two defensive improvements would help:

1. Validate the parsed result is a UUID (using the existing is_uuid) and raise/return None otherwise so the caller can skip.
2. Document the assumption in the docstring that split('-', 5)[:-1] requires exactly UUID + suffix shape.

Proposed fix

 def resource_name_to_analysis(deployment_name: str) -> str:
     ...
-    return '-'.join(deployment_name.split("analysis-")[-1].split('-', 5)[:-1])
+    candidate = '-'.join(deployment_name.split("analysis-")[-1].split('-', 5)[:-1])
+    return candidate if is_uuid(candidate) else ''

And in clean_up_the_rest, the existing known_analysis_ids check naturally treats '' as not-found — but you'd want to log/skip rather than delete to be safe.

🤖 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/utils/other.py` around lines 28 - 41, The current
resource_name_to_analysis function can return arbitrary substrings for
unexpected names and cause wrongful deletions; update resource_name_to_analysis
to validate the extracted id using is_uuid and return None (or raise) when the
pattern/UUID check fails, and update its docstring to explicitly state the
expected "analysis-{uuid}-{counter}[...]" shape and that it may return None for
non-matching names; then adjust clean_up_the_rest to treat a None result as
"skip and log" (do not call delete_k8s_resource) so unknown/malformed resource
names are safely ignored with a warning rather than deleted.

src/resources/utils.py (1)

321-360: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Unknown cleanup_type still triggers clean_up_the_rest.

When cleanup_type is not in the known list, line 358 records "Unknown cleanup type: ..." but line 360 then unconditionally runs clean_up_the_rest afterward. Combined with the critical Hub-validation issue above, an HTTP caller passing a typo'd cleanup type can inadvertently trigger destructive zombie cleanup. Consider scoping clean_up_the_rest to selectors like 'all' and 'zombies' explicitly, and rejecting unknown types with a 4xx upstream.

🤖 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/resources/utils.py` around lines 321 - 360, The code currently always
calls clean_up_the_rest(...) after processing cleanup_types, so a typo in
cleanup_type can inadvertently trigger zombie cleanup; change the flow to (1)
validate cleanup_types up front and if any unknown type is present return a 4xx
error (reject request) instead of continuing, and (2) only invoke
clean_up_the_rest and set response_content['zombies'] when one of the validated
types is 'zombies' or 'all' (e.g. check if 'zombies' in cleanup_types or 'all'
in cleanup_types before calling clean_up_the_rest). Update the logic that builds
response_content and the function that handles unknown types (the branch that
currently sets "Unknown cleanup type: ...") to perform the 4xx rejection instead
of falling through.

🧹 Nitpick comments (7)

src/main.py (1)

12-17: ⚖️ Poor tradeoff

Consider deferring side effects from module import time.

Loading cluster configuration and environment variables at module import time creates side effects that can make testing harder and produce less clear error messages if initialization fails. If these operations fail, the entire module will fail to import rather than failing with a clear error in main().

While the PR objectives indicate this reordering is intentional to "initialize logger and configs correctly," consider whether a lazy initialization pattern (e.g., initializing on first use or in main() with better error handling) would provide better testability and clearer error messages.

🤖 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/main.py` around lines 12 - 17, The module-level calls to get_logger(),
load_cluster_config(), and load_dotenv(find_dotenv()) cause import-time side
effects; move these initializations into a dedicated startup path (e.g., inside
main()) or provide lazy-initializers so imports are side-effect free: replace
the top-level invocations with functions like init_logger() / init_config() and
call them from main() (or on first use), and add try/except around
load_cluster_config() and load_dotenv(...) in that startup path to surface clear
errors instead of failing during import. Ensure references to get_logger,
load_cluster_config, load_dotenv, find_dotenv, and main are updated accordingly.

src/resources/utils.py (1)

385-390: ⚡ Quick win

Minor: shadowed builtin id and useless f-string.

• Line 385: for id in known_analysis_ids shadows the id builtin (Ruff A001). Rename to analysis_id.
• Line 390: f"No Hub client found, ..." has no placeholders (Ruff F541). Drop the f prefix.

Proposed fix

-        for id in known_analysis_ids:
-            if id not in validated_analysis_ids:
-                database.delete_analysis(id)
+        for analysis_id in known_analysis_ids:
+            if analysis_id not in validated_analysis_ids:
+                database.delete_analysis(analysis_id)
         known_analysis_ids = validated_analysis_ids
     else:
-        logger.warning(f"No Hub client found, skipping hub validation for zombie deletion.")
+        logger.warning("No Hub client found, skipping hub validation for zombie deletion.")

🤖 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/resources/utils.py` around lines 385 - 390, The loop variable shadows the
builtin id and the warning uses an unnecessary f-string: change the loop in the
block that iterates known_analysis_ids to use a non-shadowing name (e.g.,
analysis_id) and update usages inside the loop (e.g.,
database.delete_analysis(analysis_id)); also remove the f prefix from the
logger.warning call so it becomes logger.warning("No Hub client found, skipping
hub validation for zombie deletion.").

src/utils/hub_client.py (2)

102-129: ⚡ Quick win

Overloaded return type can mislead callers.

get_node_analysis_id now returns either str (when node_id_object_id is provided) or list[str] (when omitted) based on input. This polymorphic return contract is fragile: callers like _validate_analyses_with_hub (in src/resources/utils.py) and the in-file init_hub_client_and_update_hub_status rely on knowing the exact branch. Consider splitting into two functions (e.g., get_node_analysis_id and get_node_analysis_ids) to make the contract explicit and type-safe.

🤖 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/utils/hub_client.py` around lines 102 - 129, get_node_analysis_id
currently returns either a single str or list[str] depending on
node_id_object_id which makes callers fragile; split into two explicit
functions: keep get_node_analysis_id(hub_client, analysis_id, node_id_object_id)
to always return Optional[str] for the single-node lookup and add
get_node_analysis_ids(hub_client, analysis_id) to always return
Optional[list[str]] for the multi-node lookup; update callers such as
_validate_analyses_with_hub and init_hub_client_and_update_hub_status to call
the appropriate new function, preserve the same error handling/logging
(logger.error and the caught exceptions) and reuse the same
hub_client.find_analysis_nodes query logic but return consistent types for each
function.

---

158-180: 💤 Low value

Dead initialization of analysis_node_statuses.

analysis_node_statuses = {} on line 174 is only reached when node_analyzes is truthy (the populated branch). The empty/None case returns early. The dict initialization can be moved inside the if node_analyzes: branch (or the if/else simplified) for clarity.

Proposed simplification

-    analysis_node_statuses = {}
-    if node_analyzes:
-        for node in node_analyzes:
-            analysis_node_statuses[str(node.id)] = node.execution_status
-        return analysis_node_statuses
-    else:
-        return None
+    if not node_analyzes:
+        return None
+    return {str(node.id): node.execution_status for node in node_analyzes}

🤖 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/utils/hub_client.py` around lines 158 - 180, In
get_analysis_node_statuses, avoid the dead initialization of
analysis_node_statuses at top-level: remove analysis_node_statuses = {} and
instead initialize it inside the if node_analyzes: branch (e.g.,
analysis_node_statuses = {} immediately before the for loop), then populate and
return it; keep the existing except and the else return None behavior, using the
node_analyzes variable to drive the conditional.

src/status/status.py (1)

283-283: 💤 Low value

Minor: double whitespace before as e:.

except (TimeoutException, ConnectTimeout) as e: has an extra space. Drop it for consistency with line 194.

🤖 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` at line 283, Remove the extra space in the except
clause so it reads with a single space before the "as e" binder; change the
except line using the exception tuple (TimeoutException, ConnectTimeout) so the
"as e" is immediately preceded by a single space (matching the style used
elsewhere, e.g. line 194) — locate the except block that catches
TimeoutException and ConnectTimeout and fix the spacing.

src/api/api.py (2)

338-338: 💤 Low value

Minor: drop extraneous f prefix.

Lines 338 and 366 use f-strings without placeholders (Ruff F541).

Proposed fix

-                logger.warning(f"Couldn't forward logs for stopped analyses.")
+                logger.warning("Couldn't forward logs for stopped analyses.")

-                logger.warning(f"Couldn't forward logs for stopped analysis.")
+                logger.warning("Couldn't forward logs for stopped analysis.")

Also applies to: 366-366

🤖 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/api/api.py` at line 338, The logger.warning calls use unnecessary
f-strings with no placeholders (causing Ruff F541); update the two calls to
logger.warning that currently pass f"Couldn't forward logs for stopped
analyses." and the similar call at the other occurrence (around the same block)
to use a plain string literal (remove the leading f) so the message is a normal
string; locate the calls by searching for logger.warning("Couldn't forward logs
for stopped analyses.") or the exact f-string in the functions/methods that
forward logs and replace the f-prefixed strings with regular strings.

---

444-453: ⚡ Quick win

Anti-pattern: bare raise Exception / except Exception for control flow.

raise Exception followed by except Exception: (Ruff BLE001) obscures intent and swallows real errors from stream_logs (now indistinguishable from "node_id missing"). Restructure with an explicit branch and let downstream exceptions propagate to a single handler.

Proposed fix

-        try:
-            self._set_node_id_and_hub_client(max_attempts=5)
-            if self.node_id is not None:
-                return stream_logs(body, self.node_id, self.enable_hub_logging, self.database, self.hub_client)
-            else:
-                raise Exception
-        except Exception:
-            logger.error(f"Error streaming logs: node_id={self.node_id}, "
-                         f"hub_client={self.hub_client}.")
-            raise HTTPException(status_code=500, detail=f"Error streaming logs (see po logs).")
+        self._set_node_id_and_hub_client(max_attempts=5)
+        if self.node_id is None:
+            logger.error(f"Error streaming logs: node_id is unset, hub_client={self.hub_client}.")
+            raise HTTPException(status_code=500, detail="Error streaming logs (see po logs).")
+        try:
+            return stream_logs(body, self.node_id, self.enable_hub_logging, self.database, self.hub_client)
+        except Exception as e:
+            logger.error(f"Error streaming logs: {repr(e)}")
+            raise HTTPException(status_code=500, detail="Error streaming logs (see po logs).")

🤖 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/api/api.py` around lines 444 - 453, Replace the bare raise/except
control-flow with an explicit branch: call
_set_node_id_and_hub_client(max_attempts=5), then if self.node_id is None log
the missing-node error via logger.error (including hub_client and node_id) and
raise HTTPException with a clear 500 detail; otherwise return stream_logs(body,
self.node_id, self.enable_hub_logging, self.database, self.hub_client) without
catching its exceptions so downstream errors from stream_logs propagate to the
existing global handler—remove the try/except and the bare raise Exception
around stream_logs.

🤖 Prompt for all review comments with 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.

Inline comments:
In `@src/api/api.py`:
- Around line 422-426: The code calls self._set_node_id_and_hub_client(...) and
then forwards self.hub_client into cleanup(...) even when it remains None, which
allows destructive cleanup paths to run without Hub validation; after calling
_set_node_id_and_hub_client(max_attempts=5) in the method that returns
cleanup(cleanup_type,...), add a guard: if self.hub_client is None and
cleanup_type in ('all','zombies') then fail fast with a 503-style error (raise
the appropriate HTTP exception or return a 503 response) instead of invoking
cleanup; reference _set_node_id_and_hub_client, cleanup, cleanup_type,
self.hub_client and the destructive path in clean_up_the_rest when implementing
the check.
- Around line 470-491: The retry loop in _set_node_id_and_hub_client should not
short-circuit on node_id alone because that prevents rebuilding hub_client on
subsequent calls; change the loop condition to ensure both hub_client and
node_id are obtained (e.g., while self.hub_client is None or self.node_id is
None) and make callers (stop_*_call, cleanup_call, stream_logs_call) set both
self.hub_client and self.node_id to None when they detect a stale/unauthorized
client so the next call triggers a refresh; additionally, make endpoint-invoked
refreshes non-blocking by performing a single quick attempt (no long
sleep/retry) and reserve the long retry/backoff loop for __init__ or a
background task so request handlers aren’t blocked.

In `@src/resources/analysis/entity.py`:
- Around line 99-101: The stop() flow currently calls
create_analysis_tokens(...) then delete_subscription(...), which can raise
IndexError/HTTP/network exceptions and abort stop() after partial cleanup and
also remints a token unnecessarily; change stop() to reuse self.tokens if
present (avoid calling create_analysis_tokens when self.tokens exists/valid) and
wrap delete_subscription(self.analysis_id, token) in a try/except that catches
IndexError, requests/HTTP-related exceptions and a broad Exception fallback,
logging the failure via the entity's logger (e.g., self.logger.error) instead of
re-raising so stop() completes to a clean terminal state.

In `@src/resources/utils.py`:
- Around line 454-459: _validate_analyses_with_hub currently calls
get_node_analysis_id(hub_client, analysis_id) without scoping to a node so any
analysis present on any node is considered validated; update
_validate_analyses_with_hub to accept a node_id parameter (e.g., node_id: str)
and call get_node_analysis_id(hub_client, analysis_id, node_id) so validation is
limited to the local node, and update all callers accordingly; if the original
behavior was intended, instead add a clear comment inside
_validate_analyses_with_hub explaining that global validation across nodes is
deliberate.
- Around line 383-406: _validate_analyses_with_hub currently treats hub-call
failures as "not found" because get_node_analysis_id swallows exceptions, which
leads clean_up_the_rest to delete everything on transient Hub errors; change
get_node_analysis_id to surface failures (raise or return a distinct sentinel
like HubCallFailed) instead of returning None for HTTP/Hub/Attribute errors,
update _validate_analyses_with_hub to distinguish "analysis not found" from "hub
call failed" and return both validated IDs plus a failure flag/list, and modify
clean_up_the_rest to abort or skip any deletion when hub failures are detected
for all/most IDs (or require N consecutive negative validations before deleting)
and log an explicit warning; specifically adjust get_node_analysis_id,
_validate_analyses_with_hub, and clean_up_the_rest to implement these checks so
transient Hub outages cannot trigger mass deletions.

In `@src/status/status.py`:
- Around line 154-157: The current code calls clean_up_the_rest(database,
hub_client, get_current_namespace()) on every status-loop tick; change this so
zombie cleanup is throttled and only runs when the immediate Hub call for this
iteration succeeded: add a lightweight counter/backoff (e.g., run once every N
iterations or exponentially back off after failures) and a boolean check of the
last Hub poll success before invoking clean_up_the_rest; reference and modify
the status loop logic that calls clean_up_the_rest, the hub_client usage, and
get_current_namespace() so that cleanup is skipped when the Hub poll failed and
is limited to the configured cadence/backoff.
- Around line 99-102: get_node_analysis_id currently collapses "not found" and
transport/Hub errors into None, causing the status loop to call
delete_analysis(analysis_id, database) incorrectly; modify get_node_analysis_id
to surface transport/Hub failures (re-raise
HTTPStatusError/HubAPIError/AttributeError or return a distinct sentinel like
"ERROR" vs "NOT_FOUND"), then update the status.py loop that calls
get_node_analysis_id so it only invokes delete_analysis when the result
explicitly indicates "not found" (or after N consecutive explicit NOT_FOUND
results) and treats transport errors by logging and skipping/retrying instead of
deleting; reference functions get_node_analysis_id and delete_analysis and the
status loop that currently logs and calls delete_analysis.

In `@src/utils/mb_client.py`:
- Around line 5-15: The delete_subscription helper currently risks IndexError
from find_k8s_resources(...)[0], leaks the httpx Client, ignores HTTP errors,
and lacks an explicit timeout; fix by first validating the result of
find_k8s_resources('service', 'label', 'component=flame-message-broker',
namespace=...) and raising or logging a clear error if empty, then open the
httpx Client in a context manager (with Client(...) as mb_client) using an
explicit short timeout, call
mb_client.delete(f"/analyses/{analysis_id}/messages/subscriptions") inside a
try/except, call response.raise_for_status() to surface 4xx/5xx, and log or
rethrow a structured error that includes analysis_id, namespace and the caught
exception so callers (e.g., Analysis.stop) can handle failures gracefully.

---

Outside diff comments:
In `@src/resources/analysis/entity.py`:
- Around line 59-78: The initial insert persists pod_ids=None creating a race
when read_db_analysis unconditionally does json.loads(analysis.pod_ids); update
the create path so database.create_analysis(…) is called with
pod_ids=json.dumps([]) instead of None (the code around create_analysis and
create_analysis_deployment and subsequent database.update_deployment) and/or
make read_db_analysis defensive by parsing pod_ids with
json.loads(analysis.pod_ids) if analysis.pod_ids else [] so status_loop,
retrieve_history, retrieve_logs, stop_analysis and unstuck_analysis_deployments
cannot encounter TypeError during deployment creation.

In `@src/resources/utils.py`:
- Around line 321-360: The code currently always calls clean_up_the_rest(...)
after processing cleanup_types, so a typo in cleanup_type can inadvertently
trigger zombie cleanup; change the flow to (1) validate cleanup_types up front
and if any unknown type is present return a 4xx error (reject request) instead
of continuing, and (2) only invoke clean_up_the_rest and set
response_content['zombies'] when one of the validated types is 'zombies' or
'all' (e.g. check if 'zombies' in cleanup_types or 'all' in cleanup_types before
calling clean_up_the_rest). Update the logic that builds response_content and
the function that handles unknown types (the branch that currently sets "Unknown
cleanup type: ...") to perform the 4xx rejection instead of falling through.

In `@src/utils/other.py`:
- Around line 28-41: The current resource_name_to_analysis function can return
arbitrary substrings for unexpected names and cause wrongful deletions; update
resource_name_to_analysis to validate the extracted id using is_uuid and return
None (or raise) when the pattern/UUID check fails, and update its docstring to
explicitly state the expected "analysis-{uuid}-{counter}[...]" shape and that it
may return None for non-matching names; then adjust clean_up_the_rest to treat a
None result as "skip and log" (do not call delete_k8s_resource) so
unknown/malformed resource names are safely ignored with a warning rather than
deleted.

---

Nitpick comments:
In `@src/api/api.py`:
- Line 338: The logger.warning calls use unnecessary f-strings with no
placeholders (causing Ruff F541); update the two calls to logger.warning that
currently pass f"Couldn't forward logs for stopped analyses." and the similar
call at the other occurrence (around the same block) to use a plain string
literal (remove the leading f) so the message is a normal string; locate the
calls by searching for logger.warning("Couldn't forward logs for stopped
analyses.") or the exact f-string in the functions/methods that forward logs and
replace the f-prefixed strings with regular strings.
- Around line 444-453: Replace the bare raise/except control-flow with an
explicit branch: call _set_node_id_and_hub_client(max_attempts=5), then if
self.node_id is None log the missing-node error via logger.error (including
hub_client and node_id) and raise HTTPException with a clear 500 detail;
otherwise return stream_logs(body, self.node_id, self.enable_hub_logging,
self.database, self.hub_client) without catching its exceptions so downstream
errors from stream_logs propagate to the existing global handler—remove the
try/except and the bare raise Exception around stream_logs.

In `@src/main.py`:
- Around line 12-17: The module-level calls to get_logger(),
load_cluster_config(), and load_dotenv(find_dotenv()) cause import-time side
effects; move these initializations into a dedicated startup path (e.g., inside
main()) or provide lazy-initializers so imports are side-effect free: replace
the top-level invocations with functions like init_logger() / init_config() and
call them from main() (or on first use), and add try/except around
load_cluster_config() and load_dotenv(...) in that startup path to surface clear
errors instead of failing during import. Ensure references to get_logger,
load_cluster_config, load_dotenv, find_dotenv, and main are updated accordingly.

In `@src/resources/utils.py`:
- Around line 385-390: The loop variable shadows the builtin id and the warning
uses an unnecessary f-string: change the loop in the block that iterates
known_analysis_ids to use a non-shadowing name (e.g., analysis_id) and update
usages inside the loop (e.g., database.delete_analysis(analysis_id)); also
remove the f prefix from the logger.warning call so it becomes
logger.warning("No Hub client found, skipping hub validation for zombie
deletion.").

In `@src/status/status.py`:
- Line 283: Remove the extra space in the except clause so it reads with a
single space before the "as e" binder; change the except line using the
exception tuple (TimeoutException, ConnectTimeout) so the "as e" is immediately
preceded by a single space (matching the style used elsewhere, e.g. line 194) —
locate the except block that catches TimeoutException and ConnectTimeout and fix
the spacing.

In `@src/utils/hub_client.py`:
- Around line 102-129: get_node_analysis_id currently returns either a single
str or list[str] depending on node_id_object_id which makes callers fragile;
split into two explicit functions: keep get_node_analysis_id(hub_client,
analysis_id, node_id_object_id) to always return Optional[str] for the
single-node lookup and add get_node_analysis_ids(hub_client, analysis_id) to
always return Optional[list[str]] for the multi-node lookup; update callers such
as _validate_analyses_with_hub and init_hub_client_and_update_hub_status to call
the appropriate new function, preserve the same error handling/logging
(logger.error and the caught exceptions) and reuse the same
hub_client.find_analysis_nodes query logic but return consistent types for each
function.
- Around line 158-180: In get_analysis_node_statuses, avoid the dead
initialization of analysis_node_statuses at top-level: remove
analysis_node_statuses = {} and instead initialize it inside the if
node_analyzes: branch (e.g., analysis_node_statuses = {} immediately before the
for loop), then populate and return it; keep the existing except and the else
return None behavior, using the node_analyzes variable to drive the conditional.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: c6cdb1bf-fdfe-46f0-a88b-aa8b07bfed44

📥 Commits

Reviewing files that changed from the base of the PR and between 5c89ae3 and 4c6af92.

⛔ Files ignored due to path filters (1)

• poetry.lock is excluded by !**/*.lock

📒 Files selected for processing (9)

• src/api/api.py
• src/main.py
• src/resources/analysis/entity.py
• src/resources/log/entity.py
• src/resources/utils.py
• src/status/status.py
• src/utils/hub_client.py
• src/utils/mb_client.py
• src/utils/other.py

[coderabbitai]

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

cleanup_call forwards self.hub_client even when it is None.

When _set_node_id_and_hub_client(max_attempts=5) fails to resolve a client, self.hub_client stays None and is passed to cleanup(...). Combined with the destructive behavior in clean_up_the_rest (see comment in src/resources/utils.py), this means a DELETE /po/cleanup/... call during a Hub outage will run zombie cleanup with hub validation disabled (only a warning is logged). Consider failing fast with a 503 here when cleanup_type is 'all' or 'zombies' and hub_client is None, so an operator-triggered cleanup doesn't quietly skip the safety net.

🤖 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/api/api.py` around lines 422 - 426, The code calls
self._set_node_id_and_hub_client(...) and then forwards self.hub_client into
cleanup(...) even when it remains None, which allows destructive cleanup paths
to run without Hub validation; after calling
_set_node_id_and_hub_client(max_attempts=5) in the method that returns
cleanup(cleanup_type,...), add a guard: if self.hub_client is None and
cleanup_type in ('all','zombies') then fail fast with a 503-style error (raise
the appropriate HTTP exception or return a 503 response) instead of invoking
cleanup; reference _set_node_id_and_hub_client, cleanup, cleanup_type,
self.hub_client and the destructive path in clean_up_the_rest when implementing
the check.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Semantic mismatch: validating without filtering by node.

_validate_analyses_with_hub calls get_node_analysis_id(hub_client, analysis_id) without passing node_id, which (per the new overloaded contract) returns a list of all analysis-node ids across all nodes for that analysis. Any analysis that exists at any other node will be marked "validated" here, even if this node is no longer participating. If that's intentional, please add a comment; otherwise pass node_id so the validation is scoped to the local node.

🤖 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/resources/utils.py` around lines 454 - 459, _validate_analyses_with_hub
currently calls get_node_analysis_id(hub_client, analysis_id) without scoping to
a node so any analysis present on any node is considered validated; update
_validate_analyses_with_hub to accept a node_id parameter (e.g., node_id: str)
and call get_node_analysis_id(hub_client, analysis_id, node_id) so validation is
limited to the local node, and update all callers accordingly; if the original
behavior was intended, instead add a clear comment inside
_validate_analyses_with_hub explaining that global validation across nodes is
deliberate.

[coderabbitai]

⚠️ Potential issue | 🔴 Critical | 🏗️ Heavy lift

Running clean_up_the_rest every status-loop tick amplifies the Hub-validation risk.

clean_up_the_rest is now invoked on every iteration (every status_loop_interval seconds) with the live hub_client. Combined with the destructive validation logic flagged in src/resources/utils.py (Hub error → empty validated set → mass delete), this turns any Hub blip into a guaranteed wipe of local analyses and resources.

At minimum:

• Throttle zombie cleanup to a lower cadence (e.g., once per N iterations or behind a backoff after a successful Hub poll).
• Only run it when the immediately preceding Hub call in this iteration succeeded.

🤖 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 154 - 157, The current code calls
clean_up_the_rest(database, hub_client, get_current_namespace()) on every
status-loop tick; change this so zombie cleanup is throttled and only runs when
the immediate Hub call for this iteration succeeded: add a lightweight
counter/backoff (e.g., run once every N iterations or exponentially back off
after failures) and a boolean check of the last Hub poll success before invoking
clean_up_the_rest; reference and modify the status loop logic that calls
clean_up_the_rest, the hub_client usage, and get_current_namespace() so that
cleanup is skipped when the Hub poll failed and is limited to the configured
cadence/backoff.

[coderabbitai]

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Missing error handling, resource leak, and unchecked response.

Several issues in this new helper:

1. find_k8s_resources(...)[0] will raise IndexError if the message broker service is not found (e.g., during cleanup races or in a misconfigured namespace). Callers in Analysis.stop will propagate the failure and abort the stop flow.
2. httpx.Client is never closed. Use a context manager so the connection pool is released.
3. The DELETE response is not checked. A 4xx/5xx from the broker (e.g., expired token, broker not ready) will silently leave orphaned subscriptions with no log trail. Add raise_for_status() (or at least a structured error log) and wrap the call in try/except similar to other HTTP callers in this codebase.
4. No request timeout. The default httpx timeout is 5s, but consider making it explicit and shorter for cleanup paths to avoid stalling Analysis.stop.

Proposed fix

 from httpx import Client
+from httpx import HTTPStatusError, RequestError
 from src.k8s.utils import find_k8s_resources
+from src.utils.po_logging import get_logger
+
+logger = get_logger()


 def delete_subscription(analysis_id: str, keycloak_token: str, namespace: str = 'default') -> None:
-    # get the service name of the message broker
-    message_broker_service_name = find_k8s_resources('service',
-                                                     'label',
-                                                     'component=flame-message-broker',
-                                                     namespace=namespace)[0]
-    mb_client = Client(base_url=f"http://{message_broker_service_name}",
-                       headers={"Authorization": f"Bearer {keycloak_token}",
-                                "accept": "application/json"},
-                       follow_redirects=True)
-    mb_client.delete(f"/analyses/{analysis_id}/messages/subscriptions")
+    # get the service name of the message broker
+    services = find_k8s_resources('service', 'label',
+                                  'component=flame-message-broker',
+                                  namespace=namespace)
+    if not services:
+        logger.warning(f"Message broker service not found in namespace {namespace}; "
+                       f"skipping subscription cleanup for analysis {analysis_id}.")
+        return
+    message_broker_service_name = services[0]
+    try:
+        with Client(base_url=f"http://{message_broker_service_name}",
+                    headers={"Authorization": f"Bearer {keycloak_token}",
+                             "accept": "application/json"},
+                    follow_redirects=True,
+                    timeout=10.0) as mb_client:
+            response = mb_client.delete(f"/analyses/{analysis_id}/messages/subscriptions")
+            response.raise_for_status()
+    except (HTTPStatusError, RequestError) as e:
+        logger.error(f"Failed to delete message broker subscription for analysis "
+                     f"{analysis_id}: {repr(e)}")

🤖 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/utils/mb_client.py` around lines 5 - 15, The delete_subscription helper
currently risks IndexError from find_k8s_resources(...)[0], leaks the httpx
Client, ignores HTTP errors, and lacks an explicit timeout; fix by first
validating the result of find_k8s_resources('service', 'label',
'component=flame-message-broker', namespace=...) and raising or logging a clear
error if empty, then open the httpx Client in a context manager (with
Client(...) as mb_client) using an explicit short timeout, call
mb_client.delete(f"/analyses/{analysis_id}/messages/subscriptions") inside a
try/except, call response.raise_for_status() to surface 4xx/5xx, and log or
rethrow a structured error that includes analysis_id, namespace and the caught
exception so callers (e.g., Analysis.stop) can handle failures gracefully.