Kubernetes Health

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
Health Report Aggregation Script

Aggregates results from multiple health check agents into a unified report.
Calculates weighted scores and determines overall health status.

Usage:
    uv run aggregate_report.py [--reports-dir PATH] [--output summary|detailed|json]
    cat reports/*.json | uv run aggregate_report.py
"""

import json
import sys
from datetime import datetime, timezone
from typing import Any


# Component weights for overall score calculation
COMPONENT_WEIGHTS = {
    "Core": 1.0,  # Core is always weighted highest
    "default": 0.8,  # Operators weighted slightly lower
}

# Check category weights
CHECK_WEIGHTS = {
    "availability": 0.40,
    "configuration": 0.25,
    "freshness": 0.20,
    "resources": 0.15,
}

# Status thresholds
STATUS_THRESHOLDS = {
    "HEALTHY": 90,
    "DEGRADED": 60,
    "CRITICAL": 0,
}


def determine_status(score: float) -> str:
    """Determine health status from score."""
    if score >= STATUS_THRESHOLDS["HEALTHY"]:
        return "HEALTHY"
    elif score >= STATUS_THRESHOLDS["DEGRADED"]:
        return "DEGRADED"
    else:
        return "CRITICAL"


def calculate_component_score(checks: list[dict[str, Any]]) -> float:
    """Calculate score for a single component based on its checks."""
    if not checks:
        return 100.0

    total_weight = 0.0
    weighted_score = 0.0

    for check in checks:
        status = check.get("status", "OK")
        category = check.get("category", "availability")
        weight = CHECK_WEIGHTS.get(category, 0.25)

        # Convert status to score
        if status == "OK":
            check_score = 100.0
        elif status == "WARNING":
            check_score = 70.0
        else:  # ERROR
            check_score = 30.0

        weighted_score += check_score * weight
        total_weight += weight

    if total_weight == 0:
        return 100.0

    return weighted_score / total_weight


def calculate_overall_score(components: dict[str, dict[str, Any]]) -> float:
    """Calculate overall health score from all components."""
    if not components:
        return 0.0

    total_weight = 0.0
    weighted_score = 0.0

    for name, component in components.items():
        score = component.get("score", 0)
        weight = COMPONENT_WEIGHTS.get(name, COMPONENT_WEIGHTS["default"])

        weighted_score += score * weight
        total_weight += weight

    if total_weight == 0:
        return 0.0

    return weighted_score / total_weight


def extract_issues(components: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
    """Extract all issues from components, sorted by severity."""
    issues = []

    severity_order = {"ERROR": 0, "WARNING": 1, "OK": 2}

    for component_name, component in components.items():
        for check in component.get("checks", []):
            if check.get("status") in ("ERROR", "WARNING"):
                issues.append({
                    "component": component_name,
                    "check": check.get("name", "unknown"),
                    "severity": check.get("status"),
                    "message": check.get("message", ""),
                })

    # Sort by severity (ERROR first)
    issues.sort(key=lambda x: severity_order.get(x["severity"], 2))

    return issues


def extract_recommendations(components: dict[str, dict[str, Any]]) -> list[str]:
    """Extract all recommendations from components."""
    recommendations = []

    for component in components.values():
        recommendations.extend(component.get("recommendations", []))

    # Deduplicate while preserving order
    seen = set()
    unique_recommendations = []
    for rec in recommendations:
        if rec not in seen:
            seen.add(rec)
            unique_recommendations.append(rec)

    return unique_recommendations


def aggregate_reports(
    component_reports: list[dict[str, Any]],
    discovery_info: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """
    Aggregate multiple component health reports into a unified report.

    Args:
        component_reports: List of health reports from individual agents
        discovery_info: Optional discovery metadata

    Returns:
        Unified health report
    """
    components = {}

    for report in component_reports:
        component_name = report.get("component", "Unknown")

        # Calculate component score if not provided
        if "score" not in report:
            report["score"] = calculate_component_score(report.get("checks", []))

        # Determine status if not provided
        if "status" not in report:
            report["status"] = determine_status(report["score"])

        components[component_name] = report

    # Calculate overall score
    overall_score = calculate_overall_score(components)
    overall_status = determine_status(overall_score)

    # Extract issues and recommendations
    issues = extract_issues(components)
    recommendations = extract_recommendations(components)

    # Build unified report
    report = {
        "cluster": discovery_info.get("cluster", "unknown") if discovery_info else "unknown",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "discovery": discovery_info or {},
        "detected_operators": discovery_info.get("detected_operators", []) if discovery_info else [],
        "components": components,
        "overall": {
            "status": overall_status,
            "score": round(overall_score, 1),
            "critical_issues": [i for i in issues if i["severity"] == "ERROR"],
            "warnings": [i for i in issues if i["severity"] == "WARNING"],
            "recommendations": recommendations,
        },
    }

    return report


def format_summary(report: dict[str, Any]) -> str:
    """Format report as human-readable summary."""
    lines = [
        f"# Cluster Health Report - {report['cluster']}",
        f"**Timestamp**: {report['timestamp']}",
        f"**Overall Status**: {report['overall']['status']} (Score: {report['overall']['score']})",
        "",
        "## Components",
    ]

    for name, component in report["components"].items():
        status_emoji = {"HEALTHY": "+", "DEGRADED": "~", "CRITICAL": "-"}.get(component["status"], "?")
        lines.append(f"  [{status_emoji}] {name}: {component['status']} ({component['score']})")

    if report["overall"]["critical_issues"]:
        lines.append("")
        lines.append("## Critical Issues")
        for issue in report["overall"]["critical_issues"]:
            lines.append(f"  - [{issue['component']}] {issue['check']}: {issue['message']}")

    if report["overall"]["recommendations"]:
        lines.append("")
        lines.append("## Recommendations")
        for rec in report["overall"]["recommendations"][:5]:  # Top 5
            lines.append(f"  - {rec}")

    return "\n".join(lines)


def main() -> None:
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(description="Aggregate health reports")
    parser.add_argument("--reports-dir", help="Directory containing report JSON files")
    parser.add_argument("--discovery-json", help="Path to discovery JSON file")
    parser.add_argument(
        "--output",
        choices=["summary", "detailed", "json"],
        default="summary",
        help="Output format",
    )

    args = parser.parse_args()

    component_reports = []
    discovery_info = None

    # Load discovery info
    if args.discovery_json:
        with open(args.discovery_json) as f:
            discovery_info = json.load(f)

    # Load component reports from directory or stdin
    if args.reports_dir:
        import os
        for filename in os.listdir(args.reports_dir):
            if filename.endswith(".json"):
                with open(os.path.join(args.reports_dir, filename)) as f:
                    component_reports.append(json.load(f))
    elif not sys.stdin.isatty():
        # Try to read multiple JSON objects from stdin
        content = sys.stdin.read()
        try:
            # First try as JSON array
            component_reports = json.loads(content)
            if not isinstance(component_reports, list):
                component_reports = [component_reports]
        except json.JSONDecodeError:
            # Try line-by-line JSON
            for line in content.strip().split("\n"):
                if line:
                    component_reports.append(json.loads(line))

    # Aggregate reports
    report = aggregate_reports(component_reports, discovery_info)

    # Output based on format
    if args.output == "json":
        print(json.dumps(report, indent=2))
    elif args.output == "detailed":
        print(json.dumps(report, indent=2))
    else:  # summary
        print(format_summary(report))


if __name__ == "__main__":
    main()

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "kubernetes>=28.1.0",
# ]
# ///
"""
Kubernetes API Discovery Script

Discovers all API groups in a Kubernetes cluster and detects installed operators.
Returns a condensed JSON output suitable for agent consumption.

Usage:
    uv run discover_apis.py [--kubeconfig PATH] [--context NAME]
"""

import json
import sys
from typing import Any

try:
    from kubernetes import client, config
    from kubernetes.client.rest import ApiException
except ImportError:
    print(json.dumps({"error": "kubernetes package not installed", "hint": "Run: uv pip install kubernetes>=28.1.0"}))
    sys.exit(1)


# Known operator API groups and their names
KNOWN_OPERATORS = {
    "crossplane.io": "Crossplane",
    "apiextensions.crossplane.io": "Crossplane",
    "pkg.crossplane.io": "Crossplane",
    "argoproj.io": "ArgoCD",
    "cert-manager.io": "Cert-Manager",
    "acme.cert-manager.io": "Cert-Manager",
    "monitoring.coreos.com": "Prometheus",
    "flux.toolkit.fluxcd.io": "Flux",
    "source.toolkit.fluxcd.io": "Flux",
    "kustomize.toolkit.fluxcd.io": "Flux",
    "helm.toolkit.fluxcd.io": "Flux",
    "security.istio.io": "Istio",
    "networking.istio.io": "Istio",
    "gateway.networking.k8s.io": "Gateway API",
    "karpenter.sh": "Karpenter",
    "karpenter.k8s.aws": "Karpenter",
    "external-secrets.io": "External Secrets",
    "velero.io": "Velero",
}


def load_kube_config(kubeconfig: str | None = None, context: str | None = None) -> None:
    """Load Kubernetes configuration."""
    try:
        config.load_kube_config(config_file=kubeconfig, context=context)
    except config.ConfigException:
        try:
            config.load_incluster_config()
        except config.ConfigException as e:
            raise RuntimeError(f"Could not load Kubernetes config: {e}") from e


def discover_apis(kubeconfig: str | None = None, context: str | None = None) -> dict[str, Any]:
    """
    Discover all API groups and detect installed operators.

    Returns a condensed ClusterAPIMap structure.
    """
    load_kube_config(kubeconfig, context)

    api_client = client.ApiClient()

    result = {
        "cluster": "",
        "api_version": "",
        "core_resources": 0,
        "custom_resources": 0,
        "api_groups": [],
        "detected_operators": [],
    }

    # Get cluster info
    try:
        version_api = client.VersionApi(api_client)
        version_info = version_api.get_code()
        result["api_version"] = f"{version_info.major}.{version_info.minor}"
    except ApiException:
        result["api_version"] = "unknown"

    # Get current context name
    try:
        _, active_context = config.list_kube_config_contexts()
        result["cluster"] = active_context.get("context", {}).get("cluster", "unknown")
    except Exception:
        result["cluster"] = "unknown"

    # Discover core API (v1)
    try:
        core_api = client.CoreV1Api(api_client)
        api_resources = core_api.get_api_resources()
        result["core_resources"] = len(api_resources.resources)
    except ApiException as e:
        result["core_resources"] = 0
        result["error"] = f"Core API discovery failed: {e.reason}"

    # Discover all API groups
    try:
        apis_api = client.ApisApi(api_client)
        api_groups = apis_api.get_api_versions()

        seen_operators = set()

        for group in api_groups.groups:
            group_name = group.name
            result["api_groups"].append(group_name)

            # Check if this is a known operator
            if group_name in KNOWN_OPERATORS:
                operator_name = KNOWN_OPERATORS[group_name]
                if operator_name not in seen_operators:
                    seen_operators.add(operator_name)
                    result["detected_operators"].append({
                        "name": operator_name,
                        "api_group": group_name,
                        "status": "active",
                    })

        result["custom_resources"] = len(api_groups.groups)

    except ApiException as e:
        result["error"] = f"API group discovery failed: {e.reason}"

    return result


def main() -> None:
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(description="Discover Kubernetes APIs")
    parser.add_argument("--kubeconfig", help="Path to kubeconfig file")
    parser.add_argument("--context", help="Kubernetes context to use")
    parser.add_argument("--pretty", action="store_true", help="Pretty print JSON output")

    args = parser.parse_args()

    try:
        result = discover_apis(kubeconfig=args.kubeconfig, context=args.context)

        if args.pretty:
            print(json.dumps(result, indent=2))
        else:
            print(json.dumps(result))

    except RuntimeError as e:
        print(json.dumps({"error": str(e)}))
        sys.exit(1)
    except Exception as e:
        print(json.dumps({"error": f"Unexpected error: {e}"}))
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
Health Orchestrator Script

Maps discovered API groups to specialized health check agents.
Returns an agent dispatch plan with parallelization groups.

Usage:
    uv run health_orchestrator.py [--discovery-json PATH]
    echo '{"detected_operators": [...]}' | uv run health_orchestrator.py
"""

import json
import sys
from dataclasses import dataclass
from typing import Any


@dataclass
class AgentConfig:
    """Configuration for a health check agent."""
    name: str
    api_groups: list[str]
    priority: int  # 1 = highest (core), 2 = operators
    parallel_group: int  # Agents in same group can run in parallel


# Define all available health check agents
AGENT_REGISTRY: list[AgentConfig] = [
    # Core agent - always runs
    AgentConfig(
        name="k8s-core-health-agent",
        api_groups=[],  # Empty means always active
        priority=1,
        parallel_group=1,
    ),
    # Operator agents - conditional based on API discovery
    AgentConfig(
        name="k8s-crossplane-health-agent",
        api_groups=["crossplane.io", "apiextensions.crossplane.io", "pkg.crossplane.io"],
        priority=2,
        parallel_group=2,
    ),
    AgentConfig(
        name="k8s-argocd-health-agent",
        api_groups=["argoproj.io"],
        priority=2,
        parallel_group=2,
    ),
    AgentConfig(
        name="k8s-certmanager-health-agent",
        api_groups=["cert-manager.io", "acme.cert-manager.io"],
        priority=2,
        parallel_group=2,
    ),
    AgentConfig(
        name="k8s-prometheus-health-agent",
        api_groups=["monitoring.coreos.com"],
        priority=2,
        parallel_group=2,
    ),
]


def get_active_agents(discovered_apis: dict[str, Any]) -> list[dict[str, Any]]:
    """
    Determine which agents should be activated based on discovered APIs.

    Args:
        discovered_apis: Output from discover_apis.py

    Returns:
        List of agent configurations to dispatch
    """
    api_groups = set(discovered_apis.get("api_groups", []))

    active_agents = []

    for agent in AGENT_REGISTRY:
        # Core agent always runs
        if not agent.api_groups:
            active_agents.append({
                "name": agent.name,
                "priority": agent.priority,
                "parallel_group": agent.parallel_group,
                "trigger": "always",
            })
            continue

        # Check if any of the agent's API groups are present
        matching_groups = [g for g in agent.api_groups if g in api_groups]
        if matching_groups:
            active_agents.append({
                "name": agent.name,
                "priority": agent.priority,
                "parallel_group": agent.parallel_group,
                "trigger": matching_groups[0],
            })

    # Sort by priority (lower = higher priority)
    active_agents.sort(key=lambda a: (a["priority"], a["name"]))

    return active_agents


def create_dispatch_plan(active_agents: list[dict[str, Any]]) -> dict[str, Any]:
    """
    Create an execution plan for dispatching agents.

    Args:
        active_agents: List of active agent configurations

    Returns:
        Dispatch plan with parallel and sequential groups
    """
    # Group agents by parallel_group
    groups: dict[int, list[dict[str, Any]]] = {}
    for agent in active_agents:
        group = agent["parallel_group"]
        if group not in groups:
            groups[group] = []
        groups[group].append(agent)

    # Build execution plan
    plan = {
        "total_agents": len(active_agents),
        "execution_groups": [],
    }

    for group_id in sorted(groups.keys()):
        group_agents = groups[group_id]
        plan["execution_groups"].append({
            "group": group_id,
            "parallel": len(group_agents) > 1,
            "agents": [a["name"] for a in group_agents],
        })

    return plan


def main() -> None:
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(description="Map APIs to health agents")
    parser.add_argument("--discovery-json", help="Path to discovery JSON file")
    parser.add_argument("--pretty", action="store_true", help="Pretty print output")

    args = parser.parse_args()

    # Read discovery data from file or stdin
    if args.discovery_json:
        with open(args.discovery_json) as f:
            discovery_data = json.load(f)
    elif not sys.stdin.isatty():
        discovery_data = json.load(sys.stdin)
    else:
        # Default empty discovery for testing
        discovery_data = {"api_groups": [], "detected_operators": []}

    # Get active agents
    active_agents = get_active_agents(discovery_data)

    # Create dispatch plan
    dispatch_plan = create_dispatch_plan(active_agents)

    result = {
        "active_agents": active_agents,
        "dispatch_plan": dispatch_plan,
    }

    if args.pretty:
        print(json.dumps(result, indent=2))
    else:
        print(json.dumps(result))


if __name__ == "__main__":
    main()

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Kubernetes Health Report",
  "description": "Unified health report from cluster diagnostics",
  "type": "object",
  "properties": {
    "cluster": {
      "type": "string",
      "description": "Cluster name from kubectl context"
    },
    "timestamp": {
      "type": "string",
      "format": "date-time",
      "description": "ISO 8601 timestamp of report generation"
    },
    "discovery": {
      "type": "object",
      "description": "API discovery metadata",
      "properties": {
        "duration_ms": {
          "type": "integer",
          "description": "Time taken for API discovery in milliseconds"
        },
        "api_groups_found": {
          "type": "integer",
          "description": "Total number of API groups discovered"
        },
        "custom_resources": {
          "type": "integer",
          "description": "Number of custom resource definitions found"
        },
        "active_sub_agents": {
          "type": "integer",
          "description": "Number of health agents activated"
        }
      }
    },
    "detected_operators": {
      "type": "array",
      "description": "List of detected Kubernetes operators",
      "items": {
        "type": "object",
        "properties": {
          "name": {
            "type": "string",
            "description": "Operator name (e.g., Crossplane, ArgoCD)"
          },
          "version": {
            "type": "string",
            "description": "Detected operator version if available"
          },
          "status": {
            "type": "string",
            "enum": ["active", "inactive", "unknown"],
            "description": "Operator status"
          }
        },
        "required": ["name", "status"]
      }
    },
    "components": {
      "type": "object",
      "description": "Health reports per component",
      "additionalProperties": {
        "type": "object",
        "properties": {
          "status": {
            "type": "string",
            "enum": ["HEALTHY", "DEGRADED", "CRITICAL"],
            "description": "Component health status"
          },
          "score": {
            "type": "number",
            "minimum": 0,
            "maximum": 100,
            "description": "Component health score (0-100)"
          },
          "checks": {
            "type": "array",
            "description": "Individual health check results",
            "items": {
              "type": "object",
              "properties": {
                "name": {
                  "type": "string",
                  "description": "Check name"
                },
                "status": {
                  "type": "string",
                  "enum": ["OK", "WARNING", "ERROR"],
                  "description": "Check result status"
                },
                "message": {
                  "type": "string",
                  "description": "Human-readable status message"
                },
                "category": {
                  "type": "string",
                  "enum": ["availability", "configuration", "freshness", "resources"],
                  "description": "Check category for scoring"
                },
                "details": {
                  "type": "object",
                  "description": "Additional check-specific details"
                }
              },
              "required": ["name", "status"]
            }
          },
          "recommendations": {
            "type": "array",
            "description": "Recommended actions for this component",
            "items": {
              "type": "string"
            }
          }
        },
        "required": ["status", "score", "checks"]
      }
    },
    "overall": {
      "type": "object",
      "description": "Overall cluster health summary",
      "properties": {
        "status": {
          "type": "string",
          "enum": ["HEALTHY", "DEGRADED", "CRITICAL"],
          "description": "Overall cluster health status"
        },
        "score": {
          "type": "number",
          "minimum": 0,
          "maximum": 100,
          "description": "Overall health score (0-100)"
        },
        "critical_issues": {
          "type": "array",
          "description": "Critical issues requiring immediate attention",
          "items": {
            "type": "object",
            "properties": {
              "component": {
                "type": "string"
              },
              "check": {
                "type": "string"
              },
              "severity": {
                "type": "string"
              },
              "message": {
                "type": "string"
              }
            }
          }
        },
        "recommendations": {
          "type": "array",
          "description": "Prioritized recommendations across all components",
          "items": {
            "type": "string"
          }
        }
      },
      "required": ["status", "score"]
    }
  },
  "required": ["cluster", "timestamp", "components", "overall"]
}