From 19ae3ec37f491420bc48321d435036c44d1fd726 Mon Sep 17 00:00:00 2001 From: Harmon Date: Sat, 27 Jun 2026 01:20:06 +0800 Subject: [PATCH] Harden LandSAR cluster transfer e2e --- .env.example | 1 - backend/app/routers/cluster.py | 146 +++--- backend/app/services/cluster_transport.py | 470 ++++++++++++++---- backend/tests/test_cluster_transport.py | 128 +++++ ..._CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md | 17 +- nginx/nginx.conf | 15 + scripts/start_app.ps1 | 5 + 7 files changed, 629 insertions(+), 153 deletions(-) diff --git a/.env.example b/.env.example index 12dd076..d4c79f4 100644 --- a/.env.example +++ b/.env.example @@ -346,7 +346,6 @@ LANDSAR_CLUSTER_ALLOWED_WORKER_IPS=192.168.1.6 # /api/cluster input download and result upload. CLUSTER_SHARED_TOKEN= CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600 -CLUSTER_MATERIALIZE_TEMP_DIR=D:\Task_Pool\_cluster_temp # Remote LandSAR workers should point this to the main backend URL. CLUSTER_MAIN_SERVER_URL= # Empty means the worker can claim all job types. Remote LandSAR nodes should set: diff --git a/backend/app/routers/cluster.py b/backend/app/routers/cluster.py index 7cb6eff..92cbcf0 100644 --- a/backend/app/routers/cluster.py +++ b/backend/app/routers/cluster.py @@ -22,10 +22,10 @@ from fastapi import ( Form, Header, HTTPException, + Query, UploadFile, ) -from fastapi.responses import FileResponse -from starlette.background import BackgroundTask +from fastapi.responses import FileResponse, StreamingResponse from ..config import settings from ..database import get_db @@ -35,28 +35,16 @@ from ..models.orm import ( DinsarProductionRunORM, ) from ..services.cluster_transport import safe_extract_zip +from ..services.cluster_transport import ( + build_cluster_input_manifest, + iter_cluster_input_package_files, + normalize_cluster_relative_path, + stream_zip_files, +) router = APIRouter() -def _cluster_materialize_temp_dir() -> str: - explicit = str(os.environ.get("CLUSTER_MATERIALIZE_TEMP_DIR") or "").strip() - if not explicit: - explicit = str( - getattr(settings, "CLUSTER_MATERIALIZE_TEMP_DIR", "") or "" - ).strip() - if explicit: - return os.path.normpath(explicit) - task_pool = str(getattr(settings, "TASK_POOL_ROOT", "") or "").strip() - if task_pool: - return os.path.normpath(os.path.join(task_pool, "_cluster_temp")) - return os.path.normpath( - os.path.join( - os.path.dirname(__file__), "..", "..", "runtime", "_cluster_temp" - ) - ) - - def _cluster_shared_token() -> str: return str(os.environ.get("CLUSTER_SHARED_TOKEN") or "").strip() @@ -78,6 +66,61 @@ def _require_cluster_token( # Download input package # --------------------------------------------------------------------------- +async def _get_cluster_item_source_dir( + item_id: int, + db: AsyncSession, +) -> tuple[DinsarProductionRunItemORM, str]: + item = await db.get(DinsarProductionRunItemORM, int(item_id)) + if item is None: + raise HTTPException(status_code=404, detail="Cluster item not found.") + + source_dir = os.path.normpath(str(item.source_task_dir or "")) + if not source_dir or not os.path.isdir(source_dir): + raise HTTPException( + status_code=404, + detail=f"Source task directory not found: {source_dir}", + ) + return item, source_dir + + +@router.get("/cluster/input-manifest/{item_id}") +async def get_cluster_input_manifest( + item_id: int, + _cluster_token: None = Depends(_require_cluster_token), + db: AsyncSession = Depends(get_db), +): + _, source_dir = await _get_cluster_item_source_dir(item_id, db) + manifest = build_cluster_input_manifest(source_dir) + if int(manifest.get("file_count") or 0) <= 0: + raise HTTPException( + status_code=404, + detail=f"No packageable LandSAR input files found: {source_dir}", + ) + return manifest + + +@router.get("/cluster/input-file/{item_id}") +async def download_cluster_input_file( + item_id: int, + relative_path: str = Query(...), + _cluster_token: None = Depends(_require_cluster_token), + db: AsyncSession = Depends(get_db), +): + _, source_dir = await _get_cluster_item_source_dir(item_id, db) + normalized_rel = normalize_cluster_relative_path(relative_path) + package_files = { + normalize_cluster_relative_path(rel_path): abs_path + for abs_path, rel_path in iter_cluster_input_package_files(source_dir) + } + source_path = package_files.get(normalized_rel) + if not source_path or not os.path.isfile(source_path): + raise HTTPException(status_code=404, detail="Cluster input file not found.") + return FileResponse( + path=source_path, + media_type="application/octet-stream", + filename=os.path.basename(source_path), + ) + @router.get("/cluster/input-package/{item_id}") async def download_cluster_input_package( item_id: int, @@ -95,56 +138,23 @@ async def download_cluster_input_package( orbit/ ... pair_metadata.json """ - item = await db.get(DinsarProductionRunItemORM, int(item_id)) - if item is None: - raise HTTPException(status_code=404, detail="Cluster item not found.") - - source_dir = os.path.normpath(str(item.source_task_dir or "")) - if not source_dir or not os.path.isdir(source_dir): - raise HTTPException( - status_code=404, - detail=f"Source task directory not found: {source_dir}", - ) + _, source_dir = await _get_cluster_item_source_dir(item_id, db) task_name = os.path.basename(source_dir) - parent_dir = os.path.dirname(source_dir) - temp_root = _cluster_materialize_temp_dir() - os.makedirs(temp_root, exist_ok=True) - - package_root = tempfile.mkdtemp( - prefix=f"cluster_input_{item_id}_", - dir=temp_root, - ) - tmp_base = os.path.join(package_root, task_name) - try: - zip_path = shutil.make_archive( - tmp_base, - "zip", - root_dir=parent_dir, - base_dir=task_name, - ) - except Exception as exc: - try: - shutil.rmtree(package_root) - except Exception: - pass + package_files = list(iter_cluster_input_package_files(source_dir)) + if not package_files: raise HTTPException( - status_code=500, - detail=f"Failed to create input package: {exc}", + status_code=404, + detail=f"No packageable LandSAR input files found: {source_dir}", ) - def _cleanup(): - try: - if os.path.isdir(package_root): - shutil.rmtree(package_root) - except Exception: - pass - - return FileResponse( - path=zip_path, + return StreamingResponse( + stream_zip_files(package_files, top_level_dir=task_name), media_type="application/zip", - filename=f"{task_name}.zip", - background=BackgroundTask(_cleanup), + headers={ + "Content-Disposition": f'attachment; filename="{task_name}.zip"', + "X-Cluster-Package-File-Count": str(len(package_files)), + }, ) @@ -304,7 +314,12 @@ async def upload_cluster_result( f"current pointer was not created: {current_pointer_path or ''}" ) - run = await db.get(DinsarProductionRunORM, str(item.run_id or "")) + run_result = await db.execute( + select(DinsarProductionRunORM).where( + DinsarProductionRunORM.run_id == str(item.run_id or "").strip() + ) + ) + run = run_result.scalar_one_or_none() if run is None: raise RuntimeError(f"Cluster run not found for item {item_id}: {item.run_id}") execution_result = await db.execute( @@ -319,6 +334,7 @@ async def upload_cluster_result( f"Cluster execution not found for item={item.id} run_key={normalized_run_key}" ) execution.output_dir = extract_dir + execution.error_message = None item.latest_output_dir = extract_dir await dinsar_production_service.mark_item_completed( run=run, diff --git a/backend/app/services/cluster_transport.py b/backend/app/services/cluster_transport.py index 79b149f..6e8d24e 100644 --- a/backend/app/services/cluster_transport.py +++ b/backend/app/services/cluster_transport.py @@ -6,16 +6,19 @@ data from the main server and push results back via HTTP. """ from __future__ import annotations +import asyncio import io import json import os +import queue import shutil import tempfile +import threading import urllib.error import urllib.parse import urllib.request import zipfile -from typing import TYPE_CHECKING +from typing import Any, Iterable, Iterator, TYPE_CHECKING if TYPE_CHECKING: from ..models.orm import DinsarProductionRunItemORM, DinsarProductionRunORM @@ -46,6 +49,22 @@ def _cluster_request_headers() -> dict[str, str]: return {"X-Cluster-Token": token} +def normalize_cluster_relative_path(relative_path: Any) -> str: + text = str(relative_path or "").strip().replace("\\", "/") + if not text or text.startswith("/") or os.path.splitdrive(text)[0]: + raise ValueError(f"Unsafe cluster relative path: {relative_path}") + parts = [] + for part in text.split("/"): + if not part or part == ".": + continue + if part == ".." or os.path.splitdrive(part)[0]: + raise ValueError(f"Unsafe cluster relative path: {relative_path}") + parts.append(part) + if not parts: + raise ValueError(f"Unsafe cluster relative path: {relative_path}") + return "/".join(parts) + + def safe_extract_zip(zf: zipfile.ZipFile, target_dir: str) -> None: """Extract a zip after verifying all members stay inside target_dir.""" target_root = os.path.abspath(target_dir) @@ -69,7 +88,7 @@ def safe_extract_zip(zf: zipfile.ZipFile, target_dir: str) -> None: def _zip_directory_contents(source_dir: str, zip_path: str) -> None: source_root = os.path.abspath(source_dir) - with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: + with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_STORED, allowZip64=True) as zf: for current, _, files in os.walk(source_root): for name in files: path = os.path.join(current, name) @@ -77,6 +96,268 @@ def _zip_directory_contents(source_dir: str, zip_path: str) -> None: zf.write(path, arcname) +def _build_multipart_form_data( + *, + fields: dict[str, str], + files: list[tuple[str, str, str, bytes]], + boundary: str, +) -> bytes: + body = io.BytesIO() + + def _write_field( + name: str, + filename: str, + content_type: str, + data: bytes, + ) -> None: + body.write(f"--{boundary}\r\n".encode("utf-8")) + if filename: + body.write( + f'Content-Disposition: form-data; name="{name}"; ' + f'filename="{filename}"\r\n'.encode("utf-8") + ) + body.write(f"Content-Type: {content_type}\r\n".encode("utf-8")) + else: + body.write( + f'Content-Disposition: form-data; name="{name}"\r\n'.encode( + "utf-8" + ) + ) + body.write(b"\r\n") + body.write(data) + body.write(b"\r\n") + + for name, value in fields.items(): + _write_field(name, "", "text/plain", str(value or "").encode("utf-8")) + for name, filename, content_type, data in files: + _write_field(name, filename, content_type, data) + body.write(f"--{boundary}--\r\n".encode("utf-8")) + return body.getvalue() + + +def _is_lt1_source_file(name: str) -> bool: + lower_name = str(name or "").lower() + return lower_name.startswith("lt1") and lower_name.endswith( + (".xml", ".tif", ".tiff", ".jpg", ".jpeg", ".rpc") + ) + + +def _extract_date_from_name(name: str) -> str: + import re + + match = re.search(r"((?:19|20)\d{6})", str(name or "")) + return match.group(1) if match else "" + + +def _has_landsar_input_pair(input_data_dir: str) -> bool: + if not os.path.isdir(input_data_dir): + return False + + by_date: dict[str, set[str]] = {} + for entry in os.scandir(input_data_dir): + if not entry.is_file(): + continue + lower_name = entry.name.lower() + if not lower_name.endswith((".xml", ".tif", ".tiff")): + continue + if lower_name.endswith(".meta.xml") or lower_name.endswith("_check.xml"): + continue + if lower_name.endswith(".xml") and "_slc" not in lower_name: + continue + date_text = _extract_date_from_name(entry.name) + if not date_text: + continue + kinds = by_date.setdefault(date_text, set()) + if lower_name.endswith(".xml"): + kinds.add("xml") + elif lower_name.endswith((".tif", ".tiff")): + kinds.add("tif") + + return sum(1 for kinds in by_date.values() if {"xml", "tif"} <= kinds) >= 2 + + +def _has_direct_landsar_source_file(directory: str) -> bool: + if not os.path.isdir(directory): + return False + try: + for entry in os.scandir(directory): + if entry.is_file() and _is_lt1_source_file(entry.name): + return True + except OSError: + return False + return False + + +def _iter_tree_files(root_dir: str, relative_root: str = "") -> Iterator[tuple[str, str]]: + if not os.path.isdir(root_dir): + return + root_abs = os.path.abspath(root_dir) + for current, _, files in os.walk(root_abs): + for name in sorted(files): + path = os.path.join(current, name) + rel = os.path.relpath(path, root_abs) + if relative_root: + rel = os.path.join(relative_root, rel) + yield path, rel + + +def _iter_direct_files(root_dir: str, relative_root: str = "") -> Iterator[tuple[str, str]]: + if not os.path.isdir(root_dir): + return + for entry in sorted(os.scandir(root_dir), key=lambda item: item.name.lower()): + if not entry.is_file(): + continue + rel = os.path.join(relative_root, entry.name) if relative_root else entry.name + yield entry.path, rel + + +def iter_cluster_input_package_files(source_task_dir: str) -> Iterator[tuple[str, str]]: + """Yield the minimal LandSAR input package file set. + + A Task_Pool item can contain both raw LT-1 ``master/slave`` files and a + derived ``Input_Data`` directory. Shipping both copies is wasteful for + remote workers. Prefer a valid LandSAR ``Input_Data`` pair; otherwise ship + only direct raw files under ``master`` and ``slave``. This keeps the + extracted layout compatible with LandSAR validation while avoiding nested + duplicate scene copies. + """ + source_dir = os.path.abspath(source_task_dir) + if not os.path.isdir(source_dir): + return + + pair_meta = os.path.join(source_dir, ".dinsar_pair.json") + if os.path.isfile(pair_meta): + yield pair_meta, ".dinsar_pair.json" + + input_data_dir = os.path.join(source_dir, "Input_Data") + if _has_landsar_input_pair(input_data_dir): + yield from _iter_tree_files(input_data_dir, "Input_Data") + else: + master_dir = os.path.join(source_dir, "master") + slave_dir = os.path.join(source_dir, "slave") + if _has_direct_landsar_source_file(master_dir) and _has_direct_landsar_source_file(slave_dir): + yield from _iter_direct_files(master_dir, "master") + yield from _iter_direct_files(slave_dir, "slave") + else: + yield from _iter_tree_files(source_dir) + return + + orbit_dir = os.path.join(source_dir, "orbit") + if os.path.isdir(orbit_dir): + yield from _iter_tree_files(orbit_dir, "orbit") + + +def build_cluster_input_manifest(source_task_dir: str) -> dict[str, Any]: + files = [] + total_bytes = 0 + for abs_path, rel_path in iter_cluster_input_package_files(source_task_dir): + if not os.path.isfile(abs_path): + continue + normalized_rel = normalize_cluster_relative_path(rel_path) + stat = os.stat(abs_path) + size = int(stat.st_size) + total_bytes += size + files.append( + { + "relative_path": normalized_rel, + "size": size, + "mtime": float(stat.st_mtime), + } + ) + return { + "task_name": os.path.basename(os.path.normpath(source_task_dir)), + "file_count": len(files), + "total_bytes": total_bytes, + "files": files, + } + + +class _QueueZipWriter: + def __init__(self, output_queue: "queue.Queue[object]") -> None: + self._queue = output_queue + self._closed = False + + def write(self, data: bytes) -> int: + if self._closed: + raise BrokenPipeError("zip stream is closed") + chunk = bytes(data) + if not chunk: + return 0 + while not self._closed: + try: + self._queue.put(chunk, timeout=1) + return len(chunk) + except queue.Full: + continue + raise BrokenPipeError("zip stream is closed") + + def flush(self) -> None: + return None + + def close(self) -> None: + self._closed = True + + +def stream_zip_files( + files: Iterable[tuple[str, str]], + *, + top_level_dir: str, +) -> Iterator[bytes]: + """Stream a zip archive for *files* without materializing it first.""" + output_queue: "queue.Queue[object]" = queue.Queue(maxsize=8) + sentinel = object() + writer = _QueueZipWriter(output_queue) + safe_top = str(top_level_dir or "Task").strip().strip("/\\") or "Task" + + def _put_control(item: object) -> None: + while True: + try: + output_queue.put(item, timeout=1) + return + except queue.Full: + if writer._closed: + return + + def _producer() -> None: + try: + with zipfile.ZipFile( + writer, + mode="w", + compression=zipfile.ZIP_STORED, + allowZip64=True, + ) as zf: + for abs_path, rel_path in files: + if writer._closed: + raise BrokenPipeError("zip stream is closed") + if not os.path.isfile(abs_path): + continue + arcname = os.path.join(safe_top, rel_path).replace("\\", "/") + zf.write(abs_path, arcname) + except Exception as exc: + if not writer._closed: + _put_control(exc) + finally: + _put_control(sentinel) + + producer = threading.Thread( + target=_producer, + name="cluster-input-zip-stream", + daemon=True, + ) + producer.start() + + try: + while True: + item = output_queue.get() + if item is sentinel: + break + if isinstance(item, Exception): + raise item + yield item # type: ignore[misc] + finally: + writer.close() + + def _resolve_cluster_server_url() -> str: """Return the main-server HTTP base URL for cluster data transport. @@ -142,58 +423,92 @@ async def materialize_cluster_input( local_task_dir: str, task_id: str, ) -> None: - """Download and extract the input data for a cluster item. - - Calls ``GET /api/cluster/input-package/{item_id}`` on the main - server, retrieves a zip containing the Task_Pool directory tree, - and extracts it so that *local_task_dir* exists locally. - """ + """Download the input data for a cluster item file by file.""" from .task_service import task_service server_url = _resolve_cluster_server_url() - download_url = f"{server_url}/api/cluster/input-package/{item.id}" + manifest_url = f"{server_url}/api/cluster/input-manifest/{item.id}" parent_dir = os.path.dirname(local_task_dir) task_name = os.path.basename(local_task_dir) - package_task_name = os.path.basename(str(item.source_task_dir or "")) or task_name + staging_dir = os.path.join(parent_dir, f".{task_name}.download") await task_service.add_log( task_id, "INFO", - f"[cluster] Downloading input data from {download_url} ...", + f"[cluster] Fetching input manifest from {manifest_url} ...", ) - tmp_zip = os.path.join( - tempfile.mkdtemp(prefix=f"cluster_input_{item.id}_"), - f"{package_task_name}.zip", - ) - try: + def _request_json(url: str) -> dict[str, Any]: req = urllib.request.Request( - download_url, + url, headers=_cluster_request_headers(), method="GET", ) with urllib.request.urlopen(req, timeout=_cluster_transfer_timeout()) as resp: - with open(tmp_zip, "wb") as fh: + return json.loads(resp.read().decode("utf-8")) + + def _download_file(relative_path: str, expected_size: int) -> None: + normalized_rel = normalize_cluster_relative_path(relative_path) + file_url = ( + f"{server_url}/api/cluster/input-file/{item.id}" + f"?relative_path={urllib.parse.quote(normalized_rel, safe='')}" + ) + target_path = os.path.join(staging_dir, *normalized_rel.split("/")) + tmp_path = f"{target_path}.part" + os.makedirs(os.path.dirname(target_path), exist_ok=True) + req = urllib.request.Request( + file_url, + headers=_cluster_request_headers(), + method="GET", + ) + with urllib.request.urlopen(req, timeout=_cluster_transfer_timeout()) as resp: + with open(tmp_path, "wb") as fh: shutil.copyfileobj(resp, fh, 8 * 1024 * 1024) - - os.makedirs(parent_dir, exist_ok=True) - if os.path.isdir(local_task_dir): - shutil.rmtree(local_task_dir) - with zipfile.ZipFile(tmp_zip, "r") as zf: - safe_extract_zip(zf, parent_dir) - - extracted_task_dir = os.path.join(parent_dir, package_task_name) - if ( - os.path.isdir(extracted_task_dir) - and os.path.normcase(os.path.abspath(extracted_task_dir)) - != os.path.normcase(os.path.abspath(local_task_dir)) - ): - os.replace(extracted_task_dir, local_task_dir) - - if not os.path.isdir(local_task_dir): + actual_size = os.path.getsize(tmp_path) + if expected_size >= 0 and actual_size != expected_size: raise RuntimeError( - f"Extraction did not create expected directory: {local_task_dir}" + f"Downloaded file size mismatch for {normalized_rel}: " + f"expected={expected_size} actual={actual_size}" ) + os.replace(tmp_path, target_path) + + try: + manifest = await asyncio.to_thread(_request_json, manifest_url) + files = list(manifest.get("files") or []) + if not files: + raise RuntimeError(f"Input manifest contains no files: {manifest_url}") + + total_bytes = int(manifest.get("total_bytes") or 0) + await task_service.add_log( + task_id, + "INFO", + f"[cluster] Downloading input files: count={len(files)} bytes={total_bytes}", + ) + + if os.path.isdir(staging_dir): + await asyncio.to_thread(shutil.rmtree, staging_dir) + os.makedirs(staging_dir, exist_ok=True) + + for index, file_info in enumerate(files, start=1): + rel_path = normalize_cluster_relative_path(file_info.get("relative_path")) + expected_size = int(file_info.get("size") or -1) + await task_service.add_log( + task_id, + "INFO", + f"[cluster] Downloading input file {index}/{len(files)}: " + f"{rel_path} ({expected_size} bytes)", + ) + await asyncio.to_thread(_download_file, rel_path, expected_size) + await task_service.add_log( + task_id, + "INFO", + f"[cluster] Downloaded input file {index}/{len(files)}: {rel_path}", + ) + + if os.path.isdir(local_task_dir): + await asyncio.to_thread(shutil.rmtree, local_task_dir) + os.makedirs(parent_dir, exist_ok=True) + os.replace(staging_dir, local_task_dir) await task_service.add_log( task_id, @@ -216,9 +531,8 @@ async def materialize_cluster_input( ) from exc finally: try: - tmp_root = os.path.dirname(tmp_zip) - if os.path.isdir(tmp_root): - shutil.rmtree(tmp_root) + if os.path.isdir(staging_dir): + shutil.rmtree(staging_dir) except Exception: pass @@ -247,51 +561,32 @@ async def upload_cluster_result( tmp_root = tempfile.mkdtemp(prefix=f"cluster_result_{item.id}_") tmp_zip = os.path.join(tmp_root, "result.zip") - try: + + def _package_and_upload() -> dict: _zip_directory_contents(managed_run_dir, tmp_zip) - await task_service.add_log( - task_id, - "INFO", - "[cluster] Uploading results to main server ...", - ) - boundary = "----ClusterUploadBoundary" - body = io.BytesIO() - - def _write_field(name, filename, content_type, data): - body.write(f"--{boundary}\r\n".encode("utf-8")) - if filename: - body.write( - f'Content-Disposition: form-data; name="{name}"; ' - f'filename="{filename}"\r\n'.encode("utf-8") - ) - body.write(f"Content-Type: {content_type}\r\n".encode("utf-8")) - else: - body.write( - f'Content-Disposition: form-data; name="{name}"\r\n'.encode( - "utf-8" - ) - ) - body.write(b"\r\n") - body.write(data) - body.write(b"\r\n") - - _write_field("run_id", "", "text/plain", (run.run_id or "").encode("utf-8")) - _write_field("run_key", "", "text/plain", str(run_key or "").encode("utf-8")) with open(tmp_zip, "rb") as fh: - _write_field( - "result_zip", - os.path.basename(tmp_zip), - "application/zip", - fh.read(), + body = _build_multipart_form_data( + fields={ + "run_id": str(run.run_id or ""), + "run_key": str(run_key or ""), + }, + files=[ + ( + "result_zip", + os.path.basename(tmp_zip), + "application/zip", + fh.read(), + ), + ], + boundary=boundary, ) - body.write(f"--{boundary}--\r\n".encode("utf-8")) upload_url = f"{server_url}/api/cluster/upload-result/{item.id}" req = urllib.request.Request( upload_url, - data=body.getvalue(), + data=body, headers={ "Content-Type": f"multipart/form-data; boundary={boundary}", **_cluster_request_headers(), @@ -300,15 +595,24 @@ async def upload_cluster_result( ) with urllib.request.urlopen(req, timeout=_cluster_transfer_timeout()) as resp: - result = json.loads(resp.read().decode("utf-8")) - await task_service.add_log( - task_id, - "INFO", - f"[cluster] Results uploaded: " - f"registered={result.get('registered', False)} " - f"processed={result.get('processed', 0)}", - ) - return bool(result.get("registered", False)) + return json.loads(resp.read().decode("utf-8")) + + try: + await task_service.add_log( + task_id, + "INFO", + "[cluster] Uploading results to main server ...", + ) + + result = await asyncio.to_thread(_package_and_upload) + await task_service.add_log( + task_id, + "INFO", + f"[cluster] Results uploaded: " + f"registered={result.get('registered', False)} " + f"processed={result.get('processed', 0)}", + ) + return bool(result.get("registered", False)) except urllib.error.HTTPError as exc: body_text = "" diff --git a/backend/tests/test_cluster_transport.py b/backend/tests/test_cluster_transport.py index 615567e..ec4613e 100644 --- a/backend/tests/test_cluster_transport.py +++ b/backend/tests/test_cluster_transport.py @@ -3,20 +3,63 @@ import os import tempfile import unittest import zipfile +from io import BytesIO from types import SimpleNamespace from unittest import mock +from multipart import parse_form + from backend.app.services.cluster_transport import ( + _build_multipart_form_data, _zip_directory_contents, + build_cluster_input_manifest, + iter_cluster_input_package_files, + normalize_cluster_relative_path, resolve_cluster_local_run_dir, resolve_cluster_local_task_dir, safe_extract_zip, + stream_zip_files, ) from backend.app.services.dinsar_completion_files import repair_managed_completion_files from backend.app.services.dinsar_naming import RUN_META_FILENAME class ClusterTransportTests(unittest.TestCase): + def test_build_multipart_form_data_is_parseable(self): + boundary = "----ClusterUploadBoundary" + body = _build_multipart_form_data( + fields={"run_id": "run-1", "run_key": "key-1"}, + files=[("result_zip", "result.zip", "application/zip", b"zip-bytes")], + boundary=boundary, + ) + fields = {} + files = {} + + def on_field(field): + fields[field.field_name.decode("utf-8")] = field.value.decode("utf-8") + + def on_file(file): + files[file.field_name.decode("utf-8")] = { + "file_name": file.file_name.decode("utf-8"), + "size": file.size, + "content": file.file_object.getvalue(), + } + + parse_form( + { + "Content-Type": f"multipart/form-data; boundary={boundary}".encode("utf-8"), + "Content-Length": str(len(body)).encode("utf-8"), + }, + BytesIO(body), + on_field, + on_file, + ) + + self.assertEqual(fields, {"run_id": "run-1", "run_key": "key-1"}) + self.assertEqual(files["result_zip"]["file_name"], "result.zip") + self.assertEqual(files["result_zip"]["size"], len(b"zip-bytes")) + self.assertEqual(files["result_zip"]["content"], b"zip-bytes") + def test_zip_directory_contents_excludes_top_level_run_dir(self): with tempfile.TemporaryDirectory() as root: run_dir = os.path.join(root, "run_abc") @@ -40,6 +83,91 @@ class ClusterTransportTests(unittest.TestCase): self.assertIn("native/raw.txt", names) self.assertFalse(any(name.startswith("run_abc/") for name in names)) + def test_cluster_input_package_prefers_input_data(self): + with tempfile.TemporaryDirectory() as root: + task_dir = os.path.join(root, "Task_20260101_20260113") + input_dir = os.path.join(task_dir, "Input_Data") + master_dir = os.path.join(task_dir, "master") + slave_dir = os.path.join(task_dir, "slave") + orbit_dir = os.path.join(task_dir, "orbit") + os.makedirs(input_dir) + os.makedirs(master_dir) + os.makedirs(slave_dir) + os.makedirs(orbit_dir) + with open(os.path.join(task_dir, ".dinsar_pair.json"), "w", encoding="utf-8") as fp: + json.dump({"pair_key": "pair"}, fp) + for date_text in ("20260101", "20260113"): + base = f"LT1A_MONO_TEST_{date_text}_SLC" + with open(os.path.join(input_dir, f"{base}.xml"), "wb") as fp: + fp.write(b"xml") + with open(os.path.join(input_dir, f"{base}.tiff"), "wb") as fp: + fp.write(b"tif") + with open(os.path.join(master_dir, "LT1A_raw_20260101.tiff"), "wb") as fp: + fp.write(b"raw-master") + with open(os.path.join(slave_dir, "LT1A_raw_20260113.tiff"), "wb") as fp: + fp.write(b"raw-slave") + with open(os.path.join(orbit_dir, "orbit.txt"), "wb") as fp: + fp.write(b"orbit") + + names = {rel.replace("\\", "/") for _, rel in iter_cluster_input_package_files(task_dir)} + + self.assertIn(".dinsar_pair.json", names) + self.assertIn("Input_Data/LT1A_MONO_TEST_20260101_SLC.xml", names) + self.assertIn("Input_Data/LT1A_MONO_TEST_20260113_SLC.tiff", names) + self.assertIn("orbit/orbit.txt", names) + self.assertFalse(any(name.startswith("master/") for name in names)) + self.assertFalse(any(name.startswith("slave/") for name in names)) + + def test_cluster_input_manifest_reports_files_and_size(self): + with tempfile.TemporaryDirectory() as root: + task_dir = os.path.join(root, "Task_20260101_20260113") + input_dir = os.path.join(task_dir, "Input_Data") + os.makedirs(input_dir) + total = 0 + for date_text, payload in (("20260101", b"xml1"), ("20260113", b"xml2")): + base = f"LT1A_MONO_TEST_{date_text}_SLC" + for suffix, content in ((".xml", payload), (".tiff", payload * 2)): + path = os.path.join(input_dir, f"{base}{suffix}") + with open(path, "wb") as fp: + fp.write(content) + total += len(content) + + manifest = build_cluster_input_manifest(task_dir) + + self.assertEqual(manifest["task_name"], "Task_20260101_20260113") + self.assertEqual(manifest["file_count"], 4) + self.assertEqual(manifest["total_bytes"], total) + self.assertTrue( + all("\\" not in item["relative_path"] for item in manifest["files"]) + ) + + def test_normalize_cluster_relative_path_rejects_escape(self): + for value in ("../a.txt", "a/../../b.txt", "/abs.txt", r"C:\abs.txt"): + with self.subTest(value=value): + with self.assertRaises(ValueError): + normalize_cluster_relative_path(value) + + self.assertEqual( + normalize_cluster_relative_path(r"Input_Data\scene.tif"), + "Input_Data/scene.tif", + ) + + def test_stream_zip_files_preserves_task_top_level(self): + with tempfile.TemporaryDirectory() as root: + payload_path = os.path.join(root, "payload.txt") + with open(payload_path, "wb") as fp: + fp.write(b"payload") + + zip_bytes = b"".join( + stream_zip_files([(payload_path, "Input_Data/payload.txt")], top_level_dir="Task_A") + ) + zip_path = os.path.join(root, "streamed.zip") + with open(zip_path, "wb") as fp: + fp.write(zip_bytes) + + with zipfile.ZipFile(zip_path, "r") as zf: + self.assertEqual(zf.read("Task_A/Input_Data/payload.txt"), b"payload") + def test_safe_extract_zip_rejects_path_escape(self): with tempfile.TemporaryDirectory() as root: zip_path = os.path.join(root, "unsafe.zip") diff --git a/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md b/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md index 2c13c6b..ef483c9 100644 --- a/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md +++ b/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md @@ -189,10 +189,9 @@ ```env # 集群传输配置 - CLUSTER_SHARED_TOKEN= - CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600 - CLUSTER_MATERIALIZE_TEMP_DIR=D:\Task_Pool\_cluster_temp - ``` +CLUSTER_SHARED_TOKEN= +CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600 +``` Worker 端 `.env` 新增（或保持现有模板字段）： @@ -241,3 +240,13 @@ - [LANDSAR_DEM_PREPARATION_CONTRACT_20260618.md](LANDSAR_DEM_PREPARATION_CONTRACT_20260618.md) — LandSAR DEM 准备约定 - [THREE_SENSOR_LOCAL_PRODUCTION_CONTRACT_20260616.md](THREE_SENSOR_LOCAL_PRODUCTION_CONTRACT_20260616.md) — 三数据本机生产约定 - [DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md](DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md) — Task_Pool 与三引擎 refactor + +## 2026-06-26 Remote 1.6 E2E Findings + +- Remote worker `landsar-worker-192-168-1-6` can reach PostgreSQL on `192.168.1.62:5432`, authenticate to `/api/cluster/...`, heartbeat into `system_worker_heartbeats`, and claim `LANDSAR_CLUSTER_ITEM` jobs. +- The first remote E2E attempt failed before LandSAR execution with `HTTP 504` while downloading `/api/cluster/input-package/{item_id}` through nginx. +- Root cause: the old input endpoint synchronously built a full Task directory zip inside the FastAPI request. The tested Task directory was 15.47 GB because it contained duplicate `Input_Data`, `master`, and `slave` copies. nginx's default 60s upstream timeout returned 504 while the backend was still packaging, and the synchronous packaging could also block normal UI API responses. +- Current behavior: the input endpoint streams a ZIP directly to the worker and uses `ZIP_STORED` to avoid wasting CPU on already-compressed TIFF data. +- Current package selection: if `Input_Data` contains a valid LandSAR SLC xml/tif pair, only `Input_Data`, `orbit`, and `.dinsar_pair.json` are shipped. Otherwise, only direct raw files from `master` and `slave`, plus `orbit` and `.dinsar_pair.json`, are shipped. For item `135`, this reduced the transfer set from 15.47 GB to 5.18 GB. +- nginx now has a dedicated `/api/cluster/` location with large body support, disabled proxy buffering for cluster transfers, and 7200s send/read timeouts. `scripts/start_app.ps1` also rewrites this location's backend port when `PORT` changes. +- Upload registration now resolves `DinsarProductionRunORM` by business `run_id`; using `db.get()` with the string run id was invalid because the table primary key is integer. diff --git a/nginx/nginx.conf b/nginx/nginx.conf index 8a08c86..935569c 100644 --- a/nginx/nginx.conf +++ b/nginx/nginx.conf @@ -66,6 +66,21 @@ http { try_files $uri $uri/ /index.html; } + location /api/cluster/ { + proxy_pass http://127.0.0.1:18000; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + client_max_body_size 64g; + proxy_request_buffering off; + proxy_buffering off; + proxy_read_timeout 7200s; + proxy_send_timeout 7200s; + send_timeout 7200s; + } + location /api/ { proxy_pass http://127.0.0.1:18000; proxy_http_version 1.1; diff --git a/scripts/start_app.ps1 b/scripts/start_app.ps1 index 8cbaf57..8f68bc2 100644 --- a/scripts/start_app.ps1 +++ b/scripts/start_app.ps1 @@ -709,6 +709,11 @@ if (Test-Path -LiteralPath "$NginxConfPath") { '(location\s+/api/tasks/active/stream\s*\{[\s\S]*?proxy_pass\s+)http://(127\.0\.0\.1|localhost):\d+(;)', "`${1}$BackendProxy`${3}" ) + $NewConfContent = [regex]::Replace( + $NewConfContent, + '(location\s+/api/cluster/\s*\{[\s\S]*?proxy_pass\s+)http://(127\.0\.0\.1|localhost):\d+(;)', + "`${1}$BackendProxy`${3}" + ) # 使用 UTF8 无 BOM 编码写入 $Utf8NoBom = New-Object System.Text.UTF8Encoding $false [System.IO.File]::WriteAllText("$NginxConfPath", $NewConfContent, $Utf8NoBom)