Develop #16

Merged
tao.chen merged 273 commits from develop into main 2026-08-21 10:42:09 +08:00
4 changed files with 64 additions and 11 deletions
Showing only changes of commit be475dd0d9 - Show all commits
+4 -6
View File
@@ -1,6 +1,5 @@
from __future__ import annotations from __future__ import annotations
import asyncio
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import Any, Literal from typing import Any, Literal
from urllib.parse import quote from urllib.parse import quote
@@ -19,6 +18,7 @@ from pydantic import Field
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from common.config import settings
from common.db.models import ( from common.db.models import (
ScheduleNodeRuns, ScheduleNodeRuns,
ScheduleRuns, ScheduleRuns,
@@ -199,7 +199,7 @@ async def _visible_artifact(
) )
if ( if (
item is None item is None
or item.storage_backend != "rustfs" or item.storage_backend != settings.storage_backend
or not item.bucket_name or not item.bucket_name
or not item.object_key or not item.object_key
): ):
@@ -223,10 +223,8 @@ async def _artifact_bytes(
item: StorageObjects, item: StorageObjects,
request: Request, request: Request,
) -> bytes: ) -> bytes:
return await asyncio.to_thread( return await request.app.state.object_stores[item.bucket_name].get(
request.app.state.object_store.get_bytes, item.object_key,
bucket_name=item.bucket_name,
object_key=item.object_key,
) )
+11 -3
View File
@@ -48,7 +48,6 @@ services:
volumes: volumes:
- ${PWD}:/app - ${PWD}:/app
- ./default.conf:/etc/nginx/conf.d/default.conf.template:ro - ./default.conf:/etc/nginx/conf.d/default.conf.template:ro
- ./scripts/nginx-entrypoint.sh:/usr/local/bin/model-platform-entrypoint.sh:ro
healthcheck: healthcheck:
test: ["CMD-SHELL", "wget -qO- http://127.0.0.1/ >/dev/null"] test: ["CMD-SHELL", "wget -qO- http://127.0.0.1/ >/dev/null"]
interval: 10s interval: 10s
@@ -66,6 +65,13 @@ services:
max-size: "200m" max-size: "200m"
max-file: "10" max-file: "10"
restart: unless-stopped restart: unless-stopped
# The source tree is bind-mounted for local development while /app/.venv
# is an anonymous Docker volume. Sync first so restarts never keep an
# older installed backend package.
command:
- sh
- -c
- uv sync --no-dev --no-editable --package backend && uv pip install --python /app/.venv/bin/python --reinstall --no-deps ./common ./backend && exec gunicorn --config backend/gunicorn.conf.py backend.main:app
# No host port: architecture §2.2 — only Nginx is externally reachable. # No host port: architecture §2.2 — only Nginx is externally reachable.
# No local-FS volume: backend stores everything in S3 (S3_*). # No local-FS volume: backend stores everything in S3 (S3_*).
environment: environment:
@@ -95,7 +101,7 @@ services:
runtime: runtime:
condition: service_healthy condition: service_healthy
ports: ports:
- 8891:8000 - 9121:8000
volumes: volumes:
- ${PWD}:/app - ${PWD}:/app
- ./data:/data - ./data:/data
@@ -118,7 +124,7 @@ services:
max-file: "10" max-file: "10"
restart: unless-stopped restart: unless-stopped
ports: ports:
- 8892:8000 - 9122:8000
cap_add: cap_add:
- SYS_ADMIN - SYS_ADMIN
devices: devices:
@@ -192,6 +198,8 @@ services:
S3_WORKSPACE_BUCKET: ${S3_WORKSPACE_BUCKET:-workspace} S3_WORKSPACE_BUCKET: ${S3_WORKSPACE_BUCKET:-workspace}
S3_VERSION_BUCKET: ${S3_VERSION_BUCKET:-version} S3_VERSION_BUCKET: ${S3_VERSION_BUCKET:-version}
S3_RUN_LOG_BUCKET: ${S3_RUN_LOG_BUCKET:-run-log} S3_RUN_LOG_BUCKET: ${S3_RUN_LOG_BUCKET:-run-log}
# Backend's health endpoint already verifies MySQL; the scheduler only
# needs MySQL and Backend to be ready in either local or S3 mode.
READINESS_TARGETS: ${MYSQL_HOST:?MYSQL_HOST is required}:${MYSQL_PORT:-3306},backend:8000 READINESS_TARGETS: ${MYSQL_HOST:?MYSQL_HOST is required}:${MYSQL_PORT:-3306},backend:8000
depends_on: depends_on:
backend: backend:
+6 -2
View File
@@ -7,6 +7,7 @@ surface is two endpoints; everything else is lifespan orchestration.
from __future__ import annotations from __future__ import annotations
import asyncio
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Literal from typing import Literal
@@ -48,7 +49,10 @@ async def lifespan(app: FastAPI):
f"removed_meta={counters['removed_meta']} " f"removed_meta={counters['removed_meta']} "
f"live_in_registry={counters['live_in_registry']}" f"live_in_registry={counters['live_in_registry']}"
) )
await scan_workspaces() # Scanning a remote FUSE mount may take longer than the HTTP health-check
# timeout. Keep recovery asynchronous so dependent services can start as
# soon as the runtime API is ready.
asyncio.create_task(scan_workspaces(), name="workspace-startup-scan")
# P1-3: spawn background idle reaper. Stopped in the lifespan # P1-3: spawn background idle reaper. Stopped in the lifespan
# finally block; awaits the cancellation to avoid a leaked task. # finally block; awaits the cancellation to avoid a leaked task.
start_reaper() start_reaper()
@@ -112,4 +116,4 @@ async def handle_jupyter_action(req: JupyterActionRequest) -> dict:
status_code=400, status_code=400,
detail="'workspace_id' is required for action='get'", detail="'workspace_id' is required for action='get'",
) )
return await get_workspace(req.workspace_id) return await get_workspace(req.workspace_id)
+43
View File
@@ -37,6 +37,7 @@ import asyncio
import json import json
import os import os
import secrets import secrets
import socket
import subprocess import subprocess
import time import time
from typing import TypedDict from typing import TypedDict
@@ -163,6 +164,37 @@ def _bump_last_used(record: JupyterProcessRecord) -> None:
record["last_used_at"] = time.time() record["last_used_at"] = time.time()
async def _wait_for_jupyter_ready(
process: subprocess.Popen,
port: int,
*,
timeout_seconds: float = 30,
) -> None:
"""Wait until a newly spawned Jupyter server accepts TCP connections.
Starting the subprocess only means that Python has been forked. Jupyter
still needs several seconds to load extensions and bind its port. The
runtime API must not advertise the workspace as ``running`` before that
happens, otherwise callers immediately receive a transient connection
failure while creating or opening the first notebook.
"""
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
if process.poll() is not None:
raise RuntimeError(
f"Jupyter exited during startup (exit code {process.returncode})"
)
try:
with socket.create_connection(("127.0.0.1", port), timeout=0.2):
return
except OSError:
await asyncio.sleep(0.2)
raise RuntimeError(
f"Timed out after {timeout_seconds:g}s waiting for Jupyter on port {port}"
)
async def start_workspace(ws_id: str) -> dict: async def start_workspace(ws_id: str) -> dict:
async with get_workspace_lock(ws_id): async with get_workspace_lock(ws_id):
workspace_path = (WORKSPACES_ROOT / ws_id).resolve() workspace_path = (WORKSPACES_ROOT / ws_id).resolve()
@@ -297,6 +329,17 @@ async def start_workspace(ws_id: str) -> dict:
detail=f"Failed to start Jupyter: {e}", detail=f"Failed to start Jupyter: {e}",
) )
try:
await _wait_for_jupyter_ready(process, port)
except Exception as exc:
logger.exception(f"Jupyter did not become ready for workspace={ws_id}")
if process.poll() is None:
process.terminate()
raise HTTPException(
status_code=503,
detail=f"Jupyter failed to start: {exc}",
) from exc
# --------------------------------------------------------- # ---------------------------------------------------------
# Save process metadata # Save process metadata
# --------------------------------------------------------- # ---------------------------------------------------------