671 lines
27 KiB
Python
671 lines
27 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import time
|
|
|
|
from app.core.db import db_read_retry, get_db
|
|
from app.core.config import settings
|
|
from app.services.debug_event_service import push_debug_event
|
|
from app.services.detect_job_service import (
|
|
get_active_detect_job_summary,
|
|
get_detect_queue_health,
|
|
get_latest_detect_job_summary,
|
|
get_latest_unprojected_detect_job_summary,
|
|
list_recent_detect_run_events,
|
|
process_detect_pipeline_now,
|
|
)
|
|
from app.services.sync_record_service import append_detect_result_projection_if_changed
|
|
from app.services.sync_push_service import (
|
|
_load_local_detect_backlog_snapshot,
|
|
pull_detect_task_batch_now,
|
|
push_runtime_projection_now,
|
|
)
|
|
from app.services.worker_control_service import (
|
|
_expand_local_linux_worker_target_node_codes,
|
|
send_worker_command,
|
|
)
|
|
|
|
|
|
logger = logging.getLogger("domaincheck.sync_agent")
|
|
|
|
_IDLE_SYNC_KEYWORDS = (
|
|
"当前没有可推送",
|
|
"当前没有需要立即推送",
|
|
"已全部同步完成",
|
|
"无需重复发送",
|
|
"进行中",
|
|
"等待下个重试窗口",
|
|
"暂停拉取",
|
|
)
|
|
|
|
_LAST_OVERLAP_JOB_ID = 0
|
|
_LAST_OVERLAP_TRIGGERED_AT = 0.0
|
|
|
|
|
|
def _append_detect_result_projection_snapshot(active_job: dict) -> None:
|
|
if not active_job:
|
|
return
|
|
append_detect_result_projection_if_changed(
|
|
detect={
|
|
"active_job": active_job,
|
|
"progress": {
|
|
"pending": int(active_job.get("items_pending", 0) or 0),
|
|
"running": int(active_job.get("items_running", 0) or 0),
|
|
"completed": int(active_job.get("items_completed", 0) or 0),
|
|
"blacklisted": int(active_job.get("items_blacklisted", 0) or 0),
|
|
"failed": int(active_job.get("items_failed", 0) or 0),
|
|
},
|
|
"phase_label": str(active_job.get("status") or "").strip(),
|
|
"phase_detail": f"sync-agent snapshot for {active_job.get('job_code', '')}",
|
|
}
|
|
)
|
|
|
|
|
|
def _is_idle_sync_message(message: str) -> bool:
|
|
normalized = str(message or "").strip()
|
|
return any(keyword in normalized for keyword in _IDLE_SYNC_KEYWORDS)
|
|
|
|
|
|
def _filter_runtime_events_for_job(events: list[dict], *, job_code: str = "", job_id: int = 0, limit: int = 8) -> list[dict]:
|
|
target_job_code = str(job_code or "").strip()
|
|
target_job_id = int(job_id or 0)
|
|
safe_limit = max(1, min(int(limit or 8), 50))
|
|
filtered: list[dict] = []
|
|
for raw_event in list(events or []):
|
|
if not isinstance(raw_event, dict):
|
|
continue
|
|
payload = raw_event.get("payload") if isinstance(raw_event.get("payload"), dict) else {}
|
|
event_job_code = str(payload.get("job_code") or "").strip()
|
|
event_job_id = int(raw_event.get("job_id") or 0)
|
|
if target_job_code and event_job_code != target_job_code and (target_job_id <= 0 or event_job_id != target_job_id):
|
|
continue
|
|
filtered.append(raw_event)
|
|
if len(filtered) >= safe_limit:
|
|
break
|
|
return filtered
|
|
|
|
|
|
def _build_aligned_queue_health_snapshot(active_job: dict, queue_health: dict | None) -> dict:
|
|
snapshot = dict(queue_health or {})
|
|
if not active_job:
|
|
return snapshot
|
|
|
|
active_job_code = str(active_job.get("runtime_job_code") or active_job.get("job_code") or "").strip()
|
|
queue_job = dict(snapshot.get("job") or {})
|
|
queue_job_code = str(queue_job.get("runtime_job_code") or queue_job.get("job_code") or "").strip()
|
|
|
|
if active_job_code and queue_job_code and active_job_code == queue_job_code:
|
|
return snapshot
|
|
|
|
active_job_items_total = int(active_job.get("items_total", 0) or 0)
|
|
active_job_pending = int(active_job.get("items_pending", 0) or 0)
|
|
active_job_claimed = int(active_job.get("items_claimed", 0) or 0)
|
|
active_job_running = int(active_job.get("items_running", 0) or 0)
|
|
active_job_completed = int(active_job.get("items_completed", 0) or 0)
|
|
active_job_blacklisted = int(active_job.get("items_blacklisted", 0) or 0)
|
|
active_job_failed = int(active_job.get("items_failed", 0) or 0)
|
|
active_job_terminal = int(
|
|
active_job.get("items_terminal", active_job_completed + active_job_blacklisted + active_job_failed) or 0
|
|
)
|
|
display_claimed = int(active_job.get("display_items_claimed", active_job_claimed) or active_job_claimed)
|
|
display_running = int(active_job.get("display_items_running", active_job_running) or active_job_running)
|
|
|
|
node_entries: list[dict] = []
|
|
for node in list(active_job.get("node_stats") or []):
|
|
node_entries.append(
|
|
{
|
|
"node_code": str(node.get("node_code") or "").strip(),
|
|
"items_total": int(node.get("items_total", 0) or 0),
|
|
"items_pending": int(node.get("items_pending", 0) or 0),
|
|
"items_claimed": int(node.get("items_claimed", 0) or 0),
|
|
"items_running": int(node.get("items_running", 0) or 0),
|
|
"items_completed": int(node.get("items_completed", 0) or 0),
|
|
"items_blacklisted": int(node.get("items_blacklisted", 0) or 0),
|
|
"items_failed": int(node.get("items_failed", 0) or 0),
|
|
"processed_recent": int(node.get("processed_recent", 0) or 0),
|
|
"processed_per_minute": float(node.get("processed_per_minute", 0) or 0),
|
|
"completed_recent": int(node.get("completed_recent", 0) or 0),
|
|
"blacklisted_recent": int(node.get("blacklisted_recent", 0) or 0),
|
|
"failed_recent": int(node.get("failed_recent", 0) or 0),
|
|
"metrics_source": str(node.get("metrics_source") or "runtime"),
|
|
}
|
|
)
|
|
|
|
assigned_total = sum(int(item.get("items_total", 0) or 0) for item in node_entries)
|
|
unassigned_total = max(0, active_job_items_total - assigned_total)
|
|
if unassigned_total > 0:
|
|
node_entries.append(
|
|
{
|
|
"node_code": "unassigned",
|
|
"items_total": unassigned_total,
|
|
"items_pending": active_job_pending,
|
|
"items_claimed": 0,
|
|
"items_running": 0,
|
|
"items_completed": 0,
|
|
"items_blacklisted": 0,
|
|
"items_failed": 0,
|
|
"processed_recent": 0,
|
|
"processed_per_minute": 0.0,
|
|
"completed_recent": 0,
|
|
"blacklisted_recent": 0,
|
|
"failed_recent": 0,
|
|
"metrics_source": "central_queue",
|
|
}
|
|
)
|
|
|
|
snapshot["job"] = {
|
|
"job_id": active_job.get("job_id"),
|
|
"job_code": str(active_job.get("job_code") or "").strip(),
|
|
"runtime_job_code": active_job_code,
|
|
"status": str(active_job.get("status") or "").strip(),
|
|
"progress_percent": float(active_job.get("progress_percent", 0) or 0),
|
|
}
|
|
snapshot["queue"] = {
|
|
**dict(snapshot.get("queue") or {}),
|
|
"items_total": active_job_items_total,
|
|
"pending": active_job_pending,
|
|
"claimed": active_job_claimed,
|
|
"running": active_job_running,
|
|
"display_claimed": display_claimed,
|
|
"display_running": display_running,
|
|
"completed": active_job_completed,
|
|
"blacklisted": active_job_blacklisted,
|
|
"failed": active_job_failed,
|
|
"terminal": active_job_terminal,
|
|
"terminal_percent": round((active_job_terminal / active_job_items_total) * 100, 2) if active_job_items_total else 0.0,
|
|
}
|
|
snapshot["nodes"] = node_entries
|
|
return snapshot
|
|
|
|
|
|
def _select_projection_job_snapshot() -> dict | None:
|
|
active_job = get_active_detect_job_summary(event_limit=10)
|
|
if active_job:
|
|
return active_job
|
|
return get_latest_detect_job_summary(
|
|
event_limit=10,
|
|
statuses=("completed", "partial_failed", "failed"),
|
|
recent_minutes=20,
|
|
)
|
|
|
|
|
|
def _select_projection_job_snapshots() -> list[dict]:
|
|
snapshots: list[dict] = []
|
|
seen_job_ids: set[int] = set()
|
|
|
|
active_job = get_active_detect_job_summary(event_limit=10)
|
|
if active_job:
|
|
active_job_id = int(active_job.get("job_id") or 0)
|
|
if active_job_id > 0 and active_job_id not in seen_job_ids:
|
|
snapshots.append(active_job)
|
|
seen_job_ids.add(active_job_id)
|
|
|
|
latest_finished_job = get_latest_unprojected_detect_job_summary(
|
|
event_limit=10,
|
|
statuses=("completed", "partial_failed", "failed"),
|
|
recent_minutes=180,
|
|
)
|
|
if latest_finished_job:
|
|
latest_finished_job_id = int(latest_finished_job.get("job_id") or 0)
|
|
if latest_finished_job_id > 0 and latest_finished_job_id not in seen_job_ids:
|
|
snapshots.append(latest_finished_job)
|
|
seen_job_ids.add(latest_finished_job_id)
|
|
|
|
return snapshots
|
|
|
|
|
|
def _emit_structured_tick(
|
|
*,
|
|
base_event_type: str,
|
|
ok: bool,
|
|
message: str,
|
|
data: dict | None = None,
|
|
) -> None:
|
|
payload = {"ok": ok, "data": data or {}}
|
|
event_type = f"{base_event_type}_failed"
|
|
level = "warning"
|
|
if isinstance(data, dict) and str(data.get("pull_state") or "").strip() == "throttled":
|
|
event_type = f"{base_event_type}_idle"
|
|
level = "info"
|
|
elif not ok and _is_idle_sync_message(message):
|
|
event_type = f"{base_event_type}_idle"
|
|
level = "info"
|
|
elif ok:
|
|
event_type = f"{base_event_type}_success"
|
|
level = "info"
|
|
if "但远端确认失败" in str(message or ""):
|
|
event_type = f"{base_event_type}_partial"
|
|
level = "warning"
|
|
results = list((data or {}).get("results") or [])
|
|
if results and any(not bool(item.get("ok")) for item in results):
|
|
event_type = f"{base_event_type}_partial"
|
|
level = "warning"
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type=event_type,
|
|
level=level,
|
|
message=message,
|
|
payload=payload,
|
|
)
|
|
|
|
|
|
def _emit_sync_result_breakdown(data: dict | None) -> None:
|
|
results = list((data or {}).get("results") or [])
|
|
for item in results:
|
|
sync_type = str(item.get("sync_type") or "").strip() or "unknown"
|
|
ok = bool(item.get("ok"))
|
|
message = str(item.get("message") or "").strip() or f"{sync_type} sync result"
|
|
result_data = item.get("data") or {}
|
|
event_type = f"{sync_type}_sync_failed"
|
|
level = "warning"
|
|
if not ok and _is_idle_sync_message(message):
|
|
event_type = f"{sync_type}_sync_idle"
|
|
level = "info"
|
|
elif ok:
|
|
event_type = f"{sync_type}_sync_success"
|
|
level = "info"
|
|
if isinstance(result_data, dict) and result_data.get("success_count") is not None:
|
|
batch_count = int(result_data.get("batch_count") or 0)
|
|
success_count = int(result_data.get("success_count") or 0)
|
|
if batch_count > 0 and success_count < batch_count:
|
|
event_type = f"{sync_type}_sync_partial"
|
|
level = "warning"
|
|
if "成功 1/" in message or "但" in message:
|
|
event_type = f"{sync_type}_sync_partial"
|
|
level = "warning"
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type=event_type,
|
|
level=level,
|
|
message=message,
|
|
payload={
|
|
"sync_type": sync_type,
|
|
"ok": ok,
|
|
"data": result_data,
|
|
},
|
|
)
|
|
|
|
|
|
def _bool_env(name: str, default: bool) -> bool:
|
|
raw = str(os.getenv(name, "1" if default else "0") or ("1" if default else "0")).strip().lower()
|
|
return raw not in {"0", "false", "off", "no"}
|
|
|
|
|
|
def _int_env(name: str, default: int, *, minimum: int = 0, maximum: int | None = None) -> int:
|
|
try:
|
|
value = int(os.getenv(name, str(default)) or default)
|
|
except (TypeError, ValueError):
|
|
value = int(default)
|
|
value = max(minimum, value)
|
|
if maximum is not None:
|
|
value = min(maximum, value)
|
|
return value
|
|
|
|
|
|
@db_read_retry(attempts=3, initial_delay_seconds=0.05, backoff=2.0)
|
|
def _select_overlap_start_candidate() -> dict | None:
|
|
if settings.node_region != "mainland" or settings.node_role != "control":
|
|
return None
|
|
if not _bool_env("DOMAINCHECK_OVERLAP_HANDOFF_ENABLED", True):
|
|
return None
|
|
|
|
min_running_jobs = _int_env("DOMAINCHECK_OVERLAP_HANDOFF_MIN_RUNNING_JOBS", 1, minimum=1, maximum=16)
|
|
running_age_seconds = _int_env("DOMAINCHECK_OVERLAP_HANDOFF_MIN_RUNNING_AGE_SECONDS", 300, minimum=30, maximum=7200)
|
|
with get_db() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
WITH running_state AS (
|
|
SELECT
|
|
COUNT(*) AS running_jobs,
|
|
COALESCE(
|
|
MAX(
|
|
EXTRACT(
|
|
EPOCH FROM (CURRENT_TIMESTAMP - COALESCE(job.started_at, job.created_at))
|
|
)
|
|
),
|
|
0
|
|
) AS max_running_age_seconds
|
|
FROM detect_jobs AS job
|
|
WHERE job.status = 'running'
|
|
),
|
|
pending_candidate AS (
|
|
SELECT
|
|
job.id,
|
|
job.job_code,
|
|
job.task_mode,
|
|
COALESCE(job.started_at, job.created_at) AS activity_at
|
|
FROM detect_jobs AS job
|
|
WHERE job.status = 'pending'
|
|
ORDER BY
|
|
COALESCE(job.started_at, job.created_at) DESC,
|
|
job.id DESC
|
|
LIMIT 1
|
|
)
|
|
SELECT
|
|
pending.id,
|
|
pending.job_code,
|
|
pending.task_mode,
|
|
running_state.running_jobs,
|
|
running_state.max_running_age_seconds,
|
|
'overlap_tail_handoff' AS selection_reason
|
|
FROM pending_candidate AS pending
|
|
CROSS JOIN running_state
|
|
WHERE running_state.running_jobs >= %s
|
|
AND running_state.max_running_age_seconds >= %s
|
|
""",
|
|
(
|
|
min_running_jobs,
|
|
running_age_seconds,
|
|
),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return None
|
|
return {
|
|
"job_id": int(row[0] or 0),
|
|
"job_code": str(row[1] or "").strip(),
|
|
"task_mode": str(row[2] or "domain_pipeline").strip() or "domain_pipeline",
|
|
"running_jobs": int(row[3] or 0),
|
|
"max_running_age_seconds": int(float(row[4] or 0) or 0),
|
|
"selection_reason": str(row[5] or "overlap_tail_handoff").strip() or "overlap_tail_handoff",
|
|
}
|
|
|
|
|
|
@db_read_retry(attempts=3, initial_delay_seconds=0.05, backoff=2.0)
|
|
def _select_overlap_target_node_codes() -> list[str]:
|
|
if settings.node_region != "mainland" or settings.node_role != "control":
|
|
return []
|
|
|
|
target_limit = _int_env("DOMAINCHECK_OVERLAP_HANDOFF_TARGETS", 12, minimum=1, maximum=32)
|
|
target_scan_limit = _int_env(
|
|
"DOMAINCHECK_OVERLAP_HANDOFF_TARGET_SCAN_LIMIT",
|
|
max(target_limit * 4, 16),
|
|
minimum=target_limit,
|
|
maximum=256,
|
|
)
|
|
max_current_load = _int_env("DOMAINCHECK_OVERLAP_HANDOFF_TARGET_MAX_CURRENT_LOAD", 0, minimum=0, maximum=4096)
|
|
|
|
expanded_targets = _expand_local_linux_worker_target_node_codes(str(settings.worker_service_name or "").strip())
|
|
if not expanded_targets:
|
|
return []
|
|
|
|
with get_db() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT node_code, current_load
|
|
FROM detect_worker_nodes
|
|
WHERE node_code = ANY(%s)
|
|
ORDER BY
|
|
CASE WHEN COALESCE(current_load, 0) <= %s THEN 0 ELSE 1 END ASC,
|
|
COALESCE(current_load, 0) ASC,
|
|
COALESCE(update_time, last_heartbeat_at) DESC,
|
|
node_code ASC
|
|
LIMIT %s
|
|
""",
|
|
(
|
|
expanded_targets,
|
|
max_current_load,
|
|
target_scan_limit,
|
|
),
|
|
)
|
|
rows = list(cur.fetchall() or [])
|
|
|
|
preferred: list[str] = []
|
|
fallback: list[str] = []
|
|
seen: set[str] = set()
|
|
for raw_node_code, raw_current_load in rows:
|
|
node_code = str(raw_node_code or "").strip()
|
|
if not node_code or node_code in seen:
|
|
continue
|
|
seen.add(node_code)
|
|
fallback.append(node_code)
|
|
if int(raw_current_load or 0) <= max_current_load:
|
|
preferred.append(node_code)
|
|
|
|
ordered = preferred + [node_code for node_code in fallback if node_code not in set(preferred)]
|
|
if not ordered:
|
|
ordered = list(expanded_targets or [])
|
|
return ordered[:target_limit]
|
|
|
|
|
|
def _maybe_trigger_overlap_start() -> tuple[bool, str, dict]:
|
|
global _LAST_OVERLAP_JOB_ID, _LAST_OVERLAP_TRIGGERED_AT
|
|
|
|
candidate = _select_overlap_start_candidate()
|
|
if not candidate:
|
|
return False, "当前没有满足尾盘接棒条件的 pending job", {}
|
|
|
|
cooldown_seconds = _int_env("DOMAINCHECK_OVERLAP_HANDOFF_COOLDOWN_SECONDS", 120, minimum=15, maximum=1800)
|
|
job_id = int(candidate.get("job_id") or 0)
|
|
now_ts = time.time()
|
|
if (
|
|
job_id > 0
|
|
and job_id == _LAST_OVERLAP_JOB_ID
|
|
and now_ts - float(_LAST_OVERLAP_TRIGGERED_AT or 0.0) < cooldown_seconds
|
|
):
|
|
return False, f"overlap handoff 冷却中: job_id={job_id}", candidate
|
|
|
|
target_node_codes = _select_overlap_target_node_codes()
|
|
if not target_node_codes:
|
|
return False, f"overlap handoff 未找到可接棒 worker: job_id={job_id}", candidate
|
|
|
|
payload = {
|
|
"job_id": job_id,
|
|
"job_code": str(candidate.get("job_code") or "").strip(),
|
|
"target_job_id": job_id,
|
|
"target_job_code": str(candidate.get("job_code") or "").strip(),
|
|
"task_mode": str(candidate.get("task_mode") or "domain_pipeline").strip() or "domain_pipeline",
|
|
"source": "overlap-handoff",
|
|
"selection_reason": str(candidate.get("selection_reason") or "overlap_tail_handoff").strip() or "overlap_tail_handoff",
|
|
"tail_handoff_candidate": True,
|
|
"target_node_codes": target_node_codes,
|
|
}
|
|
ok, message = send_worker_command("start_detection", payload=payload)
|
|
if ok and job_id > 0:
|
|
_LAST_OVERLAP_JOB_ID = job_id
|
|
_LAST_OVERLAP_TRIGGERED_AT = now_ts
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type="overlap_handoff_started",
|
|
level="warning",
|
|
message=message,
|
|
payload={**candidate, "cooldown_seconds": cooldown_seconds, "target_node_codes": target_node_codes},
|
|
)
|
|
return ok, message, candidate
|
|
|
|
|
|
def _run_pipeline_stage_processor() -> tuple[bool, str, dict]:
|
|
process_limit = max(500, min(int(settings.sync_pipeline_process_limit or 5000), 5000))
|
|
ok, message, data = process_detect_pipeline_now(limit=process_limit)
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type="pipeline_tick_success" if ok else "pipeline_tick_failed",
|
|
level="info" if ok else "warning",
|
|
message=message,
|
|
payload={
|
|
"ok": ok,
|
|
"limit": process_limit,
|
|
"data": data or {},
|
|
},
|
|
)
|
|
return ok, message, data
|
|
|
|
|
|
def _emit_runtime_debug_snapshots() -> None:
|
|
active_job = get_active_detect_job_summary(event_limit=10)
|
|
for projection_job in _select_projection_job_snapshots():
|
|
_append_detect_result_projection_snapshot(projection_job)
|
|
|
|
if not active_job:
|
|
return
|
|
|
|
queue_health = _build_aligned_queue_health_snapshot(
|
|
active_job,
|
|
get_detect_queue_health(window_minutes=15),
|
|
)
|
|
recent_events = _filter_runtime_events_for_job(
|
|
list_recent_detect_run_events(limit=24),
|
|
job_code=str(active_job.get("runtime_job_code") or active_job.get("job_code") or "").strip(),
|
|
job_id=int(active_job.get("job_id", 0) or 0),
|
|
limit=8,
|
|
)
|
|
push_debug_event(
|
|
service="detect-runtime",
|
|
event_type="active_job_snapshot",
|
|
level="info",
|
|
message=f"active job {active_job.get('job_code', '')} status={active_job.get('status', '')}",
|
|
payload={
|
|
"job": {
|
|
"job_id": active_job.get("job_id"),
|
|
"job_code": active_job.get("job_code", ""),
|
|
"status": active_job.get("status", ""),
|
|
"items_total": active_job.get("items_total", 0),
|
|
"items_pending": active_job.get("items_pending", 0),
|
|
"items_claimed": active_job.get("items_claimed", 0),
|
|
"items_running": active_job.get("items_running", 0),
|
|
"items_completed": active_job.get("items_completed", 0),
|
|
"items_failed": active_job.get("items_failed", 0),
|
|
"progress_percent": active_job.get("progress_percent", 0),
|
|
"node_stats": list(active_job.get("node_stats") or []),
|
|
},
|
|
"queue_health": queue_health,
|
|
"backlog": _load_local_detect_backlog_snapshot(),
|
|
"recent_events": recent_events,
|
|
},
|
|
)
|
|
if queue_health.get("queue", {}).get("overdue_leases", 0):
|
|
push_debug_event(
|
|
service="detect-runtime",
|
|
event_type="queue_overdue_leases",
|
|
level="warning",
|
|
message=f"检测队列存在过期租约 {queue_health.get('queue', {}).get('overdue_leases', 0)} 个",
|
|
payload=queue_health,
|
|
)
|
|
for event in recent_events:
|
|
event_type = str(event.get("event_type") or "").strip()
|
|
if event_type not in {"domain_started", "domain_completed", "domain_failed", "domain_blacklisted"}:
|
|
continue
|
|
push_debug_event(
|
|
service="worker-event",
|
|
event_type=event_type,
|
|
level=str(event.get("level") or "info"),
|
|
message=str(event.get("message") or "").strip(),
|
|
payload={
|
|
"job_id": event.get("job_id"),
|
|
"node_code": event.get("node_code", ""),
|
|
"created_at": event.get("created_at", ""),
|
|
**(event.get("payload") or {}),
|
|
},
|
|
)
|
|
|
|
|
|
def _run_sync_tick_once() -> dict:
|
|
overlap_ok, overlap_message, overlap_data = _maybe_trigger_overlap_start()
|
|
sync_ok, sync_message, sync_data = push_runtime_projection_now()
|
|
pull_ok, pull_message, pull_data = pull_detect_task_batch_now()
|
|
pipeline_ok, pipeline_message, pipeline_data = _run_pipeline_stage_processor()
|
|
try:
|
|
_emit_runtime_debug_snapshots()
|
|
debug_snapshot_error = ""
|
|
except Exception as debug_exc: # pragma: no cover - logged by caller path
|
|
logger.warning("runtime debug snapshot skipped: %s", debug_exc)
|
|
debug_snapshot_error = str(debug_exc)
|
|
return {
|
|
"sync": {"ok": sync_ok, "message": sync_message, "data": sync_data},
|
|
"pull": {"ok": pull_ok, "message": pull_message, "data": pull_data},
|
|
"pipeline": {"ok": pipeline_ok, "message": pipeline_message, "data": pipeline_data},
|
|
"overlap": {"ok": overlap_ok, "message": overlap_message, "data": overlap_data},
|
|
"debug_snapshot_error": debug_snapshot_error,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
)
|
|
# Old env files still ship SYNC_POLL_INTERVAL_SECONDS=30. Cap the interval
|
|
# so controller pull/pipeline ticks cannot be throttled into starvation.
|
|
interval = max(2, min(int(settings.sync_poll_interval_seconds or 2), 5))
|
|
logger.info(
|
|
"sync agent started: node=%s source=%s target=%s interval=%ss enabled=%s",
|
|
settings.node_code,
|
|
settings.sync_source_region,
|
|
settings.sync_target_region,
|
|
interval,
|
|
settings.sync_push_enabled,
|
|
)
|
|
while True:
|
|
try:
|
|
tick = _run_sync_tick_once()
|
|
sync_ok = bool((tick.get("sync") or {}).get("ok"))
|
|
sync_message = str((tick.get("sync") or {}).get("message") or "")
|
|
sync_data = (tick.get("sync") or {}).get("data") or {}
|
|
logger.info("sync tick: ok=%s message=%s data=%s", sync_ok, sync_message, sync_data)
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type="sync_tick",
|
|
level="info" if sync_ok else "warning",
|
|
message=sync_message,
|
|
payload={"ok": sync_ok, "data": sync_data},
|
|
)
|
|
_emit_structured_tick(base_event_type="sync_push", ok=sync_ok, message=sync_message, data=sync_data)
|
|
_emit_sync_result_breakdown(sync_data)
|
|
|
|
pull_ok = bool((tick.get("pull") or {}).get("ok"))
|
|
pull_message = str((tick.get("pull") or {}).get("message") or "")
|
|
pull_data = (tick.get("pull") or {}).get("data") or {}
|
|
logger.info("task pull tick: ok=%s message=%s data=%s", pull_ok, pull_message, pull_data)
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type="task_pull_tick",
|
|
level="info" if pull_ok else "warning",
|
|
message=pull_message,
|
|
payload={"ok": pull_ok, "data": pull_data},
|
|
)
|
|
_emit_structured_tick(base_event_type="task_pull", ok=pull_ok, message=pull_message, data=pull_data)
|
|
|
|
pipeline_ok = bool((tick.get("pipeline") or {}).get("ok"))
|
|
pipeline_message = str((tick.get("pipeline") or {}).get("message") or "")
|
|
pipeline_data = (tick.get("pipeline") or {}).get("data") or {}
|
|
logger.info(
|
|
"pipeline tick: ok=%s message=%s data=%s",
|
|
pipeline_ok,
|
|
pipeline_message,
|
|
pipeline_data,
|
|
)
|
|
|
|
overlap_ok = bool((tick.get("overlap") or {}).get("ok"))
|
|
overlap_message = str((tick.get("overlap") or {}).get("message") or "")
|
|
overlap_data = (tick.get("overlap") or {}).get("data") or {}
|
|
logger.info(
|
|
"overlap tick: ok=%s message=%s data=%s",
|
|
overlap_ok,
|
|
overlap_message,
|
|
overlap_data,
|
|
)
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type="overlap_tick",
|
|
level="info" if overlap_ok else "warning",
|
|
message=overlap_message,
|
|
payload={"ok": overlap_ok, "data": overlap_data},
|
|
)
|
|
except Exception as exc:
|
|
logger.exception("sync tick failed: %s", exc)
|
|
push_debug_event(
|
|
service="sync-agent",
|
|
event_type="sync_tick_failed",
|
|
level="error",
|
|
message=str(exc),
|
|
payload={},
|
|
)
|
|
time.sleep(interval)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|