cluster_state: tidy signals input

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(
+def _build_signals(context: SignalContext) -> list[dict[str, Any]]:
-    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]]:
     signals = (
-        _node_delta_signals(node_context)
+        _node_delta_signals(context.node_context)
-        + _namespace_delta_signals(namespace_context)
+        + _namespace_delta_signals(context.namespace_context)
-        + _workload_health_signals(workloads_health)
+        + _workload_health_signals(context.workloads_health)
-        + _pod_issue_signals(pod_issues)
+        + _pod_issue_signals(context.pod_issues)
-        + _kustomization_signals(kustomizations)
+        + _kustomization_signals(context.kustomizations)
-        + _pvc_pressure_signals(metrics)
+        + _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,
+        SignalContext(
-        node_context,
+            metrics=metrics,
-        namespace_context,
+            node_context=node_context,
-        workload_health,
+            namespace_context=namespace_context,
-        pod_issues,
+            workloads_health=workload_health,
-        kustomizations,
+            pod_issues=pod_issues,
+            kustomizations=kustomizations,
+        )
     )
     profiles = _build_profiles(node_context, namespace_context, node_pods, workloads, node_workloads)
     summary = {