ariadne/ariadne/services/cluster_state_profiles.py

105 lines
4.3 KiB
Python

from __future__ import annotations
from typing import Any
from .cluster_state_contract import ClusterStateSummary, SignalContext, _NODE_WORKLOAD_TOP, _PROFILE_LIMIT
ProfileRows = list[dict[str, Any]]
NodeWorkloadMap = dict[str, dict[str, int]]
def _node_profiles(node_context: ProfileRows, node_pods: ProfileRows, node_workloads: NodeWorkloadMap) -> ProfileRows:
pod_map = {entry.get("node"): entry for entry in node_pods if isinstance(entry, dict)}
workload_map = node_workloads or {}
profiles: list[dict[str, Any]] = []
for entry in node_context:
if not isinstance(entry, dict):
continue
node = entry.get("node")
if not isinstance(node, str) or not node:
continue
pods = pod_map.get(node, {})
workloads = workload_map.get(node, {})
workloads_top = sorted(workloads.items(), key=lambda item: (-item[1], item[0]))[:_NODE_WORKLOAD_TOP]
profiles.append(
{
"node": node,
"ready": entry.get("ready"),
"hardware": entry.get("hardware"),
"arch": entry.get("arch"),
"roles": entry.get("roles"),
"pods_total": pods.get("pods_total"),
"pods_running": pods.get("pods_running"),
"namespaces_top": pods.get("namespaces_top") or [],
"workloads_top": workloads_top,
"load_index": entry.get("load_index"),
"cpu": entry.get("cpu"),
"ram": entry.get("ram"),
"net": entry.get("net"),
"io": entry.get("io"),
"disk": entry.get("disk"),
"baseline_delta": entry.get("baseline_delta") or {},
}
)
profiles.sort(key=lambda item: (-(item.get("load_index") or 0), item.get("node") or ""))
return profiles[:_PROFILE_LIMIT]
def _namespace_profiles(namespace_context: list[dict[str, Any]]) -> list[dict[str, Any]]:
entries = [entry for entry in namespace_context if isinstance(entry, dict)]
entries.sort(key=lambda item: (-(item.get("pods_total") or 0), item.get("namespace") or ""))
output: list[dict[str, Any]] = []
for entry in entries[:_PROFILE_LIMIT]:
output.append(
{
"namespace": entry.get("namespace"),
"pods_total": entry.get("pods_total"),
"pods_running": entry.get("pods_running"),
"primary_node": entry.get("primary_node"),
"nodes_top": entry.get("nodes_top") or [],
"cpu_usage": entry.get("cpu_usage"),
"mem_usage": entry.get("mem_usage"),
"cpu_ratio": entry.get("cpu_ratio"),
"mem_ratio": entry.get("mem_ratio"),
"baseline_delta": entry.get("baseline_delta") or {},
}
)
return output
def _workload_profiles(workloads: list[dict[str, Any]]) -> list[dict[str, Any]]:
entries = [entry for entry in workloads if isinstance(entry, dict)]
entries.sort(
key=lambda item: (-(item.get("pods_total") or 0), item.get("namespace") or "", item.get("workload") or ""),
)
output: list[dict[str, Any]] = []
for entry in entries[:_PROFILE_LIMIT]:
nodes = entry.get("nodes")
nodes_top = (
sorted(nodes.items(), key=lambda item: (-item[1], item[0]))[:3]
if isinstance(nodes, dict)
else []
)
output.append(
{
"namespace": entry.get("namespace"),
"workload": entry.get("workload"),
"source": entry.get("source"),
"pods_total": entry.get("pods_total"),
"pods_running": entry.get("pods_running"),
"primary_node": entry.get("primary_node"),
"nodes_top": nodes_top,
}
)
return output
def _build_profiles(node_context: ProfileRows, namespace_context: ProfileRows, node_pods: ProfileRows, workloads: ProfileRows, node_workloads: NodeWorkloadMap) -> dict[str, Any]:
return {
"nodes": _node_profiles(node_context, node_pods, node_workloads),
"namespaces": _namespace_profiles(namespace_context),
"workloads": _workload_profiles(workloads),
}
__all__ = [name for name in globals() if (name.startswith("_") and not name.startswith("__")) or name in {"ClusterStateSummary", "SignalContext"}]