Kubernetes and actors routes

src/lvmapi/app.py

@@ -14,9 +14,11 @@
 from lvmapi import auth
 from lvmapi.broker import broker, broker_shutdown, broker_startup
 from lvmapi.routers import (
+    actors,
     alerts,
     enclosure,
     ephemeris,
+    kubernetes,
     macros,
     overwatcher,
     slack,
@@ -25,6 +27,7 @@
     telescopes,
     weather,
 )
+from lvmapi.tools.kubernetes import Kubernetes
 
 
 app = FastAPI()
@@ -40,6 +43,8 @@
 app.include_router(enclosure.router)
 app.include_router(alerts.router)
 app.include_router(tasks.router)
+app.include_router(kubernetes.router)
+app.include_router(actors.router)
 
 
 # Lifecycle events for the broker.
@@ -49,6 +54,9 @@
 # Integration with FastAPI.
 taskiq_fastapi.init(broker, "lvmapi.app:app")
 
+# Add kubernetes API instance to state.
+app.state.kubernetes = Kubernetes()
+
 
 @app.get("/")
 def root():

src/lvmapi/config.yaml

@@ -11,3 +11,56 @@ influxdb:
 slack:
   token: null
   channel: lvm-alerts
+
+actors:
+  list:
+    - lvm.sci.agcam
+    - lvm.spec.agcam
+    - lvm.skye.agcam
+    - lvm.skyw.agcam
+    - lvmscp.sp1
+    - lvmscp.sp2
+    - lvmscp.sp3
+    - lvmecp
+    - lvm.sci.pwi
+    - lvm.spec.pwi
+    - lvm.skye.pwi
+    - lvm.skyw.pwi
+    - lvm.sci.foc
+    - lvm.spec.foc
+    - lvm.skye.foc
+    - lvm.skyw.foc
+    - lvm.sci.km
+    - lvm.skye.km
+    - lvm.skyw.km
+    - lvm.sci.guider
+    - lvm.spec.guider
+    - lvm.skye.guider
+    - lvm.skyw.guider
+  actor_to_deployment:
+    lvm.sci.agcam: lvmagcam
+    lvm.spec.agcam: lvmagcam
+    lvm.skye.agcam: lvmagcam
+    lvm.skyw.agcam: lvmagcam
+    lvmscp.sp1: lvmscp
+    lvmscp.sp2: lvmscp
+    lvmscp.sp3: lvmscp
+    lvmecp: lvmecp
+    lvm.sci.pwi: lvmpwi-sci
+    lvm.spec.pwi: lvmpwi-spec
+    lvm.skye.pwi: lvmpwi-skye
+    lvm.skyw.pwi: lvmpwi-skyw
+    lvm.sci.foc: lvmtan
+    lvm.spec.foc: lvmtan
+    lvm.skye.foc: lvmtan
+    lvm.skyw.foc: lvmtan
+    lvm.sci.km: lvmtan
+    lvm.skye.km: lvmtan
+    lvm.skyw.km: lvmtan
+    lvm.sci.guider: lvmguider
+    lvm.spec.guider: lvmguider
+    lvm.skye.guider: lvmguider
+    lvm.skyw.guider: lvmguider
+
+kubernetes:
+  deployments_path: /home/sdss5/config/kube

src/lvmapi/routers/actors.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2024-07-25
+# @Filename: actors.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+from pydantic import BaseModel
+
+from lvmapi import config
+from lvmapi.tasks import restart_kubernetes_deployment_task
+from lvmapi.tools.rabbitmq import ping_actors
+
+
+class HealthResponse(BaseModel):
+    """Health response for an actor."""
+
+    actor: str
+    deployment_name: str
+    is_deployed: bool
+    ping: bool
+
+
+router = APIRouter(prefix="/actors", tags=["actors"])
+
+
+@router.get("/")
+async def get_actors_route() -> list[str]:
+    """Returns a list of actors."""
+
+    return config["actors"]["list"]
+
+
+@router.get("/health")
+async def get_actor_health(request: Request) -> list[HealthResponse]:
+    """Returns the health of all actors."""
+
+    actors = config["actors"]["list"]
+    health: list[HealthResponse] = []
+
+    deployments = request.app.state.kubernetes.list_deployments()
+    ping_response = await ping_actors(actors=actors)
+
+    for actor in actors:
+        deployment_name = config["actors.actor_to_deployment"][actor]
+        ping = ping_response.get(actor, False)
+        is_deployed = deployment_name in deployments
+        health.append(
+            HealthResponse(
+                actor=actor,
+                deployment_name=deployment_name,
+                is_deployed=is_deployed,
+                ping=ping,
+            )
+        )
+
+    return health
+
+
+@router.get("/ping")
+async def ping_route(actors: list[str] | None = None) -> dict[str, bool]:
+    """Pings a list of actors."""
+
+    return await ping_actors(actors=actors)
+
+
+@router.get("/restart/{actor}")
+async def restart_actor_route(actor: str) -> str:
+    """Restarts an actor. Scheduled as a task and returns the task ID"""
+
+    deployment = config["actors.actor_to_deployment"][actor]
+    if deployment is None:
+        raise ValueError(f"Actor {actor} does not have a deployment.")
+
+    task = await restart_kubernetes_deployment_task.kiq(deployment)
+    return task.task_id

src/lvmapi/routers/kubernetes.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2024-07-25
+# @Filename: kubernetes.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from fastapi import APIRouter, Request
+
+from lvmapi.tasks import restart_kubernetes_deployment_task
+
+
+if TYPE_CHECKING:
+    from lvmapi.tools.kubernetes import Kubernetes
+
+
+router = APIRouter(prefix="/kubernetes", tags=["kubernetes"])
+
+
+@router.get("/deployments")
+@router.get("/deployments/list")
+async def list_deployments(request: Request) -> list[str]:
+    """Lists the deployments in all namespaces."""
+
+    kube: Kubernetes = request.app.state.kubernetes
+
+    return kube.list_deployments()
+
+
+@router.get("/deployments/{deployment}/restart")
+async def restart_deployment(deployment: str) -> str:
+    """Restarts a deployment. Scheduled as a task (returns task ID)."""
+
+    task = await restart_kubernetes_deployment_task.kiq(deployment)
+    return task.task_id

src/lvmapi/tasks.py

@@ -8,6 +8,8 @@
 
 from __future__ import annotations
 
+import asyncio
+
 from typing import Literal
 
 from lvmapi.app import broker
@@ -43,3 +45,23 @@ async def shutdown_task():
         await gort.shutdown(park_telescopes=True)
 
     return True
+
+
+@broker.task()
+async def restart_kubernetes_deployment_task(deployment: str, confirm: bool = True):
+    """Restarts a Kubernetes deployment."""
+
+    from lvmapi.app import app
+
+    app.state.kubernetes.restart_deployment(deployment)
+
+    if confirm:
+        for _ in range(15):
+            if deployment in app.state.kubernetes.list_deployments():
+                return True
+            await asyncio.sleep(1)
+
+    else:
+        return True
+
+    raise TimeoutError(f"Timed out waiting for {deployment} to start.")