839 lines
42 KiB
Python
839 lines
42 KiB
Python
from __future__ import annotations
|
|
|
|
from app.core.config import settings
|
|
from app.core.db import get_db
|
|
from app.services.cluster_runtime_service import get_cluster_snapshot
|
|
from app.services.detect_service import get_detect_status
|
|
from app.services.detect_job_service import (
|
|
_build_step_bucket,
|
|
order_step_buckets,
|
|
get_active_detect_job_summary,
|
|
get_detect_capacity_plan,
|
|
get_detect_queue_health,
|
|
)
|
|
from app.services.runtime_settings_service import get_runtime_settings
|
|
from app.services.worker_control_service import detect_worker_runtime
|
|
|
|
|
|
def _empty_active_jobs_aggregate(window_minutes: int) -> dict:
|
|
return {
|
|
"window_minutes": int(window_minutes or 15),
|
|
"active_jobs_total": 0,
|
|
"queue": {
|
|
"items_total": 0,
|
|
"pending": 0,
|
|
"claimed": 0,
|
|
"running": 0,
|
|
"completed": 0,
|
|
"blacklisted": 0,
|
|
"failed": 0,
|
|
"terminal": 0,
|
|
},
|
|
"throughput": {
|
|
"processed_recent": 0,
|
|
"processed_per_minute": 0,
|
|
"completed_recent": 0,
|
|
"blacklisted_recent": 0,
|
|
"failed_recent": 0,
|
|
},
|
|
"steps": [],
|
|
"nodes": [],
|
|
"retry_total": 0,
|
|
}
|
|
|
|
|
|
def _merge_step_queues_with_runtime_activity(
|
|
base_steps: list[dict] | None,
|
|
*,
|
|
runtime_activity: dict | None = None,
|
|
window_minutes: int = 15,
|
|
limit: int = 8,
|
|
) -> list[dict]:
|
|
safe_window_minutes = max(1, int(window_minutes or 15))
|
|
normalized_limit = max(1, int(limit or 8))
|
|
step_map: dict[str, dict] = {}
|
|
|
|
for item in list(base_steps or []):
|
|
step_code = str(item.get("step_code") or "").strip()
|
|
if not step_code:
|
|
continue
|
|
bucket = _build_step_bucket(step_code)
|
|
bucket.update(
|
|
{
|
|
"items_total": int(item.get("items_total", 0) or 0),
|
|
"items_pending": int(item.get("items_pending", 0) or 0),
|
|
"items_claimed": int(item.get("items_claimed", 0) or 0),
|
|
"items_running": int(item.get("items_running", 0) or 0),
|
|
"items_completed": int(item.get("items_completed", 0) or 0),
|
|
"items_blacklisted": int(item.get("items_blacklisted", 0) or 0),
|
|
"items_failed": int(item.get("items_failed", 0) or 0),
|
|
"started_recent": int(item.get("started_recent", 0) or 0),
|
|
"processed_recent": int(item.get("processed_recent", 0) or 0),
|
|
"processed_per_minute": float(item.get("processed_per_minute", 0) or 0),
|
|
"completed_recent": int(item.get("completed_recent", 0) or 0),
|
|
"blacklisted_recent": int(item.get("blacklisted_recent", 0) or 0),
|
|
"failed_recent": int(item.get("failed_recent", 0) or 0),
|
|
}
|
|
)
|
|
step_map[step_code] = bucket
|
|
|
|
runtime_step_stats = dict((runtime_activity or {}).get("step_stats") or {})
|
|
for step_code, stats in runtime_step_stats.items():
|
|
normalized_step_code = str(step_code or "").strip()
|
|
if not normalized_step_code:
|
|
continue
|
|
bucket = step_map.setdefault(normalized_step_code, _build_step_bucket(normalized_step_code))
|
|
started_recent = int((stats or {}).get("started_recent", 0) or 0)
|
|
processed_recent = int((stats or {}).get("processed_recent", 0) or 0)
|
|
completed_recent = int((stats or {}).get("completed_recent", 0) or 0)
|
|
blacklisted_recent = int((stats or {}).get("blacklisted_recent", 0) or 0)
|
|
failed_recent = int((stats or {}).get("failed_recent", 0) or 0)
|
|
bucket["started_recent"] = max(int(bucket.get("started_recent", 0) or 0), started_recent)
|
|
bucket["processed_recent"] = max(int(bucket.get("processed_recent", 0) or 0), processed_recent)
|
|
bucket["completed_recent"] = max(int(bucket.get("completed_recent", 0) or 0), completed_recent)
|
|
bucket["blacklisted_recent"] = max(int(bucket.get("blacklisted_recent", 0) or 0), blacklisted_recent)
|
|
bucket["failed_recent"] = max(int(bucket.get("failed_recent", 0) or 0), failed_recent)
|
|
bucket["processed_per_minute"] = max(
|
|
float(bucket.get("processed_per_minute", 0) or 0),
|
|
round(processed_recent / safe_window_minutes, 2),
|
|
)
|
|
|
|
return order_step_buckets(list(step_map.values()), limit=normalized_limit)
|
|
|
|
|
|
def _align_active_jobs_aggregate_with_runtime(
|
|
aggregate: dict,
|
|
*,
|
|
runtime: dict,
|
|
queue_health: dict,
|
|
) -> dict:
|
|
normalized = dict(aggregate or {})
|
|
node_payload = dict((runtime or {}).get("node") or {})
|
|
if str(node_payload.get("region") or "").strip() != "overseas" or str(node_payload.get("role") or "").strip() != "control":
|
|
return normalized
|
|
|
|
backlog = dict(((runtime or {}).get("detect") or {}).get("backlog") or {})
|
|
snapshot_backlog = dict(queue_health.get("runtime_snapshot_backlog") or {})
|
|
|
|
def _backlog_value(key: str) -> int:
|
|
return max(int(backlog.get(key, 0) or 0), int(snapshot_backlog.get(key, 0) or 0))
|
|
|
|
pending_total = _backlog_value("pending_total")
|
|
claimed_total = _backlog_value("claimed_total")
|
|
running_total = _backlog_value("running_total")
|
|
completed_total = _backlog_value("completed_total")
|
|
blacklisted_total = _backlog_value("blacklisted_total")
|
|
failed_total = _backlog_value("failed_total")
|
|
|
|
queue = dict(queue_health.get("queue") or {})
|
|
throughput = dict(queue_health.get("throughput") or {})
|
|
terminal_total = max(
|
|
completed_total + blacklisted_total + failed_total,
|
|
int(queue.get("completed", 0) or 0) + int(queue.get("blacklisted", 0) or 0) + int(queue.get("failed", 0) or 0),
|
|
)
|
|
items_total = pending_total + claimed_total + running_total + terminal_total
|
|
has_runtime_work = items_total > 0 or bool(queue_health.get("has_active_job"))
|
|
|
|
normalized["active_jobs_total"] = max(
|
|
int(normalized.get("active_jobs_total", 0) or 0),
|
|
1 if has_runtime_work else 0,
|
|
)
|
|
normalized["queue"] = {
|
|
"items_total": items_total,
|
|
"pending": pending_total,
|
|
"claimed": claimed_total,
|
|
"running": running_total,
|
|
"completed": max(completed_total, int(queue.get("completed", 0) or 0)),
|
|
"blacklisted": max(blacklisted_total, int(queue.get("blacklisted", 0) or 0)),
|
|
"failed": max(failed_total, int(queue.get("failed", 0) or 0)),
|
|
"terminal": terminal_total,
|
|
}
|
|
normalized["throughput"] = {
|
|
"processed_recent": int(throughput.get("processed_recent", 0) or 0),
|
|
"processed_per_minute": float(throughput.get("processed_per_minute", 0) or 0),
|
|
"completed_recent": int(throughput.get("completed_recent", 0) or 0),
|
|
"blacklisted_recent": int(throughput.get("blacklisted_recent", 0) or 0),
|
|
"failed_recent": int(throughput.get("failed_recent", 0) or 0),
|
|
}
|
|
normalized["steps"] = _merge_step_queues_with_runtime_activity(
|
|
list(queue_health.get("steps") or []),
|
|
runtime_activity=dict(queue_health.get("runtime_activity") or {}),
|
|
window_minutes=int(queue_health.get("window_minutes", normalized.get("window_minutes", 15)) or 15),
|
|
limit=8,
|
|
)
|
|
normalized["nodes"] = list(queue_health.get("nodes") or [])
|
|
return normalized
|
|
|
|
|
|
def _build_dashboard_runtime_summary(*, queue_health: dict) -> dict:
|
|
runtime_settings = get_runtime_settings()
|
|
worker_runtime = detect_worker_runtime()
|
|
worker_expected_on_this_node = not (
|
|
str(settings.node_region or "").strip() == "overseas"
|
|
and str(settings.node_role or "").strip() == "control"
|
|
)
|
|
return {
|
|
"node": {
|
|
"region": settings.node_region,
|
|
"role": settings.node_role,
|
|
},
|
|
"worker": {
|
|
"running": bool(worker_runtime.get("running", False)),
|
|
"mode": worker_runtime.get("mode", runtime_settings.get("worker_mode", "windows-local")),
|
|
"expected_on_this_node": worker_expected_on_this_node,
|
|
},
|
|
"cluster": get_cluster_snapshot(),
|
|
"detect": {
|
|
"backlog": dict((queue_health or {}).get("runtime_snapshot_backlog") or {}),
|
|
},
|
|
}
|
|
|
|
|
|
def _resolve_server_code(node_code: str | None) -> str:
|
|
normalized_node_code = str(node_code or "").strip()
|
|
if not normalized_node_code:
|
|
return ""
|
|
parent_node_code, separator, suffix = normalized_node_code.rpartition("-")
|
|
if separator and parent_node_code and suffix.isalpha() and len(suffix) <= 3:
|
|
if any(char.isdigit() for char in parent_node_code):
|
|
return parent_node_code
|
|
return normalized_node_code
|
|
|
|
|
|
def _count_active_execution_servers(rows: list[dict] | None) -> int:
|
|
active_servers: set[str] = set()
|
|
for item in list(rows or []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
server_code = _resolve_server_code(item.get("node_code"))
|
|
if not server_code:
|
|
continue
|
|
if (
|
|
int(item.get("items_running", 0) or 0) > 0
|
|
or int(item.get("items_claimed", 0) or 0) > 0
|
|
or int(item.get("processed_recent", 0) or 0) > 0
|
|
):
|
|
active_servers.add(server_code)
|
|
return len(active_servers)
|
|
|
|
|
|
def _fetch_active_jobs_aggregate(window_minutes: int = 15) -> dict:
|
|
safe_window_minutes = max(5, min(int(window_minutes or 15), 120))
|
|
payload = _empty_active_jobs_aggregate(safe_window_minutes)
|
|
with get_db() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id
|
|
FROM detect_jobs
|
|
WHERE status IN ('pending', 'running', 'partial_failed')
|
|
ORDER BY
|
|
CASE
|
|
WHEN status = 'running' THEN 0
|
|
WHEN status = 'pending' THEN 1
|
|
WHEN status = 'partial_failed' THEN 2
|
|
ELSE 3
|
|
END ASC,
|
|
COALESCE(started_at, created_at) DESC,
|
|
id DESC
|
|
"""
|
|
)
|
|
job_ids = [int(row[0]) for row in cur.fetchall() if row and row[0] is not None]
|
|
if not job_ids:
|
|
return payload
|
|
|
|
payload["active_jobs_total"] = len(job_ids)
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
COUNT(*) AS items_total,
|
|
COUNT(*) FILTER (WHERE status = 'pending') AS items_pending,
|
|
COUNT(*) FILTER (WHERE status = 'claimed') AS items_claimed,
|
|
COUNT(*) FILTER (WHERE status = 'running') AS items_running,
|
|
COUNT(*) FILTER (WHERE status = 'completed') AS items_completed,
|
|
COUNT(*) FILTER (WHERE status = 'blacklisted') AS items_blacklisted,
|
|
COUNT(*) FILTER (WHERE status = 'failed') AS items_failed
|
|
FROM detect_job_items
|
|
WHERE job_id = ANY(%s)
|
|
""",
|
|
(job_ids,),
|
|
)
|
|
queue_row = cur.fetchone() or (0, 0, 0, 0, 0, 0, 0)
|
|
payload["queue"] = {
|
|
"items_total": int(queue_row[0] or 0),
|
|
"pending": int(queue_row[1] or 0),
|
|
"claimed": int(queue_row[2] or 0),
|
|
"running": int(queue_row[3] or 0),
|
|
"completed": int(queue_row[4] or 0),
|
|
"blacklisted": int(queue_row[5] or 0),
|
|
"failed": int(queue_row[6] or 0),
|
|
"terminal": int(queue_row[4] or 0) + int(queue_row[5] or 0) + int(queue_row[6] or 0),
|
|
}
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
COUNT(*) AS processed_recent,
|
|
COUNT(*) FILTER (WHERE event_type = 'domain_completed') AS completed_recent,
|
|
COUNT(*) FILTER (WHERE event_type = 'domain_blacklisted') AS blacklisted_recent,
|
|
COUNT(*) FILTER (WHERE event_type = 'domain_failed') AS failed_recent
|
|
FROM detect_run_events
|
|
WHERE job_id = ANY(%s)
|
|
AND event_type IN ('domain_completed', 'domain_blacklisted', 'domain_failed')
|
|
AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
|
|
""",
|
|
(job_ids, safe_window_minutes),
|
|
)
|
|
throughput_row = cur.fetchone() or (0, 0, 0, 0)
|
|
processed_recent = int(throughput_row[0] or 0)
|
|
payload["throughput"] = {
|
|
"processed_recent": processed_recent,
|
|
"processed_per_minute": round(processed_recent / safe_window_minutes, 2),
|
|
"completed_recent": int(throughput_row[1] or 0),
|
|
"blacklisted_recent": int(throughput_row[2] or 0),
|
|
"failed_recent": int(throughput_row[3] or 0),
|
|
}
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM detect_job_items
|
|
WHERE job_id = ANY(%s)
|
|
AND attempt_count > 1
|
|
""",
|
|
(job_ids,),
|
|
)
|
|
payload["retry_total"] = int((cur.fetchone() or [0])[0] or 0)
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
COALESCE(NULLIF(step_code, ''), 'domain_pipeline') AS step_code,
|
|
COUNT(*) AS items_total,
|
|
COUNT(*) FILTER (WHERE status = 'pending') AS items_pending,
|
|
COUNT(*) FILTER (WHERE status = 'claimed') AS items_claimed,
|
|
COUNT(*) FILTER (WHERE status = 'running') AS items_running,
|
|
COUNT(*) FILTER (WHERE status = 'completed') AS items_completed,
|
|
COUNT(*) FILTER (WHERE status = 'blacklisted') AS items_blacklisted,
|
|
COUNT(*) FILTER (WHERE status = 'failed') AS items_failed,
|
|
COUNT(*) FILTER (
|
|
WHERE status IN ('completed', 'blacklisted', 'failed')
|
|
AND finished_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
|
|
) AS processed_recent
|
|
FROM detect_job_items
|
|
WHERE job_id = ANY(%s)
|
|
GROUP BY COALESCE(NULLIF(step_code, ''), 'domain_pipeline')
|
|
ORDER BY
|
|
COUNT(*) FILTER (WHERE status = 'pending') DESC,
|
|
COUNT(*) FILTER (WHERE status = 'running') DESC,
|
|
COUNT(*) DESC,
|
|
COALESCE(NULLIF(step_code, ''), 'domain_pipeline') ASC
|
|
LIMIT 8
|
|
""",
|
|
(safe_window_minutes, job_ids),
|
|
)
|
|
steps: list[dict] = []
|
|
for row in cur.fetchall():
|
|
step_code = str(row[0] or "domain_pipeline")
|
|
bucket = _build_step_bucket(step_code)
|
|
bucket.update(
|
|
{
|
|
"items_total": int(row[1] or 0),
|
|
"items_pending": int(row[2] or 0),
|
|
"items_claimed": int(row[3] or 0),
|
|
"items_running": int(row[4] or 0),
|
|
"items_completed": int(row[5] or 0),
|
|
"items_blacklisted": int(row[6] or 0),
|
|
"items_failed": int(row[7] or 0),
|
|
"processed_recent": int(row[8] or 0),
|
|
"processed_per_minute": round(int(row[8] or 0) / safe_window_minutes, 2),
|
|
}
|
|
)
|
|
steps.append(bucket)
|
|
payload["steps"] = order_step_buckets(steps, limit=8)
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
COALESCE(NULLIF(claimed_by, ''), 'unassigned') AS node_code,
|
|
COUNT(*) FILTER (WHERE status = 'claimed') AS items_claimed,
|
|
COUNT(*) FILTER (WHERE status = 'running') AS items_running,
|
|
COUNT(*) FILTER (WHERE status = 'completed') AS items_completed_total
|
|
FROM detect_job_items
|
|
WHERE job_id = ANY(%s)
|
|
GROUP BY COALESCE(NULLIF(claimed_by, ''), 'unassigned')
|
|
""",
|
|
(job_ids,),
|
|
)
|
|
node_map = {
|
|
str(row[0] or "unassigned"): {
|
|
"node_code": str(row[0] or "unassigned"),
|
|
"items_running": int(row[2] or 0),
|
|
"items_claimed": int(row[1] or 0),
|
|
"items_completed": int(row[3] or 0),
|
|
"processed_recent": 0,
|
|
"processed_per_minute": 0,
|
|
"completed_recent": 0,
|
|
"failed_recent": 0,
|
|
"blacklisted_recent": 0,
|
|
}
|
|
for row in cur.fetchall()
|
|
}
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
COALESCE(NULLIF(node_code, ''), 'unassigned') AS node_code,
|
|
COUNT(*) AS processed_recent,
|
|
COUNT(*) FILTER (WHERE event_type = 'domain_completed') AS completed_recent,
|
|
COUNT(*) FILTER (WHERE event_type = 'domain_blacklisted') AS blacklisted_recent,
|
|
COUNT(*) FILTER (WHERE event_type = 'domain_failed') AS failed_recent
|
|
FROM detect_run_events
|
|
WHERE job_id = ANY(%s)
|
|
AND event_type IN ('domain_completed', 'domain_blacklisted', 'domain_failed')
|
|
AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
|
|
GROUP BY COALESCE(NULLIF(node_code, ''), 'unassigned')
|
|
""",
|
|
(job_ids, safe_window_minutes),
|
|
)
|
|
for row in cur.fetchall():
|
|
node_code = str(row[0] or "unassigned")
|
|
bucket = node_map.setdefault(
|
|
node_code,
|
|
{
|
|
"node_code": node_code,
|
|
"items_running": 0,
|
|
"items_claimed": 0,
|
|
"items_completed": 0,
|
|
"processed_recent": 0,
|
|
"processed_per_minute": 0,
|
|
"completed_recent": 0,
|
|
"failed_recent": 0,
|
|
"blacklisted_recent": 0,
|
|
},
|
|
)
|
|
processed_recent = int(row[1] or 0)
|
|
bucket["processed_recent"] = processed_recent
|
|
bucket["processed_per_minute"] = round(processed_recent / safe_window_minutes, 2)
|
|
bucket["completed_recent"] = int(row[2] or 0)
|
|
bucket["blacklisted_recent"] = int(row[3] or 0)
|
|
bucket["failed_recent"] = int(row[4] or 0)
|
|
|
|
payload["nodes"] = sorted(
|
|
node_map.values(),
|
|
key=lambda item: (
|
|
-int(item.get("items_running", 0) or 0),
|
|
-int(item.get("processed_recent", 0) or 0),
|
|
-int(item.get("items_claimed", 0) or 0),
|
|
str(item.get("node_code") or ""),
|
|
),
|
|
)[:8]
|
|
return payload
|
|
|
|
|
|
def fetch_overview() -> dict:
|
|
window_minutes = 15
|
|
queries = {
|
|
"domains_total": "select count(*) from domains",
|
|
"pending_total": "select count(*) from domains where detect_status = 0",
|
|
"completed_total": "select count(*) from domains where detect_status = 1",
|
|
"running_total": "select count(*) from domains where detect_status = 2",
|
|
"blacklist_total": "select count(*) from domains where detect_status = 3",
|
|
"failed_total": "select count(*) from domains where detect_status = 4",
|
|
"registerable_total": "select count(*) from domains where detect_status = 1 and register_status = 2",
|
|
"purchasable_total": "select count(*) from domains where detect_status = 1 and register_status = 2 and coalesce(use_status, 0) = 0",
|
|
"sensitive_words_total": "select count(*) from sensitive_words",
|
|
}
|
|
result: dict[str, int | str] = {}
|
|
with get_db() as conn:
|
|
with conn.cursor() as cur:
|
|
for key, query in queries.items():
|
|
try:
|
|
cur.execute(query)
|
|
result[key] = cur.fetchone()[0]
|
|
except Exception:
|
|
result[key] = 0
|
|
|
|
active_jobs_aggregate = _fetch_active_jobs_aggregate(window_minutes=window_minutes)
|
|
active_job = get_active_detect_job_summary(event_limit=20) or {}
|
|
aggregate_queue = active_jobs_aggregate.get("queue") or {}
|
|
active_job_display_claimed = int(active_job.get("display_items_claimed", active_job.get("items_claimed", 0)) or 0)
|
|
active_job_display_running = int(
|
|
active_job.get("display_items_running", active_job.get("display_active_threads", active_job.get("items_running", 0)))
|
|
or 0
|
|
)
|
|
active_job_display_active_threads = int(
|
|
active_job.get("display_active_threads", active_job.get("display_items_running", active_job.get("items_running", 0)))
|
|
or 0
|
|
)
|
|
active_job_display_max_threads = int(active_job.get("display_max_threads", 0) or 0)
|
|
active_job_distributed_node_stats = [
|
|
dict(item)
|
|
for item in list(active_job.get("distributed_node_stats") or [])
|
|
if isinstance(item, dict)
|
|
]
|
|
|
|
queue_health = get_detect_queue_health(window_minutes=window_minutes)
|
|
runtime = _build_dashboard_runtime_summary(queue_health=queue_health)
|
|
try:
|
|
detect_status = get_detect_status()
|
|
except Exception:
|
|
detect_status = {}
|
|
cluster_summary = ((runtime.get("cluster") or {}).get("summary") or {})
|
|
online_worker_nodes = int(cluster_summary.get("online_worker_nodes", 0) or 0)
|
|
dedicated_online_worker_nodes = int(cluster_summary.get("dedicated_online_worker_nodes", 0) or 0)
|
|
result["worker_status"] = "online" if runtime["worker"]["running"] else "offline"
|
|
result["local_worker_status"] = "not-applicable" if not runtime["worker"].get("expected_on_this_node", True) else result["worker_status"]
|
|
result["cluster_worker_status"] = "online" if online_worker_nodes > 0 else "offline"
|
|
result["cluster_online_worker_nodes"] = online_worker_nodes
|
|
result["cluster_dedicated_online_worker_nodes"] = dedicated_online_worker_nodes
|
|
result["cluster_online_control_nodes"] = int(cluster_summary.get("online_control_nodes", 0) or 0)
|
|
result["api_status"] = "online"
|
|
result["worker_mode"] = runtime["worker"]["mode"]
|
|
result["node_region"] = runtime["node"]["region"]
|
|
result["node_role"] = runtime["node"]["role"]
|
|
|
|
active_jobs_aggregate = _align_active_jobs_aggregate_with_runtime(
|
|
active_jobs_aggregate,
|
|
runtime=runtime,
|
|
queue_health=queue_health,
|
|
)
|
|
runtime_snapshot_backlog = dict(queue_health.get("runtime_snapshot_backlog") or {})
|
|
aggregate_queue = active_jobs_aggregate.get("queue") or {}
|
|
aggregate_queue_health = {
|
|
"has_active_job": bool(int(active_jobs_aggregate.get("active_jobs_total", 0) or 0) > 0),
|
|
"queue": aggregate_queue,
|
|
"throughput": active_jobs_aggregate.get("throughput") or {},
|
|
}
|
|
selected_queue_health = aggregate_queue_health if aggregate_queue_health["has_active_job"] else queue_health
|
|
if float((queue_health.get("throughput") or {}).get("processed_per_minute", 0) or 0) > float(
|
|
(selected_queue_health.get("throughput") or {}).get("processed_per_minute", 0) or 0
|
|
):
|
|
selected_queue_health = queue_health
|
|
capacity_plan = get_detect_capacity_plan(
|
|
queue_health=selected_queue_health,
|
|
online_worker_nodes=online_worker_nodes,
|
|
)
|
|
|
|
retry_total = int(active_jobs_aggregate.get("retry_total", 0) or 0)
|
|
step_queue: list[dict] = []
|
|
node_throughput: list[dict] = []
|
|
bottleneck_step: dict | None = None
|
|
active_job_summary: dict | None = None
|
|
|
|
if queue_health.get("has_active_job"):
|
|
job_payload = queue_health.get("job") or {}
|
|
queue_payload = queue_health.get("queue") or {}
|
|
throughput_payload = queue_health.get("throughput") or {}
|
|
runtime_job_code = str(job_payload.get("runtime_job_code") or "").strip()
|
|
display_job_code = runtime_job_code or str(job_payload.get("job_code") or "")
|
|
active_job_matches_display_job = display_job_code == str(active_job.get("job_code") or "").strip()
|
|
queue_nodes = [
|
|
dict(item)
|
|
for item in list(queue_health.get("nodes") or [])
|
|
if isinstance(item, dict)
|
|
]
|
|
summary_display_running = max(
|
|
int(queue_payload.get("display_running", queue_payload.get("running", 0)) or 0),
|
|
active_job_display_running if active_job_matches_display_job else 0,
|
|
)
|
|
summary_display_active_threads = max(
|
|
int(queue_payload.get("display_running", queue_payload.get("running", 0)) or 0),
|
|
active_job_display_active_threads if active_job_matches_display_job else 0,
|
|
)
|
|
summary_display_max_threads = max(
|
|
sum(int(item.get("max_threads", 0) or 0) for item in queue_nodes),
|
|
active_job_display_max_threads if active_job_matches_display_job else 0,
|
|
)
|
|
summary_distributed_node_stats = active_job_distributed_node_stats if active_job_matches_display_job else queue_nodes
|
|
active_job_summary = {
|
|
"job_id": int(job_payload.get("job_id", 0) or 0),
|
|
"job_code": display_job_code,
|
|
"db_job_code": str(job_payload.get("job_code") or ""),
|
|
"runtime_job_code": runtime_job_code,
|
|
"status": str(job_payload.get("status") or ""),
|
|
"progress_percent": float(job_payload.get("progress_percent", 0) or 0),
|
|
"items_total": int(queue_payload.get("items_total", 0) or 0),
|
|
"items_pending": int(queue_payload.get("pending", 0) or 0),
|
|
"items_claimed": max(
|
|
int(queue_payload.get("display_claimed", queue_payload.get("claimed", 0)) or 0),
|
|
active_job_display_claimed if active_job_matches_display_job else 0,
|
|
),
|
|
"items_running": max(
|
|
int(queue_payload.get("running", 0) or 0),
|
|
int(active_job.get("items_running", 0) or 0) if active_job_matches_display_job else 0,
|
|
),
|
|
"items_display_running": summary_display_running,
|
|
"display_items_running": summary_display_running,
|
|
"display_active_threads": summary_display_active_threads,
|
|
"display_max_threads": summary_display_max_threads,
|
|
"distributed_node_stats": summary_distributed_node_stats,
|
|
"items_completed": int(queue_payload.get("completed", 0) or 0),
|
|
"items_blacklisted": int(queue_payload.get("blacklisted", 0) or 0),
|
|
"items_failed": int(queue_payload.get("failed", 0) or 0),
|
|
"processed_per_minute": float(throughput_payload.get("processed_per_minute", 0) or 0),
|
|
"processed_recent": int(throughput_payload.get("processed_recent", 0) or 0),
|
|
"completed_recent": int(throughput_payload.get("completed_recent", 0) or 0),
|
|
"failed_recent": int(throughput_payload.get("failed_recent", 0) or 0),
|
|
"blacklisted_recent": int(throughput_payload.get("blacklisted_recent", 0) or 0),
|
|
"active_jobs_total": int(active_jobs_aggregate.get("active_jobs_total", 0) or 0),
|
|
}
|
|
elif active_job:
|
|
active_job_summary = {
|
|
"job_id": int(active_job.get("job_id", 0) or 0),
|
|
"job_code": str(active_job.get("job_code") or active_job.get("runtime_job_code") or ""),
|
|
"db_job_code": str(active_job.get("job_code") or ""),
|
|
"runtime_job_code": str(active_job.get("runtime_job_code") or ""),
|
|
"status": str(active_job.get("status") or ""),
|
|
"progress_percent": float(active_job.get("progress_percent", 0) or 0),
|
|
"items_total": int(active_job.get("items_total", 0) or 0),
|
|
"items_pending": int(active_job.get("items_pending", 0) or 0),
|
|
"items_claimed": active_job_display_claimed,
|
|
"items_running": int(active_job.get("items_running", 0) or 0),
|
|
"items_display_running": active_job_display_running,
|
|
"display_items_running": active_job_display_running,
|
|
"display_active_threads": active_job_display_active_threads,
|
|
"display_max_threads": active_job_display_max_threads,
|
|
"distributed_node_stats": active_job_distributed_node_stats,
|
|
"items_completed": int(active_job.get("items_completed", 0) or 0),
|
|
"items_blacklisted": int(active_job.get("items_blacklisted", 0) or 0),
|
|
"items_failed": int(active_job.get("items_failed", 0) or 0),
|
|
"processed_per_minute": float((active_jobs_aggregate.get("throughput") or {}).get("processed_per_minute", 0) or 0),
|
|
"processed_recent": int((active_jobs_aggregate.get("throughput") or {}).get("processed_recent", 0) or 0),
|
|
"completed_recent": int((active_jobs_aggregate.get("throughput") or {}).get("completed_recent", 0) or 0),
|
|
"failed_recent": int((active_jobs_aggregate.get("throughput") or {}).get("failed_recent", 0) or 0),
|
|
"blacklisted_recent": int((active_jobs_aggregate.get("throughput") or {}).get("blacklisted_recent", 0) or 0),
|
|
"active_jobs_total": int(active_jobs_aggregate.get("active_jobs_total", 0) or 0),
|
|
}
|
|
|
|
queue_pending_total = 0
|
|
queue_claimed_total = 0
|
|
queue_running_total = 0
|
|
queue_display_running_total = 0
|
|
queue_completed_total = 0
|
|
queue_blacklist_total = 0
|
|
queue_failed_total = 0
|
|
backlog_payload = ((runtime.get("detect") or {}).get("backlog") or {})
|
|
backlog_pending_total = max(
|
|
int(backlog_payload.get("pending_total", 0) or 0),
|
|
int(runtime_snapshot_backlog.get("pending_total", 0) or 0),
|
|
)
|
|
backlog_claimed_total = max(
|
|
int(backlog_payload.get("claimed_total", 0) or 0),
|
|
int(runtime_snapshot_backlog.get("claimed_total", 0) or 0),
|
|
)
|
|
backlog_running_total = max(
|
|
int(backlog_payload.get("running_total", 0) or 0),
|
|
int(runtime_snapshot_backlog.get("running_total", 0) or 0),
|
|
)
|
|
backlog_register_pending_total = max(
|
|
int(backlog_payload.get("register_pending", 0) or 0),
|
|
int(runtime_snapshot_backlog.get("register_pending", 0) or 0),
|
|
)
|
|
backlog_downstream_pending_total = max(
|
|
int(backlog_payload.get("downstream_pending", 0) or 0),
|
|
int(runtime_snapshot_backlog.get("downstream_pending", 0) or 0),
|
|
)
|
|
if queue_health.get("has_active_job"):
|
|
queue_payload = queue_health.get("queue") or {}
|
|
queue_pending_total = int(queue_payload.get("pending", 0) or 0)
|
|
queue_claimed_total = max(
|
|
int(queue_payload.get("display_claimed", queue_payload.get("claimed", 0)) or 0),
|
|
active_job_display_claimed,
|
|
)
|
|
queue_running_total = max(int(queue_payload.get("running", 0) or 0), int(active_job.get("items_running", 0) or 0))
|
|
queue_display_running_total = max(
|
|
int(queue_payload.get("display_running", queue_payload.get("running", 0)) or 0),
|
|
active_job_display_running,
|
|
)
|
|
queue_completed_total = int(queue_payload.get("completed", 0) or 0)
|
|
queue_blacklist_total = int(queue_payload.get("blacklisted", 0) or 0)
|
|
queue_failed_total = int(queue_payload.get("failed", 0) or 0)
|
|
elif int(active_jobs_aggregate.get("active_jobs_total", 0) or 0) > 0:
|
|
queue_pending_total = int(aggregate_queue.get("pending", 0) or 0)
|
|
queue_claimed_total = int(aggregate_queue.get("claimed", 0) or 0)
|
|
queue_running_total = int(aggregate_queue.get("running", 0) or 0)
|
|
queue_display_running_total = queue_running_total
|
|
queue_completed_total = int(aggregate_queue.get("completed", 0) or 0)
|
|
queue_blacklist_total = int(aggregate_queue.get("blacklisted", 0) or 0)
|
|
queue_failed_total = int(aggregate_queue.get("failed", 0) or 0)
|
|
elif active_job:
|
|
queue_pending_total = int(active_job.get("items_pending", 0) or 0)
|
|
queue_claimed_total = int(active_job.get("items_claimed", 0) or 0)
|
|
queue_running_total = int(active_job.get("items_running", 0) or 0)
|
|
queue_display_running_total = int(
|
|
active_job.get("display_items_running", active_job.get("items_running", 0)) or 0
|
|
)
|
|
queue_completed_total = int(active_job.get("items_completed", 0) or 0)
|
|
queue_blacklist_total = int(active_job.get("items_blacklisted", 0) or 0)
|
|
queue_failed_total = int(active_job.get("items_failed", 0) or 0)
|
|
|
|
aggregate_step_queue = [
|
|
{
|
|
"step_code": str(item.get("step_code") or ""),
|
|
"step_name": str(item.get("step_name") or ""),
|
|
"items_pending": int(item.get("items_pending", 0) or 0),
|
|
"items_running": int(item.get("items_running", 0) or 0),
|
|
"items_claimed": int(item.get("items_claimed", 0) or 0),
|
|
"items_completed": int(item.get("items_completed", 0) or 0),
|
|
"items_blacklisted": int(item.get("items_blacklisted", 0) or 0),
|
|
"items_failed": int(item.get("items_failed", 0) or 0),
|
|
"started_recent": int(item.get("started_recent", 0) or 0),
|
|
"processed_per_minute": float(item.get("processed_per_minute", 0) or 0),
|
|
"processed_recent": int(item.get("processed_recent", 0) or 0),
|
|
"completed_recent": int(item.get("completed_recent", 0) or 0),
|
|
"blacklisted_recent": int(item.get("blacklisted_recent", 0) or 0),
|
|
"failed_recent": int(item.get("failed_recent", 0) or 0),
|
|
}
|
|
for item in list(active_jobs_aggregate.get("steps") or [])[:8]
|
|
]
|
|
aggregate_node_throughput_all = [
|
|
{
|
|
"node_code": str(item.get("node_code") or ""),
|
|
"items_pending": int(item.get("items_pending", 0) or 0),
|
|
"items_running": int(item.get("items_running", 0) or 0),
|
|
"display_running": int(item.get("display_running", item.get("items_running", 0)) or 0),
|
|
"items_claimed": int(item.get("items_claimed", 0) or 0),
|
|
"current_load": int(item.get("current_load", item.get("display_running", 0)) or 0),
|
|
"active_threads": int(item.get("active_threads", 0) or 0),
|
|
"max_threads": int(item.get("max_threads", 0) or 0),
|
|
"processed_recent": int(item.get("processed_recent", 0) or 0),
|
|
"processed_per_minute": float(item.get("processed_per_minute", 0) or 0),
|
|
"completed_recent": int(item.get("completed_recent", 0) or 0),
|
|
"failed_recent": int(item.get("failed_recent", 0) or 0),
|
|
"blacklisted_recent": int(item.get("blacklisted_recent", 0) or 0),
|
|
}
|
|
for item in list(active_jobs_aggregate.get("nodes") or [])
|
|
]
|
|
aggregate_node_throughput = aggregate_node_throughput_all[:8]
|
|
queue_step_queue = [
|
|
{
|
|
"step_code": str(item.get("step_code") or ""),
|
|
"step_name": str(item.get("step_name") or ""),
|
|
"items_pending": int(item.get("items_pending", 0) or 0),
|
|
"items_running": int(item.get("items_running", 0) or 0),
|
|
"items_claimed": int(item.get("items_claimed", 0) or 0),
|
|
"items_completed": int(item.get("items_completed", 0) or 0),
|
|
"items_blacklisted": int(item.get("items_blacklisted", 0) or 0),
|
|
"items_failed": int(item.get("items_failed", 0) or 0),
|
|
"started_recent": int(item.get("started_recent", 0) or 0),
|
|
"processed_per_minute": float(item.get("processed_per_minute", 0) or 0),
|
|
"processed_recent": int(item.get("processed_recent", 0) or 0),
|
|
"completed_recent": int(item.get("completed_recent", 0) or 0),
|
|
"blacklisted_recent": int(item.get("blacklisted_recent", 0) or 0),
|
|
"failed_recent": int(item.get("failed_recent", 0) or 0),
|
|
}
|
|
for item in _merge_step_queues_with_runtime_activity(
|
|
list(queue_health.get("steps") or []),
|
|
runtime_activity=dict(queue_health.get("runtime_activity") or {}),
|
|
window_minutes=window_minutes,
|
|
limit=8,
|
|
)
|
|
]
|
|
queue_node_throughput_all = [
|
|
{
|
|
"node_code": str(item.get("node_code") or ""),
|
|
"items_pending": int(item.get("items_pending", 0) or 0),
|
|
"items_running": int(item.get("items_running", 0) or 0),
|
|
"display_running": int(item.get("display_running", item.get("items_running", 0)) or 0),
|
|
"items_claimed": int(item.get("items_claimed", 0) or 0),
|
|
"current_load": int(item.get("current_load", item.get("display_running", 0)) or 0),
|
|
"active_threads": int(item.get("active_threads", 0) or 0),
|
|
"max_threads": int(item.get("max_threads", 0) or 0),
|
|
"processed_recent": int(item.get("processed_recent", 0) or 0),
|
|
"processed_per_minute": float(item.get("processed_per_minute", 0) or 0),
|
|
"completed_recent": int(item.get("completed_recent", 0) or 0),
|
|
"failed_recent": int(item.get("failed_recent", 0) or 0),
|
|
"blacklisted_recent": int(item.get("blacklisted_recent", 0) or 0),
|
|
}
|
|
for item in list(queue_health.get("nodes") or [])
|
|
]
|
|
queue_node_throughput = queue_node_throughput_all[:8]
|
|
step_queue = aggregate_step_queue
|
|
node_throughput = aggregate_node_throughput
|
|
active_execution_node_source = aggregate_node_throughput_all
|
|
aggregate_ppm = float((active_jobs_aggregate.get("throughput") or {}).get("processed_per_minute", 0) or 0)
|
|
queue_ppm = float((queue_health.get("throughput") or {}).get("processed_per_minute", 0) or 0)
|
|
if queue_health.get("has_active_job") or queue_ppm > aggregate_ppm:
|
|
step_queue = queue_step_queue
|
|
node_throughput = queue_node_throughput
|
|
active_execution_node_source = queue_node_throughput_all
|
|
if step_queue:
|
|
bottleneck_step = max(
|
|
step_queue,
|
|
key=lambda item: (
|
|
int(item.get("items_pending", 0) or 0),
|
|
int(item.get("items_running", 0) or 0),
|
|
-float(item.get("processed_per_minute", 0) or 0),
|
|
),
|
|
)
|
|
|
|
result["active_job"] = active_job_summary or {}
|
|
result["queue_health"] = queue_health
|
|
result["active_jobs_aggregate"] = active_jobs_aggregate
|
|
result["capacity_plan"] = capacity_plan
|
|
result["step_queue"] = step_queue
|
|
result["node_throughput"] = node_throughput
|
|
result["retry_total"] = retry_total
|
|
result["bottleneck_step"] = bottleneck_step or {}
|
|
aggregate_throughput = active_jobs_aggregate.get("throughput") or {}
|
|
queue_throughput = queue_health.get("throughput") or {}
|
|
ops_processed_per_minute = float(aggregate_throughput.get("processed_per_minute", 0) or 0)
|
|
ops_processed_recent = int(aggregate_throughput.get("processed_recent", 0) or 0)
|
|
ops_completed_recent = int(aggregate_throughput.get("completed_recent", 0) or 0)
|
|
ops_failed_recent = int(aggregate_throughput.get("failed_recent", 0) or 0)
|
|
ops_blacklisted_recent = int(aggregate_throughput.get("blacklisted_recent", 0) or 0)
|
|
if queue_health.get("has_active_job") or float(queue_throughput.get("processed_per_minute", 0) or 0) > ops_processed_per_minute:
|
|
ops_processed_per_minute = float(queue_throughput.get("processed_per_minute", 0) or 0)
|
|
ops_processed_recent = int(queue_throughput.get("processed_recent", 0) or 0)
|
|
ops_completed_recent = int(queue_throughput.get("completed_recent", 0) or 0)
|
|
ops_failed_recent = int(queue_throughput.get("failed_recent", 0) or 0)
|
|
ops_blacklisted_recent = int(queue_throughput.get("blacklisted_recent", 0) or 0)
|
|
result["ops_summary"] = {
|
|
"active_jobs_total": int(active_jobs_aggregate.get("active_jobs_total", 0) or 0),
|
|
"processed_per_minute": ops_processed_per_minute,
|
|
"processed_recent": ops_processed_recent,
|
|
"completed_recent": ops_completed_recent,
|
|
"failed_recent": ops_failed_recent,
|
|
"blacklisted_recent": ops_blacklisted_recent,
|
|
"estimated_hours_remaining": float(capacity_plan.get("estimated_hours_remaining", 0) or 0),
|
|
"remaining_items": int(capacity_plan.get("remaining_items", 0) or 0),
|
|
"recommended_additional_workers": int(capacity_plan.get("recommended_additional_workers", 0) or 0),
|
|
"online_worker_nodes": online_worker_nodes,
|
|
"dedicated_online_worker_nodes": dedicated_online_worker_nodes,
|
|
"active_execution_nodes": _count_active_execution_servers(active_execution_node_source),
|
|
}
|
|
result["processed_per_minute"] = ops_processed_per_minute
|
|
result["processed_recent"] = ops_processed_recent
|
|
result["completed_recent"] = ops_completed_recent
|
|
result["failed_recent"] = ops_failed_recent
|
|
result["blacklisted_recent"] = ops_blacklisted_recent
|
|
result["active_execution_nodes"] = int(result["ops_summary"]["active_execution_nodes"] or 0)
|
|
result["queue_pending_total"] = queue_pending_total
|
|
result["queue_claimed_total"] = queue_claimed_total
|
|
result["queue_running_total"] = queue_running_total
|
|
result["queue_display_running_total"] = max(queue_display_running_total, queue_running_total)
|
|
result["queue_display_max_threads"] = max(
|
|
int((active_job_summary or {}).get("display_max_threads", 0) or 0),
|
|
sum(int(item.get("max_threads", 0) or 0) for item in list(queue_health.get("nodes") or []) if isinstance(item, dict)),
|
|
)
|
|
result["queue_completed_total"] = queue_completed_total
|
|
result["queue_blacklist_total"] = queue_blacklist_total
|
|
result["queue_failed_total"] = queue_failed_total
|
|
result["current_job_blacklisted"] = queue_blacklist_total
|
|
result["recent_blacklisted_total"] = ops_blacklisted_recent
|
|
result["cumulative_blacklisted_total"] = int(result.get("blacklist_total", 0) or 0)
|
|
result["backlog_pending_total"] = max(backlog_pending_total, queue_pending_total)
|
|
result["backlog_claimed_total"] = max(backlog_claimed_total, queue_claimed_total)
|
|
result["backlog_running_total"] = max(backlog_running_total, queue_running_total)
|
|
result["backlog_register_pending_total"] = backlog_register_pending_total
|
|
result["backlog_downstream_pending_total"] = backlog_downstream_pending_total
|
|
result["cluster_proxy_available_count"] = int(detect_status.get("available_proxy_count", 0) or 0)
|
|
result["cluster_proxy_runtime_label"] = str(detect_status.get("proxy_runtime_label") or "").strip()
|
|
result["cluster_proxy_runtime_detail"] = str(detect_status.get("proxy_runtime_detail") or "").strip()
|
|
result["cluster_proxy_last_refresh_status"] = str(detect_status.get("proxy_last_refresh_status") or "").strip()
|
|
result["aggregate_process_count"] = int(detect_status.get("aggregate_process_count", 0) or 0)
|
|
result["aggregate_participating_node_count"] = int(detect_status.get("aggregate_participating_node_count", 0) or 0)
|
|
result["aggregate_active_thread_count"] = int(detect_status.get("active_thread_count", 0) or 0)
|
|
return result
|