feat: add schedule package

2026-07-29 13:35:32 +08:00
parent 128aa7563f
commit 233dc90efe
10 changed files with 1038 additions and 830 deletions
@@ -3,6 +3,18 @@ FROM python:3.12-slim

 WORKDIR /app

+# 安装系统依赖（fuse3 是 rclone mount 的核心底层依赖）
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    fuse3 \
+    ca-certificates \
+    curl \
+    procps \
+    && sed -i 's/#user_allow_other/user_allow_other/g' /etc/fuse.conf \
+    && rm -rf /var/lib/apt/lists/*
+
+# 从官方 Rclone 镜像直接复制 rclone 二进制文件（高效、稳定）
+COPY --from=rclone/rclone:latest /usr/local/bin/rclone /usr/local/bin/rclone
+
 # 安装 uv
 COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv

@@ -15,6 +15,10 @@ dependencies = [
    "uvicorn>=0.51.0",
 ]

+[project.scripts]
+runtime = "runtime:main"
+
+
 [[tool.uv.index]]
 url = "https://pypi.tuna.tsinghua.edu.cn/simple/"
 default = true
@@ -19,8 +19,147 @@ from loguru import logger
 # 全局内存字典：记录运行中的 Jupyter 进程信息
 JUPYTER_PROCESSES: Dict[str, dict] = {}

-WORKSPACES_ROOT = os.getenv("WORKSPACES_ROOT", "/app/workspaces")
+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:
@@ -76,10 +215,9 @@ def start_process(cmd, workspace_path, log_dir="/tmp/process_logs"):


 def _handle_start(ws_id: str):
-    workspace_path = Path(f"./test/workspaces/{ws_id}")
-    workspace_path.mkdir(parents=True, exist_ok=True)
+    workspace_path = WORKSPACES_ROOT / ws_id

-    # 1. 如果已存在，校验进程状态并复用
+    # 如果已存在，校验进程状态并复用
    if ws_id in JUPYTER_PROCESSES:
        p_info = JUPYTER_PROCESSES[ws_id]
        if p_info["process"].poll() is None:
@@ -196,6 +334,7 @@ def _handle_list():
        }
    return {"status": "success", "instances": active_instances}

+
 def _handle_get(ws_id: str):
    """【新增】获取指定 Workspace 的 Jupyter 运行状态与完整 URL"""
    if ws_id not in JUPYTER_PROCESSES:
@@ -229,44 +368,39 @@ def _handle_get(ws_id: str):
 # ==================== FastAPI Lifespan 定义 ====================
@asynccontextmanager
 async def lifespan(app: FastAPI):
-    # ---------------- 1. 启动阶段 (Startup) ----------------
-    logger.info(
-        f"Initializing workspace processes from root: '{WORKSPACES_ROOT}'..."
-    )
-    os.makedirs(WORKSPACES_ROOT, exist_ok=True)
+    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")

-    # 遍历总路径下的每一个子目录（每个子目录名即 workspace_id）
-    for entry in os.listdir(WORKSPACES_ROOT):
-        full_path = os.path.join(WORKSPACES_ROOT, entry)
-        if os.path.isdir(full_path):
-            logger.info(f"Auto-starting Jupyter for workspace ID: '{entry}'")
+    # ==================== 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_start(entry)
+                _handle_stop(ws_id)
            except Exception as err:
-                logger.error(
-                    f"Startup failed for workspace '{entry}': {err}"
-                )
+                logger.error(f"Error terminating Jupyter process for '{ws_id}': {err}")
+        logger.info("All Jupyter sub-processes have been terminated.")
+        JUPYTER_PROCESSES.clear()

-    yield  # 服务在此处保持正常运行，等待 API 请求
-
-    # ---------------- 2. 停止阶段 (Shutdown) ----------------
-    logger.info(
-        "Service is shutting down. Terminating all active Jupyter sub-processes..."
-    )
-
-    # 复制 key 列表，防止字典在迭代清理过程中因大小改变报错
-    active_workspaces = list(JUPYTER_PROCESSES.keys())
-    for ws_id in active_workspaces:
+        # 卸载 Rclone 挂载点（强制将 VFS 缓存刷新同步至对象存储）
        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 clean terminated.")
+            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():