diff --git a/ariadne/services/cluster_state.py b/ariadne/services/cluster_state.py index d438d9a..27b7d39 100644 --- a/ariadne/services/cluster_state.py +++ b/ariadne/services/cluster_state.py @@ -56,6 +56,7 @@ _WORKLOAD_INDEX_LIMIT = 20 _NODE_WORKLOAD_LIMIT = 12 _NODE_WORKLOAD_TOP = 3 _EVENTS_SUMMARY_LIMIT = 5 +_PVC_CRITICAL_THRESHOLD = 90.0 _CAPACITY_KEYS = { "cpu", "memory", @@ -137,6 +138,16 @@ class ClusterStateSummary: errors: int +@dataclass(frozen=True) +class SignalContext: + metrics: dict[str, Any] + node_context: list[dict[str, Any]] + namespace_context: list[dict[str, Any]] + workloads_health: dict[str, Any] + pod_issues: dict[str, Any] + kustomizations: dict[str, Any] + + def _items(payload: dict[str, Any]) -> list[dict[str, Any]]: items = payload.get("items") if isinstance(payload.get("items"), list) else [] return [item for item in items if isinstance(item, dict)] @@ -2326,21 +2337,14 @@ def _workload_health_signals(workloads_health: dict[str, Any]) -> list[dict[str, return output -def _build_signals( - metrics: dict[str, Any], - node_context: list[dict[str, Any]], - namespace_context: list[dict[str, Any]], - workloads_health: dict[str, Any], - pod_issues: dict[str, Any], - kustomizations: dict[str, Any], -) -> list[dict[str, Any]]: +def _build_signals(context: SignalContext) -> list[dict[str, Any]]: signals = ( - _node_delta_signals(node_context) - + _namespace_delta_signals(namespace_context) - + _workload_health_signals(workloads_health) - + _pod_issue_signals(pod_issues) - + _kustomization_signals(kustomizations) - + _pvc_pressure_signals(metrics) + _node_delta_signals(context.node_context) + + _namespace_delta_signals(context.namespace_context) + + _workload_health_signals(context.workloads_health) + + _pod_issue_signals(context.pod_issues) + + _kustomization_signals(context.kustomizations) + + _pvc_pressure_signals(context.metrics) ) signals.sort(key=lambda item: (_severity_rank(item.get("severity")), item.get("scope") or "")) return signals[:_SIGNAL_LIMIT] @@ -2473,7 +2477,7 @@ def _pvc_pressure_signals(metrics: dict[str, Any]) -> list[dict[str, Any]]: "target": f"{entry.get('namespace')}/{entry.get('pvc')}", "metric": "used_percent", "current": used, - "severity": "warning" if used < 90 else "critical", + "severity": "warning" if used < _PVC_CRITICAL_THRESHOLD else "critical", } ) return output @@ -2860,12 +2864,14 @@ def collect_cluster_state() -> tuple[dict[str, Any], ClusterStateSummary]: node_workloads, ) signals = _build_signals( - metrics, - node_context, - namespace_context, - workload_health, - pod_issues, - kustomizations, + SignalContext( + metrics=metrics, + node_context=node_context, + namespace_context=namespace_context, + workloads_health=workload_health, + pod_issues=pod_issues, + kustomizations=kustomizations, + ) ) profiles = _build_profiles(node_context, namespace_context, node_pods, workloads, node_workloads) summary = {