diff --git a/.env.example b/.env.example index 9b71072..07bbc05 100644 --- a/.env.example +++ b/.env.example @@ -19,3 +19,5 @@ NETX_UME_TOKEN_HANDSHAKE_PATH=/restconf/operations/zte-security:oauth_handshake NETX_UME_TOKEN_LOGOUT_PATH=/restconf/operations/zte-security:oauth_token NETX_UME_NE_PATH=/restconf/data/zte-resources-module:network-elements NETX_UME_ALARMS_PATH=/restconf/data/zte-alarms:alarms/alarm-list +NETX_UME_SYNC_ALARMS_CURRENT_ENABLED=true +NETX_UME_SYNC_ALARMS_CURRENT_INTERVAL_S=300 diff --git a/netx_api/config.py b/netx_api/config.py index dcb2b0d..ca2c12c 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -40,6 +40,8 @@ class Settings(BaseSettings): ume_keepalive_enabled: bool = True ume_keepalive_interval_s: int = 600 ume_keepalive_renew_before_s: int = 900 + ume_sync_alarms_current_enabled: bool = True + ume_sync_alarms_current_interval_s: int = 300 ume_token_path: str = "/restconf/operations/zte-security:oauth_token" ume_token_handshake_path: str = "/restconf/operations/zte-security:oauth_handshake" ume_token_logout_path: str = "/restconf/operations/zte-security:oauth_token" diff --git a/netx_api/main.py b/netx_api/main.py index 1d534c4..d6fb7ca 100644 --- a/netx_api/main.py +++ b/netx_api/main.py @@ -67,6 +67,37 @@ _SQL_FORBIDDEN_RE = re.compile( r"\b(insert|update|delete|drop|alter|create|truncate|grant|revoke|call|copy|vacuum|analyze)\b", flags=re.IGNORECASE, ) +_UME_RUNTIME_TASKS: dict[str, dict[str, Any]] = { + "token_keepalive": {"task": "token_keepalive", "status": "init", "last_run_at": None, "last_error": ""}, + "alarms_current_auto_sync": {"task": "alarms_current_auto_sync", "status": "init", "last_run_at": None, "last_error": ""}, +} +_UME_RUNTIME_LOCK = threading.Lock() + + +def _set_runtime_task(task: str, *, status: str, last_run_at: datetime | None = None, last_error: str = "") -> None: + with _UME_RUNTIME_LOCK: + item = _UME_RUNTIME_TASKS.get(task, {"task": task, "status": "init", "last_run_at": None, "last_error": ""}) + item["status"] = str(status or "unknown") + if last_run_at is not None: + item["last_run_at"] = last_run_at + item["last_error"] = str(last_error or "") + _UME_RUNTIME_TASKS[task] = item + + +def _list_runtime_tasks() -> list[dict[str, Any]]: + with _UME_RUNTIME_LOCK: + out: list[dict[str, Any]] = [] + for v in _UME_RUNTIME_TASKS.values(): + ts = _ensure_utc(v.get("last_run_at")) if isinstance(v.get("last_run_at"), datetime) else None + out.append( + { + "task": str(v.get("task") or ""), + "status": str(v.get("status") or "unknown"), + "last_run_at": ts.isoformat() if ts else None, + "last_error": str(v.get("last_error") or ""), + } + ) + return out def _ensure_utc(dt: datetime | None) -> datetime | None: @@ -264,14 +295,48 @@ def on_startup() -> None: expires_in = int(st.get("expires_in_s") or 0) if bool(st.get("has_token")) and expires_in > 0 and expires_in < renew_before_s: client.renew_token() + _set_runtime_task("token_keepalive", status="running", last_run_at=datetime.now(timezone.utc), last_error="") except Exception: - pass + _set_runtime_task("token_keepalive", status="error", last_run_at=datetime.now(timezone.utc), last_error="keepalive_failed") time.sleep(interval_s) t = threading.Thread(target=_keepalive_loop, name="ume-token-keepalive", daemon=True) t.start() except Exception: pass + try: + if bool(getattr(settings, "ume_sync_alarms_current_enabled", True)): + interval_s = int(getattr(settings, "ume_sync_alarms_current_interval_s", 300) or 300) + interval_s = max(30, min(interval_s, 86400)) + + def _alarms_current_sync_loop() -> None: + while True: + try: + db = SessionLocal() + try: + client = _ume_client() + sync_alarms_current(db, client, trigger_mode="schedule") + _set_runtime_task( + "alarms_current_auto_sync", + status="running", + last_run_at=datetime.now(timezone.utc), + last_error="", + ) + finally: + db.close() + except Exception as exc: + _set_runtime_task( + "alarms_current_auto_sync", + status="error", + last_run_at=datetime.now(timezone.utc), + last_error=str(exc)[:240], + ) + time.sleep(interval_s) + + t2 = threading.Thread(target=_alarms_current_sync_loop, name="ume-alarms-current-sync", daemon=True) + t2.start() + except Exception: + pass @app.get("/health") @@ -402,7 +467,14 @@ def ume_sync_status( items.append(item) if item["domain"] and item["domain"] not in latest_by_domain: latest_by_domain[item["domain"]] = item - return {"total": total, "page": page, "page_size": page_size, "items": items, "latest_by_domain": latest_by_domain} + return { + "total": total, + "page": page, + "page_size": page_size, + "items": items, + "latest_by_domain": latest_by_domain, + "runtime_tasks": _list_runtime_tasks(), + } @app.get("/v1/ume/inventory/ne") diff --git a/web/src/pages/UmePage.tsx b/web/src/pages/UmePage.tsx index 21c2997..2276da1 100644 --- a/web/src/pages/UmePage.tsx +++ b/web/src/pages/UmePage.tsx @@ -4,7 +4,6 @@ import { apiPost, disconnectUmeToken, fetchUmeCurrentAlarms, - fetchUmeHistoryAlarms, fetchUmeNe, fetchUmeSyncStatus, fetchUmeTokenStatus, @@ -12,10 +11,6 @@ import { } from "../services/api"; import { formatSystemTime } from "../utils/time"; -function toIsoNoMs(d: Date): string { - return d.toISOString().replace(/\.\d{3}Z$/, "Z"); -} - export function UmePage() { const queryClient = useQueryClient(); const [tokenOpError, setTokenOpError] = useState(""); @@ -33,29 +28,12 @@ export function UmePage() { const [curPage, setCurPage] = useState(1); const [curPageSize, setCurPageSize] = useState(50); - const [hisSeverity, setHisSeverity] = useState(""); - const [hisNeId, setHisNeId] = useState(""); - const [hisKeyword, setHisKeyword] = useState(""); - const [hisTimeFrom, setHisTimeFrom] = useState(""); - const [hisTimeTo, setHisTimeTo] = useState(""); - const [hisPage, setHisPage] = useState(1); - const [hisPageSize, setHisPageSize] = useState(50); - - const applyQuickRange = (hours: number) => { - const now = new Date(); - const from = new Date(now.getTime() - hours * 3600 * 1000); - setHisTimeFrom(toIsoNoMs(from)); - setHisTimeTo(toIsoNoMs(now)); - setHisPage(1); - }; - const syncMutation = useMutation({ mutationFn: async (domains: string[]) => apiPost<{ ok: boolean; jobs: unknown[] }>("/v1/ume/sync", { domains }), onSuccess: async () => { await queryClient.invalidateQueries({ queryKey: ["umeSyncStatus"] }); await queryClient.invalidateQueries({ queryKey: ["umeNE"] }); await queryClient.invalidateQueries({ queryKey: ["umeCurrentAlarms"] }); - await queryClient.invalidateQueries({ queryKey: ["umeHistoryAlarms"] }); }, }); @@ -63,6 +41,7 @@ export function UmePage() { queryKey: ["umeSyncStatus", syncPage, syncPageSize], queryFn: () => fetchUmeSyncStatus({ page: syncPage, pageSize: syncPageSize }), staleTime: 5000, + refetchInterval: 5000, }); const tokenStatusQuery = useQuery({ queryKey: ["umeTokenStatus"], @@ -125,20 +104,8 @@ export function UmePage() { }), staleTime: 5000, }); - const historyQuery = useQuery({ - queryKey: ["umeHistoryAlarms", hisSeverity, hisNeId, hisKeyword, hisTimeFrom, hisTimeTo, hisPage, hisPageSize], - queryFn: () => - fetchUmeHistoryAlarms({ - severity: hisSeverity, - neId: hisNeId, - keyword: hisKeyword, - timeFrom: hisTimeFrom, - timeTo: hisTimeTo, - page: hisPage, - pageSize: hisPageSize, - }), - staleTime: 5000, - }); + const runningTasks = (syncStatusQuery.data?.items || []).filter((x) => String(x.status || "").toLowerCase() === "running"); + const runtimeTasks = syncStatusQuery.data?.runtime_tasks || []; return ( <> @@ -183,10 +150,7 @@ export function UmePage() { - - @@ -194,6 +158,73 @@ export function UmePage() { +
+

当前任务

+
+ 0 ? "unknown" : "up"}`}> + running: {runningTasks.length} + + +
+ + + + + + + + + + + + + {runningTasks.map((x) => ( + + + + + + + + + ))} + {!syncStatusQuery.isLoading && runningTasks.length === 0 && ( + + + + )} + +
IDdomaintrigger_modestatusstarted_aterror
{x.id}{x.domain}{x.trigger_mode}{x.status}{formatSystemTime(x.started_at)}{x.error_message || "-"}
当前无运行中的任务
+

后台任务

+ + + + + + + + + + + {runtimeTasks.map((x) => ( + + + + + + + ))} + {!syncStatusQuery.isLoading && runtimeTasks.length === 0 && ( + + + + )} + +
taskstatuslast_run_atlast_error
{x.task}{x.status}{x.last_run_at ? formatSystemTime(x.last_run_at) : "-"}{x.last_error || "-"}
暂无后台任务状态
+
+

同步状态

@@ -453,102 +484,6 @@ export function UmePage() {
- -
-

历史告警

-
- setHisKeyword(e.target.value)} /> - setHisNeId(e.target.value)} /> - setHisTimeFrom(e.target.value)} /> - setHisTimeTo(e.target.value)} /> - -
-
- - - - -
- - - - - - - - - - - - {(historyQuery.data?.items || []).map((x) => ( - - - - - - - - ))} - -
time_createdseverityuser_labelobject_namecause
{x.time_created}{x.perceived_severity} - - {x.object_name}{x.native_probable_cause}
-
-
- 共 {historyQuery.data?.total || 0} 条 · 第 {hisPage}/ - {Math.max(1, Math.ceil(Math.max(0, Number(historyQuery.data?.total || 0)) / Math.max(1, hisPageSize)))} 页 -
-
- - - -
-
-
); } diff --git a/web/src/types.ts b/web/src/types.ts index 82c236f..edf20d6 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -93,6 +93,12 @@ export type UmeSyncStatusResponse = { page_size?: number; items: UmeSyncJobItem[]; latest_by_domain?: Record; + runtime_tasks?: Array<{ + task: string; + status: string; + last_run_at?: string | null; + last_error?: string; + }>; }; export type UmeNeItem = {