diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index 466d672..4483364 100644 --- a/backend/app/api/v1/endpoints/platform.py +++ b/backend/app/api/v1/endpoints/platform.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import asyncio +import hashlib import uuid from datetime import datetime, timedelta, timezone from pathlib import Path @@ -16,6 +18,7 @@ from app.core.config import get_settings from app.db.platform_store import get_platform_store from app.modules.compute_gateway.client import ComputeNodeClient from app.modules.compute_gateway.sync import fetch_eval_result_content, poll_compute_jobs_once +from app.modules.storage.minio_store import ObjectStorageError, get_object_storage router = APIRouter() @@ -33,6 +36,22 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None: return None +async def _wait_for_object_storage() -> None: + """Wait for MinIO before starting a resource task.""" + settings = get_settings() + deadline = datetime.now(timezone.utc).timestamp() + max(0, settings.storage_wait_seconds) + last_error = "MinIO unavailable" + while True: + try: + get_object_storage().ensure_bucket() + return + except Exception as exc: # noqa: BLE001 - retry until the configured deadline + last_error = str(exc) + if datetime.now(timezone.utc).timestamp() >= deadline: + raise RuntimeError(f"MinIO unavailable after {settings.storage_wait_seconds}s: {last_error}") + await asyncio.sleep(max(1, settings.storage_check_interval_seconds)) + + def _select_eval_node(store: Any, preferred_node_id: str | None = None) -> dict[str, Any] | None: """Select the compute node for an eval job. @@ -330,6 +349,11 @@ async def _fine_tune_preflight_with_job_payload( ) except Exception as exc: # noqa: BLE001 - return as preflight error for page visibility sync_errors.append(str(exc)) + if get_settings().minio_enabled and get_settings().compute_mode != "simulator": + try: + await _wait_for_object_storage() + except Exception as exc: # noqa: BLE001 - preflight exposes node storage failure + sync_errors.append(f"shared storage health check failed: {exc}") if get_settings().compute_mode == "simulator": preview = { "valid": True, @@ -877,6 +901,11 @@ async def merge_model(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: raise fail(400, "adapter_path is required") requested_node_id = payload.get("requested_node_id") or payload.get("compute_node_id") or (trained_model and trained_model.get("compute_node_id")) node = store.schedule_node({**payload, "requested_node_id": requested_node_id, "gpus": payload.get("gpus") or []}) + if get_settings().minio_enabled and get_settings().compute_mode != "simulator": + try: + await _wait_for_object_storage() + except RuntimeError as exc: + raise fail(503, str(exc)) health = node.get("health_detail") or {} output_root = str(health.get("output_root") or f"{node['data_root'].rstrip('/')}/outputs") output_name = str(payload.get("output_model_name") or payload.get("merged_model_name") or f"{trained_model_id or 'model'}-merged") @@ -981,7 +1010,7 @@ async def _sync_dataset_file_to_compute_nodes( content: bytes, ) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] - if get_settings().compute_mode == "simulator": + if get_settings().compute_mode == "simulator" or get_settings().minio_enabled: return results target_name = Path(filename or f"{file_id}.jsonl").name target_relative_path = f"datasets/{dataset_id}/{target_name}" @@ -1029,6 +1058,27 @@ async def _sync_training_dataset_to_compute_node( node: dict[str, Any], dataset_id: str, ) -> list[dict[str, Any]]: + if get_settings().minio_enabled: + files = store.training_dataset_files(dataset_id) + objects = store.storage_objects_for_resource("dataset", dataset_id) + object_by_name = {Path(str(item.get("file_name") or item.get("object_key") or "")).name: item for item in objects} + results: list[dict[str, Any]] = [] + client = ComputeNodeClient(node["api_base_url"]) + for item in files: + target_name = Path(str(item.get("name") or f"{item['id']}.jsonl")).name + obj = object_by_name.get(target_name) + if not obj: + raise RuntimeError(f"dataset file is not available in MinIO: {target_name}") + url = get_object_storage().presigned_get(obj["object_key"]) + result = await client.prepare_cache({ + "resource_id": dataset_id, + "version_id": obj["version_id"], + "download_url": url, + "checksum_sha256": obj.get("checksum_sha256") or "", + "relative_path": f"datasets/{dataset_id}/{target_name}", + }) + results.append({**result, "file_id": item.get("id"), "name": target_name, "node_id": node["id"]}) + return results if not dataset_id: raise RuntimeError("train_dataset_id is required") files = store.training_dataset_files(dataset_id) @@ -1092,6 +1142,17 @@ async def upload_dataset_files( created_file = store.add_dataset_file(conn, dataset_id, file.filename or "upload.jsonl", content) created.append(created_file) pending_sync.append((created_file["id"], created_file["name"], raw)) + if get_settings().minio_enabled: + object_key = f"datasets/{dataset_id}/versions/{created_file.get('active_version_id') or created_file['id']}/{Path(created_file['name']).name}" + uploaded = get_object_storage().put_bytes(object_key, raw, file.content_type or "application/octet-stream") + get_platform_store().create_storage_object({ + "resource_type": "dataset", "resource_id": dataset_id, + "version_id": created_file.get("active_version_id") or created_file["id"], + "bucket": uploaded["bucket"], "object_key": object_key, + "file_name": created_file["name"], "content_type": file.content_type, + "byte_size": len(raw), "checksum_sha256": hashlib.sha256(raw).hexdigest(), + "status": "available", + }) if sync_to_compute: for file_id, file_name, raw in pending_sync: compute_sync.extend( @@ -1865,6 +1926,8 @@ async def model_compare_load(task_id: str) -> dict[str, Any]: errors = [] for node in _candidate_online_nodes(store, preferred_node_id): try: + if get_settings().minio_enabled: + await _wait_for_object_storage() client = ComputeNodeClient(node["api_base_url"]) await client.inference_load(load_payload) store.mark_inference_loaded(node["id"]) @@ -2035,6 +2098,73 @@ async def compute_nodes() -> dict[str, Any]: return ok(get_platform_store().compute_nodes()) +@router.post("/storage/objects/presign") +async def presign_storage_object(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: + """Create a short-lived MinIO upload/download URL for a platform resource.""" + if not get_settings().minio_enabled: + raise fail(503, "MinIO object storage is disabled") + resource_type = str(payload.get("resource_type") or "") + resource_id = str(payload.get("resource_id") or "") + version_id = str(payload.get("version_id") or uuid.uuid4().hex) + object_key = str(payload.get("object_key") or f"{resource_type}/{resource_id}/versions/{version_id}/resource") + if not resource_type or not resource_id: + raise fail(400, "resource_type and resource_id are required") + if payload.get("method", "put").lower() == "get" and not has_resource_access(resource_type, resource_id, current_user, "read"): + raise fail(403, "no permission to read this resource") + try: + storage = get_object_storage() + url = storage.presigned_get(object_key) if payload.get("method", "put").lower() == "get" else storage.presigned_put(object_key) + record = get_platform_store().create_storage_object({ + "resource_type": resource_type, "resource_id": resource_id, "version_id": version_id, + "bucket": storage.bucket, "object_key": object_key, "file_name": payload.get("file_name"), + "content_type": payload.get("content_type"), "created_by": current_user.get("id"), + }) + return ok({"url": url, "method": payload.get("method", "put").lower(), "expires_seconds": 3600, "object": record}) + except ObjectStorageError as exc: + raise fail(503, str(exc)) + + +@router.get("/storage/resources/{resource_type}/{resource_id}") +async def storage_resource_objects(resource_type: str, resource_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]: + if not has_resource_access(resource_type, resource_id, current_user, "read"): + raise fail(403, "no permission to read this resource") + return ok(get_platform_store().storage_objects_for_resource(resource_type, resource_id)) + + +@router.post("/storage/resources/{resource_type}/{resource_id}/prepare/{node_id}") +async def prepare_storage_resource( + resource_type: str, + resource_id: str, + node_id: str, + current_user: dict = Depends(get_current_user), +) -> dict[str, Any]: + if not has_resource_access(resource_type, resource_id, current_user, "execute"): + raise fail(403, "no permission to execute this resource") + store = get_platform_store() + node = next((item for item in store.compute_nodes() if item["id"] == node_id), None) + if not node: + raise fail(404, "compute node not found") + if not get_settings().minio_enabled: + raise fail(503, "MinIO object storage is disabled") + objects = store.storage_objects_for_resource(resource_type, resource_id) + if not objects: + raise fail(404, "resource has no MinIO objects") + client = ComputeNodeClient(node["api_base_url"]) + prepared = [] + for obj in objects: + url = get_object_storage().presigned_get(obj["object_key"]) + filename = Path(str(obj.get("file_name") or obj["object_key"])).name + result = await client.prepare_cache({ + "resource_id": resource_id, + "version_id": obj["version_id"], + "download_url": url, + "checksum_sha256": obj.get("checksum_sha256") or "", + "relative_path": f"{resource_type}s/{resource_id}/{filename}", + }) + prepared.append({**result, "storage_object_id": obj["id"], "node_id": node_id}) + return ok({"resource_type": resource_type, "resource_id": resource_id, "node_id": node_id, "status": "ready", "items": prepared}) + + @router.get("/compute/nodes/{node_id}") async def compute_node_detail(node_id: str) -> dict[str, Any]: node = next((item for item in get_platform_store().compute_nodes() if item["id"] == node_id), None) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 9052789..89dfbcb 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -60,6 +60,14 @@ class Settings: compute_status_sync_mode: str = os.getenv("COMPUTE_STATUS_SYNC_MODE", "polling") compute_poll_interval_seconds: int = _int_env("COMPUTE_POLL_INTERVAL_SECONDS", 3) compute_request_timeout_seconds: int = _int_env("COMPUTE_REQUEST_TIMEOUT_SECONDS", 5) + minio_enabled: bool = _bool_env("MINIO_ENABLED", False) + minio_endpoint: str = os.getenv("MINIO_ENDPOINT", "http://minio:9000") + minio_access_key: str = os.getenv("MINIO_ACCESS_KEY", "minioadmin") + minio_secret_key: str = os.getenv("MINIO_SECRET_KEY", "minioadmin") + minio_bucket: str = os.getenv("MINIO_BUCKET", "yg-ft-resources") + minio_secure: bool = _bool_env("MINIO_SECURE", False) + storage_wait_seconds: int = _int_env("STORAGE_WAIT_SECONDS", 300) + storage_check_interval_seconds: int = _int_env("STORAGE_CHECK_INTERVAL_SECONDS", 10) compute_service_token: str = os.getenv("COMPUTE_SERVICE_TOKEN", "") log_level: str = os.getenv("LOG_LEVEL", "INFO") log_dir: str = os.getenv("LOG_DIR", "./logs") diff --git a/backend/app/db/platform_store.py b/backend/app/db/platform_store.py index 5eef7c3..5394be5 100644 --- a/backend/app/db/platform_store.py +++ b/backend/app/db/platform_store.py @@ -505,6 +505,20 @@ class PlatformStore: ) self._ensure_columns(conn, "gpus", {"last_seen_at": "TEXT"}) self._ensure_columns(conn, "fine_tune_tasks", {"compute_job_id": "TEXT"}) + self._ensure_columns( + conn, + "sessions", + { + "username": "TEXT", + "login_at": "TEXT", + "logout_at": "TEXT", + "duration_seconds": "INTEGER", + "issued_at": "TEXT", + "expires_at": "TEXT", + "ip": "TEXT", + "create_time": "TEXT", + }, + ) self._ensure_columns( conn, "trained_models", @@ -525,7 +539,12 @@ class PlatformStore: }, ) schema_dir = Path(__file__).with_name("sql") - for extra in ("002_governance.sql", "003_tenant_quota.sql", "004_permissions.sql"): + for extra in ( + "002_governance.sql", + "003_model_path_governance.sql", + "003_tenant_quota.sql", + "004_permissions.sql", + ): extra_path = schema_dir / extra if extra_path.exists(): conn.executescript(extra_path.read_text(encoding="utf-8")) @@ -2863,6 +2882,54 @@ class PlatformStore: result.append(dict(row)) return result + def create_storage_object(self, payload: dict[str, Any]) -> dict[str, Any]: + object_id = str(payload.get("id") or new_id("object")) + with self.connect() as conn: + conn.execute( + """ + INSERT INTO storage_objects + (id, resource_type, resource_id, version_id, bucket, object_key, file_name, + content_type, checksum_sha256, byte_size, status, created_by, create_time) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (resource_type, resource_id, version_id, object_key) + DO UPDATE SET file_name=EXCLUDED.file_name, content_type=EXCLUDED.content_type, + checksum_sha256=EXCLUDED.checksum_sha256, byte_size=EXCLUDED.byte_size, + status=EXCLUDED.status, created_by=EXCLUDED.created_by + """, + ( + object_id, payload["resource_type"], payload["resource_id"], payload["version_id"], + payload["bucket"], payload["object_key"], payload.get("file_name"), payload.get("content_type"), + payload.get("checksum_sha256"), int(payload.get("byte_size") or 0), payload.get("status", "pending"), + payload.get("created_by"), payload.get("create_time") or utcnow(), + ), + ) + row = conn.execute( + "SELECT * FROM storage_objects WHERE resource_type=? AND resource_id=? AND version_id=? AND object_key=?", + (payload["resource_type"], payload["resource_id"], payload["version_id"], payload["object_key"]), + ).fetchone() + return dict(row) + + def storage_objects_for_resource(self, resource_type: str, resource_id: str) -> list[dict[str, Any]]: + with self.connect() as conn: + rows = conn.execute( + "SELECT * FROM storage_objects WHERE resource_type=? AND resource_id=? AND status='available' ORDER BY version_id, object_key", + (resource_type, resource_id), + ).fetchall() + return [dict(row) for row in rows] + + def update_storage_object(self, object_id: str, payload: dict[str, Any]) -> dict[str, Any]: + allowed = {"status", "checksum_sha256", "byte_size", "content_type"} + fields = {key: value for key, value in payload.items() if key in allowed} + if fields: + assignments = ", ".join(f"{key}=?" for key in fields) + with self.connect() as conn: + conn.execute(f"UPDATE storage_objects SET {assignments} WHERE id=?", (*fields.values(), object_id)) + with self.connect() as conn: + row = conn.execute("SELECT * FROM storage_objects WHERE id=?", (object_id,)).fetchone() + if not row: + raise KeyError(object_id) + return dict(row) + def update_resource_replica_sync_result( self, replica_id: str, diff --git a/backend/app/db/sql/000_full_init.sql b/backend/app/db/sql/000_full_init.sql index c1aa8d0..0274529 100644 --- a/backend/app/db/sql/000_full_init.sql +++ b/backend/app/db/sql/000_full_init.sql @@ -276,6 +276,39 @@ CREATE TABLE IF NOT EXISTS resource_sync_jobs ( completed_at TEXT ); +CREATE TABLE IF NOT EXISTS storage_objects ( + id TEXT PRIMARY KEY, + resource_type TEXT NOT NULL, + resource_id TEXT NOT NULL, + version_id TEXT NOT NULL, + bucket TEXT NOT NULL, + object_key TEXT NOT NULL, + file_name TEXT, + content_type TEXT, + checksum_sha256 TEXT, + byte_size BIGINT NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'pending', + created_by TEXT, + create_time TEXT NOT NULL, + UNIQUE (resource_type, resource_id, version_id, object_key) +); + +CREATE TABLE IF NOT EXISTS storage_cache_jobs ( + id TEXT PRIMARY KEY, + storage_object_id TEXT NOT NULL REFERENCES storage_objects(id) ON DELETE CASCADE, + node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE, + direction TEXT NOT NULL, + status TEXT NOT NULL, + progress INTEGER NOT NULL DEFAULT 0, + local_path TEXT, + error TEXT, + create_time TEXT NOT NULL, + completed_at TEXT +); + +CREATE INDEX IF NOT EXISTS idx_storage_objects_resource ON storage_objects(resource_type, resource_id, version_id); +CREATE INDEX IF NOT EXISTS idx_storage_cache_jobs_node_status ON storage_cache_jobs(node_id, status); + CREATE TABLE IF NOT EXISTS eval_tasks ( id TEXT PRIMARY KEY, name TEXT NOT NULL, @@ -361,9 +394,14 @@ CREATE TABLE IF NOT EXISTS roles ( CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, - issued_at TEXT NOT NULL, - expires_at TEXT NOT NULL, - ip TEXT + username TEXT, + login_at TEXT, + logout_at TEXT, + duration_seconds INTEGER, + issued_at TEXT, + expires_at TEXT, + ip TEXT, + create_time TEXT ); CREATE TABLE IF NOT EXISTS acls ( @@ -376,6 +414,25 @@ CREATE TABLE IF NOT EXISTS acls ( create_time TEXT ); +-- ---- 权限扩展(来源:004_permissions.sql) ---- + +CREATE TABLE IF NOT EXISTS gpu_assignments ( + id TEXT PRIMARY KEY, + node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE, + gpu_index INTEGER NOT NULL, + user_id TEXT NOT NULL, + assigned_by TEXT, + assigned_at TEXT NOT NULL, + UNIQUE (node_id, gpu_index, user_id) +); +CREATE INDEX IF NOT EXISTS idx_gpu_assignments_user ON gpu_assignments(user_id); +CREATE INDEX IF NOT EXISTS idx_gpu_assignments_gpu ON gpu_assignments(node_id, gpu_index); + +ALTER TABLE datasets ADD COLUMN IF NOT EXISTS created_by TEXT; +ALTER TABLE models ADD COLUMN IF NOT EXISTS created_by TEXT; +ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS created_by TEXT; +ALTER TABLE eval_tasks ADD COLUMN IF NOT EXISTS created_by TEXT; + -- ============================================================================ -- 二、治理表(来源:002_governance.sql) -- ============================================================================ @@ -462,6 +519,8 @@ ALTER TABLE tenants ADD COLUMN IF NOT EXISTS storage_quota TEXT; ALTER TABLE models ADD COLUMN IF NOT EXISTS can_train INTEGER NOT NULL DEFAULT 0; ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS artifact_dir TEXT; +ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_id TEXT; +ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_name TEXT; -- 按规则推定已有模型的 can_train(新库为空表,此语句为 no-op) UPDATE models diff --git a/backend/app/db/sql/001_platform_runtime.sql b/backend/app/db/sql/001_platform_runtime.sql index 84cf674..0e1ec80 100644 --- a/backend/app/db/sql/001_platform_runtime.sql +++ b/backend/app/db/sql/001_platform_runtime.sql @@ -234,6 +234,39 @@ CREATE TABLE IF NOT EXISTS resource_sync_jobs ( completed_at TEXT ); +CREATE TABLE IF NOT EXISTS storage_objects ( + id TEXT PRIMARY KEY, + resource_type TEXT NOT NULL, + resource_id TEXT NOT NULL, + version_id TEXT NOT NULL, + bucket TEXT NOT NULL, + object_key TEXT NOT NULL, + file_name TEXT, + content_type TEXT, + checksum_sha256 TEXT, + byte_size BIGINT NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'pending', + created_by TEXT, + create_time TEXT NOT NULL, + UNIQUE (resource_type, resource_id, version_id, object_key) +); + +CREATE TABLE IF NOT EXISTS storage_cache_jobs ( + id TEXT PRIMARY KEY, + storage_object_id TEXT NOT NULL REFERENCES storage_objects(id) ON DELETE CASCADE, + node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE, + direction TEXT NOT NULL, + status TEXT NOT NULL, + progress INTEGER NOT NULL DEFAULT 0, + local_path TEXT, + error TEXT, + create_time TEXT NOT NULL, + completed_at TEXT +); + +CREATE INDEX IF NOT EXISTS idx_storage_objects_resource ON storage_objects(resource_type, resource_id, version_id); +CREATE INDEX IF NOT EXISTS idx_storage_cache_jobs_node_status ON storage_cache_jobs(node_id, status); + CREATE TABLE IF NOT EXISTS eval_tasks ( id TEXT PRIMARY KEY, name TEXT NOT NULL, @@ -319,9 +352,14 @@ CREATE TABLE IF NOT EXISTS roles ( CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, - issued_at TEXT NOT NULL, - expires_at TEXT NOT NULL, - ip TEXT + username TEXT, + login_at TEXT, + logout_at TEXT, + duration_seconds INTEGER, + issued_at TEXT, + expires_at TEXT, + ip TEXT, + create_time TEXT ); CREATE TABLE IF NOT EXISTS acls ( diff --git a/backend/app/db/sql/003_model_path_governance.sql b/backend/app/db/sql/003_model_path_governance.sql index 0a1b795..4511d82 100644 --- a/backend/app/db/sql/003_model_path_governance.sql +++ b/backend/app/db/sql/003_model_path_governance.sql @@ -16,3 +16,5 @@ END; -- 3. 给 trained_models 增加 artifact_dir(训练产物目录扫描结果目录) ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS artifact_dir TEXT; +ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_id TEXT; +ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_name TEXT; diff --git a/backend/app/modules/compute_gateway/client.py b/backend/app/modules/compute_gateway/client.py index 2e7343b..aad8112 100644 --- a/backend/app/modules/compute_gateway/client.py +++ b/backend/app/modules/compute_gateway/client.py @@ -192,6 +192,27 @@ class ComputeNodeClient: response.raise_for_status() return _unwrap_dict(response.json()) + async def prepare_cache(self, payload: dict[str, Any]) -> dict[str, Any]: + async with httpx.AsyncClient(timeout=httpx.Timeout(900, connect=30), headers=self.headers()) as client: + response = await client.post( + _join_url(self.api_base_url, f"{self.route_prefix}/compute/cache/prepare"), + json=payload, + ) + response.raise_for_status() + return _unwrap_dict(response.json()) + + async def cache_status(self, resource_id: str, version_id: str | None = None) -> dict[str, Any]: + params = {"resource_id": resource_id} + if version_id: + params["version_id"] = version_id + async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client: + response = await client.get( + _join_url(self.api_base_url, f"{self.route_prefix}/compute/cache/status"), + params=params, + ) + response.raise_for_status() + return _unwrap_dict(response.json()) + async def _request( self, method: str, diff --git a/backend/app/modules/storage/__init__.py b/backend/app/modules/storage/__init__.py new file mode 100644 index 0000000..0406e52 --- /dev/null +++ b/backend/app/modules/storage/__init__.py @@ -0,0 +1 @@ +"""Central object storage integration.""" diff --git a/backend/app/modules/storage/minio_store.py b/backend/app/modules/storage/minio_store.py new file mode 100644 index 0000000..89fc43c --- /dev/null +++ b/backend/app/modules/storage/minio_store.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +from datetime import timedelta +from functools import lru_cache +from io import BytesIO +from typing import Any + +from minio import Minio +from minio.error import S3Error + +from app.core.config import get_settings + + +class ObjectStorageError(RuntimeError): + pass + + +class MinioObjectStorage: + def __init__(self) -> None: + settings = get_settings() + endpoint = settings.minio_endpoint.replace("http://", "").replace("https://", "").rstrip("/") + self.client = Minio(endpoint, access_key=settings.minio_access_key, secret_key=settings.minio_secret_key, secure=settings.minio_secure) + self.bucket = settings.minio_bucket + + def _ensure_enabled(self) -> None: + if not get_settings().minio_enabled: + raise ObjectStorageError("MinIO object storage is disabled") + + def ensure_bucket(self) -> None: + self._ensure_enabled() + try: + if not self.client.bucket_exists(self.bucket): + self.client.make_bucket(self.bucket) + except S3Error as exc: + raise ObjectStorageError(str(exc)) from exc + + def presigned_put(self, object_key: str, expires_seconds: int = 3600) -> str: + self._ensure_enabled() + self.ensure_bucket() + return self.client.presigned_put_object(self.bucket, object_key, expires=timedelta(seconds=expires_seconds)) + + def presigned_get(self, object_key: str, expires_seconds: int = 3600) -> str: + self._ensure_enabled() + self.ensure_bucket() + return self.client.presigned_get_object(self.bucket, object_key, expires=timedelta(seconds=expires_seconds)) + + def stat(self, object_key: str) -> dict[str, Any]: + self._ensure_enabled() + self.ensure_bucket() + try: + result = self.client.stat_object(self.bucket, object_key) + return {"object_key": object_key, "byte_size": result.size, "etag": result.etag, "last_modified": result.last_modified.isoformat() if result.last_modified else None} + except S3Error as exc: + raise ObjectStorageError(str(exc)) from exc + + def put_bytes(self, object_key: str, content: bytes, content_type: str = "application/octet-stream") -> dict[str, Any]: + self._ensure_enabled() + self.ensure_bucket() + try: + result = self.client.put_object(self.bucket, object_key, BytesIO(content), len(content), content_type=content_type) + return {"bucket": self.bucket, "object_key": object_key, "etag": result.etag, "byte_size": len(content)} + except S3Error as exc: + raise ObjectStorageError(str(exc)) from exc + + +@lru_cache +def get_object_storage() -> MinioObjectStorage: + return MinioObjectStorage() diff --git a/backend/requirements.txt b/backend/requirements.txt index 2065541..cfb38b7 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -8,6 +8,7 @@ psycopg-pool>=3.2.1 alembic>=1.13.1 redis>=5.0.4 httpx>=0.27.0 +minio>=7.2.7 PyJWT>=2.8.0 passlib[bcrypt]>=1.7.4 python-dotenv>=1.0.1 diff --git a/compute/api/main.py b/compute/api/main.py index 081d8b6..86b218d 100644 --- a/compute/api/main.py +++ b/compute/api/main.py @@ -11,6 +11,8 @@ import time from pathlib import Path from typing import Any +import httpx + from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFile from fastapi.responses import FileResponse, JSONResponse, StreamingResponse @@ -511,13 +513,27 @@ def create_app() -> FastAPI: llama_factory_home = Path(os.getenv("LLAMA_FACTORY_HOME", "/app/LLaMA-Factory")) gpu_items = gpu_resources() torch_cuda = torch_cuda_status() + storage_available = False + storage_error = "" + try: + data_root.mkdir(parents=True, exist_ok=True) + probe = data_root / ".yg-ft-storage-healthcheck" + probe.write_text(host_id(), encoding="utf-8") + storage_available = probe.read_text(encoding="utf-8").strip() == host_id() + probe.unlink(missing_ok=True) + except Exception as exc: # noqa: BLE001 - health endpoint must remain available + storage_error = str(exc) return { - "status": "ok", + "status": "ok" if storage_available else "storage_unavailable", "api_version": "v1", "compute_host_id": os.getenv("COMPUTE_HOST_ID", "unknown"), "app_callback_enabled": os.getenv("ENABLE_APP_CALLBACK", "false").lower() == "true", "data_root": str(data_root), "data_root_exists": data_root.exists(), + "storage_mode": os.getenv("STORAGE_MODE", "minio-cache"), + "storage_available": storage_available, + "storage_error": storage_error, + "storage_root": str(data_root), "model_root": os.getenv("YG_FT_MODEL_ROOT", str(data_root / "models")), "dataset_root": str(dataset_root), "dataset_root_exists": dataset_root.exists(), @@ -532,7 +548,7 @@ def create_app() -> FastAPI: "nvidia_gpu_count": len(gpu_items), "torch_cuda": torch_cuda, "gpu_discovery_endpoint": f"{route_prefix}/compute/resources/gpus", - "capabilities": ["gpu_discovery", "torch_cuda_diagnostics", "llama_factory", "file_gateway", "job_polling"], + "capabilities": ["gpu_discovery", "torch_cuda_diagnostics", "llama_factory", "file_gateway", "job_polling", "storage_health"], } @app.get(f"{route_prefix}/v1/compute/jobs") @@ -810,6 +826,82 @@ def create_app() -> FastAPI: "checksum_sha256": hashlib.sha256(target.read_bytes()).hexdigest() if target.is_file() else "", } + @app.post(f"{route_prefix}/compute/cache/prepare") + async def prepare_cache(payload: dict[str, Any]) -> dict[str, Any]: + """Download one MinIO object into the node-local cache atomically.""" + download_url = str(payload.get("download_url") or "") + resource_id = str(payload.get("resource_id") or "") + version_id = str(payload.get("version_id") or "latest") + if not download_url or not resource_id: + raise HTTPException(status_code=400, detail="download_url and resource_id are required") + data_root = Path(os.getenv("YG_FT_DATA_ROOT", "/data/yg-ft")) + cache_root = Path(os.getenv("YG_FT_CACHE_ROOT", str(data_root))) + relative_path = str(payload.get("relative_path") or f"resources/{resource_id}/{version_id}/resource") + target = (cache_root / relative_path.lstrip("/\\")).resolve() + if not _path_inside(cache_root, target): + raise HTTPException(status_code=400, detail="cache path must stay inside cache root") + target.parent.mkdir(parents=True, exist_ok=True) + temp_target = target.with_name(f".{target.name}.part") + expected_checksum = str(payload.get("checksum_sha256") or "").lower() + digest = hashlib.sha256() + byte_size = 0 + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(900, connect=30), follow_redirects=True) as client: + async with client.stream("GET", download_url) as response: + response.raise_for_status() + with temp_target.open("wb") as output: + async for chunk in response.aiter_bytes(1024 * 1024): + output.write(chunk) + digest.update(chunk) + byte_size += len(chunk) + checksum = digest.hexdigest() + if expected_checksum and checksum != expected_checksum: + temp_target.unlink(missing_ok=True) + raise HTTPException(status_code=502, detail="cache checksum mismatch") + temp_target.replace(target) + except HTTPException: + raise + except Exception as exc: + temp_target.unlink(missing_ok=True) + raise HTTPException(status_code=502, detail=f"cache download failed: {exc}") from exc + return { + "resource_id": resource_id, + "version_id": version_id, + "status": "ready", + "local_path": str(target), + "byte_size": byte_size, + "checksum_sha256": checksum, + } + + @app.get(f"{route_prefix}/compute/cache/status") + async def cache_status(resource_id: str = Query(...), version_id: str = Query(default="latest")) -> dict[str, Any]: + data_root = Path(os.getenv("YG_FT_DATA_ROOT", "/data/yg-ft")) + cache_root = Path(os.getenv("YG_FT_CACHE_ROOT", str(data_root))) + target = cache_root / "resources" / resource_id / version_id / "resource" + return { + "resource_id": resource_id, + "version_id": version_id, + "status": "ready" if target.is_file() else "missing", + "local_path": str(target), + "byte_size": target.stat().st_size if target.is_file() else 0, + } + + @app.delete(f"{route_prefix}/compute/cache") + async def clear_cache(resource_id: str | None = Query(default=None), version_id: str | None = Query(default=None)) -> dict[str, Any]: + data_root = Path(os.getenv("YG_FT_DATA_ROOT", "/data/yg-ft")) + cache_root = Path(os.getenv("YG_FT_CACHE_ROOT", str(data_root))) + target = cache_root / "resources" + if resource_id: + target = target / resource_id + if version_id: + target = target / version_id + target = target.resolve() + if not _path_inside(cache_root, target): + raise HTTPException(status_code=400, detail="cache path must stay inside cache root") + if target.exists(): + shutil.rmtree(target) + return {"status": "cleared", "resource_id": resource_id, "version_id": version_id} + @app.post(f"{route_prefix}/compute/files/import-local") async def import_local_file(payload: dict[str, Any]) -> dict[str, Any]: source = Path(str(payload.get("source_path") or "")) diff --git a/compute/requirements.txt b/compute/requirements.txt index 9f7bef7..704b78e 100644 --- a/compute/requirements.txt +++ b/compute/requirements.txt @@ -4,6 +4,7 @@ python-multipart>=0.0.9 pydantic>=2.7.0 python-dotenv>=1.0.1 httpx>=0.27.0 +# Compute Agent downloads MinIO objects through presigned HTTP URLs; no MinIO SDK is required. # 模型评测指标 sacrebleu>=2.4.0 rouge-score>=0.1.2 diff --git a/docker/app/.env b/docker/app/.env index 2542cde..1e092b8 100644 --- a/docker/app/.env +++ b/docker/app/.env @@ -48,3 +48,13 @@ COMPUTE_STATUS_SYNC_MODE=polling COMPUTE_POLL_INTERVAL_SECONDS=10 COMPUTE_POLL_BATCH_SIZE=100 COMPUTE_REQUEST_TIMEOUT_SECONDS=5 + +# MinIO object storage. Enable after the MinIO service is reachable. +MINIO_ENABLED=false +MINIO_ENDPOINT=http://minio:9000 +MINIO_ACCESS_KEY=minioadmin +MINIO_SECRET_KEY=minioadmin +MINIO_BUCKET=yg-ft-resources +MINIO_SECURE=false +STORAGE_WAIT_SECONDS=300 +STORAGE_CHECK_INTERVAL_SECONDS=10 diff --git a/docker/app/Dockerfile.backend b/docker/app/Dockerfile.backend index 3d654f3..752ab6e 100644 --- a/docker/app/Dockerfile.backend +++ b/docker/app/Dockerfile.backend @@ -12,7 +12,7 @@ RUN pip install --upgrade pip -i https://pypi.tuna.tsinghua.edu.cn/simple \ && pip install -r /tmp/requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple \ && rm -f /tmp/requirements.txt -RUN python -c "import fastapi, uvicorn, psycopg, psycopg_pool, sqlalchemy, redis, jwt, passlib, httpx, alembic; print('backend dependency check ok')" +RUN python -c "import fastapi, uvicorn, psycopg, psycopg_pool, sqlalchemy, redis, jwt, passlib, httpx, minio, alembic; print('backend dependency check ok')" RUN mkdir -p /opt/yg-ft/logs/backend /data/yg-ft \ && chmod -R 0775 /opt/yg-ft /data/yg-ft diff --git a/docker/app/docker-compose.yml b/docker/app/docker-compose.yml index 1dfae9b..af721be 100644 --- a/docker/app/docker-compose.yml +++ b/docker/app/docker-compose.yml @@ -62,6 +62,14 @@ services: COMPUTE_POLL_INTERVAL_SECONDS: ${COMPUTE_POLL_INTERVAL_SECONDS:-3} COMPUTE_POLL_BATCH_SIZE: ${COMPUTE_POLL_BATCH_SIZE:-100} COMPUTE_REQUEST_TIMEOUT_SECONDS: ${COMPUTE_REQUEST_TIMEOUT_SECONDS:-5} + MINIO_ENABLED: ${MINIO_ENABLED:-false} + MINIO_ENDPOINT: ${MINIO_ENDPOINT:-http://minio:9000} + MINIO_ACCESS_KEY: ${MINIO_ACCESS_KEY:-minioadmin} + MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:-minioadmin} + MINIO_BUCKET: ${MINIO_BUCKET:-yg-ft-resources} + MINIO_SECURE: ${MINIO_SECURE:-false} + STORAGE_WAIT_SECONDS: ${STORAGE_WAIT_SECONDS:-300} + STORAGE_CHECK_INTERVAL_SECONDS: ${STORAGE_CHECK_INTERVAL_SECONDS:-10} PYTHONPATH: /app volumes: - ../../backend:/app:ro diff --git a/docker/compute/.env.example b/docker/compute/.env.example index 5afdb04..50f762d 100644 --- a/docker/compute/.env.example +++ b/docker/compute/.env.example @@ -8,6 +8,7 @@ ENABLE_DOCS=false COMPUTE_API_PORT=19100 FILE_GATEWAY_PORT=19101 COMPUTE_API_IMAGE=yg-ft-compute-api:latest +YG_FT_CACHE_ROOT=/data/yg-ft # The application server actively polls Compute API; compute server does not need reverse access. COMPUTE_AUTH_ENABLED=true diff --git a/docker/compute/docker-compose.yml b/docker/compute/docker-compose.yml index 0081db8..4e13372 100644 --- a/docker/compute/docker-compose.yml +++ b/docker/compute/docker-compose.yml @@ -30,6 +30,7 @@ services: CUDA_VISIBLE_DEVICES: ${CUDA_VISIBLE_DEVICES:-all} NVIDIA_VISIBLE_DEVICES: ${NVIDIA_VISIBLE_DEVICES:-all} NVIDIA_DRIVER_CAPABILITIES: ${NVIDIA_DRIVER_CAPABILITIES:-compute,utility} + YG_FT_CACHE_ROOT: ${YG_FT_CACHE_ROOT:-/data/yg-ft} PYTHONPATH: /app volumes: - ../../compute:/app/compute:ro diff --git a/docker/minio/.env.example b/docker/minio/.env.example new file mode 100644 index 0000000..c37aaa9 --- /dev/null +++ b/docker/minio/.env.example @@ -0,0 +1,5 @@ +MINIO_ACCESS_KEY=minioadmin +MINIO_SECRET_KEY=change_me_minio_secret +MINIO_API_PORT=9000 +MINIO_CONSOLE_PORT=9001 +MINIO_DATA_ROOT_HOST=./data diff --git a/docker/minio/docker-compose.yml b/docker/minio/docker-compose.yml new file mode 100644 index 0000000..41f71db --- /dev/null +++ b/docker/minio/docker-compose.yml @@ -0,0 +1,19 @@ +services: + minio: + image: minio/minio:RELEASE.2025-02-28T09-55-16Z + container_name: yg-ft-minio + command: server /data --console-address ":9001" + environment: + MINIO_ROOT_USER: ${MINIO_ACCESS_KEY:-minioadmin} + MINIO_ROOT_PASSWORD: ${MINIO_SECRET_KEY:-change_me_minio_secret} + ports: + - "${MINIO_API_PORT:-9000}:9000" + - "${MINIO_CONSOLE_PORT:-9001}:9001" + volumes: + - ${MINIO_DATA_ROOT_HOST:-./data}:/data + healthcheck: + test: ["CMD-SHELL", "curl -fsS http://127.0.0.1:9000/minio/health/live || exit 1"] + interval: 10s + timeout: 5s + retries: 5 + restart: unless-stopped diff --git a/docs/minio-compute-cache-plan.md b/docs/minio-compute-cache-plan.md new file mode 100644 index 0000000..61d6a0d --- /dev/null +++ b/docs/minio-compute-cache-plan.md @@ -0,0 +1,287 @@ +# MinIO + Compute Agent 本地缓存方案与开发计划 + +## 1. 方案结论 + +本方案用 MinIO 作为模型、数据集、checkpoint、评测结果和训练产物的唯一正式存储,算力节点不挂载 NFS,也不要求安装 NFS 客户端。 + +算力节点只保留任务运行所需的本地缓存: + +```text +MinIO + -> Backend API 生成资源授权和版本信息 + -> Compute Agent 按任务下载到本地缓存 + -> LLaMA-Factory / 推理 / 合并任务使用本地路径 + -> 任务产物上传回 MinIO + -> PostgreSQL 更新资源和任务状态 +``` + +MinIO 是唯一正式数据源,本地缓存不是正式资源,节点失效或缓存被清理不会丢失模型和数据。 + +## 2. 为什么适合当前项目 + +当前项目的 Compute API 已经负责: + +- 训练、评测、推理和权重合并任务启动; +- 本地文件网关; +- GPU 和任务状态管理; +- 训练日志和产物路径管理。 + +因此不需要把 LLaMA-Factory 改成直接读取远程对象,只需要在 Compute Agent 启动任务前准备本地路径,继续把原来的 `model_name_or_path`、`dataset_dir`、`output_dir` 传给训练引擎。 + +## 3. MinIO 部署 + +MinIO 独立部署在 Linux 存储服务器或专用存储节点: + +```text +9000 S3 API +9001 MinIO Console(仅管理员网络开放) +``` + +建议创建 bucket: + +```text +yg-ft-resources +``` + +对象前缀建议: + +```text +models/{model_id}/versions/{version_id}/... +datasets/{dataset_id}/versions/{version_id}/... +outputs/{task_id}/... +evaluations/{task_id}/... +logs/{task_id}/... +``` + +MinIO 可以使用官方 Docker 镜像,不需要在 Linux 主机安装 MinIO 软件包。需要持久化挂载 MinIO 的数据目录。 + +## 4. 是否需要额外安装包 + +### MinIO 服务端 + +不需要安装额外系统包,使用 Docker 镜像即可: + +```text +minio/minio +``` + +### Backend API + +建议增加 Python SDK: + +```text +minio>=7.2.0 +``` + +Backend 用 SDK 生成预签名上传、下载 URL,并负责 bucket、对象元数据和权限控制。 + +### Compute API / Compute Agent + +推荐第一版只使用现有 `httpx` 访问预签名 URL,不额外安装 MinIO SDK。流程是: + +```text +Backend -> 返回预签名 URL +Compute Agent -> httpx 下载/上传 +``` + +这样算力节点不需要 MinIO 客户端、AWS CLI 或 NFS 客户端。 + +如果后续需要 Agent 直接操作 bucket、列目录或分片上传,再增加: + +```text +minio>=7.2.0 +``` + +但不建议第一阶段让算力节点持有 MinIO 管理密钥。 + +## 5. 权限模型 + +继续沿用当前项目的: + +- `projects`; +- `project_members`; +- `acls`。 + +MinIO 只负责对象访问凭证,Backend 负责业务授权: + +1. 用户请求模型或数据集; +2. Backend 校验项目成员关系和资源 ACL; +3. 校验通过后生成短时预签名 URL; +4. Compute Agent 使用 URL 下载; +5. URL 过期后自动失效。 + +MinIO bucket 不直接向前端或普通算力节点开放长期 Access Key。 + +## 6. 本地缓存目录 + +Compute API 容器继续以 root 运行,本地缓存挂载到: + +```text +/data/yg-ft/cache/models +/data/yg-ft/cache/datasets +/data/yg-ft/cache/adapters +/data/yg-ft/cache/outputs +``` + +每个缓存资源必须包含: + +```text +resource_id +version_id +sha256 +byte_size +last_used_at +status +``` + +缓存状态: + +- `missing`:本地不存在; +- `downloading`:正在下载; +- `ready`:校验成功; +- `corrupted`:校验失败; +- `evicting`:正在清理。 + +任务只能使用 `ready` 状态的缓存。 + +## 7. 任务流程 + +### 7.1 训练 + +```text +校验项目/用户权限 + -> 获取基座模型版本和数据集版本 + -> 检查本地缓存 + -> 缺失则下载并校验 SHA256 + -> 启动 LLaMA-Factory + -> checkpoint 写入本地临时目录 + -> 任务完成后上传 outputs 到 MinIO + -> MinIO 上传完成并校验后更新数据库 +``` + +### 7.2 推理 + +```text +校验模型权限 + -> 下载或复用本地模型缓存 + -> 使用本地模型路径加载 + -> 推理服务只绑定当前节点缓存 +``` + +### 7.3 权重合并 + +```text +下载 base model 和 adapter/checkpoint + -> 在指定节点执行 CPU 合并 + -> 上传 merged model 到 MinIO + -> 数据库记录新的模型版本 +``` + +### 7.4 NFS 故障规则对应关系 + +MinIO 不可达时,节点不再依赖本地残留文件直接启动新任务: + +- 已有完整缓存且资源版本仍有效:允许继续执行当前任务; +- 新任务无法确认资源版本:等待 MinIO 恢复; +- 等待超过配置窗口:任务失败; +- 产物无法上传:任务不得标记为最终成功,进入 `storage_error`。 + +如果严格执行“共享存储故障时节点不能正常工作”,则即使本地缓存完整,也应禁止启动新任务。建议当前项目采用这一严格规则。 + +## 8. 数据库建议 + +现有 `resource_replicas` 可扩展为缓存索引,建议增加: + +```text +storage_backend -- minio +storage_bucket +storage_object_key +version_id +cache_path +cache_status +last_used_at +download_progress +``` + +`resource_sync_jobs` 可继续用于下载和上传任务,但建议增加方向字段: + +```text +direction -- download / upload +``` + +模型、数据集、checkpoint 和评测结果均使用 `resource_id + version_id`,不再把节点本地路径作为唯一资源标识。 + +## 9. 开发计划 + +### 阶段一:MinIO 服务和配置 + +- 增加 `docker/minio/docker-compose.yml`; +- 配置 MinIO endpoint、bucket、Access Key 和 Secret Key; +- 增加 Backend `minio` 依赖; +- 增加 MinIO 健康检查; +- 创建统一 bucket 和对象前缀规则。 + +### 阶段二:Backend 资源服务 + +- 实现对象上传、下载、删除和 HEAD 校验; +- 生成短时预签名 URL; +- 接入项目成员和 ACL 校验; +- 建立资源版本、SHA256 和大小记录; +- 上传成功后再更新数据库资源状态。 + +### 阶段三:Compute Agent 缓存服务 + +- 增加缓存目录管理器; +- 实现预签名 URL 下载; +- 支持临时文件下载和原子改名; +- 实现 SHA256 校验、失败重试和断点续传; +- 增加缓存状态查询和清理接口。 + +### 阶段四:接入业务任务 + +- 训练前准备基座模型和数据集; +- 推理前准备模型和 adapter; +- 权重合并前准备 base model 和 checkpoint; +- 评测前准备模型和数据集; +- 训练产物、合并模型和评测结果上传 MinIO; +- MinIO 故障时统一等待并超时失败。 + +### 阶段五:前端和管理员功能 + +- 显示资源版本和对象存储状态; +- 显示节点缓存状态; +- 支持手动预热模型; +- 支持缓存清理; +- 显示下载、上传和校验失败原因。 + +### 阶段六:测试和切换 + +- 单节点下载和缓存复用测试; +- 多节点同时下载同一模型测试; +- MinIO 重启和网络中断测试; +- SHA256 损坏文件测试; +- 训练、推理、权重合并全流程测试; +- 关闭旧的逐节点上传逻辑。 + +## 10. 预计工作量 + +```text +MinIO 部署和配置 1~2 人日 +Backend 对象存储服务 4~7 人日 +Compute Agent 缓存 6~10 人日 +训练/推理/合并/评测接入 8~15 人日 +权限、数据库和前端 5~10 人日 +故障和回归测试 5~8 人日 +总计 29~52 人日 +``` + +## 11. 推荐结论 + +当前项目建议采用 MinIO + HTTP 预签名 URL + Compute Agent 本地缓存: + +- MinIO 服务端使用 Docker,不安装主机软件包; +- Backend 增加 `minio` Python SDK; +- Compute Agent 第一阶段继续使用现有 `httpx`,不增加 MinIO SDK; +- 算力节点不安装 NFS 客户端; +- Compute API 继续以 root 运行; +- 训练、推理和权重合并继续使用本地路径,改造风险低于直接让训练框架读取对象存储。