1709 lines
65 KiB
Python
1709 lines
65 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import socket
|
||
import subprocess
|
||
import traceback
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
from uuid import uuid4
|
||
|
||
from app.services.ops_action_executor_core import (
|
||
build_service_name_map,
|
||
execute_structured_action,
|
||
supports_structured_action,
|
||
)
|
||
from app.services.ops_release_executor_core import (
|
||
execute_release_action,
|
||
normalize_release_health_check_services as _release_normalize_health_check_services,
|
||
normalize_text_list as _release_normalize_text_list,
|
||
)
|
||
|
||
|
||
CONTROL_PLANE_BASE_URL = str(os.getenv("OPS_CONTROL_PLANE_BASE_URL", "")).strip().rstrip("/")
|
||
AGENT_TOKEN = str(os.getenv("OPS_AGENT_TOKEN", "")).strip()
|
||
NODE_CODE = str(os.getenv("NODE_CODE", "")).strip()
|
||
NODE_REGION = str(os.getenv("NODE_REGION", "mainland")).strip() or "mainland"
|
||
NODE_ROLE = str(os.getenv("NODE_ROLE", "worker")).strip() or "worker"
|
||
AGENT_POLL_INTERVAL_SECONDS = max(2, int(os.getenv("OPS_AGENT_POLL_INTERVAL_SECONDS", "5") or 5))
|
||
AGENT_RUNTIME_CONFIG_SYNC_INTERVAL_SECONDS = max(
|
||
10,
|
||
int(os.getenv("OPS_AGENT_RUNTIME_CONFIG_SYNC_INTERVAL_SECONDS", "30") or 30),
|
||
)
|
||
AGENT_HTTP_TIMEOUT_SECONDS = max(10, int(os.getenv("OPS_AGENT_HTTP_TIMEOUT_SECONDS", "60") or 60))
|
||
AGENT_REGISTER_TIMEOUT_SECONDS = max(
|
||
10,
|
||
int(os.getenv("OPS_AGENT_REGISTER_TIMEOUT_SECONDS", str(AGENT_HTTP_TIMEOUT_SECONDS)) or AGENT_HTTP_TIMEOUT_SECONDS),
|
||
)
|
||
AGENT_HEARTBEAT_TIMEOUT_SECONDS = max(
|
||
10,
|
||
int(os.getenv("OPS_AGENT_HEARTBEAT_TIMEOUT_SECONDS", str(AGENT_HTTP_TIMEOUT_SECONDS)) or AGENT_HTTP_TIMEOUT_SECONDS),
|
||
)
|
||
AGENT_PULL_TIMEOUT_SECONDS = max(
|
||
10,
|
||
int(os.getenv("OPS_AGENT_PULL_TIMEOUT_SECONDS", str(AGENT_HTTP_TIMEOUT_SECONDS)) or AGENT_HTTP_TIMEOUT_SECONDS),
|
||
)
|
||
AGENT_RUNTIME_CONFIG_TIMEOUT_SECONDS = max(
|
||
10,
|
||
int(
|
||
os.getenv("OPS_AGENT_RUNTIME_CONFIG_TIMEOUT_SECONDS", str(AGENT_HTTP_TIMEOUT_SECONDS))
|
||
or AGENT_HTTP_TIMEOUT_SECONDS
|
||
),
|
||
)
|
||
AGENT_JOB_COMPLETE_TIMEOUT_SECONDS = max(
|
||
10,
|
||
int(
|
||
os.getenv("OPS_AGENT_JOB_COMPLETE_TIMEOUT_SECONDS", str(AGENT_HTTP_TIMEOUT_SECONDS))
|
||
or AGENT_HTTP_TIMEOUT_SECONDS
|
||
),
|
||
)
|
||
AGENT_JOB_EVENT_TIMEOUT_SECONDS = max(
|
||
10,
|
||
int(
|
||
os.getenv("OPS_AGENT_JOB_EVENT_TIMEOUT_SECONDS", str(max(15, AGENT_HTTP_TIMEOUT_SECONDS // 2)))
|
||
or max(15, AGENT_HTTP_TIMEOUT_SECONDS // 2)
|
||
),
|
||
)
|
||
WORKER_SERVICE_NAME = str(os.getenv("WORKER_SERVICE_NAME", os.getenv("WORKER_SERVICE", "domaincheck-worker"))).strip() or "domaincheck-worker"
|
||
API_SERVICE_NAME = str(os.getenv("API_SERVICE_NAME", "domaincheck-api")).strip() or "domaincheck-api"
|
||
SYNC_AGENT_SERVICE_NAME = str(os.getenv("SYNC_AGENT_SERVICE_NAME", "domaincheck-sync-agent")).strip() or "domaincheck-sync-agent"
|
||
NODE_AGENT_SERVICE_NAME = str(os.getenv("NODE_AGENT_SERVICE_NAME", "domaincheck-node-agent")).strip() or "domaincheck-node-agent"
|
||
AGENT_VERSION = "0.1.0"
|
||
_APP_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
_PROJECT_DIR = os.path.dirname(_APP_DIR)
|
||
_DEFAULT_QUEUE_DIR = os.path.join(_PROJECT_DIR, "runtime", "node-agent-queue", NODE_CODE or "unbound")
|
||
AGENT_QUEUE_DIR = str(os.getenv("OPS_AGENT_QUEUE_DIR", _DEFAULT_QUEUE_DIR)).strip() or _DEFAULT_QUEUE_DIR
|
||
AGENT_QUEUE_FLUSH_LIMIT = max(1, int(os.getenv("OPS_AGENT_QUEUE_FLUSH_LIMIT", "20") or 20))
|
||
_PERMANENT_DELIVERY_DETAIL_CODES = {
|
||
"agent_node_code_required",
|
||
"ops_job_not_owned_by_agent",
|
||
"ops_job_invalid_status",
|
||
}
|
||
_LAST_QUEUE_FLUSH_SUMMARY = {
|
||
"scanned": 0,
|
||
"delivered": 0,
|
||
"deferred": 0,
|
||
"dead_letter": 0,
|
||
"last_flush_at": "",
|
||
}
|
||
_LAST_RUNTIME_CONFIG_HASH = ""
|
||
|
||
|
||
def _normalize_text_list(raw_value: object) -> list[str]:
|
||
"""Keep node-agent payload normalization aligned with release executor helpers."""
|
||
return _release_normalize_text_list(raw_value)
|
||
|
||
|
||
def _normalize_release_health_check_services(
|
||
payload: dict | None,
|
||
restart_services: list[str] | None,
|
||
) -> list[str]:
|
||
"""Backward-compatible wrapper used by node-agent tests and payload shaping."""
|
||
return _release_normalize_health_check_services(
|
||
dict(payload or {}),
|
||
list(restart_services or []),
|
||
default_api_service_name=API_SERVICE_NAME,
|
||
)
|
||
|
||
|
||
def _log(message: str) -> None:
|
||
print(f"{datetime.now().isoformat(sep=' ', timespec='seconds')} [node-agent] {message}", flush=True)
|
||
|
||
|
||
def _json_env(name: str, fallback: object) -> object:
|
||
raw = str(os.getenv(name, "")).strip()
|
||
if not raw:
|
||
return fallback
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception:
|
||
return fallback
|
||
|
||
|
||
AGENT_CAPABILITIES = _json_env(
|
||
"OPS_AGENT_CAPABILITIES",
|
||
[
|
||
"service.start",
|
||
"service.stop",
|
||
"service.restart",
|
||
"service.status",
|
||
"runtime.start_worker",
|
||
"runtime.stop_worker",
|
||
"runtime.start_detection",
|
||
"runtime.stop_detection",
|
||
"runtime.pull_tasks",
|
||
"runtime.restart_api",
|
||
"runtime.start_sync_agent",
|
||
"runtime.stop_sync_agent",
|
||
"runtime.reset_lab_state",
|
||
"health.snapshot",
|
||
"logs.collect",
|
||
"diagnostics.collect",
|
||
"delivery.queue.flush",
|
||
"delivery.queue.replay",
|
||
"delivery.queue.discard",
|
||
"deploy.release",
|
||
],
|
||
)
|
||
AGENT_LABELS = _json_env("OPS_AGENT_LABELS", {})
|
||
|
||
|
||
class AgentResponseError(RuntimeError):
|
||
def __init__(self, message: str, *, detail_code: str = "", payload: dict | None = None):
|
||
super().__init__(message)
|
||
self.detail_code = str(detail_code or "").strip()
|
||
self.payload = dict(payload or {})
|
||
|
||
|
||
def _now_text() -> str:
|
||
return datetime.now().isoformat(sep=" ", timespec="seconds")
|
||
|
||
|
||
def _queue_pending_dir() -> str:
|
||
return os.path.join(AGENT_QUEUE_DIR, "pending")
|
||
|
||
|
||
def _queue_dead_letter_dir() -> str:
|
||
return os.path.join(AGENT_QUEUE_DIR, "dead-letter")
|
||
|
||
|
||
def _queue_discarded_dir() -> str:
|
||
return os.path.join(AGENT_QUEUE_DIR, "discarded")
|
||
|
||
|
||
def _ensure_queue_dirs() -> None:
|
||
for path in (AGENT_QUEUE_DIR, _queue_pending_dir(), _queue_dead_letter_dir(), _queue_discarded_dir()):
|
||
os.makedirs(path, exist_ok=True)
|
||
|
||
|
||
def _write_json_file(path: str, payload: dict) -> None:
|
||
temp_path = f"{path}.tmp"
|
||
with open(temp_path, "w", encoding="utf-8") as handle:
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||
os.replace(temp_path, path)
|
||
|
||
|
||
def _read_json_file(path: str) -> dict:
|
||
with open(path, "r", encoding="utf-8") as handle:
|
||
return json.load(handle)
|
||
|
||
|
||
def _new_request_id(prefix: str, job_id: int) -> str:
|
||
normalized_prefix = str(prefix or "request").strip() or "request"
|
||
return f"{normalized_prefix}-{int(job_id or 0)}-{int(time.time() * 1000)}-{uuid4().hex[:10]}"
|
||
|
||
|
||
def _queue_file_names(directory: str) -> list[str]:
|
||
_ensure_queue_dirs()
|
||
try:
|
||
return sorted(file_name for file_name in os.listdir(directory) if file_name.endswith(".json"))
|
||
except FileNotFoundError:
|
||
return []
|
||
|
||
|
||
def _queue_head_record(directory: str) -> dict:
|
||
file_names = _queue_file_names(directory)
|
||
if not file_names:
|
||
return {}
|
||
file_path = os.path.join(directory, file_names[0])
|
||
try:
|
||
return _read_json_file(file_path)
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _queue_record_id(record: dict, *, file_name: str = "") -> str:
|
||
request_id = str((record or {}).get("request_id") or "").strip()
|
||
if request_id:
|
||
return request_id
|
||
normalized_file_name = str(file_name or "").strip()
|
||
if normalized_file_name.endswith(".json"):
|
||
return normalized_file_name[:-5]
|
||
return normalized_file_name
|
||
|
||
|
||
def _queue_entries(state: str) -> list[dict]:
|
||
normalized_state = str(state or "").strip()
|
||
if normalized_state == "pending":
|
||
directory = _queue_pending_dir()
|
||
elif normalized_state == "dead_letter":
|
||
directory = _queue_dead_letter_dir()
|
||
else:
|
||
return []
|
||
|
||
entries: list[dict] = []
|
||
for file_name in _queue_file_names(directory):
|
||
file_path = os.path.join(directory, file_name)
|
||
try:
|
||
record = _read_json_file(file_path)
|
||
except Exception:
|
||
continue
|
||
entries.append(
|
||
{
|
||
"state": normalized_state,
|
||
"file_name": file_name,
|
||
"file_path": file_path,
|
||
"record_id": _queue_record_id(record, file_name=file_name),
|
||
"record": dict(record or {}),
|
||
}
|
||
)
|
||
return entries
|
||
|
||
|
||
def _matches_delivery_selector(entry: dict, selector: dict, *, allowed_states: set[str] | None = None) -> bool:
|
||
normalized_selector = dict(selector or {})
|
||
record = dict(entry.get("record") or {})
|
||
state = str(entry.get("state") or "").strip()
|
||
if allowed_states and state not in allowed_states:
|
||
return False
|
||
|
||
selected_state = str(normalized_selector.get("state") or "").strip()
|
||
if selected_state and state != selected_state:
|
||
return False
|
||
|
||
selected_record_id = str(
|
||
normalized_selector.get("record_id")
|
||
or normalized_selector.get("request_id")
|
||
or ""
|
||
).strip()
|
||
if selected_record_id and str(entry.get("record_id") or "").strip() != selected_record_id:
|
||
return False
|
||
|
||
selected_request_kind = str(normalized_selector.get("request_kind") or "").strip()
|
||
if selected_request_kind and str(record.get("kind") or "").strip() != selected_request_kind:
|
||
return False
|
||
|
||
selected_detail_code = str(normalized_selector.get("detail_code") or "").strip()
|
||
if selected_detail_code and str(record.get("last_detail_code") or "").strip() != selected_detail_code:
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def _trim_queue_preview(records: list[dict], *, limit: int = 5) -> list[dict]:
|
||
safe_limit = max(1, int(limit or 5))
|
||
preview: list[dict] = []
|
||
for entry in list(records or [])[:safe_limit]:
|
||
record = dict(entry.get("record") or {})
|
||
preview.append(
|
||
{
|
||
"record_id": str(entry.get("record_id") or "").strip(),
|
||
"state": str(entry.get("state") or "").strip(),
|
||
"request_kind": str(record.get("kind") or "").strip(),
|
||
"request_id": str(record.get("request_id") or "").strip(),
|
||
"detail_code": str(record.get("last_detail_code") or "").strip(),
|
||
"created_at": str(record.get("created_at") or "").strip(),
|
||
"updated_at": str(record.get("updated_at") or "").strip(),
|
||
}
|
||
)
|
||
return preview
|
||
|
||
|
||
def _normalize_queue_limit(raw_value: object, *, default: int = 20, minimum: int = 1, maximum: int = 200) -> int:
|
||
try:
|
||
value = int(raw_value or default)
|
||
except Exception:
|
||
value = int(default)
|
||
return max(minimum, min(maximum, value))
|
||
|
||
|
||
def _replay_dead_letter_records(payload: dict | None = None) -> tuple[bool, str, dict]:
|
||
normalized_payload = dict(payload or {})
|
||
selector = dict(normalized_payload.get("selector") or {})
|
||
selector["state"] = str(selector.get("state") or "dead_letter").strip() or "dead_letter"
|
||
limit = _normalize_queue_limit(normalized_payload.get("limit"), default=20)
|
||
flush_after_replay = bool(normalized_payload.get("flush_after_replay", True))
|
||
reason = str(normalized_payload.get("reason") or "").strip()
|
||
|
||
matched_entries = [
|
||
entry
|
||
for entry in _queue_entries("dead_letter")
|
||
if _matches_delivery_selector(entry, selector, allowed_states={"dead_letter"})
|
||
][:limit]
|
||
if not matched_entries:
|
||
return False, "未找到符合条件的死信记录", {
|
||
"selector": selector,
|
||
"limit": limit,
|
||
"queue": _delivery_queue_snapshot(),
|
||
}
|
||
|
||
replayed_total = 0
|
||
for entry in matched_entries:
|
||
record = dict(entry.get("record") or {})
|
||
record["updated_at"] = _now_text()
|
||
record["replay_count"] = int(record.get("replay_count") or 0) + 1
|
||
record["last_replay_at"] = _now_text()
|
||
if reason:
|
||
record["last_replay_reason"] = reason
|
||
record.pop("dead_letter_at", None)
|
||
record.pop("dead_letter_reason", None)
|
||
_store_pending_delivery(record)
|
||
try:
|
||
os.remove(str(entry.get("file_path") or ""))
|
||
except FileNotFoundError:
|
||
pass
|
||
replayed_total += 1
|
||
|
||
flush_summary = {}
|
||
if flush_after_replay and replayed_total > 0:
|
||
flush_summary = _flush_delivery_queue(limit=replayed_total)
|
||
|
||
result = {
|
||
"selector": selector,
|
||
"limit": limit,
|
||
"replayed_total": replayed_total,
|
||
"flush_after_replay": flush_after_replay,
|
||
"flush_summary": flush_summary,
|
||
"matched_records": _trim_queue_preview(matched_entries),
|
||
"queue": _delivery_queue_snapshot(),
|
||
}
|
||
return True, f"已重放 {replayed_total} 条死信记录", result
|
||
|
||
|
||
def _store_discarded_delivery(record: dict, *, discarded_by: str = "", reason: str = "") -> str:
|
||
_ensure_queue_dirs()
|
||
discarded_record = dict(record or {})
|
||
discarded_record["discarded_at"] = _now_text()
|
||
discarded_record["discarded_by"] = str(discarded_by or "").strip()
|
||
discarded_record["discard_reason"] = str(reason or "").strip()
|
||
file_name = f"{int(time.time() * 1000)}-{uuid4().hex}.json"
|
||
file_path = os.path.join(_queue_discarded_dir(), file_name)
|
||
_write_json_file(file_path, discarded_record)
|
||
return file_path
|
||
|
||
|
||
def _discard_dead_letter_records(payload: dict | None = None) -> tuple[bool, str, dict]:
|
||
normalized_payload = dict(payload or {})
|
||
selector = dict(normalized_payload.get("selector") or {})
|
||
selector["state"] = str(selector.get("state") or "dead_letter").strip() or "dead_letter"
|
||
limit = _normalize_queue_limit(normalized_payload.get("limit"), default=20)
|
||
discarded_by = str(normalized_payload.get("discarded_by") or "").strip()
|
||
reason = str(normalized_payload.get("reason") or "").strip()
|
||
|
||
matched_entries = [
|
||
entry
|
||
for entry in _queue_entries("dead_letter")
|
||
if _matches_delivery_selector(entry, selector, allowed_states={"dead_letter"})
|
||
][:limit]
|
||
if not matched_entries:
|
||
return False, "未找到符合条件的死信记录", {
|
||
"selector": selector,
|
||
"limit": limit,
|
||
"queue": _delivery_queue_snapshot(),
|
||
}
|
||
|
||
discarded_total = 0
|
||
for entry in matched_entries:
|
||
_store_discarded_delivery(
|
||
dict(entry.get("record") or {}),
|
||
discarded_by=discarded_by,
|
||
reason=reason,
|
||
)
|
||
try:
|
||
os.remove(str(entry.get("file_path") or ""))
|
||
except FileNotFoundError:
|
||
pass
|
||
discarded_total += 1
|
||
|
||
result = {
|
||
"selector": selector,
|
||
"limit": limit,
|
||
"discarded_total": discarded_total,
|
||
"discarded_by": discarded_by,
|
||
"reason": reason,
|
||
"matched_records": _trim_queue_preview(matched_entries),
|
||
"queue": _delivery_queue_snapshot(),
|
||
}
|
||
return True, f"已丢弃 {discarded_total} 条死信记录", result
|
||
|
||
|
||
def _delivery_queue_snapshot() -> dict:
|
||
pending_dir = _queue_pending_dir()
|
||
dead_letter_dir = _queue_dead_letter_dir()
|
||
pending_files = _queue_file_names(pending_dir)
|
||
dead_letter_files = _queue_file_names(dead_letter_dir)
|
||
pending_head = _queue_head_record(pending_dir)
|
||
dead_letter_head = _queue_head_record(dead_letter_dir)
|
||
pending_count = len(pending_files)
|
||
dead_letter_count = len(dead_letter_files)
|
||
|
||
state = "healthy"
|
||
label = "正常"
|
||
reason = "当前没有待重试回执，也没有死信记录。"
|
||
if dead_letter_count > 0:
|
||
state = "dead_letter"
|
||
label = f"死信 {dead_letter_count}"
|
||
reason = "存在语义失败的回执/事件，自动重试已停止，建议人工查看。"
|
||
elif pending_count > 0:
|
||
state = "retrying"
|
||
label = f"待重试 {pending_count}"
|
||
reason = "存在待重试的回执/事件，Node Agent 会在后续 heartbeat/poll 周期继续回放。"
|
||
elif not pending_head and not dead_letter_head:
|
||
label = "正常"
|
||
|
||
return {
|
||
"state": state,
|
||
"label": label,
|
||
"reason": reason,
|
||
"pending_count": pending_count,
|
||
"dead_letter_count": dead_letter_count,
|
||
"oldest_pending_at": str(pending_head.get("created_at") or "").strip(),
|
||
"oldest_pending_request_id": str(pending_head.get("request_id") or "").strip(),
|
||
"oldest_pending_kind": str(pending_head.get("kind") or "").strip(),
|
||
"oldest_dead_letter_at": str(
|
||
dead_letter_head.get("dead_letter_at") or dead_letter_head.get("created_at") or ""
|
||
).strip(),
|
||
"oldest_dead_letter_request_id": str(dead_letter_head.get("request_id") or "").strip(),
|
||
"oldest_dead_letter_kind": str(dead_letter_head.get("kind") or "").strip(),
|
||
"last_flush_at": str(_LAST_QUEUE_FLUSH_SUMMARY.get("last_flush_at") or "").strip(),
|
||
"last_flush_delivered": int(_LAST_QUEUE_FLUSH_SUMMARY.get("delivered") or 0),
|
||
"last_flush_deferred": int(_LAST_QUEUE_FLUSH_SUMMARY.get("deferred") or 0),
|
||
"last_flush_dead_letter": int(_LAST_QUEUE_FLUSH_SUMMARY.get("dead_letter") or 0),
|
||
}
|
||
|
||
|
||
def _response_detail_code(response: dict) -> str:
|
||
if not isinstance(response, dict):
|
||
return ""
|
||
top_level = str(response.get("detail_code") or "").strip()
|
||
if top_level:
|
||
return top_level
|
||
data = response.get("data")
|
||
if isinstance(data, dict):
|
||
return str(data.get("detail_code") or "").strip()
|
||
return ""
|
||
|
||
|
||
def _ensure_ok_response(response: dict, fallback_message: str) -> dict:
|
||
raw_code = response.get("code", 1)
|
||
try:
|
||
normalized_code = int(raw_code if raw_code not in (None, "") else 1)
|
||
except Exception:
|
||
normalized_code = 1
|
||
if normalized_code == 0:
|
||
return response
|
||
raise AgentResponseError(
|
||
str(response.get("message") or fallback_message),
|
||
detail_code=_response_detail_code(response),
|
||
payload=(response.get("data") if isinstance(response.get("data"), dict) else {}),
|
||
)
|
||
|
||
|
||
def _build_delivery_record(kind: str, path: str, payload: dict, request_id: str) -> dict:
|
||
return {
|
||
"kind": str(kind or "").strip() or "delivery",
|
||
"path": str(path or "").strip(),
|
||
"payload": dict(payload or {}),
|
||
"request_id": str(request_id or "").strip(),
|
||
"attempt_count": 0,
|
||
"created_at": _now_text(),
|
||
"updated_at": _now_text(),
|
||
"last_error": "",
|
||
"last_detail_code": "",
|
||
}
|
||
|
||
|
||
def _store_pending_delivery(record: dict) -> str:
|
||
_ensure_queue_dirs()
|
||
file_name = f"{int(time.time() * 1000)}-{uuid4().hex}.json"
|
||
file_path = os.path.join(_queue_pending_dir(), file_name)
|
||
_write_json_file(file_path, record)
|
||
return file_path
|
||
|
||
|
||
def _store_dead_letter_delivery(record: dict, *, reason: str, detail_code: str = "") -> str:
|
||
_ensure_queue_dirs()
|
||
dead_record = dict(record or {})
|
||
dead_record["dead_letter_reason"] = str(reason or "").strip()
|
||
dead_record["dead_letter_at"] = _now_text()
|
||
dead_record["last_error"] = str(reason or "").strip()
|
||
dead_record["last_detail_code"] = str(detail_code or "").strip()
|
||
file_name = f"{int(time.time() * 1000)}-{uuid4().hex}.json"
|
||
file_path = os.path.join(_queue_dead_letter_dir(), file_name)
|
||
_write_json_file(file_path, dead_record)
|
||
return file_path
|
||
|
||
|
||
def _dispatch_delivery_record(record: dict) -> dict:
|
||
kind = str(record.get("kind") or "delivery").strip() or "delivery"
|
||
path = str(record.get("path") or "").strip()
|
||
payload = record.get("payload") or {}
|
||
timeout = 15 if kind == "job_event" else 30
|
||
response = _post(path, payload, timeout=timeout)
|
||
return _ensure_ok_response(response, f"{kind} failed")
|
||
|
||
|
||
def _flush_delivery_queue(limit: int | None = None) -> dict:
|
||
global _LAST_QUEUE_FLUSH_SUMMARY
|
||
_ensure_queue_dirs()
|
||
safe_limit = max(1, int(limit or AGENT_QUEUE_FLUSH_LIMIT or 1))
|
||
summary = {
|
||
"scanned": 0,
|
||
"delivered": 0,
|
||
"deferred": 0,
|
||
"dead_letter": 0,
|
||
}
|
||
file_names = sorted(
|
||
file_name
|
||
for file_name in os.listdir(_queue_pending_dir())
|
||
if file_name.endswith(".json")
|
||
)[:safe_limit]
|
||
for file_name in file_names:
|
||
summary["scanned"] += 1
|
||
file_path = os.path.join(_queue_pending_dir(), file_name)
|
||
try:
|
||
record = _read_json_file(file_path)
|
||
except Exception as exc:
|
||
_store_dead_letter_delivery({"file_name": file_name}, reason=f"invalid queue record: {exc}")
|
||
try:
|
||
os.remove(file_path)
|
||
except FileNotFoundError:
|
||
pass
|
||
summary["dead_letter"] += 1
|
||
continue
|
||
|
||
try:
|
||
_dispatch_delivery_record(record)
|
||
try:
|
||
os.remove(file_path)
|
||
except FileNotFoundError:
|
||
pass
|
||
summary["delivered"] += 1
|
||
except AgentResponseError as exc:
|
||
record["attempt_count"] = int(record.get("attempt_count") or 0) + 1
|
||
record["updated_at"] = _now_text()
|
||
record["last_error"] = str(exc)
|
||
record["last_detail_code"] = exc.detail_code
|
||
if exc.detail_code in _PERMANENT_DELIVERY_DETAIL_CODES:
|
||
_store_dead_letter_delivery(record, reason=str(exc), detail_code=exc.detail_code)
|
||
try:
|
||
os.remove(file_path)
|
||
except FileNotFoundError:
|
||
pass
|
||
summary["dead_letter"] += 1
|
||
_log(
|
||
f"delivery moved to dead-letter: kind={record.get('kind')} request_id={record.get('request_id')} "
|
||
f"detail_code={exc.detail_code or '-'} message={exc}"
|
||
)
|
||
else:
|
||
_write_json_file(file_path, record)
|
||
summary["deferred"] += 1
|
||
except Exception as exc:
|
||
record["attempt_count"] = int(record.get("attempt_count") or 0) + 1
|
||
record["updated_at"] = _now_text()
|
||
record["last_error"] = str(exc)
|
||
_write_json_file(file_path, record)
|
||
summary["deferred"] += 1
|
||
summary["last_flush_at"] = _now_text()
|
||
_LAST_QUEUE_FLUSH_SUMMARY = dict(summary)
|
||
return summary
|
||
|
||
|
||
def _deliver_or_queue(
|
||
*,
|
||
kind: str,
|
||
path: str,
|
||
payload: dict,
|
||
request_id: str,
|
||
timeout: int = 30,
|
||
) -> dict:
|
||
record = _build_delivery_record(kind, path, payload, request_id)
|
||
try:
|
||
response = _post(path, payload, timeout=timeout)
|
||
_ensure_ok_response(response, f"{kind} failed")
|
||
return {"state": "delivered", "request_id": request_id, "detail_code": "", "response": response}
|
||
except AgentResponseError as exc:
|
||
if exc.detail_code in _PERMANENT_DELIVERY_DETAIL_CODES:
|
||
file_path = _store_dead_letter_delivery(record, reason=str(exc), detail_code=exc.detail_code)
|
||
_log(
|
||
f"{kind} dead-lettered: request_id={request_id} detail_code={exc.detail_code or '-'} "
|
||
f"message={exc} file={file_path}"
|
||
)
|
||
return {"state": "dead_letter", "request_id": request_id, "detail_code": exc.detail_code, "file_path": file_path}
|
||
file_path = _store_pending_delivery({**record, "last_error": str(exc), "last_detail_code": exc.detail_code})
|
||
_log(
|
||
f"{kind} queued for retry: request_id={request_id} detail_code={exc.detail_code or '-'} "
|
||
f"message={exc} file={file_path}"
|
||
)
|
||
return {"state": "queued", "request_id": request_id, "detail_code": exc.detail_code, "file_path": file_path}
|
||
except Exception as exc:
|
||
file_path = _store_pending_delivery({**record, "last_error": str(exc)})
|
||
_log(f"{kind} queued for retry: request_id={request_id} message={exc} file={file_path}")
|
||
return {"state": "queued", "request_id": request_id, "detail_code": "", "file_path": file_path}
|
||
|
||
|
||
def _headers() -> dict[str, str]:
|
||
return {
|
||
"Content-Type": "application/json",
|
||
"X-Domaincheck-Agent-Token": AGENT_TOKEN,
|
||
}
|
||
|
||
|
||
def _request(method: str, path: str, payload: dict | None = None, timeout: int = AGENT_HTTP_TIMEOUT_SECONDS) -> dict:
|
||
if not CONTROL_PLANE_BASE_URL:
|
||
raise RuntimeError("OPS_CONTROL_PLANE_BASE_URL 未配置")
|
||
if not AGENT_TOKEN:
|
||
raise RuntimeError("OPS_AGENT_TOKEN 未配置")
|
||
url = f"{CONTROL_PLANE_BASE_URL}{path}"
|
||
data = json.dumps(payload or {}, ensure_ascii=False).encode("utf-8")
|
||
request = urllib.request.Request(url=url, data=data, headers=_headers(), method=method.upper())
|
||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||
body = response.read().decode("utf-8", errors="ignore")
|
||
return json.loads(body or "{}")
|
||
|
||
|
||
def _post(path: str, payload: dict, timeout: int = AGENT_HTTP_TIMEOUT_SECONDS) -> dict:
|
||
return _request("POST", path, payload, timeout=timeout)
|
||
|
||
|
||
def _pull_runtime_config(timeout: int = AGENT_RUNTIME_CONFIG_TIMEOUT_SECONDS) -> dict:
|
||
response = _post("/api/v1/ops/agent/runtime-config", {"node_code": NODE_CODE}, timeout=timeout)
|
||
_ensure_ok_response(response, "agent runtime config pull failed")
|
||
data = response.get("data") or {}
|
||
return dict(data.get("bundle") or {})
|
||
|
||
|
||
def _publish_local_config_update(config_type: str) -> None:
|
||
try:
|
||
from app.core.redis_client import get_redis
|
||
|
||
redis_client = get_redis()
|
||
redis_client.publish("domain_tool:config_update", str(config_type or "").strip() or "config")
|
||
except Exception:
|
||
return
|
||
|
||
|
||
def _normalize_process_count(value: object, default: int = 1) -> int:
|
||
try:
|
||
normalized = int(value)
|
||
except Exception:
|
||
return max(1, int(default or 1))
|
||
return max(1, normalized)
|
||
|
||
|
||
def _resolve_desired_process_count(bundle: dict) -> int:
|
||
normalized_bundle = dict(bundle or {})
|
||
node_process_counts = normalized_bundle.get("node_process_counts")
|
||
if isinstance(node_process_counts, dict):
|
||
override_value = node_process_counts.get(NODE_CODE)
|
||
if override_value not in (None, ""):
|
||
return _normalize_process_count(override_value, default=80)
|
||
return _normalize_process_count(normalized_bundle.get("process_count", 80), default=80)
|
||
|
||
|
||
def _worker_instance_suffixes(extra_count: int) -> list[str]:
|
||
alphabet = "abcdefghijklmnopqrstuvwxyz"
|
||
suffixes: list[str] = []
|
||
normalized_extra_count = max(0, int(extra_count or 0))
|
||
if normalized_extra_count <= 0:
|
||
return suffixes
|
||
|
||
for char in alphabet:
|
||
suffixes.append(char)
|
||
if len(suffixes) >= normalized_extra_count:
|
||
return suffixes
|
||
for first in alphabet:
|
||
for second in alphabet:
|
||
suffixes.append(f"{first}{second}")
|
||
if len(suffixes) >= normalized_extra_count:
|
||
return suffixes
|
||
return suffixes[:normalized_extra_count]
|
||
|
||
|
||
def _worker_env_path(service_name: str, suffix: str = "") -> Path:
|
||
normalized_service_name = str(service_name or "").strip() or WORKER_SERVICE_NAME
|
||
suffix_text = f"-{suffix}" if str(suffix or "").strip() else ""
|
||
return Path("/etc/default") / f"{normalized_service_name}{suffix_text}"
|
||
|
||
|
||
def _worker_instance_unit_path(service_name: str) -> Path:
|
||
normalized_service_name = str(service_name or "").strip() or WORKER_SERVICE_NAME
|
||
return Path("/etc/systemd/system") / f"{normalized_service_name}@.service"
|
||
|
||
|
||
def _worker_instance_unit_name(service_name: str, suffix: str) -> str:
|
||
normalized_service_name = str(service_name or "").strip() or WORKER_SERVICE_NAME
|
||
return f"{normalized_service_name}@{suffix}"
|
||
|
||
|
||
def _template_worker_unit_source_path() -> Path:
|
||
return Path(_PROJECT_DIR) / "deploy" / "systemd" / "domain-worker@.service"
|
||
|
||
|
||
def _render_worker_instance_env(base_env_text: str, *, instance_node_code: str, parent_node_code: str) -> str:
|
||
lines: list[str] = []
|
||
saw_node_code = False
|
||
saw_parent = False
|
||
for raw_line in str(base_env_text or "").splitlines():
|
||
if raw_line.startswith("NODE_CODE="):
|
||
lines.append(f"NODE_CODE={instance_node_code}")
|
||
saw_node_code = True
|
||
continue
|
||
if raw_line.startswith("WORKER_PARENT_NODE_CODE="):
|
||
lines.append(f"WORKER_PARENT_NODE_CODE={parent_node_code}")
|
||
saw_parent = True
|
||
continue
|
||
lines.append(raw_line)
|
||
if not saw_node_code:
|
||
lines.append(f"NODE_CODE={instance_node_code}")
|
||
if not saw_parent:
|
||
lines.append(f"WORKER_PARENT_NODE_CODE={parent_node_code}")
|
||
return "\n".join(lines).rstrip() + "\n"
|
||
|
||
|
||
def _ensure_worker_instance_unit_template(service_name: str) -> tuple[bool, str]:
|
||
unit_path = _worker_instance_unit_path(service_name)
|
||
source_path = _template_worker_unit_source_path()
|
||
if not source_path.exists():
|
||
return False, f"worker instance template missing: {source_path}"
|
||
try:
|
||
template_text = source_path.read_text(encoding="utf-8")
|
||
normalized_service_name = str(service_name or "").strip() or WORKER_SERVICE_NAME
|
||
rendered = (
|
||
template_text
|
||
.replace("domaincheck-worker-%i", f"{normalized_service_name}-%i")
|
||
.replace("domaincheck-worker@%i", f"{normalized_service_name}@%i")
|
||
)
|
||
if unit_path.exists():
|
||
existing = unit_path.read_text(encoding="utf-8")
|
||
if existing == rendered:
|
||
return True, str(unit_path)
|
||
unit_path.write_text(rendered, encoding="utf-8")
|
||
return True, str(unit_path)
|
||
except Exception as exc:
|
||
return False, f"write worker instance template failed: {exc}"
|
||
|
||
|
||
def _worker_instance_start_batch_size() -> int:
|
||
raw_value = str(os.getenv("NODE_AGENT_WORKER_RECONCILE_BATCH_SIZE", "") or "").strip()
|
||
try:
|
||
return max(1, min(32, int(raw_value or 6)))
|
||
except Exception:
|
||
return 6
|
||
|
||
|
||
def _worker_instance_start_batch_delay_seconds() -> float:
|
||
raw_value = str(os.getenv("NODE_AGENT_WORKER_RECONCILE_BATCH_DELAY_SECONDS", "") or "").strip()
|
||
try:
|
||
return max(0.0, min(30.0, float(raw_value or 1.0)))
|
||
except Exception:
|
||
return 1.0
|
||
|
||
|
||
def _chunked_units(units: list[str], size: int) -> list[list[str]]:
|
||
batch_size = max(1, int(size or 1))
|
||
return [units[index:index + batch_size] for index in range(0, len(units), batch_size)]
|
||
|
||
|
||
def _reconcile_worker_instances(bundle: dict) -> dict:
|
||
runtime_settings = dict(bundle.get("runtime_settings") or {})
|
||
worker_mode = str(runtime_settings.get("worker_mode") or "").strip() or "windows-local"
|
||
worker_service_name = str(runtime_settings.get("worker_service_name") or WORKER_SERVICE_NAME).strip() or WORKER_SERVICE_NAME
|
||
desired_process_count = _resolve_desired_process_count(bundle)
|
||
|
||
if worker_mode != "linux-systemd":
|
||
return {
|
||
"applied": False,
|
||
"reason": f"worker_mode={worker_mode}",
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
if NODE_REGION != "mainland":
|
||
return {
|
||
"applied": False,
|
||
"reason": f"region={NODE_REGION}",
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
|
||
base_env_path = _worker_env_path(worker_service_name)
|
||
if not base_env_path.exists():
|
||
return {
|
||
"applied": False,
|
||
"reason": f"base env missing: {base_env_path}",
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
|
||
ok, template_message = _ensure_worker_instance_unit_template(worker_service_name)
|
||
if not ok:
|
||
return {
|
||
"applied": False,
|
||
"reason": template_message,
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
|
||
desired_suffixes = _worker_instance_suffixes(max(0, desired_process_count - 1))
|
||
desired_units = [_worker_instance_unit_name(worker_service_name, suffix) for suffix in desired_suffixes]
|
||
desired_env_paths = {_worker_env_path(worker_service_name, suffix) for suffix in desired_suffixes}
|
||
managed_prefix = f"{worker_service_name}-"
|
||
|
||
try:
|
||
base_env_text = base_env_path.read_text(encoding="utf-8")
|
||
for suffix in desired_suffixes:
|
||
env_path = _worker_env_path(worker_service_name, suffix)
|
||
env_path.write_text(
|
||
_render_worker_instance_env(
|
||
base_env_text,
|
||
instance_node_code=f"{NODE_CODE}-{suffix}",
|
||
parent_node_code=NODE_CODE,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
except Exception as exc:
|
||
return {
|
||
"applied": False,
|
||
"reason": f"write worker env failed: {exc}",
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
|
||
existing_env_paths: list[Path] = []
|
||
try:
|
||
for candidate in Path("/etc/default").iterdir():
|
||
if not candidate.is_file():
|
||
continue
|
||
if not candidate.name.startswith(managed_prefix):
|
||
continue
|
||
existing_env_paths.append(candidate)
|
||
except Exception:
|
||
existing_env_paths = []
|
||
|
||
stale_env_paths = [
|
||
candidate
|
||
for candidate in existing_env_paths
|
||
if candidate not in desired_env_paths
|
||
]
|
||
|
||
rc, stdout, stderr = _run(["systemctl", "daemon-reload"], timeout=90)
|
||
if rc != 0:
|
||
return {
|
||
"applied": False,
|
||
"reason": stderr or stdout or "systemctl daemon-reload failed",
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
|
||
if desired_units:
|
||
rc, stdout, stderr = _run(["systemctl", "enable", *desired_units], timeout=180)
|
||
if rc != 0:
|
||
return {
|
||
"applied": False,
|
||
"reason": stderr or stdout or "systemctl enable worker instances failed",
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
batch_size = _worker_instance_start_batch_size()
|
||
batch_delay_seconds = _worker_instance_start_batch_delay_seconds()
|
||
for batch_index, unit_batch in enumerate(_chunked_units(desired_units, batch_size), start=1):
|
||
rc, stdout, stderr = _run(
|
||
["systemctl", "start", *unit_batch],
|
||
timeout=max(120, 30 * len(unit_batch)),
|
||
)
|
||
if rc != 0:
|
||
return {
|
||
"applied": False,
|
||
"reason": (
|
||
stderr
|
||
or stdout
|
||
or f"systemctl start worker instances failed at batch {batch_index}"
|
||
),
|
||
"desired_process_count": desired_process_count,
|
||
}
|
||
if batch_delay_seconds > 0 and batch_index * batch_size < len(desired_units):
|
||
time.sleep(batch_delay_seconds)
|
||
|
||
stale_units = [
|
||
f"{worker_service_name}@{candidate.name[len(managed_prefix):]}"
|
||
for candidate in stale_env_paths
|
||
if candidate.name[len(managed_prefix):]
|
||
]
|
||
if stale_units:
|
||
_run(["systemctl", "stop", *stale_units], timeout=180)
|
||
_run(["systemctl", "disable", *stale_units], timeout=180)
|
||
for candidate in stale_env_paths:
|
||
try:
|
||
candidate.unlink()
|
||
except Exception:
|
||
continue
|
||
|
||
return {
|
||
"applied": True,
|
||
"reason": "reconciled",
|
||
"desired_process_count": desired_process_count,
|
||
"instance_service_name": f"{worker_service_name}@.service",
|
||
"extra_instances": len(desired_suffixes),
|
||
"stale_instances_removed": len(stale_units),
|
||
"template_path": template_message,
|
||
}
|
||
|
||
|
||
def _apply_runtime_config(bundle: dict) -> bool:
|
||
global _LAST_RUNTIME_CONFIG_HASH
|
||
|
||
normalized_bundle = dict(bundle or {})
|
||
bundle_hash = str(normalized_bundle.get("config_hash") or "").strip()
|
||
if not bundle_hash:
|
||
bundle_hash = hashlib.sha256(
|
||
json.dumps(normalized_bundle, ensure_ascii=False, sort_keys=True).encode("utf-8")
|
||
).hexdigest()
|
||
if bundle_hash and bundle_hash == _LAST_RUNTIME_CONFIG_HASH:
|
||
reconcile_summary = _reconcile_worker_instances(normalized_bundle)
|
||
if reconcile_summary.get("applied"):
|
||
_log(
|
||
"worker instance reconcile refreshed: "
|
||
f"desired={reconcile_summary.get('desired_process_count', 1)} "
|
||
f"extra={reconcile_summary.get('extra_instances', 0)}"
|
||
)
|
||
return False
|
||
|
||
from app.core.files import write_json
|
||
from app.services.runtime_settings_service import update_runtime_settings
|
||
from app.services.sensitive_words_service import save_sensitive_words_payload
|
||
|
||
detect_options = dict(normalized_bundle.get("detect_options") or {})
|
||
proxy_config = dict(normalized_bundle.get("proxy_config") or {})
|
||
thread_count = int(normalized_bundle.get("thread_count", 1000) or 1000)
|
||
node_thread_counts = dict(normalized_bundle.get("node_thread_counts") or {})
|
||
process_count = int(normalized_bundle.get("process_count", 80) or 80)
|
||
node_process_counts = dict(normalized_bundle.get("node_process_counts") or {})
|
||
runtime_settings = dict(normalized_bundle.get("runtime_settings") or {})
|
||
sensitive_words = dict(normalized_bundle.get("sensitive_words") or {})
|
||
sensitive_words_text = str(sensitive_words.get("text") or "")
|
||
sensitive_word_items = list(sensitive_words.get("items") or [])
|
||
|
||
write_json("detect_options.json", detect_options)
|
||
write_json("proxy_config.json", proxy_config)
|
||
write_json("thread_count.json", {"thread_count": str(thread_count)})
|
||
write_json("node_thread_counts.json", node_thread_counts)
|
||
write_json("process_count.json", {"process_count": str(process_count)})
|
||
write_json("node_process_counts.json", node_process_counts)
|
||
write_json("runtime_settings.json", runtime_settings)
|
||
write_json("runtime/runtime_settings.json", runtime_settings)
|
||
write_json(
|
||
"runtime/sensitive_words.json",
|
||
{
|
||
"items": sensitive_word_items,
|
||
"text": sensitive_words_text,
|
||
"total": int(sensitive_words.get("total", 0) or 0),
|
||
},
|
||
)
|
||
|
||
try:
|
||
update_runtime_settings(runtime_settings)
|
||
except Exception as exc:
|
||
_log(f"runtime settings local api sync skipped: {exc}")
|
||
|
||
try:
|
||
save_sensitive_words_payload({"text": sensitive_words_text})
|
||
except Exception as exc:
|
||
_log(f"sensitive words db sync skipped: {exc}")
|
||
|
||
try:
|
||
from app.core.redis_client import get_redis
|
||
|
||
redis_client = get_redis()
|
||
redis_client.set("domain_tool:detect_options", json.dumps(detect_options, ensure_ascii=False))
|
||
redis_client.set("domain_tool:proxy_config", json.dumps(proxy_config, ensure_ascii=False))
|
||
redis_client.set("domain_tool:thread_count", thread_count)
|
||
redis_client.set("domain_tool:node_thread_counts", json.dumps(node_thread_counts, ensure_ascii=False))
|
||
redis_client.set("domain_tool:process_count", process_count)
|
||
redis_client.set("domain_tool:node_process_counts", json.dumps(node_process_counts, ensure_ascii=False))
|
||
redis_client.set("domain_tool:runtime_settings", json.dumps(runtime_settings, ensure_ascii=False))
|
||
redis_client.set("domain_tool:sensitive_words", json.dumps(sensitive_word_items, ensure_ascii=False))
|
||
redis_client.publish("domain_tool:config_update", "detect_options")
|
||
redis_client.publish("domain_tool:config_update", "proxy_config")
|
||
redis_client.publish("domain_tool:config_update", "thread_count")
|
||
redis_client.publish("domain_tool:config_update", "node_thread_counts")
|
||
redis_client.publish("domain_tool:config_update", "process_count")
|
||
redis_client.publish("domain_tool:config_update", "node_process_counts")
|
||
redis_client.publish("domain_tool:config_update", "runtime_settings")
|
||
redis_client.publish("domain_tool:config_update", "sensitive_words")
|
||
except Exception:
|
||
_publish_local_config_update("detect_options")
|
||
_publish_local_config_update("proxy_config")
|
||
_publish_local_config_update("thread_count")
|
||
_publish_local_config_update("node_thread_counts")
|
||
_publish_local_config_update("process_count")
|
||
_publish_local_config_update("node_process_counts")
|
||
_publish_local_config_update("runtime_settings")
|
||
_publish_local_config_update("sensitive_words")
|
||
|
||
reconcile_summary = _reconcile_worker_instances(normalized_bundle)
|
||
_LAST_RUNTIME_CONFIG_HASH = bundle_hash
|
||
_log(
|
||
"runtime config applied: "
|
||
f"thread_count={thread_count} "
|
||
f"process_count={process_count} "
|
||
f"node_override={node_thread_counts.get(NODE_CODE)} "
|
||
f"process_override={node_process_counts.get(NODE_CODE)} "
|
||
f"sensitive_words={int(sensitive_words.get('total', 0) or 0)}"
|
||
)
|
||
if reconcile_summary.get("applied"):
|
||
_log(
|
||
"worker instance reconcile: "
|
||
f"desired={reconcile_summary.get('desired_process_count', 1)} "
|
||
f"extra={reconcile_summary.get('extra_instances', 0)} "
|
||
f"removed={reconcile_summary.get('stale_instances_removed', 0)}"
|
||
)
|
||
elif reconcile_summary.get("reason"):
|
||
_log(f"worker instance reconcile skipped: {reconcile_summary.get('reason')}")
|
||
return True
|
||
|
||
|
||
def _hostname() -> str:
|
||
generic_values = {"localhost", "localhost.localdomain", "ip6-localhost", "localhost6"}
|
||
candidates: list[str] = []
|
||
try:
|
||
candidates.append(socket.gethostname())
|
||
except Exception:
|
||
pass
|
||
try:
|
||
candidates.append(socket.getfqdn())
|
||
except Exception:
|
||
pass
|
||
try:
|
||
candidates.append(os.uname().nodename)
|
||
except Exception:
|
||
pass
|
||
candidates.append(str(os.getenv("HOSTNAME", "")).strip())
|
||
|
||
for value in candidates:
|
||
normalized = str(value or "").strip()
|
||
if not normalized:
|
||
continue
|
||
if normalized.lower() in generic_values:
|
||
continue
|
||
return normalized
|
||
return NODE_CODE or ""
|
||
|
||
|
||
def _control_plane_host_port() -> tuple[str, int]:
|
||
parsed = urlparse(CONTROL_PLANE_BASE_URL if "://" in CONTROL_PLANE_BASE_URL else f"http://{CONTROL_PLANE_BASE_URL}")
|
||
host = str(parsed.hostname or "").strip()
|
||
if not host:
|
||
return "", 0
|
||
port = int(parsed.port or (443 if parsed.scheme == "https" else 80))
|
||
return host, port
|
||
|
||
|
||
def _first_non_loopback_ip(values: list[str]) -> str:
|
||
for value in values:
|
||
normalized = str(value or "").strip()
|
||
if not normalized or normalized.startswith("127.") or normalized == "::1":
|
||
continue
|
||
return normalized
|
||
return ""
|
||
|
||
|
||
def _ip() -> str:
|
||
explicit_ip = str(os.getenv("OPS_AGENT_IP", "")).strip()
|
||
if explicit_ip:
|
||
return explicit_ip
|
||
|
||
host, port = _control_plane_host_port()
|
||
if host and port:
|
||
for family in (socket.AF_INET, socket.AF_INET6):
|
||
sock = None
|
||
try:
|
||
sock = socket.socket(family, socket.SOCK_DGRAM)
|
||
sock.connect((host, port))
|
||
local_ip = str(sock.getsockname()[0] or "").strip()
|
||
if local_ip and not local_ip.startswith("127.") and local_ip != "::1":
|
||
return local_ip
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
if sock is not None:
|
||
try:
|
||
sock.close()
|
||
except Exception:
|
||
pass
|
||
|
||
hostname = _hostname()
|
||
if hostname:
|
||
try:
|
||
addrinfo = socket.getaddrinfo(hostname, None)
|
||
resolved_ips = [str(item[4][0] or "").strip() for item in addrinfo if item and len(item) >= 5]
|
||
best_ip = _first_non_loopback_ip(resolved_ips)
|
||
if best_ip:
|
||
return best_ip
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
fallback_ip = str(socket.gethostbyname(socket.gethostname()) or "").strip()
|
||
if fallback_ip and not fallback_ip.startswith("127."):
|
||
return fallback_ip
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _detect_runtime_snapshot() -> dict:
|
||
worker_runtime = {}
|
||
try:
|
||
from app.services.worker_control_service import detect_worker_runtime
|
||
|
||
worker_runtime = detect_worker_runtime() or {}
|
||
except Exception as exc:
|
||
worker_runtime = {
|
||
"running": False,
|
||
"process_count": 0,
|
||
"latest_start_time": "",
|
||
"message": "",
|
||
"error": str(exc),
|
||
}
|
||
|
||
try:
|
||
from app.services.detect_service import get_detect_status
|
||
|
||
detect_status = get_detect_status() or {}
|
||
runtime_state = detect_status.get("runtime_state") if isinstance(detect_status.get("runtime_state"), dict) else {}
|
||
active_job = detect_status.get("active_job") if isinstance(detect_status.get("active_job"), dict) else {}
|
||
|
||
active_threads = max(0, int(detect_status.get("active_thread_count", 0) or 0))
|
||
max_threads = max(0, int(detect_status.get("max_thread_count", 0) or 0))
|
||
current_load = max(
|
||
active_threads,
|
||
int(active_job.get("items_running", 0) or 0),
|
||
)
|
||
phase_label = str(detect_status.get("phase_label") or "").strip()
|
||
phase_detail = str(
|
||
detect_status.get("phase_detail")
|
||
or detect_status.get("recent_event")
|
||
or detect_status.get("worker_runtime_message")
|
||
or ""
|
||
).strip()
|
||
|
||
inferred_worker_online = bool(
|
||
worker_runtime.get("running", False)
|
||
or detect_status.get("worker_online", False)
|
||
or runtime_state.get("service_running", False)
|
||
or current_load > 0
|
||
or active_threads > 0
|
||
)
|
||
return {
|
||
"worker_online": inferred_worker_online,
|
||
"service_running": bool(
|
||
runtime_state.get("service_running", False)
|
||
or worker_runtime.get("running", False)
|
||
or inferred_worker_online
|
||
),
|
||
"detecting": bool(
|
||
detect_status.get("detecting", False)
|
||
or runtime_state.get("detecting", False)
|
||
or current_load > 0
|
||
or active_threads > 0
|
||
),
|
||
"active_threads": active_threads,
|
||
"max_threads": max_threads,
|
||
"current_load": current_load,
|
||
"phase_label": phase_label,
|
||
"phase_detail": phase_detail,
|
||
"recent_warning": str(detect_status.get("recent_warning") or "").strip(),
|
||
"updated_at": str(runtime_state.get("updated_at") or "").strip(),
|
||
"available_proxy_count": int(detect_status.get("available_proxy_count", 0) or 0),
|
||
"proxy_runtime_label": str(detect_status.get("proxy_runtime_label") or "").strip(),
|
||
"proxy_runtime_reason": str(detect_status.get("proxy_runtime_reason") or "").strip(),
|
||
"proxy_last_refresh_status": str(detect_status.get("proxy_last_refresh_status") or "").strip(),
|
||
"proxy_last_refresh_time": str(detect_status.get("proxy_last_refresh_time") or "").strip(),
|
||
"proxy_last_refresh_source_count": int(detect_status.get("proxy_last_refresh_source_count", 0) or 0),
|
||
"proxy_last_refresh_total_items": int(detect_status.get("proxy_last_refresh_total_items", 0) or 0),
|
||
"proxy_last_validated_count": int(detect_status.get("proxy_last_validated_count", 0) or 0),
|
||
"proxy_last_available_count": int(detect_status.get("proxy_last_available_count", 0) or 0),
|
||
"detect_participating": bool(
|
||
detect_status.get("detect_participating", False)
|
||
or current_load > 0
|
||
or active_threads > 0
|
||
),
|
||
}
|
||
except Exception as exc:
|
||
worker_online = bool(worker_runtime.get("running", False))
|
||
worker_message = str(worker_runtime.get("message") or "").strip()
|
||
return {
|
||
"worker_online": worker_online,
|
||
"service_running": worker_online,
|
||
"detecting": False,
|
||
"active_threads": 0,
|
||
"max_threads": 0,
|
||
"current_load": 0,
|
||
"phase_label": "",
|
||
"phase_detail": worker_message,
|
||
"recent_warning": "",
|
||
"updated_at": str(worker_runtime.get("latest_start_time") or "").strip(),
|
||
"available_proxy_count": 0,
|
||
"proxy_runtime_label": "",
|
||
"proxy_runtime_reason": "",
|
||
"proxy_last_refresh_status": "",
|
||
"proxy_last_refresh_time": "",
|
||
"proxy_last_refresh_source_count": 0,
|
||
"proxy_last_refresh_total_items": 0,
|
||
"proxy_last_validated_count": 0,
|
||
"proxy_last_available_count": 0,
|
||
"detect_participating": False,
|
||
"error": str(exc),
|
||
}
|
||
|
||
|
||
def _base_payload() -> dict:
|
||
return {
|
||
"node_code": NODE_CODE,
|
||
"region": NODE_REGION,
|
||
"role": NODE_ROLE,
|
||
"title": NODE_CODE,
|
||
"hostname": _hostname(),
|
||
"ip": _ip(),
|
||
"agent_version": AGENT_VERSION,
|
||
"capabilities": AGENT_CAPABILITIES,
|
||
"labels": AGENT_LABELS,
|
||
"metadata": {
|
||
"service_names": {
|
||
"api": API_SERVICE_NAME,
|
||
"worker": WORKER_SERVICE_NAME,
|
||
"sync_agent": SYNC_AGENT_SERVICE_NAME,
|
||
"node_agent": NODE_AGENT_SERVICE_NAME,
|
||
},
|
||
"delivery_queue": _delivery_queue_snapshot(),
|
||
"detect_runtime": _detect_runtime_snapshot(),
|
||
},
|
||
}
|
||
|
||
|
||
def _run(command: list[str], timeout: int = 60) -> tuple[int, str, str]:
|
||
completed = subprocess.run(command, capture_output=True, text=True, timeout=timeout)
|
||
normalized_command = [str(part or "").strip() for part in command]
|
||
combined_output = f"{completed.stdout or ''}\n{completed.stderr or ''}".lower()
|
||
needs_sudo_retry = (
|
||
normalized_command
|
||
and normalized_command[0] == "systemctl"
|
||
and completed.returncode != 0
|
||
and "sudo" not in normalized_command
|
||
and any(
|
||
marker in combined_output
|
||
for marker in (
|
||
"interactive authentication required",
|
||
"authentication is required",
|
||
"authorization not available",
|
||
"polkit",
|
||
)
|
||
)
|
||
)
|
||
if needs_sudo_retry:
|
||
completed = subprocess.run(
|
||
["sudo", "-n", *normalized_command],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
)
|
||
return completed.returncode, completed.stdout.strip(), completed.stderr.strip()
|
||
|
||
|
||
def _execute_action(
|
||
action: str,
|
||
payload: dict,
|
||
*,
|
||
job_id: int | None = None,
|
||
job_context: dict | None = None,
|
||
) -> tuple[bool, str, dict]:
|
||
normalized_action = str(action or "").strip()
|
||
if normalized_action == "delivery.queue.flush":
|
||
limit = _normalize_queue_limit((payload or {}).get("limit"), default=20)
|
||
summary = _flush_delivery_queue(limit=limit)
|
||
return True, "Delivery Queue 已执行冲刷", {
|
||
"action": normalized_action,
|
||
"limit": limit,
|
||
"flush_summary": summary,
|
||
"queue": _delivery_queue_snapshot(),
|
||
}
|
||
if normalized_action == "delivery.queue.replay":
|
||
return _replay_dead_letter_records(payload)
|
||
if normalized_action == "delivery.queue.discard":
|
||
return _discard_dead_letter_records(payload)
|
||
if supports_structured_action(normalized_action):
|
||
return execute_structured_action(
|
||
normalized_action,
|
||
payload,
|
||
service_names=build_service_name_map(
|
||
api_service_name=API_SERVICE_NAME,
|
||
worker_service_name=WORKER_SERVICE_NAME,
|
||
sync_agent_service_name=SYNC_AGENT_SERVICE_NAME,
|
||
node_agent_service_name=NODE_AGENT_SERVICE_NAME,
|
||
),
|
||
runner=_run,
|
||
host_context={
|
||
"hostname": _hostname(),
|
||
"ip": _ip(),
|
||
},
|
||
)
|
||
if normalized_action == "deploy.release":
|
||
return execute_release_action(
|
||
dict(payload or {}),
|
||
run_command=_run,
|
||
default_api_service_name=API_SERVICE_NAME,
|
||
event_callback=(
|
||
(lambda event_type, message, level="info", payload=None: _job_event(
|
||
int(job_id),
|
||
event_type=event_type,
|
||
message=message,
|
||
level=level,
|
||
payload={
|
||
**dict(payload or {}),
|
||
"release_context": dict((job_context or {}).get("release_context") or {}),
|
||
"step_key": str((job_context or {}).get("step_key") or "").strip(),
|
||
},
|
||
summary_text=message,
|
||
focus_ref=dict((job_context or {}).get("focus_ref") or {}),
|
||
occurred_at=_now_text(),
|
||
))
|
||
if job_id
|
||
else None
|
||
),
|
||
urlopen_func=urllib.request.urlopen,
|
||
user_agent=f"domaincheck-node-agent/{AGENT_VERSION}",
|
||
)
|
||
return False, f"unsupported action: {normalized_action}", {"action": normalized_action}
|
||
|
||
|
||
def _register() -> None:
|
||
response = _post("/api/v1/ops/agent/register", _base_payload(), timeout=AGENT_REGISTER_TIMEOUT_SECONDS)
|
||
_ensure_ok_response(response, "agent register failed")
|
||
_log(f"registered: {response.get('message')}")
|
||
|
||
|
||
def _heartbeat() -> None:
|
||
response = _post("/api/v1/ops/agent/heartbeat", _base_payload(), timeout=AGENT_HEARTBEAT_TIMEOUT_SECONDS)
|
||
_ensure_ok_response(response, "agent heartbeat failed")
|
||
|
||
|
||
def _pull_jobs() -> list[dict]:
|
||
response = _post(
|
||
f"/api/v1/ops/agent/pull?limit=1",
|
||
{"node_code": NODE_CODE},
|
||
timeout=AGENT_PULL_TIMEOUT_SECONDS,
|
||
)
|
||
_ensure_ok_response(response, "agent pull failed")
|
||
data = response.get("data") or {}
|
||
return list(data.get("jobs") or [])
|
||
|
||
|
||
def _normalize_agent_job(job: dict) -> dict:
|
||
normalized_job = dict(job or {})
|
||
focus_ref = normalized_job.get("focus_ref") if isinstance(normalized_job.get("focus_ref"), dict) else {}
|
||
release_context = (
|
||
normalized_job.get("release_context") if isinstance(normalized_job.get("release_context"), dict) else {}
|
||
)
|
||
step_ref = normalized_job.get("step_ref") if isinstance(normalized_job.get("step_ref"), dict) else {}
|
||
job_id = int(normalized_job.get("job_id") or normalized_job.get("id") or 0)
|
||
step_key = str(
|
||
normalized_job.get("step_key")
|
||
or step_ref.get("step_key")
|
||
or "dispatch"
|
||
).strip() or "dispatch"
|
||
step_title = str(
|
||
normalized_job.get("step_title")
|
||
or step_ref.get("step_title")
|
||
or normalized_job.get("summary")
|
||
or normalized_job.get("action")
|
||
or step_key
|
||
).strip() or step_key
|
||
if not focus_ref:
|
||
focus_ref = {
|
||
"kind": "ops_job",
|
||
"job_id": job_id,
|
||
"job_code": str(normalized_job.get("job_code") or "").strip(),
|
||
"action": str(normalized_job.get("action") or "").strip(),
|
||
"target_node_code": str(normalized_job.get("target_node_code") or "").strip(),
|
||
}
|
||
return {
|
||
**normalized_job,
|
||
"job_id": job_id,
|
||
"job_code": str(normalized_job.get("job_code") or "").strip(),
|
||
"job_type": str(normalized_job.get("job_type") or "ops_action").strip() or "ops_action",
|
||
"action": str(normalized_job.get("action") or "").strip(),
|
||
"payload": dict(normalized_job.get("payload") or {}),
|
||
"policy": dict(normalized_job.get("policy") or {}),
|
||
"focus_ref": focus_ref,
|
||
"release_context": dict(release_context or {}),
|
||
"step_key": step_key,
|
||
"step_title": step_title,
|
||
"step_ref": {
|
||
"step_id": int(step_ref.get("step_id") or 0),
|
||
"step_key": step_key,
|
||
"step_title": step_title,
|
||
},
|
||
}
|
||
|
||
|
||
def _job_start(job_id: int, *, job: dict | None = None) -> None:
|
||
normalized_job = _normalize_agent_job(job or {"job_id": job_id})
|
||
response = _post(
|
||
f"/api/v1/ops/agent/jobs/{job_id}/start",
|
||
{
|
||
"node_code": NODE_CODE,
|
||
"job_code": normalized_job.get("job_code") or "",
|
||
"action": normalized_job.get("action") or "",
|
||
"step_key": normalized_job.get("step_key") or "",
|
||
"focus_ref": normalized_job.get("focus_ref") or {},
|
||
},
|
||
)
|
||
_ensure_ok_response(response, "job start failed")
|
||
|
||
|
||
def _job_complete(
|
||
job_id: int,
|
||
*,
|
||
status: str,
|
||
stdout: str,
|
||
stderr: str,
|
||
result: dict,
|
||
error_message: str = "",
|
||
client_request_id: str | None = None,
|
||
duration_ms: int | None = None,
|
||
summary_text: str = "",
|
||
focus_ref: dict | None = None,
|
||
step_ref: dict | None = None,
|
||
release_context: dict | None = None,
|
||
) -> dict:
|
||
request_id = str(client_request_id or "").strip() or _new_request_id("complete", job_id)
|
||
payload = {
|
||
"node_code": NODE_CODE,
|
||
"status": status,
|
||
"stdout": stdout,
|
||
"stderr": stderr,
|
||
"result": result,
|
||
"error_message": error_message,
|
||
"client_request_id": request_id,
|
||
}
|
||
if duration_ms is not None:
|
||
payload["duration_ms"] = max(0, int(duration_ms or 0))
|
||
if str(summary_text or "").strip():
|
||
payload["summary_text"] = str(summary_text).strip()
|
||
if isinstance(focus_ref, dict) and focus_ref:
|
||
payload["focus_ref"] = dict(focus_ref)
|
||
if isinstance(step_ref, dict) and step_ref:
|
||
payload["step_ref"] = dict(step_ref)
|
||
if isinstance(release_context, dict) and release_context:
|
||
payload["release_context"] = dict(release_context)
|
||
return _deliver_or_queue(
|
||
kind="job_complete",
|
||
path=f"/api/v1/ops/agent/jobs/{job_id}/complete",
|
||
payload=payload,
|
||
request_id=request_id,
|
||
timeout=AGENT_JOB_COMPLETE_TIMEOUT_SECONDS,
|
||
)
|
||
|
||
|
||
def _job_event(
|
||
job_id: int,
|
||
*,
|
||
event_type: str,
|
||
message: str,
|
||
level: str = "info",
|
||
payload: dict | None = None,
|
||
client_event_id: str | None = None,
|
||
summary_text: str = "",
|
||
focus_ref: dict | None = None,
|
||
occurred_at: str = "",
|
||
) -> dict:
|
||
request_id = str(client_event_id or "").strip() or _new_request_id("event", job_id)
|
||
delivery_payload = {
|
||
"node_code": NODE_CODE,
|
||
"event_type": event_type,
|
||
"message": message,
|
||
"level": level,
|
||
"payload": payload or {},
|
||
"client_event_id": request_id,
|
||
}
|
||
if str(summary_text or "").strip():
|
||
delivery_payload["summary_text"] = str(summary_text).strip()
|
||
if isinstance(focus_ref, dict) and focus_ref:
|
||
delivery_payload["focus_ref"] = dict(focus_ref)
|
||
if str(occurred_at or "").strip():
|
||
delivery_payload["occurred_at"] = str(occurred_at).strip()
|
||
return _deliver_or_queue(
|
||
kind="job_event",
|
||
path=f"/api/v1/ops/agent/jobs/{job_id}/events",
|
||
payload=delivery_payload,
|
||
request_id=request_id,
|
||
timeout=AGENT_JOB_EVENT_TIMEOUT_SECONDS,
|
||
)
|
||
|
||
|
||
def _process_job(job: dict) -> None:
|
||
normalized_job = _normalize_agent_job(job)
|
||
job_id = int(normalized_job.get("job_id") or 0)
|
||
action = str(normalized_job.get("action") or "").strip()
|
||
payload = dict(normalized_job.get("payload") or {})
|
||
if job_id <= 0 or not action:
|
||
return
|
||
|
||
started_at = time.monotonic()
|
||
_log(
|
||
"job start: "
|
||
f"id={job_id} code={normalized_job.get('job_code') or '-'} "
|
||
f"type={normalized_job.get('job_type') or '-'} "
|
||
f"step={normalized_job.get('step_key') or '-'} action={action}"
|
||
)
|
||
start_delivery_state = "delivered"
|
||
start_delivery_error = ""
|
||
try:
|
||
_job_start(job_id, job=normalized_job)
|
||
except Exception as exc:
|
||
start_delivery_state = "failed_local"
|
||
start_delivery_error = str(exc)
|
||
_log(
|
||
"job start delivery failed, continue locally: "
|
||
f"id={job_id} code={normalized_job.get('job_code') or '-'} action={action} error={exc}"
|
||
)
|
||
_job_event(
|
||
job_id,
|
||
event_type="executor_received",
|
||
message=f"node agent accepted action {action}",
|
||
summary_text=f"已接单 {action}",
|
||
focus_ref=dict(normalized_job.get("focus_ref") or {}),
|
||
occurred_at=_now_text(),
|
||
payload={
|
||
"action": action,
|
||
"job_code": normalized_job.get("job_code") or "",
|
||
"job_type": normalized_job.get("job_type") or "",
|
||
"step_key": normalized_job.get("step_key") or "",
|
||
"step_title": normalized_job.get("step_title") or "",
|
||
"release_context": dict(normalized_job.get("release_context") or {}),
|
||
"start_delivery_state": start_delivery_state,
|
||
"start_delivery_error": start_delivery_error,
|
||
},
|
||
)
|
||
try:
|
||
ok, message, result = _execute_action(action, payload, job_id=job_id, job_context=normalized_job)
|
||
except Exception as exc:
|
||
ok = False
|
||
message = f"executor exception: {exc}"
|
||
result = {
|
||
"stdout": "",
|
||
"stderr": traceback.format_exc(),
|
||
"summary_text": message,
|
||
"exception_type": exc.__class__.__name__,
|
||
}
|
||
_log(
|
||
"job execute exception: "
|
||
f"id={job_id} code={normalized_job.get('job_code') or '-'} "
|
||
f"action={action} error={exc}"
|
||
)
|
||
stdout = str(result.get("stdout") or "")
|
||
stderr = str(result.get("stderr") or "")
|
||
duration_ms = max(0, int((time.monotonic() - started_at) * 1000))
|
||
summary_text = str(result.get("summary_text") or result.get("summary") or message).strip()
|
||
delivery = _job_complete(
|
||
job_id,
|
||
status="success" if ok else "failed",
|
||
stdout=stdout,
|
||
stderr=stderr,
|
||
result=result,
|
||
error_message="" if ok else message,
|
||
duration_ms=duration_ms,
|
||
summary_text=summary_text,
|
||
focus_ref=dict(normalized_job.get("focus_ref") or {}),
|
||
step_ref=dict(normalized_job.get("step_ref") or {}),
|
||
release_context=dict(normalized_job.get("release_context") or {}),
|
||
)
|
||
_log(
|
||
"job complete: "
|
||
f"id={job_id} code={normalized_job.get('job_code') or '-'} "
|
||
f"action={action} ok={ok} duration_ms={duration_ms} delivery={delivery.get('state')}"
|
||
)
|
||
|
||
|
||
def _process_job_with_guard(job: dict) -> None:
|
||
normalized_job = _normalize_agent_job(job)
|
||
job_id = int(normalized_job.get("job_id") or 0)
|
||
action = str(normalized_job.get("action") or "").strip()
|
||
started_at = time.monotonic()
|
||
try:
|
||
_process_job(normalized_job)
|
||
except Exception as exc:
|
||
duration_ms = max(0, int((time.monotonic() - started_at) * 1000))
|
||
traceback_text = traceback.format_exc()
|
||
summary_text = f"node-agent fatal exception: {exc}"
|
||
_log(
|
||
"job fatal exception: "
|
||
f"id={job_id} code={normalized_job.get('job_code') or '-'} "
|
||
f"action={action or '-'} error={exc}"
|
||
)
|
||
if job_id <= 0:
|
||
raise
|
||
try:
|
||
delivery = _job_complete(
|
||
job_id,
|
||
status="failed",
|
||
stdout="",
|
||
stderr=traceback_text,
|
||
result={
|
||
"stdout": "",
|
||
"stderr": traceback_text,
|
||
"summary_text": summary_text,
|
||
"exception_type": exc.__class__.__name__,
|
||
"fatal_loop_exception": True,
|
||
},
|
||
error_message=summary_text,
|
||
duration_ms=duration_ms,
|
||
summary_text=summary_text,
|
||
focus_ref=dict(normalized_job.get("focus_ref") or {}),
|
||
step_ref=dict(normalized_job.get("step_ref") or {}),
|
||
release_context=dict(normalized_job.get("release_context") or {}),
|
||
)
|
||
_log(
|
||
"job fatal exception completion: "
|
||
f"id={job_id} code={normalized_job.get('job_code') or '-'} "
|
||
f"action={action or '-'} delivery={delivery.get('state')}"
|
||
)
|
||
except Exception as completion_exc:
|
||
_log(
|
||
"job fatal exception completion failed: "
|
||
f"id={job_id} code={normalized_job.get('job_code') or '-'} "
|
||
f"action={action or '-'} error={completion_exc}"
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
if not NODE_CODE:
|
||
raise RuntimeError("NODE_CODE 未配置")
|
||
_log(f"starting node agent: node={NODE_CODE} role={NODE_ROLE} region={NODE_REGION}")
|
||
_ensure_queue_dirs()
|
||
_register()
|
||
last_heartbeat_at = 0.0
|
||
last_runtime_config_sync_at = 0.0
|
||
|
||
while True:
|
||
now = time.time()
|
||
try:
|
||
delivery_summary = _flush_delivery_queue(limit=AGENT_QUEUE_FLUSH_LIMIT)
|
||
if delivery_summary["delivered"] or delivery_summary["dead_letter"]:
|
||
_log(f"delivery queue flush: {delivery_summary}")
|
||
if now - last_runtime_config_sync_at >= AGENT_RUNTIME_CONFIG_SYNC_INTERVAL_SECONDS:
|
||
bundle = _pull_runtime_config()
|
||
_apply_runtime_config(bundle)
|
||
last_runtime_config_sync_at = now
|
||
if now - last_heartbeat_at >= 15:
|
||
_heartbeat()
|
||
last_heartbeat_at = now
|
||
jobs = _pull_jobs()
|
||
if jobs:
|
||
for job in jobs:
|
||
_process_job_with_guard(job)
|
||
else:
|
||
time.sleep(AGENT_POLL_INTERVAL_SECONDS)
|
||
except urllib.error.HTTPError as exc:
|
||
_log(f"http error: {exc.code}")
|
||
time.sleep(AGENT_POLL_INTERVAL_SECONDS)
|
||
except urllib.error.URLError as exc:
|
||
_log(f"url error: {exc}")
|
||
time.sleep(AGENT_POLL_INTERVAL_SECONDS)
|
||
except Exception as exc:
|
||
_log(f"loop error: {exc}")
|
||
time.sleep(AGENT_POLL_INTERVAL_SECONDS)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|