Kubernetes Reporting Tools

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

def get_namespace_resources(namespace: str) -> list[dict]:
    """Collect resource requests and limits for all pods in a namespace."""
    pods = v1.list_namespaced_pod(namespace)
    resources = []

    for pod in pods.items:
        if not pod.spec.containers:
            continue
        for container in pod.spec.containers:
            resources.append({
                "namespace": namespace,
                "pod": pod.metadata.name,
                "container": container.name,
                "cpu_request": container.resources.requests.get("cpu", "0") if container.resources.requests else "0",
                "cpu_limit": container.resources.limits.get("cpu", "0") if container.resources.limits else "0",
                "mem_request": container.resources.requests.get("memory", "0") if container.resources.requests else "0",
                "mem_limit": container.resources.limits.get("memory", "0") if container.resources.limits else "0",
                "phase": pod.status.phase,
            })

    return resources

# Collect for all namespaces
namespaces = ["default", "production", "staging"]
all_resources = []
for ns in namespaces:
    all_resources.extend(get_namespace_resources(ns))

# Summarize
total_cpu_req = sum(parse_cpu(r["cpu_request"]) for r in all_resources)
total_mem_req = sum(parse_memory(r["mem_request"]) for r in all_resources)
print(f"Total CPU requests: {total_cpu_req} cores")
print(f"Total memory requests: {total_mem_req / 1024:.1f} GB")

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

# Azure AKS pricing (approximate, varies by region)
CPU_COST_PER_HOUR = 0.048  # per vCPU
MEMORY_COST_PER_HOUR = 0.006  # per GB

def estimate_namespace_cost(namespace: str) -> dict:
    """Estimate hourly cost based on resource requests."""
    pods = v1.list_namespaced_pod(namespace)
    total_cpu = 0
    total_mem_gb = 0

    for pod in pods.items:
        for container in (pod.spec.containers or []):
            if container.resources and container.resources.requests:
                total_cpu += parse_cpu(container.resources.requests.get("cpu", "0"))
                total_mem_gb += parse_memory(container.resources.requests.get("memory", "0")) / 1024

    hourly_cost = (total_cpu * CPU_COST_PER_HOUR) + (total_mem_gb * MEMORY_COST_PER_HOUR)

    return {
        "namespace": namespace,
        "cpu_cores": total_cpu,
        "memory_gb": round(total_mem_gb, 2),
        "hourly_cost": round(hourly_cost, 2),
        "daily_cost": round(hourly_cost * 24, 2),
        "monthly_cost": round(hourly_cost * 24 * 30, 2),
    }

def parse_cpu(cpu_str: str) -> float:
    """Parse CPU string like '500m' or '2' to float cores."""
    if cpu_str.endswith("m"):
        return int(cpu_str[:-1]) / 1000
    return float(cpu_str)

def parse_memory(mem_str: str) -> float:
    """Parse memory string like '256Mi' or '1Gi' to MB."""
    if mem_str.endswith("Gi"):
        return float(mem_str[:-2]) * 1024
    if mem_str.endswith("Mi"):
        return float(mem_str[:-2])
    if mem_str.endswith("Ki"):
        return float(mem_str[:-2]) / 1024
    return float(mem_str) / (1024 * 1024)

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

def check_compliance(namespace: str) -> list[dict]:
    """Check pods for common compliance issues."""
    violations = []
    pods = v1.list_namespaced_pod(namespace)

    for pod in pods.items:
        for container in (pod.spec.containers or []):
            # Check for latest tag
            image = container.image
            if image.endswith(":latest") or ":" not in image:
                violations.append({
                    "pod": pod.metadata.name,
                    "container": container.name,
                    "rule": "image-tag",
                    "message": f"Image uses latest or no tag: {image}",
                })

            # Check for privileged containers
            if container.security_context and container.security_context.privileged:
                violations.append({
                    "pod": pod.metadata.name,
                    "container": container.name,
                    "rule": "privileged",
                    "message": "Container runs in privileged mode",
                })

            # Check for missing resource limits
            if not container.resources or not container.resources.limits:
                violations.append({
                    "pod": pod.metadata.name,
                    "container": container.name,
                    "rule": "no-limits",
                    "message": "No resource limits set",
                })

            # Check for latest tag
            if container.image.endswith(":latest"):
                violations.append({
                    "pod": pod.metadata.name,
                    "container": container.name,
                    "rule": "latest-tag",
                    "message": "Uses :latest tag",
                })

    return violations

# Run compliance checks
for ns in ["production", "staging"]:
    violations = check_compliance(ns)
    print(f"\n{ns}: {len(violations)} violations")
    for v in violations:
        print(f"  [{v['rule']}] {v['pod']}: {v['message']}")

from kubernetes import client, config
from collections import defaultdict

config.load_kube_config()
v1 = client.CoreV1Api()

def generate_namespace_report(namespace: str) -> dict:
    """Generate a comprehensive namespace report."""
    pods = v1.list_namespaced_pod(namespace)
    services = v1.list_namespaced_service(namespace)
    deployments = apps_v1.list_namespaced_deployment(namespace)
    configmaps = v1.list_namespaced_configmap(namespace)
    secrets = v1.list_namespaced_secret(namespace)

    # Pod status breakdown
    pod_status = defaultdict(int)
    for pod in pods.items:
        pod_status[pod.status.phase] += 1

    # Container images
    images = set()
    for pod in pods.items:
        for c in (pod.spec.containers or []):
            images.add(c.image)

    # Resource totals
    total_cpu = 0
    total_mem = 0
    for pod in pods.items:
        for c in (pod.spec.containers or []):
            if c.resources and c.resources.requests:
                total_cpu += parse_cpu(c.resources.requests.get("cpu", "0"))
                total_mem += parse_memory(c.resources.requests.get("memory", "0"))

    return {
        "namespace": namespace,
        "pods": dict(pod_status),
        "services": len(services.items),
        "deployments": len(deployments.items),
        "configmaps": len(configmaps.items),
        "secrets": len(secrets.items),
        "unique_images": len(images),
        "total_cpu_request": f"{total_cpu:.2f}",
        "total_memory_request_mb": f"{total_mem:.0f}",
    }

# Generate reports for all namespaces
report = generate_namespace_report("production")

from jinja2 import Template
from datetime import datetime
from pathlib import Path

REPORT_TEMPLATE = Template("""
<!DOCTYPE html>
<html>
<head><title>Kubernetes Cluster Report</title></head>
<body>
<h1>Cluster Report</h1>
<p>Generated: {{ timestamp }}</p>

<h2>Namespace Summary</h2>
<table border="1" cellpadding="8">
<tr><th>Namespace</th><th>Pods</th><th>Services</th><th>CPU Request</th><th>Memory Request</th></tr>
{% for ns in namespaces %}
<tr>
  <td>{{ ns.name }}</td>
  <td>{{ ns.pod_count }}</td>
  <td>{{ ns.service_count }}</td>
  <td>{{ ns.cpu_request }}</td>
  <td>{{ ns.mem_request }} MB</td>
</tr>
{% endfor %}
</table>

<h2>Compliance Violations</h2>
{% if violations %}
<ul>
{% for v in violations %}
<li><strong>{{ v.rule }}</strong>: {{ v.pod }} — {{ v.message }}</li>
{% endfor %}
</ul>
{% else %}
<p>No violations found.</p>
{% endif %}
</body>
</html>
""")

def generate_html_report(namespaces: list[dict], violations: list[dict]) -> str:
    return REPORT_TEMPLATE.render(
        timestamp=datetime.utcnow().isoformat(),
        namespaces=namespaces,
        violations=violations,
    )

html = generate_html_report(namespace_reports, all_violations)
Path("cluster-report.html").write_text(html, encoding="utf-8")

import schedule
import time
from datetime import datetime

def weekly_cluster_report():
    """Generate and send weekly cluster report."""
    print(f"[{datetime.now()}] Generating cluster report...")

    # Collect data
    namespaces = []
    all_violations = []

    for ns in ["default", "production", "staging"]:
        report = generate_namespace_report(ns)
        namespaces.append(report)
        all_violations.extend(check_compliance(ns))

    # Generate HTML report
    html = generate_html_report(namespaces, all_violations)

    # Save to file
    filename = f"cluster-report-{datetime.now().strftime('%Y-%m-%d')}.html"
    Path(f"/reports/{filename}").write_text(html)

    # Send email
    send_report_email(
        to="platform-team@example.com",
        subject=f"Weekly Cluster Report - {datetime.now().strftime('%Y-%m-%d')}",
        body=html,
    )

# Schedule for every Monday at 9am
schedule.every().monday.at("09:00").do(weekly_cluster_report)

while True:
    schedule.run_pending()
    time.sleep(60)