tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor

Browse Source
This commit is contained in:
tao.chen
2026-07-30 18:57:18 +08:00
parent ff7fc4ef1d
commit c1e15758a3
31 changed files with 2648 additions and 3356 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/Dockerfile
+16 -1
View File
@@ -2,12 +2,27 @@ FROM python:3.12-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PYTHONPATH=/app
WORKDIR /app
# 从官方 Rclone 镜像直接复制 rclone 二进制文件（高效、稳定）
COPY --from=rclone/rclone:latest /usr/local/bin/rclone /usr/local/bin/rclone
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
# 安装系统依赖（fuse3 是 rclone mount 的核心底层依赖）
RUN apt-get update && apt-get install -y --no-install-recommends \
fuse3 \
ca-certificates \
curl \
procps \
build-essential \
python3-dev \
&& sed -i 's/#user_allow_other/user_allow_other/g' /etc/fuse.conf \
&& rm -rf /var/lib/apt/lists/*
COPY pyproject.toml uv.lock ./
COPY common ./common
COPY contracts ./contracts
COPY runtime ./runtime
RUN uv sync --frozen --no-dev --no-editable --package runtime
RUN UV_DEFAULT_INDEX="https://pypi.tuna.tsinghua.edu.cn/simple/" uv sync --frozen --no-dev --no-editable --package runtime
EXPOSE 8000
CMD ["uv", "run", "--frozen", "--package", "runtime", "uvicorn", "runtime.main:app", "--host", "0.0.0.0", "--port", "8000"]
runtime/pyproject.toml
+2 -1
View File
@@ -7,7 +7,8 @@ dependencies = [
"fastapi==0.116.1",
"uvicorn[standard]==0.35.0",
"httpx==0.28.1",
"redis==5.2.1",
"loguru==0.7.2",
"notebook",
]
[tool.uv.sources]
runtime/src/runtime/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Runtime Manager application."""
runtime/src/runtime/legacy_process_runtime.py
-449
View File
@@ -1,449 +0,0 @@
# coding=utf-8
"""
@Time :2026/7/27
@Author :tao.chen
"""
import os
import secrets
import socket
import subprocess
import time
import datetime
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Dict, Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from loguru import logger
# 全局内存字典：记录运行中的 Jupyter 进程信息
JUPYTER_PROCESSES: Dict[str, dict] = {}
WORKSPACES_ROOT = Path(os.getenv("WORKSPACES_ROOT", "/app/workspaces"))
PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", "http://localhost")
REMOTE_BUCKET = os.getenv("REMOTE_BUCKET", "rustfs:workspaces")
RCLONE_PROCESS = None
def is_mountpoint(path: Path) -> bool:
"""
判断目录是否已经挂载
"""
result = subprocess.run(
["mountpoint", "-q", str(path)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return result.returncode == 0
def start_rclone_mount():
"""
启动 rclone mount
"""
global RCLONE_PROCESS
if is_mountpoint(WORKSPACES_ROOT):
logger.info(
f"Mountpoint already exists: {WORKSPACES_ROOT}"
)
return
WORKSPACES_ROOT.mkdir(parents=True, exist_ok=True)
logger.info(
f"Starting rclone mount "
f"{REMOTE_BUCKET} -> {WORKSPACES_ROOT}"
)
log_file = open(
"/tmp/rclone-mount.log",
"a",
buffering=1,
)
cmd = [
"rclone",
"mount",
REMOTE_BUCKET,
WORKSPACES_ROOT.as_posix(),
"--allow-other",
"--vfs-cache-mode","full",
"--vfs-cache-max-size","20G",
"--vfs-write-back","5s",
"--dir-cache-time","30s",
"--poll-interval","30s",
"--log-level","INFO",
]
RCLONE_PROCESS = subprocess.Popen(
cmd,
stdout=log_file,
stderr=subprocess.STDOUT,
start_new_session=True,
)
# 等待 mount ready
timeout = 20
while timeout > 0:
if is_mountpoint(WORKSPACES_ROOT):
logger.info(f"rclone mount ready: {WORKSPACES_ROOT}" )
return
# rclone异常退出
if RCLONE_PROCESS.poll() is not None:
raise RuntimeError( "rclone mount process exited")
time.sleep(1)
timeout -= 1
raise RuntimeError( f"Timeout waiting mount: {WORKSPACES_ROOT}")
def stop_rclone_mount():
global RCLONE_PROCESS
logger.info(
"Stopping rclone mount..."
)
if RCLONE_PROCESS:
if RCLONE_PROCESS.poll() is None:
RCLONE_PROCESS.terminate()
try:
RCLONE_PROCESS.wait(timeout=10)
except subprocess.TimeoutExpired:
logger.warning("Force killing rclone")
RCLONE_PROCESS.kill()
if is_mountpoint(WORKSPACES_ROOT):
logger.info(f"Unmount {WORKSPACES_ROOT}")
result = subprocess.run(
[
"fusermount3",
"-u",
WORKSPACES_ROOT.as_posix(),
]
)
if result.returncode != 0:
subprocess.run(
[
"umount",
"-l",
WORKSPACES_ROOT.as_posix(),
]
)
logger.info("rclone stopped")
def scan_workspaces():
"""
扫描已有 workspace
"""
if not WORKSPACES_ROOT.exists():
return
try:
entries = os.listdir(WORKSPACES_ROOT)
except Exception as e:
logger.error(f"scan workspace failed: {e}")
return
for entry in entries:
path = WORKSPACES_ROOT / entry
if not path.is_dir():
continue
logger.info(f"Found workspace: {entry}" )
try:
full_path = os.path.join(WORKSPACES_ROOT, entry)
if os.path.isdir(full_path):
logger.info(f"Auto-starting Jupyter for workspace ID: '{entry}'")
try:
_handle_start(entry)
except Exception as err:
logger.error(f"Startup failed for workspace '{entry}': {err}")
except Exception as e:
logger.error(f"Start workspace {entry} failed: {e}" )
def get_free_port() -> int:
"""利用操作系统 socket 特性，动态获取当前闲置的可用端口"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("", 0))
s.listen(1)
port = s.getsockname()[1]
return port
# 统一请求 Model
class JupyterActionRequest(BaseModel):
action: str = Field(
..., description="操作类型: 'start' | 'stop' | 'list'"
)
workspace_id: Optional[str] = Field(
None, description="Workspace ID (start 和 stop 操作时必填)"
)
# 辅助处理函数：启动逻辑
def start_process(cmd, workspace_path, log_dir="/tmp/process_logs"):
log_dir = Path(log_dir)
log_dir.mkdir(parents=True, exist_ok=True)
start_time = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
# 临时日志文件
temp_log = log_dir / f"process_start_{start_time}.log"
log_file = open(temp_log, "a", buffering=1)
process = subprocess.Popen(
cmd,
cwd=workspace_path,
stdout=log_file,
stderr=subprocess.STDOUT, # stderr 合并到 stdout
start_new_session=True,
)
# 根据真实 pid 重命名
final_log = log_dir / f"process_{process.pid}_{start_time}.log"
log_file.close()
temp_log.rename(final_log)
logger.info(
f"process started pid={process.pid}, log={final_log}"
)
return process
def _handle_start(ws_id: str):
workspace_path = WORKSPACES_ROOT / ws_id
# 如果已存在，校验进程状态并复用
if ws_id in JUPYTER_PROCESSES:
p_info = JUPYTER_PROCESSES[ws_id]
if p_info["process"].poll() is None:
logger.info(f"Workspace {ws_id} already running.")
return {
"status": "running",
"workspace_id": ws_id,
"port": p_info["port"],
"full_url": p_info["full_url"],
}
else:
del JUPYTER_PROCESSES[ws_id]
# 2. 动态申请端口与 Token
port = get_free_port()
token = secrets.token_hex(16)
base_path = f"/jupyter/{ws_id}/"
cmd = [
"jupyter",
"notebook",
f"--port={port}",
"--ip=0.0.0.0",
"--no-browser",
"--allow-root",
f"--ServerApp.token={token}",
f"--ServerApp.base_url={base_path}",
"--notebook-dir=.",
# 适用于现代 Jupyter Server / JupyterLab
"--ServerApp.terminals_enabled=False",
# 兼容经典 Notebook / 旧版配置项
"--NotebookApp.terminals_enabled=False",
# 允许 Nginx 跨域代理与 WebSocket 通信（关键）
"--ServerApp.allow_origin=*",
"--NotebookApp.allow_origin=*",
"--ServerApp.disable_check_xsrf=True",
"--NotebookApp.disable_check_xsrf=True"
]
try:
process = start_process(cmd, workspace_path.as_posix())
full_url = f"{PUBLIC_BASE_URL}:{port}{base_path}?token={token}"
JUPYTER_PROCESSES[ws_id] = {
"process": process,
"base_url": PUBLIC_BASE_URL,
"port": port,
"token": token,
"full_url": full_url,
"started_at": time.time(),
}
logger.info(
f"Started Jupyter for workspace {ws_id} on port {port}"
)
return {
"pid": process.pid,
"base_url": PUBLIC_BASE_URL,
"status": "success",
"workspace_id": ws_id,
"port": port,
"token": token,
}
except Exception as e:
logger.error(f"Failed to start Jupyter for {ws_id}: {str(e)}")
raise HTTPException(
status_code=500, detail=f"Failed to start Jupyter: {str(e)}"
)
# 辅助处理函数：停止逻辑
def _handle_stop(ws_id: str):
if ws_id not in JUPYTER_PROCESSES:
raise HTTPException(
status_code=404,
detail=f"No active Jupyter process found for workspace '{ws_id}'",
)
p_info = JUPYTER_PROCESSES[ws_id]
process: subprocess.Popen = p_info["process"]
if process.poll() is None:
try:
process.terminate()
process.wait(timeout=3)
logger.info(
f"Gracefully stopped Jupyter for workspace {ws_id}"
)
except subprocess.TimeoutExpired:
logger.warning(
f"Jupyter for {ws_id} did not stop gracefully. Force killing..."
)
process.kill()
process.wait()
del JUPYTER_PROCESSES[ws_id]
return {
"status": "stopped",
"workspace_id": ws_id,
"message": "Jupyter process terminated and port released.",
}
# 辅助处理函数：列表查询逻辑
def _handle_list():
active_instances = {}
for ws_id, info in list(JUPYTER_PROCESSES.items()):
is_alive = info["process"].poll() is None
active_instances[ws_id] = {
"port": info["port"],
"full_url": info["full_url"],
"is_alive": is_alive,
}
return {"status": "success", "instances": active_instances}
def _handle_get(ws_id: str):
"""【新增】获取指定 Workspace 的 Jupyter 运行状态与完整 URL"""
if ws_id not in JUPYTER_PROCESSES:
raise HTTPException(
status_code=404,
detail=f"No active Jupyter process found for workspace '{ws_id}'",
)
p_info = JUPYTER_PROCESSES[ws_id]
is_alive = p_info["process"].poll() is None
if not is_alive:
# 进程如果挂了，清理内存字典并报 404
del JUPYTER_PROCESSES[ws_id]
raise HTTPException(
status_code=404,
detail=f"Jupyter process for workspace '{ws_id}' has terminated unexpectedly.",
)
return {
"status": "running",
"pid": p_info["process"].pid,
"base_url": PUBLIC_BASE_URL,
"workspace_id": ws_id,
"port": p_info["port"],
"token": p_info["token"],
"started_at": p_info["started_at"],
}
# ==================== FastAPI Lifespan 定义 ====================
@asynccontextmanager
async def lifespan(app: FastAPI):
global RCLONE_PROCESS
logger.info("Starting up Runtime Service...")
start_rclone_mount()
logger.info(f"Scanning workspaces: {WORKSPACES_ROOT}")
scan_workspaces()
logger.info("Runtime Service started")
# ==================== 2. 服务运行阶段 (Serving) ====================
try:
yield # 服务保持运行，等待并处理 API 请求
finally:
logger.info("Service is shutting down. Terminating all active Jupyter sub-processes...")
# 优先杀死所有 Jupyter 子进程（确保文件句柄被释放）
active_workspaces = list(JUPYTER_PROCESSES.keys())
for ws_id in active_workspaces:
try:
_handle_stop(ws_id)
except Exception as err:
logger.error(f"Error terminating Jupyter process for '{ws_id}': {err}")
logger.info("All Jupyter sub-processes have been terminated.")
JUPYTER_PROCESSES.clear()
# 卸载 Rclone 挂载点（强制将 VFS 缓存刷新同步至对象存储）
try:
stop_rclone_mount()
except Exception as e:
logger.error(f"Stop rclone failed: {e}" )
logger.info("Runtime Service stopped")
app = FastAPI(lifespan=lifespan)
# ---------------- 统一入口 POST 接口 ----------------
@app.get("/api/v1/health")
def healthz():
return {"status": "ok"}
@app.post("/api/v1/jupyter")
def handle_jupyter_action(req: JupyterActionRequest):
action = req.action.lower()
# 1. 启动操作
if action == "start":
if not req.workspace_id:
raise HTTPException(
status_code=400,
detail="'workspace_id' is required when action='start'",
)
return _handle_start(req.workspace_id)
# 2. 停止操作
elif action == "stop":
if not req.workspace_id:
raise HTTPException(
status_code=400,
detail="'workspace_id' is required when action='stop'",
)
return _handle_stop(req.workspace_id)
# 3. 列表操作
elif action == "list":
return _handle_list()
elif action == "get":
if not req.workspace_id:
raise HTTPException(
status_code=400,
detail="'workspace_id' is required for action='get'",
)
return _handle_get(req.workspace_id)
# 未知操作
else:
raise HTTPException(
status_code=400,
detail=f"Invalid action '{req.action}'. Supported actions: 'start', 'stop', 'list'",
)
runtime/src/runtime/main.py
+70 -1117
View File
File diff suppressed because it is too large Load Diff
runtime/src/runtime/mount.py
+93
View File
@@ -0,0 +1,93 @@
"""Object storage mount management.
Owns the rclone mount lifecycle for the remote workspace bucket.
``WORKSPACES_ROOT`` is defined here because this module is what makes
the directory usable; downstream consumers (e.g. process.py) import it.
"""
from __future__ import annotations
import os
import subprocess
import time
from pathlib import Path
from loguru import logger
WORKSPACES_ROOT = Path(os.getenv("WORKSPACES_ROOT", "/app/workspaces"))
REMOTE_BUCKET = os.getenv("REMOTE_BUCKET", "rustfs:workspaces")
RCLONE_PROCESS: subprocess.Popen | None = None
def is_mountpoint(path: Path) -> bool:
result = subprocess.run(
["mountpoint", "-q", str(path)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return result.returncode == 0
def start_rclone_mount() -> None:
global RCLONE_PROCESS
if is_mountpoint(WORKSPACES_ROOT):
logger.info(f"Mountpoint already exists: {WORKSPACES_ROOT}")
return
WORKSPACES_ROOT.mkdir(parents=True, exist_ok=True)
logger.info(f"Starting rclone mount {REMOTE_BUCKET} -> {WORKSPACES_ROOT}")
with open("/tmp/rclone-mount.log", "a", buffering=1) as log_file:
cmd = [
"rclone",
"mount",
REMOTE_BUCKET,
str(WORKSPACES_ROOT),
"--allow-other",
"--vfs-cache-mode", "full",
"--vfs-cache-max-size", "20G",
"--vfs-write-back", "5s",
"--dir-cache-time", "30s",
"--poll-interval", "30s",
"--log-level", "INFO",
]
RCLONE_PROCESS = subprocess.Popen(
cmd,
stdout=log_file,
stderr=subprocess.STDOUT,
start_new_session=True,
)
timeout = 20
while timeout > 0:
if is_mountpoint(WORKSPACES_ROOT):
logger.info(f"rclone mount ready: {WORKSPACES_ROOT}")
return
if RCLONE_PROCESS.poll() is not None:
raise RuntimeError("rclone mount process exited")
time.sleep(1)
timeout -= 1
raise RuntimeError(f"Timeout waiting mount: {WORKSPACES_ROOT}")
def stop_rclone_mount() -> None:
global RCLONE_PROCESS
logger.info("Stopping rclone mount...")
if RCLONE_PROCESS:
if RCLONE_PROCESS.poll() is None:
RCLONE_PROCESS.terminate()
try:
RCLONE_PROCESS.wait(timeout=10)
except subprocess.TimeoutExpired:
logger.warning("Force killing rclone")
RCLONE_PROCESS.kill()
if is_mountpoint(WORKSPACES_ROOT):
logger.info(f"Unmount {WORKSPACES_ROOT}")
result = subprocess.run(["fusermount3", "-u", str(WORKSPACES_ROOT)])
if result.returncode != 0:
subprocess.run(["umount", "-l", str(WORKSPACES_ROOT)])
logger.info("rclone stopped")
runtime/src/runtime/process.py
+261
View File
@@ -0,0 +1,261 @@
"""Jupyter process registry and lifecycle.
Owns the in-memory ``JUPYTER_PROCESSES`` dict, the ``STATE_LOCK`` that
serializes mutations to it, and the per-workspace start/stop/list/get
operations. Workspace discovery (startup scan) lives here because it is
a thin wrapper over ``start_workspace``.
"""
from __future__ import annotations
import asyncio
import os
import secrets
import subprocess
import time
from typing import TypedDict
from fastapi import HTTPException
from loguru import logger
from common.utils import get_free_port, start_process
from runtime.mount import WORKSPACES_ROOT
PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", "http://localhost")
class JupyterProcessRecord(TypedDict):
process: subprocess.Popen
port: int
token: str
base_url: str
started_at: float
JUPYTER_PROCESSES: dict[str, JupyterProcessRecord] = {}
WORKSPACE_LOCKS: dict[str, asyncio.Lock] = {}
_LOCKS_REGISTRY = asyncio.Lock()
def get_workspace_lock(ws_id: str) -> asyncio.Lock:
"""Return the per-workspace lock, creating it on first use.
``asyncio.Lock`` is created lazily per event loop; sharing it across
loops is unsafe. The instance lives for the lifetime of the process.
"""
if ws_id in WORKSPACE_LOCKS:
return WORKSPACE_LOCKS[ws_id]
# We do not need to hold the registry lock long enough to block; a
# double-check pattern prevents accidentally replacing an existing
# lock. The Lock object is itself safe to call .locked()/acquire on.
lock = WORKSPACE_LOCKS.get(ws_id)
if lock is None:
lock = asyncio.Lock()
WORKSPACE_LOCKS[ws_id] = lock
return lock
def _drop_workspace_lock(ws_id: str) -> None:
"""Remove the per-workspace lock once no process references it.
Called after a successful stop to keep the registry bounded. We only
drop a lock we created and only when it is not held.
"""
lock = WORKSPACE_LOCKS.get(ws_id)
if lock is None or lock.locked():
return
WORKSPACE_LOCKS.pop(ws_id, None)
async def start_workspace(ws_id: str) -> dict:
async with get_workspace_lock(ws_id):
workspace_path = WORKSPACES_ROOT / ws_id
if ws_id in JUPYTER_PROCESSES:
p_info = JUPYTER_PROCESSES[ws_id]
if p_info["process"].poll() is None:
if time.time() - p_info["started_at"] > 24 * 3600:
logger.warning(
f"Reusing Jupyter for {ws_id} older than 24h "
f"(started_at={p_info['started_at']})"
)
return {
"status": "running",
"workspace_id": ws_id,
"pid": p_info["process"].pid,
"port": p_info["port"],
"token": p_info["token"],
"base_url": PUBLIC_BASE_URL,
"full_url": (
f"{PUBLIC_BASE_URL}:{p_info['port']}/jupyter/{ws_id}/"
f"?token={p_info['token']}"
),
}
del JUPYTER_PROCESSES[ws_id]
port = get_free_port()
token = secrets.token_urlsafe(16)
base_path = f"/jupyter/{ws_id}/"
cmd = [
"jupyter", "notebook",
f"--port={port}",
"--ip=0.0.0.0",
"--no-browser",
"--allow-root",
f"--ServerApp.token={token}",
f"--ServerApp.base_url={base_path}",
"--notebook-dir=.",
"--ServerApp.terminals_enabled=False",
"--NotebookApp.terminals_enabled=False",
"--ServerApp.allow_origin=*",
"--NotebookApp.allow_origin=*",
"--ServerApp.disable_check_xsrf=True",
"--NotebookApp.disable_check_xsrf=True",
]
try:
process, log_file = start_process(cmd, workspace_path)
except Exception as e:
logger.error(f"Failed to start Jupyter for {ws_id}: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to start Jupyter: {e}"
)
full_url = f"{PUBLIC_BASE_URL}:{port}{base_path}?token={token}"
JUPYTER_PROCESSES[ws_id] = {
"process": process,
"base_url": PUBLIC_BASE_URL,
"port": port,
"token": token,
"started_at": time.time(),
}
logger.info(
f"Started Jupyter for workspace {ws_id} "
f"pid={process.pid} port={port} log={log_file}"
)
return {
"pid": process.pid,
"base_url": PUBLIC_BASE_URL,
"status": "success",
"workspace_id": ws_id,
"port": port,
"token": token,
"full_url": full_url,
}
async def stop_workspace(ws_id: str) -> dict:
async with get_workspace_lock(ws_id):
if ws_id not in JUPYTER_PROCESSES:
raise HTTPException(
status_code=404,
detail=f"No active Jupyter process found for workspace '{ws_id}'",
)
p_info = JUPYTER_PROCESSES[ws_id]
process: subprocess.Popen = p_info["process"]
if process.poll() is None:
try:
process.terminate()
process.wait(timeout=3)
logger.info(f"Gracefully stopped Jupyter for workspace {ws_id}")
except subprocess.TimeoutExpired:
logger.warning(
f"Jupyter for {ws_id} did not stop gracefully. Force killing..."
)
try:
process.kill()
process.wait()
except Exception as err:
logger.error(f"Failed to kill Jupyter process for {ws_id}: {err}")
del JUPYTER_PROCESSES[ws_id]
_drop_workspace_lock(ws_id)
return {
"status": "stopped",
"workspace_id": ws_id,
"message": "Jupyter process terminated and port released.",
}
async def list_workspaces() -> dict:
async with _LOCKS_REGISTRY:
snapshot = dict(JUPYTER_PROCESSES)
return {
"status": "success",
"instances": {
ws_id: {
"port": info["port"],
"full_url": (
f"{PUBLIC_BASE_URL}:{info['port']}/jupyter/{ws_id}/"
f"?token={info['token']}"
),
"is_alive": info["process"].poll() is None,
}
for ws_id, info in snapshot.items()
},
}
async def get_workspace(ws_id: str) -> dict:
async with get_workspace_lock(ws_id):
if ws_id not in JUPYTER_PROCESSES:
raise HTTPException(
status_code=404,
detail=f"No active Jupyter process found for workspace '{ws_id}'",
)
p_info = JUPYTER_PROCESSES[ws_id]
is_alive = p_info["process"].poll() is None
if not is_alive:
del JUPYTER_PROCESSES[ws_id]
else:
return {
"status": "running",
"pid": p_info["process"].pid,
"base_url": PUBLIC_BASE_URL,
"workspace_id": ws_id,
"port": p_info["port"],
"token": p_info["token"],
"full_url": (
f"{PUBLIC_BASE_URL}:{p_info['port']}/jupyter/{ws_id}/"
f"?token={p_info['token']}"
),
"started_at": p_info["started_at"],
}
_drop_workspace_lock(ws_id)
raise HTTPException(
status_code=404,
detail=(
f"Jupyter process for workspace '{ws_id}' "
"has terminated unexpectedly."
),
)
async def scan_workspaces() -> None:
if not WORKSPACES_ROOT.exists():
return
try:
entries = os.listdir(WORKSPACES_ROOT)
except Exception as e:
logger.error(f"scan workspace failed: {e}")
return
for entry in entries:
path = WORKSPACES_ROOT / entry
if not path.is_dir():
continue
logger.info(f"Found workspace: {entry}")
logger.info(f"Auto-starting Jupyter for workspace ID: '{entry}'")
try:
await start_workspace(entry)
except Exception as err:
logger.error(f"Startup failed for workspace '{entry}': {err}")
runtime/src/runtime/providers/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Runtime provider implementations."""
runtime/src/runtime/providers/shared_jupyter.py
-203
View File
@@ -1,203 +0,0 @@
from __future__ import annotations
import hashlib
from datetime import UTC, datetime
from pathlib import PurePosixPath
from urllib.parse import quote
import httpx
from contracts.runtime.runtime_adapter import (
CreateSessionRequest,
EnsureRuntimeRequest,
RuntimeEndpoint,
RuntimeHealth,
RuntimeSession,
)
class RuntimeProviderError(RuntimeError):
pass
class SharedJupyterAdapter:
"""Compose provider backed by one internal Jupyter Server.
Runtime rows are scoped to a Workspace. Each Notebook has its own
Jupyter Session and Kernel inside that server. Replacing this class with
a Docker or Kubernetes Workspace provider does not change the contract.
"""
provider_name = "process"
runtime_ref = "compose:jupyter"
def __init__(
self,
client: httpx.AsyncClient,
*,
internal_url: str,
proxy_base_path: str,
) -> None:
self.client = client
self.internal_url = internal_url.rstrip("/") + "/"
self.proxy_base_path = "/" + proxy_base_path.strip("/") + "/"
async def _request(
self,
method: str,
path: str,
*,
payload: dict | None = None,
allow_not_found: bool = False,
) -> httpx.Response:
try:
response = await self.client.request(method, path, json=payload)
except httpx.RequestError as exc:
raise RuntimeProviderError(
f"Jupyter request failed: {type(exc).__name__}"
) from exc
if allow_not_found and response.status_code == 404:
return response
if response.is_error:
raise RuntimeProviderError(
f"Jupyter returned HTTP {response.status_code}"
)
return response
async def ensure_running(
self,
request: EnsureRuntimeRequest,
) -> RuntimeEndpoint:
health = await self.health(request.runtime_id)
if not health.healthy:
raise RuntimeProviderError(
health.detail or "Jupyter is not healthy"
)
return RuntimeEndpoint(
runtime_id=request.runtime_id,
runtime_type="jupyter",
provider=self.provider_name,
runtime_ref=self.runtime_ref,
internal_url=self.internal_url,
proxy_base_path=self.proxy_base_path,
)
async def stop(self, runtime_id: str, reason: str) -> None:
# Compose keeps the shared infrastructure process alive. Stopping a
# logical Runtime terminates its sessions and updates MySQL state.
return None
async def restart(self, runtime_id: str) -> RuntimeEndpoint:
health = await self.health(runtime_id)
if not health.healthy:
raise RuntimeProviderError(
health.detail or "Jupyter is not healthy"
)
return RuntimeEndpoint(
runtime_id=runtime_id,
runtime_type="jupyter",
provider=self.provider_name,
runtime_ref=self.runtime_ref,
internal_url=self.internal_url,
proxy_base_path=self.proxy_base_path,
)
async def health(self, runtime_id: str) -> RuntimeHealth:
try:
await self._request("GET", "api/status")
except RuntimeProviderError as exc:
return RuntimeHealth(
runtime_id=runtime_id,
healthy=False,
checked_at=datetime.now(UTC),
detail=str(exc),
)
return RuntimeHealth(
runtime_id=runtime_id,
healthy=True,
checked_at=datetime.now(UTC),
)
@staticmethod
def _session_path(request: CreateSessionRequest) -> str:
relative = PurePosixPath(request.relative_path.replace("\\", "/"))
if (
relative.is_absolute()
or not relative.parts
or any(part in {"", ".", ".."} for part in relative.parts)
):
raise RuntimeProviderError("invalid Jupyter relative path")
return PurePosixPath(
request.workspace_code,
*relative.parts,
).as_posix()
async def create_session(
self,
request: CreateSessionRequest,
) -> RuntimeSession:
session_path = self._session_path(request)
encoded_path = quote(session_path, safe="/")
await self._request("GET", f"api/contents/{encoded_path}")
if session_path.lower().endswith(".ipynb"):
jupyter_url = (
f"{self.proxy_base_path}doc/tree/{encoded_path}"
)
else:
jupyter_url = (
f"{self.proxy_base_path}lab/tree/{encoded_path}"
)
if not session_path.lower().endswith(".ipynb"):
logical_id = hashlib.sha256(
f"{request.runtime_id}:{session_path}".encode("utf-8")
).hexdigest()[:32]
return RuntimeSession(
runtime_id=request.runtime_id,
session_id=f"file:{logical_id}",
jupyter_url=jupyter_url,
reused=True,
)
sessions_response = await self._request("GET", "api/sessions")
sessions = sessions_response.json()
for item in sessions:
if item.get("path") == session_path:
return RuntimeSession(
runtime_id=request.runtime_id,
session_id=str(item["id"]),
jupyter_url=jupyter_url,
reused=True,
)
created = (
await self._request(
"POST",
"api/sessions",
payload={
"path": session_path,
"name": "",
"type": "notebook",
"kernel": {"name": "python3"},
},
)
).json()
return RuntimeSession(
runtime_id=request.runtime_id,
session_id=str(created["id"]),
jupyter_url=jupyter_url,
reused=False,
)
async def terminate_session(
self,
runtime_id: str,
session_id: str,
) -> None:
if session_id.startswith("file:"):
return
await self._request(
"DELETE",
f"api/sessions/{quote(session_id, safe='')}",
allow_not_found=True,
)
runtime/src/runtime/redis_lock.py
-110
View File
@@ -1,110 +0,0 @@
from __future__ import annotations
import json
from typing import Any
from redis.asyncio import Redis
HEARTBEAT_SCRIPT = """
local raw = redis.call('GET', KEYS[1])
if not raw then
return 0
end
local ok, value = pcall(cjson.decode, raw)
if not ok then
return -2
end
if value['edit_session_id'] ~= ARGV[1]
or value['token_hash'] ~= ARGV[2] then
return -1
end
redis.call('PEXPIRE', KEYS[1], ARGV[3])
return 1
"""
RELEASE_SCRIPT = """
local raw = redis.call('GET', KEYS[1])
if not raw then
return 0
end
local ok, value = pcall(cjson.decode, raw)
if not ok then
return -2
end
if value['edit_session_id'] ~= ARGV[1]
or value['token_hash'] ~= ARGV[2] then
return -1
end
return redis.call('DEL', KEYS[1])
"""
def lock_key(workspace_id: str, storage_object_id: str) -> str:
return f"lock:file:{workspace_id}:{storage_object_id}"
async def acquire(
client: Redis,
*,
key: str,
value: dict[str, Any],
ttl_ms: int,
) -> bool:
encoded = json.dumps(
value,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
)
return bool(await client.set(key, encoded, nx=True, px=ttl_ms))
async def current(client: Redis, key: str) -> tuple[dict[str, Any] | None, int]:
raw = await client.get(key)
if raw is None:
return None, -2
try:
value = json.loads(raw)
except (TypeError, json.JSONDecodeError):
return None, await client.pttl(key)
return value, await client.pttl(key)
async def heartbeat(
client: Redis,
*,
key: str,
edit_session_id: str,
token_hash: str,
ttl_ms: int,
) -> int:
return int(
await client.eval(
HEARTBEAT_SCRIPT,
1,
key,
edit_session_id,
token_hash,
ttl_ms,
)
)
async def release(
client: Redis,
*,
key: str,
edit_session_id: str,
token_hash: str,
) -> int:
return int(
await client.eval(
RELEASE_SCRIPT,
1,
key,
edit_session_id,
token_hash,
)
)
runtime/src/runtime/runtime_lifecycle.py
-308
View File
@@ -1,308 +0,0 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from common.db.models import (
RuntimeInstances,
WorkspaceOperations,
Workspaces,
)
from common.ids import new_ulid
from contracts.runtime.runtime_adapter import (
CreateSessionRequest,
RuntimeAdapter,
RuntimeHealth,
RuntimeSession,
EnsureRuntimeRequest,
)
def utcnow() -> datetime:
return datetime.now(UTC).replace(tzinfo=None)
class RuntimeLifecycle:
def __init__(
self,
adapter: RuntimeAdapter,
*,
lease_seconds: int,
) -> None:
self.adapter = adapter
self.lease_seconds = lease_seconds
def _lease_expiry(self, now: datetime) -> datetime:
return now + timedelta(seconds=self.lease_seconds)
async def _record_operation(
self,
session: AsyncSession,
*,
workspace_id: str,
runtime_id: str,
user_id: str,
operation_type: str,
request_id: str | None,
status: str = "succeeded",
error_code: str | None = None,
error_message: str | None = None,
) -> None:
now = utcnow()
operation_request_id = (
f"runtime:{operation_type}:{request_id}"
if request_id
else None
)
if operation_request_id:
existing = await session.scalar(
select(WorkspaceOperations).where(
WorkspaceOperations.request_id == operation_request_id
)
)
if existing is not None:
return
session.add(
WorkspaceOperations(
operation_id=new_ulid(),
workspace_id=workspace_id,
runtime_id=runtime_id,
operation_type=operation_type,
operation_status=status,
state_version=1,
request_id=operation_request_id,
requested_by=user_id,
started_at=now,
finished_at=now,
error_code=error_code,
error_message=error_message,
)
)
async def ensure_running(
self,
session: AsyncSession,
*,
workspace: Workspaces,
user_id: str,
request_id: str | None,
) -> tuple[RuntimeInstances, bool]:
# Serializes ensure_running for one Workspace across stateless Runtime
# Manager replicas. Provider calls remain behind the adapter boundary.
await session.execute(
select(Workspaces.workspace_id)
.where(Workspaces.workspace_id == workspace.workspace_id)
.with_for_update()
)
now = utcnow()
existing = await session.scalar(
select(RuntimeInstances)
.where(
RuntimeInstances.workspace_id == workspace.workspace_id,
RuntimeInstances.runtime_type == "jupyter",
RuntimeInstances.desired_state == "running",
RuntimeInstances.actual_state.in_(
["provisioning", "starting", "running", "unhealthy"]
),
)
.order_by(RuntimeInstances.created_at.desc())
)
if existing is not None:
# Jupyter Server is scoped to the Workspace. The user who first
# starts it is still recorded in started_by and operation audit,
# while owner_user_id=None identifies a shared Workspace Runtime.
existing.owner_user_id = None
health = await self.adapter.health(existing.runtime_id)
if health.healthy:
existing.actual_state = "running"
existing.last_heartbeat_at = now
existing.lease_expires_at = self._lease_expiry(now)
existing.state_version += 1
existing.error_message = None
await self._record_operation(
session,
workspace_id=workspace.workspace_id,
runtime_id=existing.runtime_id,
user_id=user_id,
operation_type="open",
request_id=request_id,
)
return existing, True
existing.actual_state = "unhealthy"
existing.state_version += 1
existing.error_message = health.detail
runtime_id = new_ulid()
endpoint = await self.adapter.ensure_running(
EnsureRuntimeRequest(
runtime_id=runtime_id,
workspace_id=workspace.workspace_id,
workspace_code=workspace.workspace_code,
owner_user_id=user_id,
)
)
item = RuntimeInstances(
runtime_id=runtime_id,
workspace_id=workspace.workspace_id,
owner_user_id=None,
runtime_type=endpoint.runtime_type,
runtime_provider=endpoint.provider,
runtime_ref=endpoint.runtime_ref,
host_node="compose",
internal_url=endpoint.internal_url,
proxy_base_path=endpoint.proxy_base_path,
desired_state="running",
actual_state="running",
state_version=1,
started_by=user_id,
started_at=now,
last_heartbeat_at=now,
lease_expires_at=self._lease_expiry(now),
)
session.add(item)
await session.flush()
await self._record_operation(
session,
workspace_id=workspace.workspace_id,
runtime_id=item.runtime_id,
user_id=user_id,
operation_type="start",
request_id=request_id,
)
return item, False
async def health(
self,
item: RuntimeInstances,
) -> RuntimeHealth:
health = await self.adapter.health(item.runtime_id)
now = utcnow()
item.last_heartbeat_at = now
item.actual_state = "running" if health.healthy else "unhealthy"
item.error_message = health.detail
if health.healthy:
item.lease_expires_at = self._lease_expiry(now)
item.state_version += 1
return health
async def stop(
self,
session: AsyncSession,
item: RuntimeInstances,
*,
user_id: str,
request_id: str | None,
reason: str,
) -> None:
if item.actual_state == "stopped":
return
item.desired_state = "stopped"
item.actual_state = "stopping"
item.state_version += 1
await self.adapter.stop(item.runtime_id, reason)
item.actual_state = "stopped"
item.stopped_at = utcnow()
item.lease_expires_at = None
item.state_version += 1
await self._record_operation(
session,
workspace_id=item.workspace_id,
runtime_id=item.runtime_id,
user_id=user_id,
operation_type="stop",
request_id=request_id,
)
async def restart(
self,
session: AsyncSession,
item: RuntimeInstances,
*,
user_id: str,
request_id: str | None,
) -> None:
item.desired_state = "running"
item.actual_state = "starting"
item.state_version += 1
endpoint = await self.adapter.restart(item.runtime_id)
now = utcnow()
item.runtime_ref = endpoint.runtime_ref
item.internal_url = endpoint.internal_url
item.proxy_base_path = endpoint.proxy_base_path
item.actual_state = "running"
item.started_at = item.started_at or now
item.last_heartbeat_at = now
item.lease_expires_at = self._lease_expiry(now)
item.stopped_at = None
item.error_message = None
item.state_version += 1
await self._record_operation(
session,
workspace_id=item.workspace_id,
runtime_id=item.runtime_id,
user_id=user_id,
operation_type="restart",
request_id=request_id,
)
async def create_session(
self,
item: RuntimeInstances,
*,
workspace_code: str,
relative_path: str,
) -> RuntimeSession:
result = await self.adapter.create_session(
CreateSessionRequest(
runtime_id=item.runtime_id,
workspace_code=workspace_code,
relative_path=relative_path,
)
)
now = utcnow()
item.last_heartbeat_at = now
item.lease_expires_at = self._lease_expiry(now)
item.state_version += 1
return result
def touch(self, item: RuntimeInstances) -> None:
now = utcnow()
item.last_heartbeat_at = now
item.lease_expires_at = self._lease_expiry(now)
item.state_version += 1
async def terminate_session(
self,
runtime_id: str | None,
session_id: str | None,
) -> None:
if runtime_id and session_id:
await self.adapter.terminate_session(runtime_id, session_id)
def runtime_payload(item: RuntimeInstances) -> dict[str, Any]:
def iso(value: datetime | None) -> str | None:
if value is None:
return None
return value.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
return {
"runtime_id": item.runtime_id,
"workspace_id": item.workspace_id,
"owner_user_id": item.owner_user_id,
"runtime_type": item.runtime_type,
"runtime_provider": item.runtime_provider,
"runtime_ref": item.runtime_ref,
"internal_url": item.internal_url,
"proxy_base_path": item.proxy_base_path,
"desired_state": item.desired_state,
"actual_state": item.actual_state,
"state_version": item.state_version,
"started_at": iso(item.started_at),
"last_heartbeat_at": iso(item.last_heartbeat_at),
"lease_expires_at": iso(item.lease_expires_at),
"stopped_at": iso(item.stopped_at),
"error_message": item.error_message,
}
runtime/src/runtime/schemas.py
-40
View File
@@ -1,40 +0,0 @@
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class AcquireFileLockRequest(StrictModel):
workspace_id: str = Field(min_length=26, max_length=26)
storage_object_id: str = Field(min_length=26, max_length=26)
user_id: str = Field(min_length=26, max_length=26)
request_id: str | None = Field(default=None, min_length=1, max_length=64)
class FileLockTokenRequest(StrictModel):
workspace_id: str = Field(min_length=26, max_length=26)
user_id: str = Field(min_length=26, max_length=26)
lock_token: str = Field(min_length=32, max_length=256)
class RuntimeIdentityRequest(StrictModel):
workspace_id: str = Field(min_length=26, max_length=26)
user_id: str = Field(min_length=26, max_length=26)
request_id: str | None = Field(default=None, min_length=1, max_length=64)
class EnsureRuntimeApiRequest(RuntimeIdentityRequest):
pass
class StopRuntimeApiRequest(RuntimeIdentityRequest):
reason: str = Field(default="client_request", min_length=1, max_length=64)
class CreateRuntimeSessionApiRequest(RuntimeIdentityRequest):
storage_object_id: str = Field(min_length=26, max_length=26)
relative_path: str = Field(min_length=1, max_length=1024)