server: add healthcheck for worker

README.md

@@ -133,17 +133,25 @@ TRANSLATE_URL=https://monadical-sas--reflector-translator-web.modal.run
 ZEPHYR_LLM_URL=https://monadical-sas--reflector-llm-zephyr-web.modal.run
 ```
 
-### Start the project
+### Start the API/Backend
 
-Use:
+Start the API server:
 
 ```bash
 poetry run python3 -m reflector.app
 ```
 
-And start the background worker
+Start the background worker:
 
+```bash
 celery -A reflector.worker.app worker --loglevel=info
+```
+
+For crontab (only healthcheck for now), start the celery beat:
+
+```bash
+celery -A reflector.worker.app beat
+```
 
 #### Using docker
 

server/reflector/settings.py

@@ -128,5 +128,8 @@ class Settings(BaseSettings):
     # Profiling
     PROFILING: bool = False
 
+    # Healthcheck
+    HEALTHCHECK_URL: str | None = None
+
 
 settings = Settings()

server/reflector/worker/app.py

@@ -1,6 +1,8 @@
+import structlog
 from celery import Celery
 from reflector.settings import settings
 
+logger = structlog.get_logger(__name__)
 app = Celery(__name__)
 app.conf.broker_url = settings.CELERY_BROKER_URL
 app.conf.result_backend = settings.CELERY_RESULT_BACKEND
@@ -8,5 +10,18 @@ app.conf.broker_connection_retry_on_startup = True
 app.autodiscover_tasks(
     [
         "reflector.pipelines.main_live_pipeline",
+        "reflector.worker.healthcheck",
     ]
 )
+
+# crontab
+app.conf.beat_schedule = {}
+
+if settings.HEALTHCHECK_URL:
+    app.conf.beat_schedule["healthcheck_ping"] = {
+        "task": "reflector.worker.healthcheck.healthcheck_ping",
+        "schedule": 60.0 * 10,
+    }
+    logger.info("Healthcheck enabled", url=settings.HEALTHCHECK_URL)
+else:
+    logger.warning("Healthcheck disabled, no url configured")

server/reflector/worker/healthcheck.py

@@ -0,0 +1,18 @@
+import httpx
+import structlog
+from celery import shared_task
+from reflector.settings import settings
+
+logger = structlog.get_logger(__name__)
+
+
+@shared_task
+def healthcheck_ping():
+    url = settings.HEALTHCHECK_URL
+    if not url:
+        return
+    try:
+        print("pinging healthcheck url", url)
+        httpx.get(url, timeout=10)
+    except Exception as e:
+        logger.error("healthcheck_ping", error=str(e))

server/runserver.sh

@@ -9,6 +9,8 @@ if [ "${ENTRYPOINT}" = "server" ]; then
     python -m reflector.app
 elif [ "${ENTRYPOINT}" = "worker" ]; then
     celery -A reflector.worker.app worker --loglevel=info
+elif [ "${ENTRYPOINT}" = "beat" ]; then
+    celery -A reflector.worker.app beat --loglevel=info
 else
     echo "Unknown command"
 fi