diff --git a/backend/app/config.py b/backend/app/config.py index 9dcc639..a69df4f 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -312,6 +312,7 @@ class Settings(BaseSettings): WSL_BROKER_JOB_ROOT: str = "" ISCE2_RUNTIME_ID: str = "" PYINT_RUNTIME_ID: str = "" + LANDSAR_RUNTIME_ID: str = "" ISCE2_ENABLED: bool = False ISCE2_WSL_DISTRO: str = "Ubuntu-24.04" @@ -678,6 +679,8 @@ class Settings(BaseSettings): object.__setattr__(self, "ISCE2_RUNTIME_ID", "isce2_runtime_v1") if not self.PYINT_RUNTIME_ID: object.__setattr__(self, "PYINT_RUNTIME_ID", "gamma_pyint_runtime_v1") + if not self.LANDSAR_RUNTIME_ID: + object.__setattr__(self, "LANDSAR_RUNTIME_ID", "landsar_runtime_v1") if not self.PYINT_WSL_DISTRO: object.__setattr__(self, "PYINT_WSL_DISTRO", self.WSL_DISTRO or self.ISCE2_WSL_DISTRO) if not self.PYINT_WSL_PYTHON: diff --git a/backend/app/dinsar_engines/landsar_engine.py b/backend/app/dinsar_engines/landsar_engine.py index 4bc6113..d307bae 100644 --- a/backend/app/dinsar_engines/landsar_engine.py +++ b/backend/app/dinsar_engines/landsar_engine.py @@ -204,6 +204,7 @@ _SYSTEM_EXTRA_KEYS = { "__managed_orbit_output_dir", "__managed_run_key", "__source_root_override", + "__source_task_dir_override", "__rerun_mode", "__validated_task_count", "__validated_mode", @@ -2318,7 +2319,7 @@ class LandsarEngine(DinsarEngine): "engine_code": self.engine_code, "profile_code": request.profile, "source_root": _norm_path(request.extra.get("__source_root_override") or request.root_dir), - "task_dir": _norm_path(task_dir), + "task_dir": _norm_path(request.extra.get("__source_task_dir_override") or task_dir), "work_dir": landsar_output_dir, "output_dir": _norm_path(run_dir), "native_output_dir": _norm_path(native_output_dir), diff --git a/backend/app/routers/cluster.py b/backend/app/routers/cluster.py index 896f83d..7cb6eff 100644 --- a/backend/app/routers/cluster.py +++ b/backend/app/routers/cluster.py @@ -13,6 +13,8 @@ import tempfile import zipfile from typing import Optional +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession from fastapi import ( APIRouter, Depends, @@ -24,11 +26,14 @@ from fastapi import ( ) from fastapi.responses import FileResponse from starlette.background import BackgroundTask -from sqlalchemy.ext.asyncio import AsyncSession from ..config import settings from ..database import get_db -from ..models.orm import DinsarProductionRunItemORM +from ..models.orm import ( + DinsarProductionExecutionORM, + DinsarProductionRunItemORM, + DinsarProductionRunORM, +) from ..services.cluster_transport import safe_extract_zip router = APIRouter() @@ -263,17 +268,84 @@ async def upload_cluster_result( # ----- catalog registration ------------------------------------------------ from ..services.result_catalog_service import result_catalog_service as rcs + from ..services.dinsar_production_service import dinsar_production_service try: publish_result = await rcs.publish_from_sources(db, [extract_dir]) processed = int(publish_result.get("processed", 0) or 0) + failed = int(publish_result.get("failed", 0) or 0) if processed > 0: await rcs.rebuild_catalog(db, full_rebuild=True) + if processed != 1 or failed != 0: + raise RuntimeError( + f"expected processed=1 failed=0, got processed={processed} failed={failed}" + ) + + details = publish_result.get("details") if isinstance(publish_result, dict) else [] + detail = next( + ( + item_detail + for item_detail in (details or []) + if str(item_detail.get("run_key") or "").strip() == normalized_run_key + ), + (details or [{}])[0] if details else {}, + ) + execution_manifest_path = str( + detail.get("execution_manifest_path") + or os.path.join(extract_dir, "execution_manifest.json") + ) + current_pointer_path = str(detail.get("current_pointer_path") or "") + if not os.path.isfile(execution_manifest_path): + raise RuntimeError( + f"execution manifest was not created: {execution_manifest_path}" + ) + if not current_pointer_path or not os.path.isfile(current_pointer_path): + raise RuntimeError( + f"current pointer was not created: {current_pointer_path or ''}" + ) + + run = await db.get(DinsarProductionRunORM, str(item.run_id or "")) + if run is None: + raise RuntimeError(f"Cluster run not found for item {item_id}: {item.run_id}") + execution_result = await db.execute( + select(DinsarProductionExecutionORM).where( + DinsarProductionExecutionORM.item_id == item.id, + DinsarProductionExecutionORM.run_key == normalized_run_key, + ) + ) + execution = execution_result.scalar_one_or_none() + if execution is None: + raise RuntimeError( + f"Cluster execution not found for item={item.id} run_key={normalized_run_key}" + ) + execution.output_dir = extract_dir + item.latest_output_dir = extract_dir + await dinsar_production_service.mark_item_completed( + run=run, + item=item, + execution=execution, + manifest_path=execution_manifest_path, + metrics={ + "cluster_upload": True, + "catalog_processed": processed, + "catalog_failed": failed, + "current_pointer_path": current_pointer_path, + }, + db=db, + ) + final_status = await dinsar_production_service.finalize_cluster_run_if_complete( + run, + db=db, + ) return { "registered": processed > 0, + "completed": True, + "final_status": final_status, "processed": processed, - "failed": int(publish_result.get("failed", 0) or 0), + "failed": failed, "catalog_path": extract_dir, + "execution_manifest_path": execution_manifest_path, + "current_pointer_path": current_pointer_path, } except Exception as exc: raise HTTPException( diff --git a/backend/app/services/cluster_transport.py b/backend/app/services/cluster_transport.py index 14ed155..79b149f 100644 --- a/backend/app/services/cluster_transport.py +++ b/backend/app/services/cluster_transport.py @@ -67,6 +67,16 @@ def safe_extract_zip(zf: zipfile.ZipFile, target_dir: str) -> None: zf.extractall(target_root) +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: + for current, _, files in os.walk(source_root): + for name in files: + path = os.path.join(current, name) + arcname = os.path.relpath(path, source_root) + zf.write(path, arcname) + + def _resolve_cluster_server_url() -> str: """Return the main-server HTTP base URL for cluster data transport. @@ -99,23 +109,52 @@ def _is_remote_worker() -> bool: return host not in {"", "127.0.0.1", "localhost"} +def resolve_cluster_local_task_dir(item: DinsarProductionRunItemORM) -> str: + """Return the worker-local Task_* directory for a cluster item.""" + source_task_dir = os.path.normpath(str(item.source_task_dir or "")) + worker_root = _read_cluster_env("CLUSTER_WORKER_TASK_ROOT") + if not worker_root: + return source_task_dir + task_name = os.path.basename(source_task_dir) or f"Task_item_{item.id}" + return os.path.normpath( + os.path.join(worker_root, f"item_{item.id}", task_name) + ) + + +def resolve_cluster_local_run_dir( + item: DinsarProductionRunItemORM, + run_key: str, +) -> str: + """Return the worker-local managed result run directory.""" + worker_root = _read_cluster_env("CLUSTER_WORKER_RESULT_ROOT") + if not worker_root: + return os.path.join(str(item.results_root_dir or ""), "runs", run_key) + pair_fragment = str(item.pair_key or f"item_{item.id}").strip() or f"item_{item.id}" + safe_pair = "".join( + ch if ch.isalnum() or ch in "._-" else "_" + for ch in pair_fragment + ).strip("._") or f"item_{item.id}" + return os.path.normpath(os.path.join(worker_root, safe_pair, "runs", run_key)) + + async def materialize_cluster_input( item: DinsarProductionRunItemORM, - source_task_dir: str, + 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 *source_task_dir* exists locally. + and extracts it so that *local_task_dir* exists locally. """ from .task_service import task_service server_url = _resolve_cluster_server_url() download_url = f"{server_url}/api/cluster/input-package/{item.id}" - parent_dir = os.path.dirname(source_task_dir) - task_name = os.path.basename(source_task_dir) + 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 await task_service.add_log( task_id, @@ -124,8 +163,8 @@ async def materialize_cluster_input( ) tmp_zip = os.path.join( - tempfile.gettempdir(), - f"cluster_input_{item.id}_{task_name}.zip", + tempfile.mkdtemp(prefix=f"cluster_input_{item.id}_"), + f"{package_task_name}.zip", ) try: req = urllib.request.Request( @@ -138,18 +177,28 @@ async def materialize_cluster_input( 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) - if not os.path.isdir(source_task_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): raise RuntimeError( - f"Extraction did not create expected directory: {source_task_dir}" + f"Extraction did not create expected directory: {local_task_dir}" ) await task_service.add_log( task_id, "INFO", - f"[cluster] Input data ready: {source_task_dir}", + f"[cluster] Input data ready: {local_task_dir}", ) except urllib.error.HTTPError as exc: body_text = "" @@ -167,7 +216,9 @@ async def materialize_cluster_input( ) from exc finally: try: - os.unlink(tmp_zip) + tmp_root = os.path.dirname(tmp_zip) + if os.path.isdir(tmp_root): + shutil.rmtree(tmp_root) except Exception: pass @@ -194,19 +245,10 @@ async def upload_cluster_result( "[cluster] Packaging results for upload ...", ) - run_dir_name = os.path.basename(os.path.normpath(managed_run_dir)) - tmp_zip = os.path.join( - tempfile.gettempdir(), - f"cluster_result_{item.id}.zip", - ) + tmp_root = tempfile.mkdtemp(prefix=f"cluster_result_{item.id}_") + tmp_zip = os.path.join(tmp_root, "result.zip") try: - parent = os.path.dirname(managed_run_dir) - shutil.make_archive( - tmp_zip.replace(".zip", ""), - "zip", - root_dir=parent, - base_dir=run_dir_name, - ) + _zip_directory_contents(managed_run_dir, tmp_zip) await task_service.add_log( task_id, @@ -284,7 +326,7 @@ async def upload_cluster_result( ) from exc finally: try: - if os.path.isfile(tmp_zip): - os.unlink(tmp_zip) + if os.path.isdir(tmp_root): + shutil.rmtree(tmp_root) except Exception: pass diff --git a/backend/app/services/dinsar_completion_files.py b/backend/app/services/dinsar_completion_files.py index a70a69a..815e069 100644 --- a/backend/app/services/dinsar_completion_files.py +++ b/backend/app/services/dinsar_completion_files.py @@ -71,9 +71,11 @@ def _current_pointer_path(results_root_dir: str, *, engine_code: str, profile_co def _runtime_id_for_engine(engine_code: str) -> Optional[str]: normalized = str(engine_code or "").strip().lower() if normalized == "isce2": - return settings.ISCE2_RUNTIME_ID or None + return getattr(settings, "ISCE2_RUNTIME_ID", "") or None + if normalized == "landsar": + return getattr(settings, "LANDSAR_RUNTIME_ID", "") or None if normalized in {"pyint", "gamma"}: - return settings.PYINT_RUNTIME_ID or None + return getattr(settings, "PYINT_RUNTIME_ID", "") or None return None @@ -137,9 +139,17 @@ def repair_managed_completion_files( task_name = _first_text(payload.get("task_name"), payload.get("task_alias"), run_key) task_alias = _first_text(payload.get("task_alias"), payload.get("task_name"), task_name) output_dir = _normalize_path(payload.get("output_dir") or normalized_run_dir) + if normalized_primary.startswith(normalized_run_dir + os.sep): + output_dir = normalized_run_dir native_output_dir = _normalize_path( payload.get("native_output_dir") or os.path.join(normalized_run_dir, "native") ) + if not native_output_dir.startswith(normalized_run_dir + os.sep): + local_native_dir = os.path.join(normalized_run_dir, "native") + if os.path.isdir(local_native_dir): + native_output_dir = _normalize_path(local_native_dir) + else: + native_output_dir = output_dir results_root_dir = _infer_results_root_dir(normalized_run_dir, pair_key) publish_root_dir = _normalize_path(os.path.dirname(results_root_dir)) diff --git a/backend/app/services/dinsar_production_service.py b/backend/app/services/dinsar_production_service.py index f4c92e9..e9759bd 100644 --- a/backend/app/services/dinsar_production_service.py +++ b/backend/app/services/dinsar_production_service.py @@ -1184,9 +1184,12 @@ class DinsarProductionService: run: DinsarProductionRunORM, item: DinsarProductionRunItemORM, run_key: str, + output_dir_override: Optional[str] = None, db: AsyncSession, ) -> DinsarProductionExecutionORM: - output_dir = _execution_dir(item, run_key) + output_dir = os.path.normpath( + os.path.abspath(output_dir_override) + ) if output_dir_override else _execution_dir(item, run_key) _ensure_dir(output_dir) execution = DinsarProductionExecutionORM( execution_id=run_key, diff --git a/backend/app/services/job_handlers.py b/backend/app/services/job_handlers.py index 6a300be..621aa0c 100644 --- a/backend/app/services/job_handlers.py +++ b/backend/app/services/job_handlers.py @@ -26,6 +26,8 @@ from .asset_inventory_service import asset_inventory_service from .cluster_transport import ( _is_remote_worker, materialize_cluster_input, + resolve_cluster_local_run_dir, + resolve_cluster_local_task_dir, upload_cluster_result, ) from .dinsar_compat_service import dinsar_compat_service @@ -164,15 +166,30 @@ def _normalize_positive_int(value: Any) -> Optional[int]: def _cluster_source_task_dir_ready(path: str) -> bool: if not path or not os.path.isdir(path): return False + + def _has_landsar_source_file(directory: str) -> bool: + try: + with os.scandir(directory) as entries: + for entry in entries: + if not entry.is_file(): + continue + name = entry.name.lower() + if name.startswith("lt1") and name.endswith((".xml", ".tif", ".tiff")): + return True + except OSError: + return False + return False + + input_data_dir = os.path.join(path, "Input_Data") + if os.path.isdir(input_data_dir) and _has_landsar_source_file(input_data_dir): + return True + + if not os.path.isfile(os.path.join(path, ".dinsar_pair.json")): + return False + for child_name in ("master", "slave"): child_dir = os.path.join(path, child_name) - if not os.path.isdir(child_dir): - return False - try: - with os.scandir(child_dir) as entries: - if not any(entries): - return False - except OSError: + if not os.path.isdir(child_dir) or not _has_landsar_source_file(child_dir): return False return True @@ -3275,14 +3292,14 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None: raise ValueError(f"LandSAR cluster item not found: {item_id}") await db.refresh(item) - source_task_dir = os.path.normpath(str(item.source_task_dir or "")) - if source_task_dir and not _cluster_source_task_dir_ready(source_task_dir): - await materialize_cluster_input(item, source_task_dir, job.task_id) - item_status = str(item.status or "").strip().upper() if item_status in {"COMPLETED", "FAILED", "SKIPPED", "CANCELLED"}: await dinsar_production_service.finalize_cluster_run_if_complete(run, db=db) return + source_task_dir = os.path.normpath(str(item.source_task_dir or "")) + local_task_dir = resolve_cluster_local_task_dir(item) + if local_task_dir and not _cluster_source_task_dir_ready(local_task_dir): + await materialize_cluster_input(item, local_task_dir, job.task_id) current_task = await task_service.get_task(job.task_id) task_cancelled = bool(current_task and current_task.status == "CANCELLED") @@ -3323,10 +3340,12 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None: item_index = max(1, int(item.order_index or 1)) item_label = item.task_alias or item.task_name run_key = f"{build_run_key('landsar', run.profile_code, started_at=datetime.utcnow())}_{item.id}_{uuid.uuid4().hex[:6]}" + local_run_dir = os.path.normpath(resolve_cluster_local_run_dir(item, run_key)) execution = await dinsar_production_service.begin_item_execution( run=run, item=item, run_key=run_key, + output_dir_override=local_run_dir, db=db, ) @@ -3425,7 +3444,7 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None: request = RunRequest( engine_code="landsar", profile=run.profile_code, - root_dir=str(item.source_task_dir), + root_dir=local_task_dir, job_id=job.job_id, num_to_process=1, timeout_seconds=per_task_timeout or None, @@ -3438,6 +3457,7 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None: "__managed_orbit_output_dir": managed_orbit_output_dir, "__managed_run_key": run_key, "__source_root_override": run.source_root, + "__source_task_dir_override": source_task_dir, "__rerun_mode": "rerun_all", "__cluster_item": True, }, @@ -3497,24 +3517,17 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None: native_output_dir=native_output_dir, metrics=metrics, ) - await asyncio.to_thread( - dinsar_production_service.write_current_pointer, - run=run, - item=item, - execution=execution, - manifest_path=manifest_path, - primary_file=primary_file, - source_files=source_files, - native_output_dir=native_output_dir, - ) - await dinsar_production_service.mark_item_completed( - run=run, - item=item, - execution=execution, - manifest_path=manifest_path, - metrics=metrics, - db=db, - ) + if not _is_remote_worker(): + await asyncio.to_thread( + dinsar_production_service.write_current_pointer, + run=run, + item=item, + execution=execution, + manifest_path=manifest_path, + primary_file=primary_file, + source_files=source_files, + native_output_dir=native_output_dir, + ) # ---- Post-flight: upload or local publish ---- if _is_remote_worker(): @@ -3544,6 +3557,14 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None: "WARNING", f"[cluster {item_index}/{total_items}] Result catalog publish failed for {item_label}: {publish_error}", ) + await dinsar_production_service.mark_item_completed( + run=run, + item=item, + execution=execution, + manifest_path=manifest_path, + metrics=metrics, + db=db, + ) await task_service.add_log( job.task_id, @@ -3554,28 +3575,46 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None: except Exception as exc: run_exception_text = str(exc) item_error = run_exception_text - await dinsar_production_service.mark_item_failed( - run=run, - item=item, - execution=execution, - error_message=item_error, - db=db, - ) - await task_service.add_log( - job.task_id, - "WARNING", - f"[cluster {item_index}/{total_items}] Failed {item_label}: {item_error}", - ) - dinsar_production_service.append_run_log( - run.run_id, - f"[cluster-item-failed] {item_index}/{total_items} {item_label}: {item_error}", - ) - if task_result.get("command"): - await task_service.add_log(job.task_id, "INFO", f"LandSAR command [{item_label}]: {task_result.get('command')}") - if task_result.get("stdout_tail"): - await task_service.add_log(job.task_id, "INFO", f"LandSAR stdout tail [{item_label}]:\n{task_result.get('stdout_tail')}") - if task_result.get("stderr_tail"): - await task_service.add_log(job.task_id, "WARNING", f"LandSAR stderr tail [{item_label}]:\n{task_result.get('stderr_tail')}") + remote_completed = False + if _is_remote_worker(): + try: + await db.refresh(item) + remote_completed = str(item.status or "").strip().upper() == "COMPLETED" + except Exception: + remote_completed = False + if remote_completed: + await task_service.add_log( + job.task_id, + "INFO", + f"[cluster {item_index}/{total_items}] Main server already completed {item_label}; keeping completed state after local error: {item_error}", + ) + dinsar_production_service.append_run_log( + run.run_id, + f"[cluster-item-ok-after-upload] {item_index}/{total_items} {item_label}: {item_error}", + ) + else: + await dinsar_production_service.mark_item_failed( + run=run, + item=item, + execution=execution, + error_message=item_error, + db=db, + ) + await task_service.add_log( + job.task_id, + "WARNING", + f"[cluster {item_index}/{total_items}] Failed {item_label}: {item_error}", + ) + dinsar_production_service.append_run_log( + run.run_id, + f"[cluster-item-failed] {item_index}/{total_items} {item_label}: {item_error}", + ) + if task_result.get("command"): + await task_service.add_log(job.task_id, "INFO", f"LandSAR command [{item_label}]: {task_result.get('command')}") + if task_result.get("stdout_tail"): + await task_service.add_log(job.task_id, "INFO", f"LandSAR stdout tail [{item_label}]:\n{task_result.get('stdout_tail')}") + if task_result.get("stderr_tail"): + await task_service.add_log(job.task_id, "WARNING", f"LandSAR stderr tail [{item_label}]:\n{task_result.get('stderr_tail')}") finally: keepalive_task.cancel() try: diff --git a/backend/app/services/result_catalog_service.py b/backend/app/services/result_catalog_service.py index c78cb29..b67b587 100644 --- a/backend/app/services/result_catalog_service.py +++ b/backend/app/services/result_catalog_service.py @@ -142,9 +142,11 @@ def _coerce_optional_int(value: Any) -> Optional[int]: def _runtime_id_for_engine(engine_code: Optional[str]) -> Optional[str]: normalized = str(engine_code or "").strip().lower() if normalized == "isce2": - return settings.ISCE2_RUNTIME_ID or None + return getattr(settings, "ISCE2_RUNTIME_ID", "") or None + if normalized == "landsar": + return getattr(settings, "LANDSAR_RUNTIME_ID", "") or None if normalized in {"pyint", "gamma"}: - return settings.PYINT_RUNTIME_ID or None + return getattr(settings, "PYINT_RUNTIME_ID", "") or None return None @@ -822,6 +824,13 @@ class ResultCatalogService: package_dir = _ensure_directory(os.path.join(target_root, pair_key, "runs", run_key)) source_dir = _normalize_path(candidate["source_dir"]) in_place_source = _is_path_within(package_dir, source_dir) + if candidate_meta["engine_code"] in {"isce2", "landsar"} and in_place_source: + candidate_meta = dict(candidate_meta) + candidate_meta["output_dir"] = package_dir + native_dir = os.path.join(package_dir, RUN_NATIVE_DIRNAME) + candidate_meta["native_output_dir"] = ( + native_dir if os.path.isdir(native_dir) else package_dir + ) task_item = await self._lookup_task_item( db, pair_key=pair_key, @@ -994,7 +1003,7 @@ class ResultCatalogService: manifest_path = os.path.join(package_dir, "manifest.json") with open(manifest_path, "w", encoding="utf-8") as fp: json.dump(manifest, fp, ensure_ascii=False, indent=2) - if candidate["engine_code"] == "isce2" and in_place_source: + if candidate["engine_code"] in {"isce2", "landsar"} and in_place_source: try: completion_files_result = repair_managed_completion_files( package_dir, diff --git a/backend/tests/test_cluster_transport.py b/backend/tests/test_cluster_transport.py new file mode 100644 index 0000000..615567e --- /dev/null +++ b/backend/tests/test_cluster_transport.py @@ -0,0 +1,130 @@ +import json +import os +import tempfile +import unittest +import zipfile +from types import SimpleNamespace +from unittest import mock + +from backend.app.services.cluster_transport import ( + _zip_directory_contents, + resolve_cluster_local_run_dir, + resolve_cluster_local_task_dir, + safe_extract_zip, +) +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_zip_directory_contents_excludes_top_level_run_dir(self): + with tempfile.TemporaryDirectory() as root: + run_dir = os.path.join(root, "run_abc") + os.makedirs(os.path.join(run_dir, "assets", "disp")) + os.makedirs(os.path.join(run_dir, "native")) + with open(os.path.join(run_dir, RUN_META_FILENAME), "w", encoding="utf-8") as fp: + json.dump({"run_key": "run_abc"}, fp) + with open(os.path.join(run_dir, "assets", "disp", "disp.tif"), "wb") as fp: + fp.write(b"disp") + with open(os.path.join(run_dir, "native", "raw.txt"), "w", encoding="utf-8") as fp: + fp.write("raw") + + zip_path = os.path.join(root, "result.zip") + _zip_directory_contents(run_dir, zip_path) + + with zipfile.ZipFile(zip_path, "r") as zf: + names = set(zf.namelist()) + + self.assertIn(RUN_META_FILENAME, names) + self.assertIn("assets/disp/disp.tif", names) + self.assertIn("native/raw.txt", names) + self.assertFalse(any(name.startswith("run_abc/") for name in names)) + + def test_safe_extract_zip_rejects_path_escape(self): + with tempfile.TemporaryDirectory() as root: + zip_path = os.path.join(root, "unsafe.zip") + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("../escape.txt", "bad") + + with zipfile.ZipFile(zip_path, "r") as zf: + with self.assertRaises(ValueError): + safe_extract_zip(zf, os.path.join(root, "extract")) + + def test_worker_local_paths_are_optional_overrides(self): + item = SimpleNamespace( + id=7, + source_task_dir=r"D:\Task_Pool\DInSAR\Task_20260101_20260113", + results_root_dir=r"D:\production_results\dinsar\pair_a", + pair_key="lt1/pair:a", + ) + + with mock.patch.dict(os.environ, {}, clear=True): + self.assertEqual( + os.path.normpath(resolve_cluster_local_task_dir(item)), + os.path.normpath(item.source_task_dir), + ) + self.assertEqual( + os.path.normpath(resolve_cluster_local_run_dir(item, "run_1")), + os.path.normpath(r"D:\production_results\dinsar\pair_a\runs\run_1"), + ) + + with mock.patch.dict( + os.environ, + { + "CLUSTER_WORKER_TASK_ROOT": r"E:\cluster_tasks", + "CLUSTER_WORKER_RESULT_ROOT": r"E:\cluster_results", + }, + clear=True, + ): + self.assertEqual( + os.path.normpath(resolve_cluster_local_task_dir(item)), + os.path.normpath(r"E:\cluster_tasks\item_7\Task_20260101_20260113"), + ) + self.assertEqual( + os.path.normpath(resolve_cluster_local_run_dir(item, "run_1")), + os.path.normpath(r"E:\cluster_results\lt1_pair_a\runs\run_1"), + ) + + def test_repair_completion_files_reanchors_uploaded_landsar_run(self): + with tempfile.TemporaryDirectory() as root: + run_dir = os.path.join(root, "pair_a", "runs", "run_abc") + disp_dir = os.path.join(run_dir, "assets", "disp") + native_dir = os.path.join(run_dir, "native") + os.makedirs(disp_dir) + os.makedirs(native_dir) + primary_file = os.path.join(disp_dir, "disp.tif") + with open(primary_file, "wb") as fp: + fp.write(b"disp") + with open(os.path.join(run_dir, RUN_META_FILENAME), "w", encoding="utf-8") as fp: + json.dump( + { + "run_key": "run_abc", + "pair_key": "pair_a", + "engine_code": "landsar", + "profile_code": "lt1_dinsar", + "task_name": "Task_20260101_20260113", + "task_alias": "Task_20260101_20260113", + "output_dir": r"E:\worker_results\pair_a\runs\run_abc", + "native_output_dir": r"E:\worker_results\pair_a\runs\run_abc\native", + "primary_file": primary_file, + "source_files": [primary_file], + }, + fp, + ) + + result = repair_managed_completion_files( + run_dir, + primary_file=primary_file, + source_files=[primary_file], + ) + + self.assertTrue(os.path.isfile(result["execution_manifest_path"])) + self.assertTrue(os.path.isfile(result["current_pointer_path"])) + with open(result["execution_manifest_path"], "r", encoding="utf-8") as fp: + manifest = json.load(fp) + self.assertEqual(os.path.normpath(manifest["output_dir"]), os.path.normpath(run_dir)) + self.assertEqual(os.path.normpath(manifest["native_output_dir"]), os.path.normpath(native_dir)) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md b/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md index c7444cf..2c13c6b 100644 --- a/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md +++ b/docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md @@ -14,7 +14,7 @@ ## 设计目标 - 不依赖 Windows 文件共享（SMB）、映射盘符、UNC 路径 - - 不要求主服务器和 worker 共用盘符或路径结构 +- 不要求主服务器和 worker 共用盘符或路径结构；worker 可通过 `CLUSTER_WORKER_TASK_ROOT`、`CLUSTER_WORKER_RESULT_ROOT` 使用本机任务/结果缓存根 - Worker 节点可以动态增减，配置简单 - 复用现有 `SOURCE_PRODUCT_DIRS` 压缩包源池和 `TASK_POOL_ROOT` 体系 - 传输失败利用队列系统自带的重试机制 @@ -69,13 +69,13 @@ **Worker 端流程**： - 1. 读取 `item.source_task_dir`，检查本地目录是否存在且包含 `master/`、`slave/`、`pair_metadata.json` + 1. 读取 `item.source_task_dir`，解析 worker 本地输入目录。未配置 `CLUSTER_WORKER_TASK_ROOT` 时沿用原路径；已配置时使用 `CLUSTER_WORKER_TASK_ROOT\item_\Task_*` 2. 若无 → 调用 `GET /api/cluster/input-package/{item_id}` 3. Worker 请求头携带 `X-Cluster-Token`，主服务器校验 `CLUSTER_SHARED_TOKEN` 4. 主服务器读取 `item.source_task_dir` 指向的现有 Task_Pool 目录 5. 将该 Task_Pool 目录打成 zip 流式返回给 worker - 6. Worker 解包到 `item.source_task_dir`（保持与主服务器一致的路径结构） - 7. Worker 对 zip 成员路径做目录逃逸校验，并校验解包后的 `master/`、`slave/` 目录可用 + 6. Worker 解包到本地输入目录 + 7. Worker 对 zip 成员路径做目录逃逸校验，并校验解包后的 `Input_Data` 或 `master/`、`slave/` 目录可用 **主服务器新增 API**： @@ -85,13 +85,13 @@ Response: application/zip (streaming) 内容结构: Task_YYYYMMDD_YYYYMMDD/ + .dinsar_pair.json master/ <源文件...> slave/ <源文件...> orbit/ <精轨文件...> - pair_metadata.json ``` **当前实现边界**：主服务器不在传输接口里重新从 `SOURCE_PRODUCT_DIRS` 解包源压缩包；传输接口只打包已经由 Task_Pool 准备流程生成的 `item.source_task_dir`。如果主服务器上的 `source_task_dir` 不存在，接口返回 404，队列重试会保留失败信息。 @@ -102,11 +102,12 @@ **Worker 端流程**： - 1. LandSAR 完成后，收集标准产品包文件列表（从 `task_result.source_files` 和 manifest 获取） + 1. LandSAR 完成后，收集标准产品包文件列表（从 `task_result.source_files` 和 manifest 获取）。未配置 `CLUSTER_WORKER_RESULT_ROOT` 时沿用 `item.results_root_dir\runs\`；已配置时使用 `CLUSTER_WORKER_RESULT_ROOT\\runs\` 2. 调用 `POST /api/cluster/upload-result/{item_id}` - 3. 以 multipart 或流式上传产品包（含 primary file、auxiliary files、metadata） + 3. 以 multipart 上传 managed run 目录内容（不是外层 run 目录本身） 4. 主服务器接收后写入该 item 的标准目录 `results_root_dir\runs\` - 5. 触发 catalog 登记（复用现有 `result_catalog_service.bootstrap` 或增量登记） + 5. 触发 catalog 登记，修复 `execution_manifest.json` 和 current 指针 + 6. 主服务器把 item/execution 标记为完成并尝试 finalize cluster run；远端 worker 不再在上传后自行标记完成 **主服务器新增 API**： @@ -122,11 +123,12 @@ **上传后处理**： - 1. 校验文件完整性（与 manifest 对比） - 2. 写入 `D:\production_results\dinsar\\\\` - 3. 调用 `result_catalog_service` 增量登记 - 4. 更新 `DinsarProductionExecutionORM` 指向最终结果路径 - 5. 返回登记结果给 worker + 1. 安全解压上传 zip，替换该 item 的 `results_root_dir\runs\` + 2. 调用 `result_catalog_service` 增量登记 + 3. 校验 catalog 只登记 1 条且生成 `execution_manifest.json` 和 current 指针 + 4. 更新 `DinsarProductionExecutionORM` / `DinsarProductionRunItemORM` 指向主服务器最终结果路径 + 5. 标记 item 完成，并在所有 item 终态后 finalize run + 6. 返回登记结果给 worker ### 错误处理与重试 @@ -199,10 +201,14 @@ CLUSTER_MAIN_SERVER_URL=http://192.168.1.62 CLUSTER_SHARED_TOKEN= CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600 + CLUSTER_WORKER_TASK_ROOT=D:\Cluster_Work\Task_Pool + CLUSTER_WORKER_RESULT_ROOT=D:\Cluster_Work\Results + LANDSAR_RUNTIME_ID=landsar_runtime_v1 ``` - `CLUSTER_MAIN_SERVER_URL` 用于 worker 构造下载/上传 API 的完整 URL。未配置时默认使用 `DATABASE_URL` 中的 host 推断。 - `CLUSTER_SHARED_TOKEN` 是集群传输接口的专用共享密钥，主服务器和所有远端 worker 必须一致且非空；未配置时 `/api/cluster/...` 接口返回 503。 +`CLUSTER_MAIN_SERVER_URL` 用于 worker 构造下载/上传 API 的完整 URL。未配置时默认使用 `DATABASE_URL` 中的 host 推断。 +`CLUSTER_SHARED_TOKEN` 是集群传输接口的专用共享密钥，主服务器和所有远端 worker 必须一致且非空；未配置时 `/api/cluster/...` 接口返回 503。 +`CLUSTER_WORKER_TASK_ROOT`、`CLUSTER_WORKER_RESULT_ROOT` 是远端 worker 本机缓存根。未配置时保持旧行为，直接使用数据库中的 `source_task_dir` 和 `results_root_dir`。 ## 实现路线图 @@ -224,6 +230,8 @@ - [x] 数据搬运 pre-flight（HTTP 下载 Task_Pool zip + 安全解压） - [x] 结果上传 post-flight（HTTP 上传结果 zip + catalog 登记） - [x] 集群传输接口 `CLUSTER_SHARED_TOKEN` 鉴权 + - [x] 主服务器上传登记后负责 item/execution 完成和 run finalize + - [x] 支持 worker 本机输入/结果根，避免强依赖主服务器盘符 - [ ] 远端 .6 端到端测试 - [ ] Worker 开机自启