AI-Powered Log Analysis — How LLMs Are Replacing grep for DevOps Engineers

# Using Claude API with curl
kubectl logs deployment/payment-service --since=30m | \
  head -500 | \
  jq -Rs '{
    model: "claude-sonnet-4-5-20250514",
    max_tokens: 1024,
    messages: [{
      role: "user",
      content: ("Analyze these application logs and identify the root cause of any errors. Provide a summary, root cause, and suggested fix.\n\nLogs:\n" + .)
    }]
  }' | \
  curl -s https://api.anthropic.com/v1/messages \
    -H "x-api-key: $ANTHROPIC_API_KEY" \
    -H "anthropic-version: 2023-06-01" \
    -H "content-type: application/json" \
    -d @- | jq -r '.content[0].text'

# log_analyzer.py
import subprocess
import json
from anthropic import Anthropic
 
client = Anthropic()
 
def get_logs(namespace: str, deployment: str, since: str = "30m") -> str:
    """Fetch logs from Kubernetes."""
    result = subprocess.run(
        ["kubectl", "logs", f"deployment/{deployment}",
         "-n", namespace, f"--since={since}", "--tail=1000"],
        capture_output=True, text=True
    )
    return result.stdout
 
def analyze_logs(logs: str, context: str = "") -> str:
    """Send logs to Claude for analysis."""
    prompt = f"""You are a senior DevOps engineer analyzing production logs.
 
Context: {context}
 
Analyze these logs and provide:
1. **Summary** — What's happening in 2-3 sentences
2. **Errors Found** — List each unique error with count and first occurrence
3. **Root Cause** — Your best assessment of why this is happening
4. **Impact** — What's affected and severity (P1-P4)
5. **Recommended Fix** — Step-by-step remediation
6. **Prevention** — How to prevent this in the future
 
Logs:
{logs}"""
 
    response = client.messages.create(
        model="claude-sonnet-4-5-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text
 
def chunk_and_analyze(logs: str, chunk_size: int = 50000) -> str:
    """Handle large log volumes by chunking."""
    if len(logs) <= chunk_size:
        return analyze_logs(logs)
 
    chunks = [logs[i:i+chunk_size] for i in range(0, len(logs), chunk_size)]
    chunk_summaries = []
 
    for i, chunk in enumerate(chunks):
        summary = analyze_logs(
            chunk,
            context=f"This is chunk {i+1} of {len(chunks)} from the same log stream"
        )
        chunk_summaries.append(summary)
 
    # Final synthesis
    combined = "\n---\n".join(chunk_summaries)
    synthesis_prompt = f"""You analyzed {len(chunks)} chunks of logs. Here are the per-chunk analyses.
Synthesize them into a single coherent root cause analysis.
 
{combined}"""
 
    response = client.messages.create(
        model="claude-sonnet-4-5-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return response.content[0].text
 
 
if __name__ == "__main__":
    import sys
    namespace = sys.argv[1] if len(sys.argv) > 1 else "default"
    deployment = sys.argv[2] if len(sys.argv) > 2 else "my-app"
 
    print(f"Fetching logs for {namespace}/{deployment}...")
    logs = get_logs(namespace, deployment)
 
    if not logs.strip():
        print("No logs found.")
        sys.exit(1)
 
    print(f"Analyzing {len(logs)} characters of logs...")
    analysis = chunk_and_analyze(logs)
    print("\n" + "="*60)
    print("LOG ANALYSIS REPORT")
    print("="*60 + "\n")
    print(analysis)

python log_analyzer.py production payment-service

# alert_analyzer.py
from flask import Flask, request
import json
from log_analyzer import get_logs, chunk_and_analyze
 
app = Flask(__name__)
 
# Map alert names to Kubernetes deployments
ALERT_TO_DEPLOYMENT = {
    "PaymentService5xxRate": ("production", "payment-service"),
    "AuthServiceLatency": ("production", "auth-service"),
    "OrderServiceOOM": ("production", "order-service"),
}
 
@app.route("/webhook/pagerduty", methods=["POST"])
def handle_alert():
    payload = request.json
    alert_name = payload.get("alert", {}).get("name", "")
 
    if alert_name in ALERT_TO_DEPLOYMENT:
        namespace, deployment = ALERT_TO_DEPLOYMENT[alert_name]
 
        # Fetch and analyze logs
        logs = get_logs(namespace, deployment, since="15m")
        analysis = chunk_and_analyze(logs)
 
        # Post analysis back to the incident
        post_to_slack(
            channel="#incidents",
            text=f"*AI Log Analysis for {alert_name}*\n\n{analysis}"
        )
 
        # Add as PagerDuty incident note
        add_incident_note(
            incident_id=payload["incident"]["id"],
            note=analysis
        )
 
    return {"status": "ok"}, 200

# Install
brew install k8sgpt
 
# Configure with your LLM
k8sgpt auth add --backend anthropic --model claude-sonnet-4-5-20250514
 
# Analyze cluster issues
k8sgpt analyze --explain

# Install with Helm
helm repo add robusta https://robusta-charts.storage.googleapis.com
helm install holmesgpt robusta/holmesgpt \
  --set anthropicKey=$ANTHROPIC_API_KEY

pip install openlit
 
# Auto-instrument your LLM calls
import openlit
openlit.init(otlp_endpoint="http://otel-collector:4318")

import re
 
def sanitize_logs(logs: str) -> str:
    """Remove sensitive data before LLM analysis."""
    # Remove bearer tokens
    logs = re.sub(r'Bearer [A-Za-z0-9\-._~+/]+=*', 'Bearer [REDACTED]', logs)
    # Remove email addresses
    logs = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]', logs)
    # Remove IP addresses (optional)
    logs = re.sub(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP]', logs)
    # Remove common secret patterns
    logs = re.sub(r'(password|secret|token|key)=\S+', r'\1=[REDACTED]', logs, flags=re.I)
    return logs