diff --git a/domain-api/app/services/cluster_runtime_service.py b/domain-api/app/services/cluster_runtime_service.py index 89f8daf..1feb409 100644 --- a/domain-api/app/services/cluster_runtime_service.py +++ b/domain-api/app/services/cluster_runtime_service.py @@ -122,7 +122,11 @@ def register_node_heartbeat( current_load: int = 0, worker_version: str = "0.1.0", metadata: dict | None = None, + hostname_override: str | None = None, + ip_override: str | None = None, ) -> None: + hostname = str(hostname_override or "").strip() or socket.gethostname() + ip = str(ip_override or "").strip() or _resolve_local_ip() with get_db() as conn: with conn.cursor() as cur: cur.execute( @@ -146,8 +150,8 @@ def register_node_heartbeat( node_code, region, role, - socket.gethostname(), - _resolve_local_ip(), + hostname, + ip, status, worker_version, max(0, int(current_load or 0)), diff --git a/domain-api/app/services/sync_push_service.py b/domain-api/app/services/sync_push_service.py index c03aa56..64ce824 100644 --- a/domain-api/app/services/sync_push_service.py +++ b/domain-api/app/services/sync_push_service.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import socket import urllib.error import urllib.parse import urllib.request @@ -8,6 +9,7 @@ from datetime import datetime, timedelta from app.core.config import settings from app.core.db import get_db +from app.services.cluster_runtime_service import register_node_heartbeat from app.services.sync_record_service import _decode_json, _normalize_region @@ -34,6 +36,38 @@ def _projection_ingest_type(sync_type: str) -> str: return "sync_ingest" +def _refresh_remote_runtime_node(*, source_region: str, projection: dict, received_at: datetime | None = None) -> None: + node_info = projection.get("node") or {} + node_code = str(node_info.get("node_code") or "").strip() + region = _normalize_region(node_info.get("region"), source_region) + role = str(node_info.get("role") or "control").strip() or "control" + hostname = str(node_info.get("hostname") or "").strip() or socket.gethostname() + ip = str(node_info.get("ip") or "").strip() + if not node_code: + node_code = f"{region}-{role}-imported" + + metadata = { + "service": "runtime-ingest", + "projection_source_region": source_region, + "worker_mode": projection.get("worker_mode", ""), + "phase_label": projection.get("phase_label", ""), + "phase_detail": projection.get("phase_detail", ""), + "proxy_runtime_label": projection.get("proxy_runtime_label", ""), + "proxy_runtime_reason": projection.get("proxy_runtime_reason", ""), + "updated_at": _format_time(received_at or datetime.now()), + } + register_node_heartbeat( + node_code=node_code, + region=region, + role=role, + status="online", + current_load=int(((projection.get("progress") or {}).get("running", 0) or 0)), + metadata=metadata, + hostname_override=hostname, + ip_override=ip, + ) + + def _load_latest_projection(sync_type: str) -> dict | None: source_region = _normalize_region(settings.sync_source_region, settings.node_region) target_region = _normalize_region(settings.sync_target_region, "overseas") @@ -213,6 +247,7 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) projection_hash = str(payload.get("projection_hash") or "").strip() projection = payload.get("projection") or {} target_region = _normalize_region(settings.node_region, "overseas") + received_at = datetime.now() with get_db() as conn: with conn.cursor() as cur: @@ -232,6 +267,17 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) ) existing = cur.fetchone() if existing: + if sync_type == "runtime_projection": + _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at) + cur.execute( + """ + UPDATE detect_sync_records + SET updated_at = CURRENT_TIMESTAMP + WHERE id = %s + """, + (int(existing[0]),), + ) + conn.commit() return True, "同步投影已存在，已按幂等处理", {"record_id": int(existing[0]), "deduplicated": True} stored_payload = { @@ -239,7 +285,7 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) "source_record_id": source_record_id, "projection_hash": projection_hash, "projection": projection, - "received_at": _format_time(datetime.now()), + "received_at": _format_time(received_at), } cur.execute( """ @@ -259,6 +305,8 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) ) record_id = int(cur.fetchone()[0]) conn.commit() + if sync_type == "runtime_projection": + _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at) return True, "同步投影接收成功", {"record_id": record_id, "deduplicated": False} @@ -313,12 +361,22 @@ def _push_projection_batch(sync_type: str, ingest_url: str) -> tuple[bool, str, def _push_projection_record(source_record: dict, sync_type: str, ingest_url: str) -> tuple[bool, str, dict]: latest_attempt = _latest_push_attempt(source_record["id"], source_record["target_region"], sync_type) if latest_attempt and latest_attempt["status"] == "success": - return True, "该投影已推送，无需重复发送", { - "action": "push_sync", - "sync_type": sync_type, - "source_record_id": source_record["id"], - "deduplicated": True, - } + last_created_at = latest_attempt.get("created_at") + if sync_type != "runtime_projection" or not isinstance(last_created_at, datetime): + return True, "该投影已推送，无需重复发送", { + "action": "push_sync", + "sync_type": sync_type, + "source_record_id": source_record["id"], + "deduplicated": True, + } + now = datetime.now(last_created_at.tzinfo) if last_created_at.tzinfo else datetime.now() + if now - last_created_at < timedelta(seconds=max(20, int(settings.sync_poll_interval_seconds or 30))): + return True, "该投影已推送，无需重复发送", { + "action": "push_sync", + "sync_type": sync_type, + "source_record_id": source_record["id"], + "deduplicated": True, + } if latest_attempt and latest_attempt["status"] == "pending": return True, "该投影已有同步推送进行中，暂不重复发送", { "action": "push_sync", diff --git a/domain-api/app/services/sync_record_service.py b/domain-api/app/services/sync_record_service.py index ef793cd..4399006 100644 --- a/domain-api/app/services/sync_record_service.py +++ b/domain-api/app/services/sync_record_service.py @@ -2,6 +2,7 @@ from __future__ import annotations import hashlib import json +import socket from datetime import datetime, timedelta from app.core.config import settings @@ -12,6 +13,13 @@ def _format_time(value: datetime | None) -> str: return value.isoformat(sep=" ", timespec="seconds") if value else "" +def _resolve_local_ip() -> str: + try: + return socket.gethostbyname(socket.gethostname()) + except Exception: + return "" + + def _decode_json(value: object) -> dict: if isinstance(value, dict): return value @@ -397,6 +405,13 @@ def append_runtime_projection_if_changed( normalized_target_region = _normalize_region(target_region, _normalize_region(settings.sync_target_region, "overseas")) active_job = detect.get("active_job") or {} projection = { + "node": { + "node_code": settings.node_code, + "region": settings.node_region, + "role": settings.node_role, + "hostname": socket.gethostname(), + "ip": _resolve_local_ip(), + }, "worker_online": bool(detect.get("worker_online", False)), "worker_mode": detect.get("worker_mode", ""), "phase_label": detect.get("phase_label", ""),