feat(server): add kubernetes events to provider logs (#748)

Signed-off-by: Radek Ježek <[email protected]>

Author: jezekra1

apps/beeai-server/src/beeai_server/api/routes/provider.py

@@ -20,7 +20,7 @@
 from uuid import UUID
 
 from beeai_server.domain.models.provider import ProviderWithState
-from beeai_server.api.routes.dependencies import ProviderServiceDependency
+from beeai_server.api.routes.dependencies import ProviderServiceDependency, AdminAuthDependency
 from beeai_server.api.schema.common import PaginatedResponse
 from starlette.responses import StreamingResponse
 
@@ -31,6 +31,7 @@
 
 @router.post("")
 async def create_provider(
+    _: AdminAuthDependency,
     request: CreateProviderRequest,
     provider_service: ProviderServiceDependency,
     auto_remove: bool = Query(default=False),
@@ -42,7 +43,9 @@ async def create_provider(
 
 @router.post("/register/unmanaged", deprecated=True)
 async def deprecated_create_unmanaged_provider(
-    request: CreateProviderRequest, provider_service: ProviderServiceDependency
+    _: AdminAuthDependency,
+    request: CreateProviderRequest,
+    provider_service: ProviderServiceDependency,
 ) -> ProviderWithState:
     """Backward compatibility for ACP sdk."""
     return await provider_service.create_provider(location=request.location, auto_remove=True)
@@ -67,11 +70,19 @@ async def get_provider(id: UUID, provider_service: ProviderServiceDependency) ->
 
 
 @router.delete("/{id}", status_code=fastapi.status.HTTP_204_NO_CONTENT)
-async def delete_provider(id: UUID, provider_service: ProviderServiceDependency) -> None:
+async def delete_provider(
+    _: AdminAuthDependency,
+    id: UUID,
+    provider_service: ProviderServiceDependency,
+) -> None:
     await provider_service.delete_provider(provider_id=id)
 
 
 @router.get("/{id}/logs", status_code=fastapi.status.HTTP_204_NO_CONTENT)
-async def stream_logs(id: UUID, provider_service: ProviderServiceDependency) -> StreamingResponse:
+async def stream_logs(
+    _: AdminAuthDependency,
+    id: UUID,
+    provider_service: ProviderServiceDependency,
+) -> StreamingResponse:
     logs_iterator = await provider_service.stream_logs(provider_id=id)
     return streaming_response(logs_iterator())

apps/beeai-server/src/beeai_server/infrastructure/kubernetes/provider_deployment_manager.py

@@ -21,7 +21,7 @@
 from asyncio import TaskGroup
 from contextlib import asynccontextmanager
 from datetime import timedelta
-from typing import Callable, Awaitable
+from typing import Callable, Awaitable, AsyncIterator
 from uuid import UUID
 
 import kr8s
@@ -40,11 +40,11 @@
 
 
 class KubernetesProviderDeploymentManager(IProviderDeploymentManager):
-    def __init__(self, api_factory: Callable[[], Awaitable[kr8s.Api]]):
+    def __init__(self, api_factory: Callable[[], Awaitable[kr8s.asyncio.Api]]):
         self._api_factory = api_factory
 
     @asynccontextmanager
-    async def api(self):
+    async def api(self) -> AsyncIterator[kr8s.asyncio.Api]:
         client = await self._api_factory()
         yield client
 
@@ -240,23 +240,57 @@ async def get_provider_url(self, *, provider_id: UUID) -> HttpUrl:
     async def stream_logs(self, *, provider_id: UUID, logs_container: LogsContainer):
         try:
             async with self.api() as api:
+                missing_logged = False
                 while True:
-                    [state] = await self.state(provider_ids=[provider_id])
-                    if state == ProviderDeploymentState.running:
-                        break
-                    logs_container.add_stdout("Waiting for provider startup...")
+                    try:
+                        deploy = await Deployment.get(name=self._get_k8s_name(provider_id, kind="deploy"), api=api)
+                        if pods := await deploy.pods():
+                            break
+                    except kr8s.NotFoundError:
+                        ...
+                    if not missing_logged:
+                        logs_container.add_stdout("Provider is not running, run a query to start it up...")
+                    missing_logged = True
                     await asyncio.sleep(1)
 
+                if deploy.status.get("availableReplicas", 0) == 0:
+                    async for event_stream_type, event in api.watch(
+                        kind="event",
+                        # TODO: we select for only one pod, for multi-pod agents this might hold up the logs for a while
+                        field_selector=f"involvedObject.name=={pods[0].name},involvedObject.kind==Pod",
+                    ):
+                        message = event.raw.get("message", "")
+                        logs_container.add_stdout(f"{event.raw.reason}: {message}")
+                        if event.raw.reason == "Started":
+                            break
+
+                for attempt in range(10):
+                    try:
+                        _ = [log async for log in pods[0].logs(tail_lines=1)]
+                        break
+                    except kr8s.ServerError:
+                        await asyncio.sleep(1)
+                else:
+                    logs_container.add_stdout("Container crashed or not starting up, attempting to get previous logs:")
+                    with suppress(kr8s.ServerError):
+                        previous_logs = [log async for log in pods[0].logs(previous=True)]
+                        if previous_logs:
+                            logs_container.add_stdout("Previous container logs:")
+                            for log in previous_logs:
+                                logs_container.add_stdout(f"Previous: {log}")
+                    return
+
+                # Stream logs from pods
                 async def stream_logs(pod: Pod):
                     async for line in pod.logs(follow=True):
                         logs_container.add_stdout(
                             f"{pod.name.replace(self._get_k8s_name(provider_id, 'deploy'), '')}: {line}"
                         )
 
                 async with TaskGroup() as tg:
-                    deploy = await Deployment.get(name=self._get_k8s_name(provider_id, kind="deploy"), api=api)
                     for pod in await deploy.pods():
                         tg.create_task(stream_logs(pod))
+
         except Exception as ex:
             logs_container.add(
                 ProcessLogMessage(stream=ProcessLogType.stderr, message=extract_messages(ex), error=True)

apps/beeai-server/src/beeai_server/utils/logs_container.py

@@ -91,7 +91,7 @@ def _handle_message(log: ProcessLogMessage):
             try:
                 stream_send.send_nowait(log)
             except WouldBlock:
-                logger.error("Unable to stream logs to client due to a full buffer")
+                logger.debug("Unable to stream logs to client due to a full buffer")
 
         if include_old:
             for message in self.logs:

apps/beeai-server/tasks.toml

@@ -102,18 +102,22 @@ depends = ["beeai-server:image:build"]
 run = """
 #!/bin/bash
 
-NAMESPACE=beeai
+NAMESPACE=default
 VM_NAME='{{option(name="vm-name", default="beeai-local-dev")}}'
 KUBECONFIG="${HOME}/.beeai/lima/${VM_NAME}/copied-from-guest/kubeconfig.yaml"
+LIMA_HOME=~/.beeai/lima
+
+export KUBECONFIG
+export LIMA_HOME
 
 [[ ! -f .env ]] && cp template.env .env
 
 tele="telepresence --use .*${NAMESPACE}.*"
-($tele list -n beeai --replacements 2>/dev/null | grep -q platform) && echo "Dev environment already running" && exit
+($tele list --replacements 2>/dev/null | grep -q platform) && echo "Dev environment already running" && exit
 
+# Stopping all platform deployments
 mise run beeai-server:dev:stop
 
-# Stopping all platform deployments
 if [ '{{flag(name='no-registry')}}' = 'true' ]; then
   extra_flags="--set externalRegistries=null"
 fi
@@ -122,7 +126,6 @@ mise run beeai-cli:run -- platform start \
     --vm-name=${VM_NAME} \
     --import ghcr.io/i-am-bee/beeai-platform/beeai-server:local \
     --set image.tag=local $extra_flags
-export KUBECONFIG
 
 $tele helm install
 $tele connect --namespace "$NAMESPACE"
@@ -171,7 +174,9 @@ deactivate () {
 dir = "{{config_root}}/apps/beeai-server"
 run = """
 #!/bin/bash
-NAMESPACE=beeai
+NAMESPACE=default
+
+telepresence --use ".*${NAMESPACE}.*" uninstall --all-agents || true
 
 # Stop all lima VMs
 {% raw %}

helm/beeai-platform/templates/serviceaccount/role.yaml

@@ -28,4 +28,9 @@ rules:
   # Permissions for pod logs
   - apiGroups: [ "" ]
     resources: [ "pods/log" ]
-    verbs: [ "get" ]
+    verbs: [ "get" ]
+
+  # Permissions for events (read-only)
+  - apiGroups: [ "" ]
+    resources: [ "events" ]
+    verbs: [ "get", "list", "watch" ]