YG-Soft/YG_FT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 权限与日志治理完善，MinIO 独立部署与 tiktoken 离线打包适配

Browse Source 
- 后端：强化平台/审批/资源/系统接口权限校验与操作日志，更新权限设计文档与测试用例
- 存储：新增 MinIO 独立部署适配（端口 19000/19001），外部端点与 host-gateway 互通
- 离线：打包 tiktoken cl100k_base 词表进镜像，避免无网环境联网下载
- 其他：算力节点接口微调，前端微调创建页小修，忽略 MinIO 运行时数据

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wuyongtao
2026-08-12 15:21:23 +08:00
parent 2f64086177
commit 5ecca9f0bc
26 changed files with 101600 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

334
backend/app/api/v1/endpoints/platform.py
View File

@@ -4,6 +4,7 @@ import json
import asyncio
import hashlib
import uuid
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
@@ -13,7 +14,7 @@ from fastapi.responses import PlainTextResponse, StreamingResponse
import httpx
from app.core.auth import filter_accessible_resource_ids, get_current_user, has_resource_access, is_admin
from app.core.auth import filter_accessible_resource_ids, filter_accessible_resource_ids_batch, get_current_user, has_resource_access, is_admin
from app.core.config import get_settings
from app.db.platform_store import get_platform_store
from app.modules.compute_gateway.client import ComputeNodeClient
@@ -21,6 +22,21 @@ from app.modules.compute_gateway.sync import fetch_eval_result_content, poll_com
from app.modules.storage.minio_store import ObjectStorageError, get_object_storage
router = APIRouter()
_LOGIN_FAILURES: dict[str, list[float]] = {}
_DASHBOARD_CACHE_TTL = 5.0
_DASHBOARD_CACHE: dict[str, Any] = {}
def _cached_dashboard(key: str) -> dict[str, Any] | None:
item = _DASHBOARD_CACHE.get(key)
if not item or time.monotonic() - item["created_at"] >= _DASHBOARD_CACHE_TTL:
return None
return item["value"]
def _store_dashboard_cache(key: str, value: dict[str, Any]) -> dict[str, Any]:
_DASHBOARD_CACHE[key] = {"created_at": time.monotonic(), "value": value}
return value
def ok(data: Any = None, message: str = "ok") -> dict[str, Any]:
@@ -77,6 +93,27 @@ def _candidate_online_nodes(store: Any, preferred_node_id: str | None = None) ->
return preferred + others
async def _prepare_resource_on_node(store: Any, resource_type: str, resource_id: str, node: dict[str, Any]) -> str | None:
"""Prepare MinIO resource files on a node and return the local directory."""
if not get_settings().minio_enabled or not resource_id:
return None
objects = store.storage_objects_for_resource(resource_type, resource_id)
if not objects:
return None
client = ComputeNodeClient(node["api_base_url"], timeout=900)
root_name = "trained_models" if resource_type in {"trained_model", "model_artifact"} else f"{resource_type}s"
for obj in objects:
await client.prepare_cache({
"resource_id": resource_id,
"version_id": obj["version_id"],
"download_url": get_object_storage().presigned_get(obj["object_key"]),
"checksum_sha256": obj.get("checksum_sha256") or "",
"byte_size": obj.get("byte_size") or 0,
"relative_path": f"{root_name}/{resource_id}/{Path(str(obj.get('file_name') or obj['object_key'])).name}",
})
return f"/data/yg-ft/{root_name}/{resource_id}"
def _build_messages_payload(payload: dict[str, Any]) -> dict[str, Any]:
"""Convert frontend inference payload to compute API messages format.
@@ -396,13 +433,20 @@ async def _fine_tune_preflight_with_job_payload(
@router.post("/login")
async def login(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def login(payload: dict[str, Any] = Body(...), request: Request = None) -> dict[str, Any]:
store = get_platform_store()
ip = request.client.host if request and request.client else "unknown"
now = time.time()
recent = [stamp for stamp in _LOGIN_FAILURES.get(ip, []) if now - stamp < 300]
if len(recent) >= 5:
raise fail(429, "too many login attempts, retry later")
user = store.login(payload.get("username", ""), payload.get("password", ""))
if not user:
_LOGIN_FAILURES[ip] = [*recent, now]
raise fail(401, "invalid username or password")
sess = store.create_session(user["id"])
return ok({"token": f"platform-token-{user['id']}", "user": user, "session_id": sess["session_id"]})
_LOGIN_FAILURES.pop(ip, None)
sess = store.create_session(user["id"], ip=None)
return ok({"token": f"platform-token-{user['id']}.{sess['session_id']}", "user": user, "session_id": sess["session_id"]})
@router.post("/logout")
@@ -422,7 +466,7 @@ async def me(request: Request) -> dict[str, Any]:
token = auth.replace("Bearer ", "").strip()
# token 格式: platform-token-{user_id}
if token.startswith("platform-token-"):
user_id = token[len("platform-token-"):]
user_id = token[len("platform-token-"):].split(".", 1)[0]
for u in store.users():
if u.get("id") == user_id:
return ok(u)
@@ -431,9 +475,12 @@ async def me(request: Request) -> dict[str, Any]:
@router.get("/dashboard/overview")
async def dashboard_overview() -> dict[str, Any]:
cached = _cached_dashboard("overview")
if cached is not None:
return cached
store = get_platform_store()
tasks = store.tasks()
return ok(
return _store_dashboard_cache("overview", ok(
{
"models": len(store.models()),
"datasets": len(store.datasets()),
@@ -442,11 +489,14 @@ async def dashboard_overview() -> dict[str, Any]:
"compute_nodes": len(store.compute_nodes()),
"gpus": len(store.gpus()),
}
)
))
@router.get("/dashboard/stats")
async def dashboard_stats() -> dict[str, Any]:
cached = _cached_dashboard("stats")
if cached is not None:
return cached
"""看板聚合数据：基于平台真实数据；缺项做合理近似。"""
store = get_platform_store()
tasks = store.tasks()
@@ -603,7 +653,7 @@ async def dashboard_stats() -> dict[str, Any]:
except Exception:
pass
return ok(
return _store_dashboard_cache("stats", ok(
{
"online_services": sum(s["count"] for s in service_status),
"running_tasks": len(running_ft) + eval_running,
@@ -615,7 +665,7 @@ async def dashboard_stats() -> dict[str, Any]:
"login_duration_rank": login_duration_rank,
"recent_login_users": recent_login_users,
}
)
))
@router.get("/system-info")
@@ -731,7 +781,12 @@ async def trained_models(current_user: dict = Depends(get_current_user)) -> dict
@router.delete("/model-manage/trained-models/{model_id}")
async def delete_trained_model(model_id: str, type: str = Query(default="merged")) -> dict[str, Any]:
async def delete_trained_model(model_id: str, type: str = Query(default="merged"), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not has_resource_access("trained_model", model_id, current_user, "delete"):
raise fail(403, "no permission to delete this trained model")
pending = _require_approval_or_admin("trained_model", model_id, current_user, f"删除训练模型 {model_id}")
if pending:
return pending
get_platform_store().delete_trained_model(model_id)
return ok({"deleted": model_id, "type": type})
@@ -747,7 +802,9 @@ async def trained_model_lineage(model_id: str) -> dict[str, Any]:
@router.get("/model-manage/export-jobs")
async def model_export_jobs(trained_model_id: str | None = Query(default=None)) -> dict[str, Any]:
async def model_export_jobs(trained_model_id: str | None = Query(default=None), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if trained_model_id and not has_resource_access("trained_model", trained_model_id, current_user, "read"):
raise fail(403, "no permission to access export jobs")
return ok(get_platform_store().model_export_jobs(trained_model_id))
@@ -878,7 +935,7 @@ async def delete_model(model_id: str, current_user: dict = Depends(get_current_u
@router.post("/model-manage/merge")
async def merge_model(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def merge_model(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
trained_model_id = str(payload.get("trained_model_id") or payload.get("model_id") or payload.get("model_name") or "")
trained_model = next(
@@ -889,6 +946,12 @@ async def merge_model(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
),
None,
)
if not trained_model:
raise fail(404, "trained model not found")
if not has_resource_access("trained_model", trained_model["id"], current_user, "execute"):
raise fail(403, "no permission to merge this trained model")
if payload.get("base_model_id") and not has_resource_access("model", str(payload["base_model_id"]), current_user, "execute"):
raise fail(403, "no permission to use merge base model")
base_model_path = payload.get("base_model_path") or (trained_model and trained_model.get("base_model_path"))
adapter_path = (
payload.get("adapter_path")
@@ -906,6 +969,15 @@ async def merge_model(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
await _wait_for_object_storage()
except RuntimeError as exc:
raise fail(503, str(exc))
try:
prepared_base = await _prepare_resource_on_node(store, "model", str(payload.get("base_model_id") or base_model_path), node)
if prepared_base:
base_model_path = prepared_base
prepared_adapter = await _prepare_resource_on_node(store, "trained_model", str(payload.get("adapter_model_id") or (trained_model and trained_model.get("id")) or ""), node)
if prepared_adapter:
adapter_path = prepared_adapter
except Exception as exc:
raise fail(502, f"merge resource preparation failed: {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")
@@ -1075,6 +1147,7 @@ async def _sync_training_dataset_to_compute_node(
"version_id": obj["version_id"],
"download_url": url,
"checksum_sha256": obj.get("checksum_sha256") or "",
"byte_size": obj.get("byte_size") or 0,
"relative_path": f"datasets/{dataset_id}/{target_name}",
})
results.append({**result, "file_id": item.get("id"), "name": target_name, "node_id": node["id"]})
@@ -1258,7 +1331,15 @@ async def fine_tune_list(current_user: dict = Depends(get_current_user)) -> dict
@router.post("/fine-tune")
async def create_fine_tune(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def create_fine_tune(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
payload.setdefault("created_by", current_user.get("id"))
if not is_admin(current_user):
model_id = str(payload.get("base_model") or payload.get("base_model_id") or "")
dataset_id = str(payload.get("train_dataset_id") or "")
if model_id and not has_resource_access("model", model_id, current_user, "execute"):
raise fail(403, "no permission to use this base model")
if dataset_id and not has_resource_access("dataset", dataset_id, current_user, "execute"):
raise fail(403, "no permission to use this dataset")
try:
task = get_platform_store().create_task(payload)
return ok({"id": task["id"]})
@@ -1275,11 +1356,16 @@ async def start_fine_tune(
# GPU 权限校验：普通用户只能使用被分配的 GPU
if not is_admin(current_user):
node_id = payload.get("compute_node_id") or payload.get("node_id")
gpu_indices = payload.get("gpus") or []
gpu_indices = payload.get("gpu_indices")
if gpu_indices is None:
gpu_indices = payload.get("gpus") or []
if node_id and gpu_indices:
if not store.check_gpu_access(current_user["id"], node_id, gpu_indices):
raise fail(403, "无权使用所选 GPU，请联系管理员分配")
# 记录创建者
if node_id and not gpu_indices:
payload["allowed_gpu_indices"] = store.assigned_gpu_indexes(current_user["id"], node_id)
payload["strict_node_selection"] = bool(node_id)
payload.setdefault("created_by", current_user.get("id"))
try:
return ok(await _submit_fine_tune_task(store, payload))
@@ -1411,7 +1497,9 @@ async def fine_tune_diagnostics(task_id: str) -> dict[str, Any]:
@router.put("/fine-tune/{task_id}")
async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not has_resource_access("fine-tune", task_id, current_user, "write"):
raise fail(403, "no permission to update this task")
try:
return ok(get_platform_store().update_task(task_id, payload))
except KeyError:
@@ -1442,13 +1530,15 @@ async def stop_fine_tune_alt(task_id: str, current_user: dict = Depends(get_curr
@router.post("/fine-tune/{task_id}/retry")
async def retry_fine_tune(task_id: str, payload: dict[str, Any] | None = Body(default=None)) -> dict[str, Any]:
async def retry_fine_tune(task_id: str, payload: dict[str, Any] | None = Body(default=None), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
payload = payload or {}
try:
task = store.task(task_id)
except KeyError:
raise fail(404, "fine tune task not found")
if not has_resource_access("fine-tune", task_id, current_user, "execute"):
raise fail(403, "no permission to retry this task")
if task["status"] not in {"failed", "stopped"} and not payload.get("force"):
raise fail(409, "only failed or stopped tasks can be retried without force=true")
retry_payload = {**task, **payload, "task_id": task_id, "id": task_id}
@@ -1544,7 +1634,7 @@ async def model_eval_detail(task_id: str, current_user: dict = Depends(get_curre
@router.post("/model-eval/start")
async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
"""Start an evaluation task: submit eval job to compute node."""
store = get_platform_store()
# 1. Create eval task record
@@ -1745,12 +1835,23 @@ async def dimension_delete(dimension_id: str) -> dict[str, Any]:
@router.get("/model-compare")
async def model_compare_list() -> dict[str, Any]:
return ok(get_platform_store().compare_tasks())
async def model_compare_list(current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
tasks = get_platform_store().compare_tasks()
if is_admin(current_user):
return ok(tasks)
accessible = filter_accessible_resource_ids_batch("compare", [item["id"] for item in tasks], current_user)
return ok([item for item in tasks if item["id"] in accessible])
@router.post("/model-compare")
async def model_compare_create(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def model_compare_create(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
payload.setdefault("created_by", current_user.get("id"))
model_ids = payload.get("model_ids") or payload.get("models") or []
if not is_admin(current_user):
for model_id in model_ids:
if isinstance(model_id, dict): model_id = model_id.get("id") or model_id.get("model_id")
if model_id and not has_resource_access("model", str(model_id), current_user, "execute"):
raise fail(403, "no permission to use inference model")
task = get_platform_store().create_compare_task(payload)
return ok({"id": task["id"]})
@@ -1766,9 +1867,12 @@ async def model_compare_stop_by_pid(payload: dict[str, Any] = Body(...)) -> dict
@router.get("/model-compare/{task_id}")
async def model_compare_detail(task_id: str) -> dict[str, Any]:
async def model_compare_detail(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
try:
return ok(get_platform_store().compare_task(task_id))
task = get_platform_store().compare_task(task_id)
if not has_resource_access("compare", task_id, current_user, "read"):
raise fail(403, "no permission to access inference task")
return ok(task)
except KeyError:
raise fail(404, "compare task not found")
@@ -1805,12 +1909,17 @@ async def _unload_from_compute_node(store: Any, task: dict[str, Any] | None = No
@router.delete("/model-compare/{task_id}")
async def model_compare_delete(task_id: str) -> dict[str, Any]:
async def model_compare_delete(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
# 先删记录（快），再 best-effort 释放算力节点上的模型——删除绝不被卸载阻塞
try:
task = get_platform_store().compare_task(task_id)
except KeyError:
raise fail(404, "compare task not found")
if not has_resource_access("compare", task_id, current_user, "delete"):
raise fail(403, "no permission to delete inference task")
pending = _require_approval_or_admin("compare", task_id, current_user, f"删除推理任务 {task_id}")
if pending:
return pending
get_platform_store().delete_compare_task(task_id)
try:
await _unload_from_compute_node(get_platform_store(), task=task)
@@ -1873,7 +1982,7 @@ def _invalidate_superseded_models(store: Any, task_id: str, loaded_models: list[
@router.post("/model-compare/{task_id}/load")
async def model_compare_load(task_id: str) -> dict[str, Any]:
async def model_compare_load(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
"""异步派发模型加载到算力节点，立即返回。
加载进度由轮询对账器（compute_poller → reconcile_inference_loads）推进：
@@ -1883,6 +1992,8 @@ async def model_compare_load(task_id: str) -> dict[str, Any]:
try:
store = get_platform_store()
task = store.compare_task(task_id)
if not has_resource_access("compare", task_id, current_user, "execute"):
raise fail(403, "no permission to load inference task")
models = task.get("models") or []
if isinstance(models, str):
try:
@@ -1953,10 +2064,12 @@ async def model_compare_load(task_id: str) -> dict[str, Any]:
@router.post("/model-compare/{task_id}/unload")
async def model_compare_unload(task_id: str) -> dict[str, Any]:
async def model_compare_unload(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
try:
store = get_platform_store()
task = store.compare_task(task_id)
if not has_resource_access("compare", task_id, current_user, "write"):
raise fail(403, "no permission to unload inference task")
# 任务感知卸载：只释放该任务实际加载到的节点，短超时快速返回
unload_result = await _unload_from_compute_node(store, task=task)
updated = store.update_compare_task(task_id, {"status": "pending", "load_status": {"loaded_models": []}})
@@ -2048,7 +2161,19 @@ async def model_chat_local_unload() -> dict[str, Any]:
if not store.is_inference_loaded(n["id"]):
continue
try:
result = await ComputeNodeClient(n["api_base_url"]).inference_unload()
client = ComputeNodeClient(n["api_base_url"])
last_error = ""
result = None
for attempt in range(3):
try:
result = await client.inference_unload()
break
except Exception as exc: # noqa: BLE001 - retry node cleanup
last_error = str(exc)
if attempt < 2:
await asyncio.sleep(2 ** attempt)
if result is None:
raise RuntimeError(last_error or "inference unload failed")
results.append({"node_id": n["id"], "success": True, "result": result})
except Exception as exc: # noqa: BLE001 - best-effort unload
results.append({"node_id": n["id"], "success": False, "error": str(exc)})
@@ -2073,19 +2198,31 @@ async def model_chat_local_status() -> dict[str, Any]:
@router.post("/model-chat/trained/preload")
async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def model_chat_trained_preload(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
resource_id = str(payload.get("trained_model_id") or payload.get("model_id") or payload.get("resource_id") or "")
if resource_id and not has_resource_access("trained_model", resource_id, current_user, "execute") and not has_resource_access("model", resource_id, current_user, "execute"):
raise fail(403, "no permission to load this model")
"""Load a trained model (base + adapter) on the compute node for inference."""
model_path = (payload.get("model_name_or_path") or "").strip()
if not model_path:
return ok({"loaded": False, "error": "model_name_or_path is required"})
store = get_platform_store()
node = _select_first_online_node(store)
requested_node_id = str(payload.get("compute_node_id") or payload.get("node_id") or "")
node = next((item for item in store.compute_nodes() if item.get("id") == requested_node_id and item.get("enabled") and item.get("scheduler_status") == "online"), None)
if not node:
node = _select_first_online_node(store)
if not node:
return ok({"loaded": False, "error": "no online compute node"})
try:
prepared_path = await _prepare_resource_on_node(store, "trained_model", str(payload.get("trained_model_id") or payload.get("model_id") or payload.get("resource_id") or ""), node)
if prepared_path:
payload = {**payload, "model_name_or_path": prepared_path}
prepared_path = await _prepare_resource_on_node(store, "model", str(payload.get("model_id") or payload.get("resource_id") or ""), node)
if prepared_path:
payload = {**payload, "model_name_or_path": prepared_path}
client = ComputeNodeClient(node["api_base_url"])
# 计算节点现在异步加载：HTTP 接受（loading/ready）即视为派发成功
result = await client.inference_load(payload)
result = await client.inference_load({**payload, "compute_node_id": node["id"]})
if result.get("loaded") or result.get("status") in {"loading", "ready"}:
store.mark_inference_loaded(node["id"])
return ok(result)
@@ -2094,8 +2231,13 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dic
@router.get("/compute/nodes")
async def compute_nodes() -> dict[str, Any]:
return ok(get_platform_store().compute_nodes())
async def compute_nodes(current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
nodes = get_platform_store().compute_nodes()
if is_admin(current_user):
return ok(nodes)
# 普通用户只看到自己被分配 GPU 的节点，避免泄露节点拓扑和未授权资源。
assigned = {item["node_id"] for item in get_platform_store().gpu_assignments_for_user(current_user["id"])}
return ok([node for node in nodes if node["id"] in assigned])
@router.post("/storage/objects/presign")
@@ -2154,27 +2296,90 @@ async def prepare_storage_resource(
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}",
})
cache_job = store.create_storage_cache_job({"storage_object_id": obj["id"], "node_id": node_id, "direction": "download"})
try:
result = await client.prepare_cache({
"resource_id": resource_id,
"version_id": obj["version_id"],
"download_url": url,
"checksum_sha256": obj.get("checksum_sha256") or "",
"byte_size": obj.get("byte_size") or 0,
"relative_path": f"{resource_type}s/{resource_id}/{filename}",
})
store.update_storage_cache_job(cache_job["id"], {"status": "completed", "progress": 100, "local_path": result.get("local_path"), "completed_at": utcnow()})
except Exception as exc:
store.update_storage_cache_job(cache_job["id"], {"status": "failed", "error": str(exc), "completed_at": utcnow()})
raise
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)
@router.get("/storage/cache/jobs/{node_id}")
async def storage_cache_jobs(node_id: str, limit: int = Query(default=100, ge=1, le=500), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
return ok(get_platform_store().storage_cache_jobs_for_node(node_id, limit))
@router.post("/storage/resources/{resource_type}/{resource_id}/archive-node/{node_id}")
async def archive_node_files(
resource_type: str,
resource_id: str,
node_id: str,
payload: dict[str, Any] = Body(...),
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
"""Archive completed node files to MinIO without proxying file bytes through Backend."""
if not has_resource_access(resource_type, resource_id, current_user, "execute"):
raise fail(403, "no permission to archive this resource")
store = get_platform_store()
if not is_admin(current_user):
model_id = str(payload.get("model_id") or "")
dataset_id = str(payload.get("dataset_id") or "")
if not model_id or not has_resource_access("model", model_id, current_user, "execute"):
raise fail(403, "no permission to evaluate this model")
if not dataset_id or not has_resource_access("dataset", dataset_id, current_user, "execute"):
raise fail(403, "no permission to evaluate this dataset")
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")
files = payload.get("files") or []
if not isinstance(files, list) or not files:
raise fail(400, "files is required")
client = ComputeNodeClient(node["api_base_url"], timeout=900)
archived = []
for item in files:
path = str(item.get("path") or "")
name = Path(str(item.get("file_name") or Path(path).name)).name
version_id = str(item.get("version_id") or uuid.uuid4().hex)
object_key = str(item.get("object_key") or f"{resource_type}s/{resource_id}/versions/{version_id}/{name}")
url = get_object_storage().presigned_put(object_key)
result = await client.upload_file_to_url(path, url, object_key, str(item.get("content_type") or "application/octet-stream"))
metadata = get_object_storage().stat(object_key)
record = store.create_storage_object({
"resource_type": resource_type, "resource_id": resource_id, "version_id": version_id,
"bucket": get_object_storage().bucket, "object_key": object_key, "file_name": name,
"content_type": item.get("content_type"), "checksum_sha256": result.get("checksum_sha256"),
"byte_size": metadata.get("byte_size") or result.get("byte_size") or 0, "status": "available",
"created_by": current_user.get("id"),
})
archived.append({"object": record, "node_id": node_id})
return ok({"status": "available", "items": archived})
@router.get("/compute/nodes/{node_id}")
async def compute_node_detail(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
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 is_admin(current_user) and not any(item["node_id"] == node_id for item in store.gpu_assignments_for_user(current_user["id"])):
raise fail(403, "no permission to access this compute node")
return ok(node)
@router.post("/compute/nodes")
async def create_compute_node(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def create_compute_node(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
try:
return ok(get_platform_store().create_compute_node(payload))
except KeyError as exc:
@@ -2184,7 +2389,9 @@ async def create_compute_node(payload: dict[str, Any] = Body(...)) -> dict[str,
@router.put("/compute/nodes/{node_id}")
async def update_compute_node(node_id: str, payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def update_compute_node(node_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
try:
return ok(get_platform_store().update_compute_node(node_id, payload))
except KeyError:
@@ -2194,7 +2401,9 @@ async def update_compute_node(node_id: str, payload: dict[str, Any] = Body(...))
@router.delete("/compute/nodes/{node_id}")
async def delete_compute_node(node_id: str) -> dict[str, Any]:
async def delete_compute_node(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
try:
return ok(get_platform_store().delete_compute_node(node_id))
except KeyError:
@@ -2204,7 +2413,9 @@ async def delete_compute_node(node_id: str) -> dict[str, Any]:
@router.post("/compute/nodes/{node_id}/test-connection")
async def test_compute_node(node_id: str) -> dict[str, Any]:
async def test_compute_node(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
store = get_platform_store()
node = next((item for item in store.compute_nodes() if item["id"] == node_id), None)
if not node:
@@ -2238,22 +2449,28 @@ async def test_compute_node(node_id: str) -> dict[str, Any]:
@router.post("/compute/nodes/{node_id}/health-check")
async def health_check_compute_node(node_id: str) -> dict[str, Any]:
return await test_compute_node(node_id)
async def health_check_compute_node(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
return await test_compute_node(node_id, current_user)
@router.post("/compute/nodes/{node_id}/enable")
async def enable_compute_node(node_id: str) -> dict[str, Any]:
async def enable_compute_node(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
return ok(get_platform_store().update_compute_node(node_id, {"enabled": True, "scheduler_status": "online"}))
@router.post("/compute/nodes/{node_id}/disable")
async def disable_compute_node(node_id: str) -> dict[str, Any]:
async def disable_compute_node(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
return ok(get_platform_store().update_compute_node(node_id, {"enabled": False, "scheduler_status": "offline"}))
@router.post("/compute/nodes/{node_id}/drain")
async def drain_compute_node(node_id: str) -> dict[str, Any]:
async def drain_compute_node(node_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not is_admin(current_user):
raise fail(403, "admin permission required")
return ok(get_platform_store().update_compute_node(node_id, {"scheduler_status": "draining"}))
@@ -2460,8 +2677,13 @@ async def compute_node_engines(node_id: str) -> dict[str, Any]:
@router.get("/compute/gpus")
async def compute_gpus() -> dict[str, Any]:
return ok(get_platform_store().gpus())
async def compute_gpus(current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
gpus = store.gpus()
if is_admin(current_user):
return ok(gpus)
assigned = {(item["node_id"], int(item["gpu_index"])) for item in store.gpu_assignments_for_user(current_user["id"])}
return ok([gpu for gpu in gpus if (gpu.get("node_id"), int(gpu.get("id", gpu.get("gpu_index", -1)))) in assigned])
@router.get("/compute/queue")
@@ -2485,8 +2707,10 @@ async def compute_job_detail(job_id: str) -> dict[str, Any]:
@router.post("/compute/jobs/{job_id}/stop")
async def compute_job_stop(job_id: str) -> dict[str, Any]:
async def compute_job_stop(job_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
task = _task_for_compute_job(job_id)
if task and not has_resource_access("fine-tune", task["id"], current_user, "write"):
raise fail(403, "no permission to stop this compute job")
if not task:
node = _node_for_compute_job_record(job_id)
if not node:
@@ -2521,13 +2745,13 @@ async def compute_job_logs(
@router.post("/compute/jobs/{job_id}/retry")
async def compute_job_retry(job_id: str, payload: dict[str, Any] | None = Body(default=None)) -> dict[str, Any]:
async def compute_job_retry(job_id: str, payload: dict[str, Any] | None = Body(default=None), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
payload = payload or {}
task = _task_for_compute_job(job_id)
if not task:
raise fail(404, "compute job not found")
return await retry_fine_tune(task["id"], payload)
return await retry_fine_tune(task["id"], payload, current_user)
@router.post("/compute/jobs/{job_id}/priority")