Increase model wait time for vLLM

PyWorker Error Handling

lib/backend.py

@@ -30,7 +30,7 @@ from lib.data_types import (
     BenchmarkResult
 )
 
-VERSION = "0.2.0"
+VERSION = "0.2.1"
 
 MSG_HISTORY_LEN = 100
 log = logging.getLogger(__file__)

lib/server.py

@@ -3,38 +3,58 @@ import logging
 from typing import List
 import ssl
 from asyncio import run, gather
-
+import asyncio
 
 from lib.backend import Backend
+from lib.metrics import Metrics
 from aiohttp import web
 
 log = logging.getLogger(__file__)
 
 
 def start_server(backend: Backend, routes: List[web.RouteDef], **kwargs):
-    log.debug("getting certificate...")
+    try:
-    use_ssl = os.environ.get("USE_SSL", "false") == "true"
+        log.debug("getting certificate...")
-    if use_ssl is True:
+        use_ssl = os.environ.get("USE_SSL", "false") == "true"
-        ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        if use_ssl is True:
-        ssl_context.load_cert_chain(
+            ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
-            certfile="/etc/instance.crt",
+            ssl_context.load_cert_chain(
-            keyfile="/etc/instance.key",
+                certfile="/etc/instance.crt",
-        )
+                keyfile="/etc/instance.key",
-    else:
+            )
-        ssl_context = None
+        else:
+            ssl_context = None
 
-    async def main():
+        async def main():
-        log.debug("starting server...")
+            log.debug("starting server...")
-        app = web.Application()
+            app = web.Application()
-        app.add_routes(routes)
+            app.add_routes(routes)
-        runner = web.AppRunner(app)
+            runner = web.AppRunner(app)
-        await runner.setup()
+            await runner.setup()
-        site = web.TCPSite(
+            site = web.TCPSite(
-            runner,
+                runner,
-            ssl_context=ssl_context,
+                ssl_context=ssl_context,
-            port=int(os.environ["WORKER_PORT"]),
+                port=int(os.environ["WORKER_PORT"]),
-            **kwargs
+                **kwargs
-        )
+            )
-        await gather(site.start(), backend._start_tracking())
+            await gather(site.start(), backend._start_tracking())
 
-    run(main())
+        run(main())
+
+    except Exception as e:
+        err_msg = f"PyWorker failed to launch: {e}"
+        log.error(err_msg)
+
+        async def beacon():
+            metrics = Metrics()
+            metrics._set_version(getattr(backend, "version", "0"))
+            metrics._set_mtoken(getattr(backend, "mtoken", ""))
+            try:
+                while True:
+                    metrics._model_errored(err_msg)
+                    await metrics._Metrics__send_metrics_and_reset()
+                    await asyncio.sleep(10)
+            finally:
+                await metrics.aclose()
+
+        run(beacon())

start_server.sh

@@ -132,5 +132,43 @@ cd "$SERVER_DIR"
 
 echo "launching PyWorker server"
 
-(python3 -m "workers.$BACKEND.server" |& tee -a "$PYWORKER_LOG") &
+set +e
-echo "launching PyWorker server done"
+python3 -m "workers.$BACKEND.server" |& tee -a "$PYWORKER_LOG"
+PY_STATUS=${PIPESTATUS[0]}
+set -e
+
+if [ "${PY_STATUS}" -ne 0 ]; then
+  echo "PyWorker exited with status ${PY_STATUS}; notifying autoscaler..."
+  ERROR_MSG="PyWorker exited: code ${PY_STATUS}"
+  MTOKEN="${MASTER_TOKEN:-}"
+  VERSION="${PYWORKER_VERSION:-0}"
+
+  IFS=',' read -r -a REPORT_ADDRS <<< "${REPORT_ADDR}"
+  for addr in "${REPORT_ADDRS[@]}"; do
+    curl -sS -X POST -H 'Content-Type: application/json' \
+      -d "$(cat <<JSON
+{
+  "id": ${CONTAINER_ID:-0},
+  "mtoken": "${MTOKEN}",
+  "version": "${VERSION}",
+  "loadtime": 0,
+  "new_load": 0,
+  "cur_load": 0,
+  "rej_load": 0,
+  "max_perf": 0,
+  "cur_perf": 0,
+  "error_msg": "${ERROR_MSG}",
+  "num_requests_working": 0,
+  "num_requests_recieved": 0,
+  "additional_disk_usage": 0,
+  "working_request_idxs": [],
+  "cur_capacity": 0,
+  "max_capacity": 0,
+  "url": "${URL}"
+}
+JSON
+)" "${addr%/}/worker_status/" || true
+  done
+fi
+
+echo "launching PyWorker server done"

workers/openai/server.py

@@ -11,6 +11,7 @@ MODEL_SERVER_START_LOG_MSG = [
     "llama runner started",  # Ollama
     '"message":"Connected","target":"text_generation_router"',  # TGI
     '"message":"Connected","target":"text_generation_router::server"',  # TGI
+    "main: model loaded" # llama.cpp
 ]
 
 MODEL_SERVER_ERROR_LOG_MSGS = [
@@ -34,6 +35,7 @@ backend = Backend(
     model_server_url=os.environ["MODEL_SERVER_URL"],
     model_log_file=os.environ["MODEL_LOG"],
     allow_parallel_requests=True,
+    max_wait_time=600.0,
     benchmark_handler=CompletionsHandler(benchmark_runs=3, benchmark_words=256),
     log_actions=[
         *[(LogAction.ModelLoaded, info_msg) for info_msg in MODEL_SERVER_START_LOG_MSG],