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

refactor: remove local fs, add config class to common pacakge

Browse Source
This commit is contained in:
tao.chen
2026-07-31 12:46:31 +08:00
parent 872388775a
commit f288c90d37
18 changed files with 192 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/pyproject.toml
+2 -1
View File
@@ -13,7 +13,8 @@ dependencies = [
]
[tool.uv.sources]
common = { path = "../common" }
common = { workspace = true }
[build-system]
requires = ["hatchling"]
backend/src/backend/jupyter.py
+2 -2
View File
@@ -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"
backend/src/backend/main.py
+8 -19
View File
@@ -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)
backend/src/backend/schedule_client.py
+13 -40
View File
@@ -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.
"""
backend/src/backend/schedule_runs.py
+3 -3
View File
@@ -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,
backend/src/backend/scripts.py
+116 -123
View File
@@ -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(
backend/src/backend/storage_api.py
+13 -98
View File
@@ -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(
common/pyproject.toml
+2
View File
@@ -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]
common/src/common/db/models/storage.py
+1 -1
View File
@@ -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)
common/src/common/service_app.py
+3 -2
View File
@@ -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()]
docker-compose.yml
+2 -5
View File
@@ -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
runtime/src/runtime/mount.py
+4 -3
View File
@@ -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
runtime/src/runtime/process.py
+2 -2
View File
@@ -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):
schedule/src/schedule/execution.py
+6 -12
View File
@@ -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}")
schedule/src/schedule/main.py
+4 -8
View File
@@ -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,
)
schedule/src/schedule/notebook_runner.py
+4 -3
View File
@@ -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
schedule/src/schedule/service.py
+7 -14
View File
@@ -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),
)
schedule/src/schedule/worker.py
-7
View File
@@ -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: