From e1e178da2e01e91f3cedf4d35ced9a5e69e0342d Mon Sep 17 00:00:00 2001 From: Onkar Date: Wed, 17 Jun 2026 13:16:27 +0530 Subject: [PATCH] feat(ai-sre): add 7 missing agent tools, wire into pipeline runbook_tools.py: - restart_deployment: kubectl rollout restart deployment/X -n Y (safe, no downtime) - scale_deployment: kubectl scale deployment/X --replicas=N (clamped 0-50) - delete_stuck_pod: kubectl delete pod X [--force --grace-period=0] for stuck pods - cordon_node / uncordon_node: prevent scheduling on pressured nodes - describe_cluster_resource: kubectl describe on pod/deployment/node/service/pvc/etc log_tools.py: - analyze_logs_for_namespace: parallel log analysis across all running pods in namespace, returns aggregated error counts, categories, worst offender agents.py: - log_agent: +analyze_logs_for_namespace - infra_agent: +describe_cluster_resource - runbook_agent: +restart_deployment +scale_deployment +delete_stuck_pod +cordon_node --- ai-sre/agents.py | 10 +++- ai-sre/tools/log_tools.py | 65 +++++++++++++++++++- ai-sre/tools/runbook_tools.py | 110 ++++++++++++++++++++++++++++++++++ 3 files changed, 183 insertions(+), 2 deletions(-) diff --git a/ai-sre/agents.py b/ai-sre/agents.py index 7400d6b..7ae682d 100644 --- a/ai-sre/agents.py +++ b/ai-sre/agents.py @@ -16,10 +16,12 @@ get_oom_kill_events, detect_metric_anomaly, ) from .tools.log_tools import ( - analyze_pod_logs, search_logs_pattern, loki_query, + analyze_pod_logs, analyze_logs_for_namespace, search_logs_pattern, loki_query, ) from .tools.runbook_tools import ( list_available_runbooks, load_runbook, execute_safe_command, + restart_deployment, scale_deployment, delete_stuck_pod, + cordon_node, describe_cluster_resource, ) llm = LLM( @@ -114,6 +116,7 @@ llm=llm, tools=[ analyze_pod_logs, + analyze_logs_for_namespace, search_logs_pattern, loki_query, get_pod_logs, @@ -143,6 +146,7 @@ llm=llm, tools=[ describe_pod, + describe_cluster_resource, get_resource_quotas, get_pvc_status, get_hpa_status, @@ -177,6 +181,10 @@ tools=[ list_available_runbooks, load_runbook, + restart_deployment, + scale_deployment, + delete_stuck_pod, + cordon_node, execute_safe_command, ], verbose=True, diff --git a/ai-sre/tools/log_tools.py b/ai-sre/tools/log_tools.py index aa647a2..9624825 100644 --- a/ai-sre/tools/log_tools.py +++ b/ai-sre/tools/log_tools.py @@ -2,10 +2,11 @@ Log analysis tools: pattern mining, stack trace extraction, Loki LogQL support. Falls back to kubectl logs if Loki unavailable. """ +import concurrent.futures import json import logging import re -from typing import List, Optional +from typing import Dict, List, Optional import requests from crewai.tools import tool @@ -250,3 +251,65 @@ def loki_query(logql: str = "", namespace: str = "", pod_name: str = "", return json.dumps({"error": f"Loki unreachable at {sre_config.loki_url}"}) except Exception as e: return json.dumps({"error": str(e)}) + + +@tool +def analyze_logs_for_namespace(namespace: str = "default", max_pods: int = 10) -> str: + """Analyze logs from all running pods in a namespace in parallel. + Returns aggregated error counts, top error categories, and worst offenders. + Use when you need a namespace-wide log health picture quickly without targeting a specific pod. + Format: analyze_logs_for_namespace(namespace="production", max_pods=10)""" + try: + v1 = _get_v1() + pods = v1.list_namespaced_pod(namespace) + pod_names = [ + p.metadata.name for p in pods.items + if p.status and p.status.phase == "Running" + ][:max_pods] + + if not pod_names: + return json.dumps({"namespace": namespace, "pods_analyzed": 0, + "message": "No running pods found"}) + + def analyze_one(name: str) -> dict: + raw = _fetch_pod_logs(name, namespace, tail=100) + analysis = _mine_patterns(raw) + errors = analysis.get("errors", []) + cat_freq: Dict[str, int] = {} + for e in errors: + cat = e.get("category", "other") + cat_freq[cat] = cat_freq.get(cat, 0) + 1 + return { + "pod": name, + "error_count": len(errors), + "categories": cat_freq, + "top_error": errors[0]["text"][:200] if errors else None, + } + + results = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=5) as ex: + futures = {ex.submit(analyze_one, name): name for name in pod_names} + for fut in concurrent.futures.as_completed(futures): + try: + results.append(fut.result(timeout=15)) + except Exception as e: + results.append({"pod": futures[fut], "error": str(e)}) + + results.sort(key=lambda r: r.get("error_count", 0), reverse=True) + + total_errors = sum(r.get("error_count", 0) for r in results) + agg_cats: Dict[str, int] = {} + for r in results: + for cat, cnt in r.get("categories", {}).items(): + agg_cats[cat] = agg_cats.get(cat, 0) + cnt + + return json.dumps({ + "namespace": namespace, + "pods_analyzed": len(results), + "total_errors": total_errors, + "error_categories": agg_cats, + "pod_breakdown": results, + "worst_offender": results[0]["pod"] if results and results[0].get("error_count", 0) > 0 else None, + }) + except Exception as e: + return json.dumps({"error": str(e), "namespace": namespace}) diff --git a/ai-sre/tools/runbook_tools.py b/ai-sre/tools/runbook_tools.py index ff4b614..964fcc6 100644 --- a/ai-sre/tools/runbook_tools.py +++ b/ai-sre/tools/runbook_tools.py @@ -115,6 +115,116 @@ def load_runbook(filename: str) -> str: return json.dumps({"error": str(e)}) +@tool +def restart_deployment(deployment_name: str, namespace: str = "default") -> str: + """Perform a rolling restart of a Kubernetes deployment. + Safe — triggers new rollout without downtime. + Use for CrashLoopBackOff pods or after config changes. + Format: restart_deployment(deployment_name="web", namespace="default")""" + cmd = f"kubectl rollout restart deployment/{deployment_name} -n {namespace}" + if config.dry_run: + return json.dumps({"status": "dry_run", "command": cmd, "action": "restart_deployment", + "target": f"{namespace}/{deployment_name}"}) + try: + result = subprocess.run(shlex.split(cmd), capture_output=True, text=True, timeout=30) + return json.dumps({ + "status": "executed" if result.returncode == 0 else "error", + "action": "restart_deployment", "target": f"{namespace}/{deployment_name}", + "command": cmd, "stdout": result.stdout[:500], "stderr": result.stderr[:300], + "success": result.returncode == 0, + }) + except Exception as e: + return json.dumps({"status": "error", "reason": str(e), "command": cmd}) + + +@tool +def scale_deployment(deployment_name: str, replicas: int, namespace: str = "default") -> str: + """Scale a Kubernetes deployment to the specified replica count. + Use to scale UP during high load or DOWN to free resources. Max 50 replicas. + Format: scale_deployment(deployment_name="web", replicas=3, namespace="default")""" + replicas = max(0, min(replicas, 50)) + cmd = f"kubectl scale deployment/{deployment_name} --replicas={replicas} -n {namespace}" + if config.dry_run: + return json.dumps({"status": "dry_run", "command": cmd, "action": "scale_deployment", + "target": f"{namespace}/{deployment_name}", "replicas": replicas}) + try: + result = subprocess.run(shlex.split(cmd), capture_output=True, text=True, timeout=30) + return json.dumps({ + "status": "executed" if result.returncode == 0 else "error", + "action": "scale_deployment", "target": f"{namespace}/{deployment_name}", + "replicas": replicas, "command": cmd, + "stdout": result.stdout[:500], "success": result.returncode == 0, + }) + except Exception as e: + return json.dumps({"status": "error", "reason": str(e), "command": cmd}) + + +@tool +def delete_stuck_pod(pod_name: str, namespace: str = "default", force: bool = False) -> str: + """Delete a stuck or failed pod so its controller recreates it. + Use for Evicted, Failed, or stuck Terminating pods. + Set force=True ONLY for pods stuck in Terminating state (adds --grace-period=0). + Format: delete_stuck_pod(pod_name="web-abc-123", namespace="default", force=False)""" + force_flags = " --force --grace-period=0" if force else "" + cmd = f"kubectl delete pod {pod_name} -n {namespace}{force_flags}" + if config.dry_run: + return json.dumps({"status": "dry_run", "command": cmd, "action": "delete_pod", + "target": f"{namespace}/{pod_name}"}) + try: + result = subprocess.run(shlex.split(cmd), capture_output=True, text=True, timeout=30) + return json.dumps({ + "status": "executed" if result.returncode == 0 else "error", + "action": "delete_pod", "target": f"{namespace}/{pod_name}", + "command": cmd, "stdout": result.stdout[:500], "success": result.returncode == 0, + }) + except Exception as e: + return json.dumps({"status": "error", "reason": str(e), "command": cmd}) + + +@tool +def cordon_node(node_name: str, uncordon: bool = False) -> str: + """Cordon or uncordon a Kubernetes node to control pod scheduling. + Cordon prevents NEW pods from being scheduled on the node (existing pods unaffected). + Use when node shows disk/memory pressure or needs maintenance. + Set uncordon=True to re-enable scheduling after the node recovers. + Format: cordon_node(node_name="node-1", uncordon=False)""" + action = "uncordon" if uncordon else "cordon" + cmd = f"kubectl {action} {node_name}" + if config.dry_run: + return json.dumps({"status": "dry_run", "command": cmd, "action": action, "node": node_name}) + try: + result = subprocess.run(shlex.split(cmd), capture_output=True, text=True, timeout=30) + return json.dumps({ + "status": "executed" if result.returncode == 0 else "error", + "action": action, "node": node_name, + "command": cmd, "stdout": result.stdout[:500], "success": result.returncode == 0, + }) + except Exception as e: + return json.dumps({"status": "error", "reason": str(e), "command": cmd}) + + +@tool +def describe_cluster_resource(resource_type: str, resource_name: str, namespace: str = "default") -> str: + """Run kubectl describe on any cluster resource for deep diagnosis. + Returns full status, events, conditions, and resource limits. + resource_type must be one of: pod, deployment, node, service, pvc, replicaset, statefulset, daemonset, job. + Format: describe_cluster_resource(resource_type="pod", resource_name="web-abc-123", namespace="default")""" + SAFE_TYPES = {"pod", "deployment", "node", "service", "pvc", "replicaset", "statefulset", "daemonset", "job"} + if resource_type.lower() not in SAFE_TYPES: + return json.dumps({"error": f"resource_type '{resource_type}' not allowed. Use: {sorted(SAFE_TYPES)}"}) + ns_flag = f" -n {namespace}" if resource_type.lower() != "node" else "" + cmd = f"kubectl describe {resource_type} {resource_name}{ns_flag}" + try: + result = subprocess.run(shlex.split(cmd), capture_output=True, text=True, timeout=20) + return json.dumps({ + "command": cmd, + "output": result.stdout[:4000], + "error": result.stderr[:500] if result.returncode != 0 else None, + }) + except Exception as e: + return json.dumps({"error": str(e), "command": cmd}) + + @tool def execute_safe_command(command: str, dry_run: bool = True) -> str: """