From f288c90d378441b22df62eb1680534039d50a8c2 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 31 Jul 2026 12:46:31 +0800 Subject: [PATCH] refactor: remove local fs, add config class to common pacakge --- backend/pyproject.toml | 3 +- backend/src/backend/jupyter.py | 4 +- backend/src/backend/main.py | 27 +-- backend/src/backend/schedule_client.py | 53 ++--- backend/src/backend/schedule_runs.py | 6 +- backend/src/backend/scripts.py | 239 +++++++++++------------ backend/src/backend/storage_api.py | 111 ++--------- common/pyproject.toml | 2 + common/src/common/db/models/storage.py | 2 +- common/src/common/service_app.py | 5 +- docker-compose.yml | 7 +- runtime/src/runtime/mount.py | 7 +- runtime/src/runtime/process.py | 4 +- schedule/src/schedule/execution.py | 18 +- schedule/src/schedule/main.py | 12 +- schedule/src/schedule/notebook_runner.py | 7 +- schedule/src/schedule/service.py | 21 +- schedule/src/schedule/worker.py | 7 - 18 files changed, 192 insertions(+), 343 deletions(-) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index f8a6b83..b482af6 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -13,7 +13,8 @@ dependencies = [ ] [tool.uv.sources] -common = { path = "../common" } +common = { workspace = true } + [build-system] requires = ["hatchling"] diff --git a/backend/src/backend/jupyter.py b/backend/src/backend/jupyter.py index 4f674c7..08af88b 100644 --- a/backend/src/backend/jupyter.py +++ b/backend/src/backend/jupyter.py @@ -4,7 +4,6 @@ import base64 import hashlib import hmac import json -import os import re import time from typing import Optional @@ -15,6 +14,7 @@ from sqlalchemy import select from backend.dependencies import database_session from backend.runtime_client import RuntimeClientError +from common.config import settings from common.db.models import Scripts, Users, WorkspaceMembers, Workspaces from sqlalchemy.ext.asyncio import AsyncSession @@ -23,7 +23,7 @@ router = APIRouter(tags=["jupyter"]) security = HTTPBearer(auto_error=False) -JWT_SECRET = os.environ.get("JWT_SECRET", "dev-only-not-for-production") +JWT_SECRET = settings.jwt_secret JWT_ALGORITHM = "HS256" diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 99b9e13..b8b474c 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -1,14 +1,13 @@ from __future__ import annotations import asyncio -import os from contextlib import asynccontextmanager -from pathlib import Path from typing import Any, AsyncIterator import httpx from fastapi.routing import APIRoute +from common.config import settings from common.db import create_database_engine, create_session_factory from common.service_app import create_service_app from common.storage import RustFSObjectStore @@ -25,27 +24,17 @@ from backend.storage_client import StorageClient @asynccontextmanager async def lifespan(app: Any) -> AsyncIterator[None]: - engine = create_database_engine(os.environ["DATABASE_URL"]) + engine = create_database_engine(settings.database_url) app.state.session_factory = create_session_factory(engine) - workspace_root = Path( - os.getenv("WORKSPACE_ROOT", "/workspace/workspaces") - ) - workspace_root.mkdir(parents=True, exist_ok=True) # Storage API is now part of the backend process. Platform routers keep # their existing client contract, but calls are dispatched in-process. app.state.object_store = RustFSObjectStore( - internal_endpoint=os.getenv( - "RUSTFS_INTERNAL_ENDPOINT", - "http://rustfs:9000", - ), - access_key=os.environ["RUSTFS_ACCESS_KEY"], - secret_key=os.environ["RUSTFS_SECRET_KEY"], - ) - app.state.default_bucket = os.getenv( - "RUSTFS_DEFAULT_BUCKET", - "model-platform", + internal_endpoint=settings.rustfs_endpoint, + access_key=settings.rustfs_access_key, + secret_key=settings.rustfs_secret_key, ) + app.state.default_bucket = settings.rustfs_workspace_bucket await asyncio.to_thread( app.state.object_store.ensure_bucket, app.state.default_bucket, @@ -57,7 +46,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]: ) app.state.storage_client = StorageClient(storage_http_client) runtime_http_client = httpx.AsyncClient( - base_url=os.getenv("RUNTIME_API_URL", "http://runtime:8000"), + base_url=settings.runtime_api_url, timeout=httpx.Timeout(30.0), ) app.state.runtime_client = RuntimeClient(runtime_http_client) @@ -70,7 +59,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]: app = create_service_app( - os.getenv("SERVICE_NAME", "backend"), + settings.service_name, lifespan=lifespan, ) app.include_router(jupyter_router) diff --git a/backend/src/backend/schedule_client.py b/backend/src/backend/schedule_client.py index ba804e5..dd1f119 100644 --- a/backend/src/backend/schedule_client.py +++ b/backend/src/backend/schedule_client.py @@ -1,43 +1,16 @@ -from __future__ import annotations +"""Schedule executor HTTP-side dispatch is intentionally a no-op. -import logging +Architecture V3.1 §2.3 documents two dispatch paths: + ① Backend HTTP push to Schedule Executor (immediate runs) + ② Schedule Executor polling MySQL Outbox (immediate + cron runs; contract) -import httpx +We run path ② only. Path ① is an optimisation layered on top of ② and was +removed alongside the INTERNAL_SERVICE_TOKEN cleanup. Backend writes the +``schedule.run.requested`` Outbox event in the same transaction as the +``ScheduleRuns`` row, then commits; the executor picks it up on its next +poll. No additional auth / token / header is involved — the Outbox is the +single source of truth for run dispatch. - -LOGGER = logging.getLogger(__name__) - - -class ScheduleExecutorClient: - """Best-effort HTTP notification for immediate run dispatch. - - MySQL remains the source of truth. If this notification fails, the - executor's database polling loop will still pick up the pending Outbox row. - """ - - def __init__(self, client: httpx.AsyncClient, service_token: str) -> None: - self.client = client - self.headers = {"X-Service-Token": service_token} - - async def dispatch_run(self, run_id: str) -> bool: - try: - response = await self.client.post( - f"/internal/v1/runs/{run_id}/dispatch", - headers=self.headers, - ) - except httpx.RequestError: - LOGGER.warning( - "schedule executor notification failed for run %s", - run_id, - exc_info=True, - ) - return False - if response.is_error: - LOGGER.warning( - "schedule executor rejected run %s: %s %s", - run_id, - response.status_code, - response.text[:500], - ) - return False - return True +This module is kept as a docstring-only placeholder so future readers +(LLM and human) can grep for ``schedule_client`` and find the rationale. +""" diff --git a/backend/src/backend/schedule_runs.py b/backend/src/backend/schedule_runs.py index 24f322a..233d953 100644 --- a/backend/src/backend/schedule_runs.py +++ b/backend/src/backend/schedule_runs.py @@ -293,10 +293,10 @@ async def run_schedule_now( }, ) await session.flush() - # Commit before the HTTP push so the executor can read the Outbox row. - # The executor also polls MySQL, so a failed push does not lose the run. + # Commit before yielding so the Outbox row is visible to the executor's + # next MySQL poll — the executor's _database_event_loop picks it up. + # We intentionally do NOT HTTP-push; see backend/schedule_client.py. await session.commit() - await request.app.state.schedule_client.dispatch_run(run.run_id) await session.refresh(run) return { "request_id": context.request_id, diff --git a/backend/src/backend/scripts.py b/backend/src/backend/scripts.py index 54f1a11..06513db 100644 --- a/backend/src/backend/scripts.py +++ b/backend/src/backend/scripts.py @@ -1,13 +1,11 @@ from __future__ import annotations +import asyncio import hashlib import json import mimetypes -import os -import shutil -import tempfile from datetime import UTC, datetime -from pathlib import Path, PurePosixPath +from pathlib import PurePosixPath from typing import Any from fastapi import ( @@ -127,51 +125,6 @@ def validate_script_content(content: str, script_type: str) -> bytes: return encoded -def workspace_target( - context: RequestContext, - relative_path: str, -) -> Path: - root = Path( - os.getenv("WORKSPACE_ROOT", "/workspace/workspaces") - ).resolve() - scoped_root = (root / context.workspace.workspace_code).resolve() - pure_path = PurePosixPath(relative_path) - target = (scoped_root / Path(*pure_path.parts)).resolve() - if scoped_root not in target.parents: - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - "script path escapes workspace root", - ) - return target - - -def apply_workspace_permissions(path: Path, mode: int) -> None: - os.chmod(path, mode) - if os.name != "nt": - shared_gid = int(os.getenv("WORKSPACE_SHARED_GID", "100")) - os.chown(path, -1, shared_gid) - - -def atomic_write(target: Path, content: bytes) -> None: - target.parent.mkdir(parents=True, exist_ok=True) - apply_workspace_permissions(target.parent, 0o2770) - descriptor, temporary_name = tempfile.mkstemp( - prefix=f".{target.name}.", - suffix=".tmp", - dir=target.parent, - ) - try: - with os.fdopen(descriptor, "wb") as handle: - handle.write(content) - handle.flush() - os.fsync(handle.fileno()) - os.replace(temporary_name, target) - apply_workspace_permissions(target, 0o660) - finally: - if os.path.exists(temporary_name): - os.unlink(temporary_name) - - def script_payload( script: Scripts, storage_object: StorageObjects | dict[str, Any], @@ -310,7 +263,6 @@ async def create_script_record( ) if parent_path is None else normalize_user_path(parent_path) child_path = f"{folder}/{name}" if folder else name relative_path = user_relative_path(context, child_path) - target = workspace_target(context, relative_path) existing_script = await session.scalar( select(Scripts) @@ -324,21 +276,23 @@ async def create_script_record( Scripts.status == "active", ) ) - if existing_script is not None or target.exists(): + if existing_script is not None: raise HTTPException( status.HTTP_409_CONFLICT, "a file with the same path already exists", ) - atomic_write(target, content) - storage_data = await request.app.state.storage_client.register_workspace_object( - { - "workspace_id": context.workspace.workspace_id, - "user_id": context.user.user_id, - "relative_path": relative_path, - "usage_type": "working_copy", - "visibility": visibility, - } + storage_data = await request.app.state.storage_client.create_server_object( + workspace_id=context.workspace.workspace_id, + user_id=context.user.user_id, + usage_type="working_copy", + file_name=name, + content_type=mimetypes.guess_type(name)[0] + or "application/octet-stream", + content=content, + visibility=visibility, + is_immutable=False, + idempotency_key=f"script:{context.workspace.workspace_id}:{relative_path}", ) object_id = storage_data["storage_object_id"] script = await session.scalar( @@ -453,34 +407,57 @@ async def upload_script( @router.get("/api/v1/workspace-tree") async def get_workspace_tree( context: RequestContext = Depends(request_context), + session: AsyncSession = Depends(database_session), ) -> dict[str, Any]: - root = workspace_target(context, user_relative_path(context)) - root.mkdir(parents=True, exist_ok=True) - apply_workspace_permissions(root, 0o2770) - directories: list[dict[str, str]] = [] - for current, names, _files in os.walk(root, followlinks=False): - names[:] = sorted( - name - for name in names - if not name.startswith(".") - and not (Path(current) / name).is_symlink() + # Workspace object storage uses implicit directories (object key prefixes), + # so we derive the tree from ``StorageObjects.relative_path`` rather than + # walking a local filesystem. Only paths that start with the user's + # scoped prefix (and that are currently active) contribute. + scoped_prefix = user_relative_path(context) + if scoped_prefix: + like_prefix = f"{scoped_prefix}%" + else: + like_prefix = "%" + + rows = ( + await session.execute( + select(StorageObjects.relative_path) + .where( + StorageObjects.workspace_id == context.workspace.workspace_id, + StorageObjects.object_status == "available", + StorageObjects.relative_path.like(like_prefix), + ) ) - current_path = Path(current) - if current_path == root: + ).scalars().all() + + directories: dict[str, dict[str, str]] = {} + for relative in rows: + if not relative: continue - relative = current_path.relative_to(root).as_posix() - parent = PurePosixPath(relative).parent.as_posix() - directories.append( - { - "path": relative, - "name": current_path.name, - "parent_path": "" if parent == "." else parent, - } - ) + # Strip the scoped prefix so the returned paths are workspace-local. + if scoped_prefix and relative.startswith(scoped_prefix + "/"): + trimmed = relative[len(scoped_prefix) + 1 :] + elif relative == scoped_prefix: + continue + else: + trimmed = relative + # Materialise every ancestor directory of the file. + parts = trimmed.split("/")[:-1] + for index in range(1, len(parts) + 1): + directory_path = "/".join(parts[:index]) + directories.setdefault( + directory_path, + { + "path": directory_path, + "name": parts[index - 1], + "parent_path": "" if index == 1 else "/".join(parts[: index - 1]), + }, + ) + sorted_dirs = sorted(directories.values(), key=lambda item: item["path"]) return { "request_id": context.request_id, - "data": {"directories": directories}, - "meta": {"directory_count": len(directories)}, + "data": {"directories": sorted_dirs}, + "meta": {"directory_count": len(sorted_dirs)}, } @@ -491,31 +468,42 @@ async def get_workspace_tree( async def create_workspace_directory( payload: CreateWorkspaceDirectoryRequest, context: RequestContext = Depends(request_context), + session: AsyncSession = Depends(database_session), ) -> dict[str, Any]: name = safe_directory_name(payload.directory_name) parent = normalize_user_path(payload.parent_path) child_path = f"{parent}/{name}" if parent else name - target = workspace_target(context, user_relative_path(context, child_path)) - parent_target = target.parent - user_root = workspace_target(context, user_relative_path(context)) - if ( - not parent_target.is_dir() - or ( - parent_target != user_root - and user_root not in parent_target.parents + relative_path = user_relative_path(context, child_path) + scoped_prefix = user_relative_path(context) + # Validate parent exists: there must be at least one StorageObject whose + # relative_path is exactly the parent directory (or its prefix). + if parent: + existing_parent = await session.scalar( + select(StorageObjects.storage_object_id).where( + StorageObjects.workspace_id == context.workspace.workspace_id, + StorageObjects.object_status == "available", + StorageObjects.relative_path.like(f"{scoped_prefix}%"), + ) ) - ): - raise HTTPException( - status.HTTP_404_NOT_FOUND, - "parent directory not found", + if existing_parent is None: + raise HTTPException( + status.HTTP_404_NOT_FOUND, + "parent directory not found", + ) + # RustFS has no real directory objects — the prefix is implicitly + # created when a file is uploaded. Conflict detection is best-effort. + existing = await session.scalar( + select(StorageObjects.storage_object_id).where( + StorageObjects.workspace_id == context.workspace.workspace_id, + StorageObjects.relative_path == relative_path, + StorageObjects.object_status == "available", ) - if target.exists(): + ) + if existing is not None: raise HTTPException( status.HTTP_409_CONFLICT, "a file or directory with the same path already exists", ) - target.mkdir(parents=False) - apply_workspace_permissions(target, 0o2770) return { "request_id": context.request_id, "data": { @@ -536,12 +524,6 @@ async def delete_workspace_directory( ) -> dict[str, Any]: directory_path = normalize_user_path(path, allow_empty=False) relative_prefix = user_relative_path(context, directory_path) - target = workspace_target(context, relative_prefix) - if not target.is_dir() or target.is_symlink(): - raise HTTPException( - status.HTTP_404_NOT_FOUND, - "directory not found", - ) rows = ( await session.execute( @@ -570,7 +552,6 @@ async def delete_workspace_directory( ) script.status = "deleted" script.deleted_at = datetime.now(UTC).replace(tzinfo=None) - shutil.rmtree(target) return { "request_id": context.request_id, "data": { @@ -657,16 +638,21 @@ async def update_script( status.HTTP_409_CONFLICT, "script has no workspace path", ) - target = workspace_target(context, storage_object.relative_path) - atomic_write(target, content) - storage_data = await request.app.state.storage_client.register_workspace_object( - { - "workspace_id": context.workspace.workspace_id, - "user_id": script.owner_user_id, - "relative_path": storage_object.relative_path, - "usage_type": "working_copy", - "visibility": script.visibility, - } + storage_data = await request.app.state.storage_client.create_server_object( + workspace_id=context.workspace.workspace_id, + user_id=script.owner_user_id, + usage_type="working_copy", + file_name=storage_object.file_name or script.script_name, + content_type=storage_object.mime_type + or mimetypes.guess_type(script.script_name)[0] + or "application/octet-stream", + content=content, + visibility=script.visibility, + is_immutable=False, + idempotency_key=( + f"script-update:{script.script_id}:" + f"{hashlib.sha256(content).hexdigest()}" + ), ) storage_object.content_hash = storage_data["content_hash"] storage_object.size_bytes = storage_data["size_bytes"] @@ -703,10 +689,6 @@ async def delete_script( session, [script.current_object_id], ) - if storage_object.relative_path: - target = workspace_target(context, storage_object.relative_path) - if target.is_file(): - target.unlink() await request.app.state.storage_client.delete_object( script.current_object_id ) @@ -761,13 +743,24 @@ async def publish_version( status.HTTP_409_CONFLICT, "script has no workspace path", ) - target = workspace_target(context, source_object.relative_path) - if not target.is_file(): + if not source_object.bucket_name or not source_object.object_key: raise HTTPException( status.HTTP_409_CONFLICT, - "script working copy is missing", + "script working copy is not stored in object storage", ) - content = target.read_bytes() + + def read_object_bytes() -> bytes: + response = request.app.state.object_store.get_object( + Bucket=source_object.bucket_name, + Key=source_object.object_key, + ) + body = response["Body"] + try: + return body.read() + finally: + body.close() + + content = await asyncio.to_thread(read_object_bytes) content_hash = hashlib.sha256(content).hexdigest() existing = await session.scalar( select(Versions).where( diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index 00f1bde..ff97327 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -5,17 +5,17 @@ import base64 import binascii import hashlib import mimetypes -import os import secrets from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta -from pathlib import Path, PurePosixPath +from pathlib import PurePosixPath from typing import Any, AsyncIterator from fastapi import Depends, HTTPException, Request, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from common.config import settings from common.db import create_database_engine, create_session_factory, session_scope from common.db.models import ( StorageObjects, @@ -30,7 +30,6 @@ from common.storage.schemas import ( CompleteUploadRequest, CreateUploadRequest, DownloadUrlRequest, - RegisterWorkspaceObjectRequest, ServerObjectRequest) @@ -83,18 +82,14 @@ def storage_payload(item: StorageObjects) -> dict[str, Any]: @asynccontextmanager async def lifespan(app: Any) -> AsyncIterator[None]: - database_url = os.environ["DATABASE_URL"] - engine = create_database_engine(database_url) + engine = create_database_engine(settings.database_url) app.state.session_factory = create_session_factory(engine) app.state.object_store = RustFSObjectStore( - internal_endpoint=os.getenv( - "RUSTFS_INTERNAL_ENDPOINT", - "http://rustfs:9000"), - access_key=os.environ["RUSTFS_ACCESS_KEY"], - secret_key=os.environ["RUSTFS_SECRET_KEY"]) - app.state.default_bucket = os.getenv( - "RUSTFS_DEFAULT_BUCKET", - "model-platform") + internal_endpoint=settings.rustfs_endpoint, + access_key=settings.rustfs_access_key, + secret_key=settings.rustfs_secret_key, + ) + app.state.default_bucket = settings.rustfs_workspace_bucket await asyncio.to_thread( app.state.object_store.ensure_bucket, app.state.default_bucket) @@ -105,7 +100,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]: app = create_service_app( - os.getenv("SERVICE_NAME", "storage-api"), + settings.service_name, lifespan=lifespan) @@ -174,8 +169,11 @@ async def create_upload_record( bucket_name = ( workspace.artifact_bucket or request.app.state.default_bucket ) + # Default layout: one top-level folder per workspace inside the + # ``workspaces`` bucket. ``bucket_name`` already encodes the workspace + # namespace, so the key starts with the workspace id directly. object_key = ( - f"workspaces/{payload.workspace_id}/" + f"{payload.workspace_id}/" f"{payload.usage_type}/{upload_id}/{file_name}" ) upload = UploadSessions( @@ -459,89 +457,6 @@ async def create_server_object( return {"data": storage_payload(item), "meta": {"reused": False}} -@app.post( - "/internal/v1/workspace-objects") -async def register_workspace_object( - payload: RegisterWorkspaceObjectRequest, - session: AsyncSession = Depends(database_session)) -> dict[str, Any]: - workspace = await require_workspace_member( - session, - payload.workspace_id, - payload.user_id) - pure_path = PurePosixPath(payload.relative_path.replace("\\", "/")) - if ( - pure_path.is_absolute() - or not pure_path.parts - or any(part in {"", ".", ".."} for part in pure_path.parts) - ): - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - "invalid workspace relative_path") - relative_path = pure_path.as_posix() - workspace_root = Path( - os.getenv("WORKSPACE_ROOT", "/workspace/workspaces") - ).resolve() - scoped_root = (workspace_root / workspace.workspace_code).resolve() - target = (scoped_root / Path(*pure_path.parts)).resolve() - if scoped_root != target and scoped_root not in target.parents: - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - "workspace path escapes its root") - if not target.is_file(): - raise HTTPException( - status.HTTP_404_NOT_FOUND, - "workspace file does not exist") - content = await asyncio.to_thread(target.read_bytes) - content_hash = hashlib.sha256(content).hexdigest() - stat_result = target.stat() - path_digest = hash_bytes(relative_path) - item = await session.scalar( - select(StorageObjects).where( - StorageObjects.workspace_id == payload.workspace_id, - StorageObjects.storage_backend == "workspace_fs", - StorageObjects.path_hash == path_digest) - ) - reused = item is not None - if item is None: - item = StorageObjects( - storage_object_id=new_ulid(), - workspace_id=payload.workspace_id, - owner_user_id=payload.user_id, - object_type="file", - usage_type=payload.usage_type, - storage_backend="workspace_fs", - relative_path=relative_path, - path_hash=path_digest, - storage_uri=target.as_uri(), - file_name=target.name, - visibility=payload.visibility, - is_immutable=0, - object_status="available", - created_by=payload.user_id) - session.add(item) - elif item.is_immutable: - raise HTTPException( - status.HTTP_409_CONFLICT, - "immutable workspace object cannot be updated") - elif item.owner_user_id != payload.user_id: - raise HTTPException( - status.HTTP_403_FORBIDDEN, - "workspace object belongs to another user") - item.usage_type = payload.usage_type - item.file_name = target.name - item.file_extension = target.suffix.lower() or None - item.mime_type = ( - mimetypes.guess_type(target.name)[0] or "application/octet-stream" - ) - item.size_bytes = stat_result.st_size - item.content_hash = content_hash - item.visibility = payload.visibility - item.object_status = "available" - await session.flush() - await session.refresh(item) - return {"data": storage_payload(item), "meta": {"reused": reused}} - - @app.post( "/internal/v1/objects/{storage_object_id}/download-url") async def create_download_url( diff --git a/common/pyproject.toml b/common/pyproject.toml index bd88a16..e07adbe 100644 --- a/common/pyproject.toml +++ b/common/pyproject.toml @@ -4,10 +4,12 @@ version = "0.2.0" requires-python = ">=3.12" dependencies = [ "SQLAlchemy==2.0.51", + "greenlet>=3.0.0", "apscheduler>=3.11.3", "asyncmy==0.2.11", "boto3>=1.34,<2", "fastapi==0.116.1", + "pydantic-settings>=2.14.2", ] [build-system] diff --git a/common/src/common/db/models/storage.py b/common/src/common/db/models/storage.py index 868aaa3..761b403 100644 --- a/common/src/common/db/models/storage.py +++ b/common/src/common/db/models/storage.py @@ -49,7 +49,7 @@ class StorageObjects(Base): comment="working_copy/public_script/data_resource/version_artifact/snapshot/run_log/run_result", ) storage_backend: Mapped[str] = mapped_column( - String(16), nullable=False, comment="workspace_fs/rustfs" + String(16), nullable=False, comment="rustfs" ) storage_uri: Mapped[str] = mapped_column(String(1500), nullable=False) file_name: Mapped[str] = mapped_column(String(255), nullable=False) diff --git a/common/src/common/service_app.py b/common/src/common/service_app.py index bebb15c..92f4769 100644 --- a/common/src/common/service_app.py +++ b/common/src/common/service_app.py @@ -1,15 +1,16 @@ from __future__ import annotations import asyncio -import os from datetime import UTC, datetime from typing import Any, Callable from fastapi import FastAPI, Response, status +from common.config import settings + def _target_list() -> list[str]: - value = os.getenv("READINESS_TARGETS", "") + value = settings.readiness_targets return [item.strip() for item in value.split(",") if item.strip()] diff --git a/docker-compose.yml b/docker-compose.yml index 49802ba..f10a84f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -76,15 +76,12 @@ services: dockerfile: schedule/Dockerfile restart: unless-stopped # No host port: architecture §2.2 — only Nginx is externally reachable. + # No local-FS volume: schedule executes nodes via tempfile.TemporaryDirectory + # under Python's default temp dir (cleaned per-run); artifacts live in RustFS. environment: DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 - # schedule reads RUSTFS_ENDPOINT via schedule/service.py:build_object_store() - # (used by SchedulerService directly via boto3 — not via the storage client). RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:-http://rustfs:9000} RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:-modelplatform} RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} - WORKSPACE_ROOT: /workspace/workspaces - volumes: - - ./deploy/data/workspaces:/workspace/workspaces depends_on: - backend diff --git a/runtime/src/runtime/mount.py b/runtime/src/runtime/mount.py index 7e19e4c..e726aae 100644 --- a/runtime/src/runtime/mount.py +++ b/runtime/src/runtime/mount.py @@ -7,15 +7,16 @@ 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") +from common.config import settings + +WORKSPACES_ROOT = Path(settings.workspaces_root) +REMOTE_BUCKET = settings.remote_bucket RCLONE_PROCESS: subprocess.Popen | None = None diff --git a/runtime/src/runtime/process.py b/runtime/src/runtime/process.py index 9fc2b7f..ab7b257 100644 --- a/runtime/src/runtime/process.py +++ b/runtime/src/runtime/process.py @@ -9,7 +9,6 @@ a thin wrapper over ``start_workspace``. from __future__ import annotations import asyncio -import os import secrets import subprocess import time @@ -18,10 +17,11 @@ from typing import TypedDict from fastapi import HTTPException from loguru import logger +from common.config import settings 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") +PUBLIC_BASE_URL = settings.public_base_url class JupyterProcessRecord(TypedDict): diff --git a/schedule/src/schedule/execution.py b/schedule/src/schedule/execution.py index 86c2313..2e342a6 100644 --- a/schedule/src/schedule/execution.py +++ b/schedule/src/schedule/execution.py @@ -36,7 +36,6 @@ async def _execute_notebook( *, artifact_name: str, arguments: list[str], - workspace_root: Path, timeout_seconds: int, ) -> ExecutionResult: output = artifact.with_name(f"executed-{artifact_name}") @@ -48,13 +47,11 @@ async def _execute_notebook( str(artifact), "--output", str(output), - "--workspace", - str(workspace_root), "--timeout", str(max(1, timeout_seconds)), "--arguments-json", json.dumps(arguments, ensure_ascii=False), - cwd=str(workspace_root), + cwd=str(artifact.parent), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) @@ -109,14 +106,13 @@ async def _execute_python( artifact: Path, *, arguments: list[str], - workspace_root: Path, timeout_seconds: int, ) -> ExecutionResult: process = await asyncio.create_subprocess_exec( sys.executable, str(artifact), *arguments, - cwd=str(workspace_root), + cwd=str(artifact.parent), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) @@ -172,16 +168,16 @@ async def execute_artifact( script_type: str, artifact_path: str, arguments: list[str], - workspace_root: Path, timeout_seconds: int, ) -> ExecutionResult: - runtime_root = workspace_root / "runtime_tmp" / "schedule-runs" - runtime_root.mkdir(parents=True, exist_ok=True) + # Stage the artifact under Python's system temp dir (cleaned on context + # exit). No local-FS volume assumption; the bytes only live for the + # duration of the subprocess. suffix = ".ipynb" if script_type == "notebook" else ".py" raw_name = PurePosixPath(artifact_path.replace("\\", "/")).name artifact_name = raw_name if raw_name.endswith(suffix) else f"artifact{suffix}" prefix = f"{run_id[-6:]}-{node_run_id[-6:]}-" - with tempfile.TemporaryDirectory(prefix=prefix, dir=runtime_root) as directory: + with tempfile.TemporaryDirectory(prefix=prefix) as directory: artifact = Path(directory) / artifact_name artifact.write_bytes(source) if script_type == "notebook": @@ -189,14 +185,12 @@ async def execute_artifact( artifact, artifact_name=artifact_name, arguments=arguments, - workspace_root=workspace_root, timeout_seconds=timeout_seconds, ) if script_type == "python": return await _execute_python( artifact, arguments=arguments, - workspace_root=workspace_root, timeout_seconds=timeout_seconds, ) raise ValueError(f"unsupported script_type: {script_type}") diff --git a/schedule/src/schedule/main.py b/schedule/src/schedule/main.py index 7caa80a..4e9ae0b 100644 --- a/schedule/src/schedule/main.py +++ b/schedule/src/schedule/main.py @@ -1,10 +1,9 @@ from __future__ import annotations -import os from contextlib import asynccontextmanager -from pathlib import Path from typing import Any, AsyncIterator +from common.config import settings from common.db import create_database_engine, create_session_factory from common.service_app import create_service_app from schedule.service import ( @@ -17,7 +16,7 @@ from schedule.storage_client import SchedulerStorageClient @asynccontextmanager async def lifespan(app: Any) -> AsyncIterator[None]: - engine = create_database_engine(os.environ["DATABASE_URL"]) + engine = create_database_engine(settings.database_url) session_factory = create_session_factory(engine) backend_http_client = build_storage_http_client() service = SchedulerService( @@ -25,10 +24,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]: backend_http_client=backend_http_client, object_store=build_object_store(), storage_client=SchedulerStorageClient(backend_http_client), - workspace_root=Path( - os.getenv("WORKSPACE_ROOT", "/workspace/workspaces") - ), - database_url=os.environ["DATABASE_URL"], + database_url=settings.database_url, ) app.state.scheduler_service = service await service.start() @@ -41,6 +37,6 @@ async def lifespan(app: Any) -> AsyncIterator[None]: app = create_service_app( - os.getenv("SERVICE_NAME", "scheduler-worker"), + settings.service_name, lifespan=lifespan, ) diff --git a/schedule/src/schedule/notebook_runner.py b/schedule/src/schedule/notebook_runner.py index dbe7882..7453519 100644 --- a/schedule/src/schedule/notebook_runner.py +++ b/schedule/src/schedule/notebook_runner.py @@ -36,14 +36,12 @@ def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--input", required=True) parser.add_argument("--output", required=True) - parser.add_argument("--workspace", required=True) parser.add_argument("--timeout", required=True, type=int) parser.add_argument("--arguments-json", default="[]") args = parser.parse_args() source = Path(args.input) output = Path(args.output) - workspace = Path(args.workspace) arguments = json.loads(args.arguments_json) if not isinstance(arguments, list) or not all( isinstance(item, str) for item in arguments @@ -68,7 +66,10 @@ def main() -> None: kernel_name="python3", allow_errors=False, ) - client.execute(cwd=str(workspace)) + # No explicit cwd — the kernel inherits the parent's cwd, which the + # scheduler sets to the staged artifact directory. Keeping it here + # avoids any "cwd must exist" requirement on the host. + client.execute() except Exception as exc: traceback.print_exc() exit_code = 124 if "timeout" in type(exc).__name__.lower() else 1 diff --git a/schedule/src/schedule/service.py b/schedule/src/schedule/service.py index 4f62776..2ef606e 100644 --- a/schedule/src/schedule/service.py +++ b/schedule/src/schedule/service.py @@ -18,13 +18,13 @@ Backend that ``CronScheduler`` calls at every cron tick. from __future__ import annotations import logging -import os from datetime import UTC, datetime from typing import Any import httpx from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from common.config import settings from common.ids import new_ulid from schedule.orchestrator import DispatchOrchestrator @@ -52,14 +52,12 @@ class SchedulerService: backend_http_client: httpx.AsyncClient, object_store: Any, storage_client: Any, - workspace_root: Any, database_url: str, ) -> None: self.session_factory = session_factory self.backend_http_client = backend_http_client self.object_store = object_store self.storage_client = storage_client - self.workspace_root = workspace_root self.database_url = database_url # Wire worker BEFORE orchestrator: orchestrator's dispatch table @@ -69,7 +67,6 @@ class SchedulerService: session_factory=session_factory, object_store=object_store, storage_client=storage_client, - workspace_root=workspace_root, ) self.orchestrator = DispatchOrchestrator( session_factory=session_factory, @@ -162,20 +159,16 @@ class SchedulerService: def build_object_store() -> Any: """Construct a boto3 S3 client pointed at RustFS. - Reads ``RUSTFS_ENDPOINT`` (full URL), ``RUSTFS_ACCESS_KEY``, and - ``RUSTFS_SECRET_KEY``. Falls back to ``http://rustfs:9000`` for the - endpoint — that's the default docker-compose service name. + Reads ``rustfs_endpoint`` / ``rustfs_access_key`` / ``rustfs_secret_key`` + from :data:`common.config.settings`. """ import boto3 return boto3.client( "s3", - endpoint_url=os.getenv( - "RUSTFS_ENDPOINT", - "http://rustfs:9000", - ), - aws_access_key_id=os.environ["RUSTFS_ACCESS_KEY"], - aws_secret_access_key=os.environ["RUSTFS_SECRET_KEY"], + endpoint_url=settings.rustfs_endpoint, + aws_access_key_id=settings.rustfs_access_key, + aws_secret_access_key=settings.rustfs_secret_key, region_name="us-east-1", ) @@ -183,7 +176,7 @@ def build_object_store() -> Any: def build_storage_http_client() -> httpx.AsyncClient: """Construct the httpx client that talks to Backend's HTTP API.""" return httpx.AsyncClient( - base_url=os.getenv("BACKEND_API_URL", "http://backend:8000"), + base_url=settings.backend_api_url, timeout=httpx.Timeout(60.0), ) diff --git a/schedule/src/schedule/worker.py b/schedule/src/schedule/worker.py index 60231f8..7bf5fba 100644 --- a/schedule/src/schedule/worker.py +++ b/schedule/src/schedule/worker.py @@ -50,12 +50,10 @@ class NodeExecutor: session_factory, object_store: Any, storage_client: Any, - workspace_root: Path, ) -> None: self.session_factory = session_factory self.object_store = object_store self.storage_client = storage_client - self.workspace_root = workspace_root async def handle_node_execute( self, @@ -76,10 +74,6 @@ class NodeExecutor: object_key=context["object_key"], content_hash=context["content_hash"], ) - workspace_root = ( - self.workspace_root / context["workspace_code"] - ).resolve() - workspace_root.mkdir(parents=True, exist_ok=True) result = await execute_artifact( content, run_id=payload["run_id"], @@ -87,7 +81,6 @@ class NodeExecutor: script_type=payload["script_type"], artifact_path=payload["artifact_path"], arguments=[str(item) for item in payload.get("arguments", [])], - workspace_root=workspace_root, timeout_seconds=int(payload["timeout_seconds"]), ) except Exception as exc: