How to Implement AI-Driven Log Analysis with OpenTelemetry 1.20 and Claude 3.5

# deploy_otel_collector.py
# Deploys OpenTelemetry 1.20 Collector as DaemonSet to Kubernetes
# Requires: kubernetes Python client, valid kubeconfig, cluster admin access

import argparse
import logging
from kubernetes import client, config
from kubernetes.client.rest import ApiException

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def load_kube_config():
    """Load local kubeconfig or in-cluster config for CI/CD environments"""
    try:
        config.load_kube_config()
        logger.info("Loaded local kubeconfig")
    except Exception as e:
        logger.warning(f"Failed to load local kubeconfig: {e}. Trying in-cluster config...")
        try:
            config.load_incluster_config()
            logger.info("Loaded in-cluster config")
        except Exception as e:
            logger.error(f"Failed to load any kubeconfig: {e}")
            raise

def create_otel_namespace(api):
    """Create dedicated otel-logs namespace if not exists"""
    namespace_manifest = client.V1Namespace(
        api_version="v1",
        kind="Namespace",
        metadata=client.V1ObjectMeta(name="otel-logs")
    )
    try:
        api.create_namespace(namespace_manifest)
        logger.info("Created otel-logs namespace")
    except ApiException as e:
        if e.status == 409:
            logger.info("otel-logs namespace already exists")
        else:
            logger.error(f"Failed to create namespace: {e}")
            raise

def deploy_collector_daemonset(api):
    """Deploy OpenTelemetry 1.20 Collector DaemonSet with log collection config"""
    # DaemonSet spec with resource limits for production safety
    daemonset_manifest = {
        "apiVersion": "apps/v1",
        "kind": "DaemonSet",
        "metadata": {
            "name": "otel-collector",
            "namespace": "otel-logs",
            "labels": {"app": "otel-collector"}
        },
        "spec": {
            "selector": {"matchLabels": {"app": "otel-collector"}},
            "template": {
                "metadata": {"labels": {"app": "otel-collector"}},
                "spec": {
                    "containers": [{
                        "name": "otel-collector",
                        "image": "otel/opentelemetry-collector-contrib:1.20.0",
                        "resources": {
                            "limits": {"cpu": "500m", "memory": "512Mi"},
                            "requests": {"cpu": "100m", "memory": "128Mi"}
                        },
                        "volumeMounts": [
                            {"name": "varlog", "mountPath": "/var/log"},
                            {"name": "varlibdockercontainers", "mountPath": "/var/lib/docker/containers", "readOnly": True}
                        ]
                    }],
                    "volumes": [
                        {"name": "varlog", "hostPath": {"path": "/var/log"}},
                        {"name": "varlibdockercontainers", "hostPath": {"path": "/var/lib/docker/containers"}}
                    ]
                }
            }
        }
    }
    try:
        api.create_namespaced_daemon_set(
            namespace="otel-logs",
            body=daemonset_manifest
        )
        logger.info("Deployed OpenTelemetry 1.20 Collector DaemonSet")
    except ApiException as e:
        logger.error(f"Failed to deploy DaemonSet: {e}")
        raise

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Deploy OpenTelemetry 1.20 Collector to Kubernetes")
    parser.parse_args()  # No additional args for simplicity

    try:
        load_kube_config()
        api = client.AppsV1Api()
        core_api = client.CoreV1Api()

        create_otel_namespace(core_api)
        deploy_collector_daemonset(api)

        logger.info("Step 1 complete: OpenTelemetry 1.20 Collector deployed successfully")
    except Exception as e:
        logger.error(f"Step 1 failed: {e}")
        exit(1)

// otel_log_processor.go
// Custom OpenTelemetry Log Processor to enrich logs with trace context and normalize severity
// Compatible with OpenTelemetry Collector 1.20.0 Contrib SDK
// Build: go build -o otel-log-processor otel_log_processor.go

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "strings"

    "go.opentelemetry.io/collector/component"
    "go.opentelemetry.io/collector/consumer"
    "go.opentelemetry.io/collector/logs"
    "go.opentelemetry.io/collector/pdata/plog"
    "go.opentelemetry.io/collector/service"
    "go.opentelemetry.io/collector/service/featuregate"
)

// logProcessor enriches logs with trace ID, span ID, and normalizes severity levels
type logProcessor struct {
    config component.Config
}

// newLogProcessor creates a new instance of the custom log processor
func newLogProcessor(cfg component.Config) (component.Component, error) {
    if cfg == nil {
        return nil, fmt.Errorf("nil config provided to log processor")
    }
    return &logProcessor{config: cfg}, nil
}

// processLogs implements the logs.LogProcessor interface
func (p *logProcessor) processLogs(ctx context.Context, ld plog.Logs) (plog.Logs, error) {
    // Iterate over all resource logs, scope logs, and log records
    rls := ld.ResourceLogs()
    for i := 0; i < rls.Len(); i++ {
        rl := rls.At(i)
        sls := rl.ScopeLogs()
        for j := 0; j < sls.Len(); j++ {
            sl := sls.At(j)
            lr := sl.LogRecords()
            for k := 0; k < lr.Len(); k++ {
                logRecord := lr.At(k)

                // Normalize severity: map custom severity strings to OpenTelemetry standard levels
                rawSeverity := logRecord.Body().AsString()
                if strings.Contains(strings.ToLower(rawSeverity), "err") {
                    logRecord.SetSeverityNumber(plog.SeverityNumberError)
                    logRecord.SetSeverityText("ERROR")
                } else if strings.Contains(strings.ToLower(rawSeverity), "warn") {
                    logRecord.SetSeverityNumber(plog.SeverityNumberWarn)
                    logRecord.SetSeverityText("WARN")
                } else {
                    logRecord.SetSeverityNumber(plog.SeverityNumberInfo)
                    logRecord.SetSeverityText("INFO")
                }

                // Enrich with trace context if present in attributes
                attrs := logRecord.Attributes()
                if traceID, ok := attrs.Get("trace_id"); ok {
                    logRecord.SetTraceID(traceID.Bytes())
                }
                if spanID, ok := attrs.Get("span_id"); ok {
                    logRecord.SetSpanID(spanID.Bytes())
                }
            }
            }
        }
    }
    return ld, nil
}

// main initializes and runs the custom processor as part of the Collector
func main() {
    // Enable feature gates for 1.20.0 compatibility
    featuregate.GlobalRegistry().Set("telemetry.useOtelStartEndTime", true)

    // Register the custom processor with the Collector
    factory := component.NewProcessorFactory(
        "custom.logprocessor",
        func() component.Config { return &struct{}{} },
        component.WithLogsProcessor(newLogProcessor),
    )

    // Initialize Collector service with the custom processor
    srv, err := service.New(service.Config{
        Components: component.Components{
            Processors: map[component.Type]component.ProcessorFactory{
                factory.Type(): factory,
            },
        },
    })
    if err != nil {
        log.Fatalf("Failed to create Collector service: %v", err)
    }

    // Start the service
    if err := srv.Start(context.Background()); err != nil {
        log.Fatalf("Failed to start Collector service: %v", err)
    }

    // Keep the process running
    select {}
}

# claude_log_analyzer.py
# Python service to receive normalized logs from OpenTelemetry Collector, analyze with Claude 3.5 Sonnet
# Requires: anthropic>=0.39.0, grpcio, opentelemetry-api, opentelemetry-exporter-otlp
# Start: python claude_log_analyzer.py --port 50051 --anthropic-api-key $ANTHROPIC_API_KEY

import argparse
import json
import logging
import os
import signal
import sys
from concurrent import futures
from typing import List, Dict

import grpc
from anthropic import Anthropic, AnthropicError
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.logs.exporter import OTLPLogExporter
from opentelemetry.sdk.logs import LoggerProvider
from opentelemetry.sdk.logs.export import BatchLogProcessor
from log_analyzer_pb2_grpc import add_LogAnalyzerServicer_to_server
from log_analyzer_pb2 import AnalyzeLogsResponse

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Initialize Anthropic client with error handling
def init_anthropic_client(api_key: str) -> Anthropic:
    if not api_key:
        logger.error("ANTHROPIC_API_KEY environment variable not set")
        sys.exit(1)
    try:
        client = Anthropic(api_key=api_key)
        # Test API connectivity with a minimal call
        client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=10,
            messages=[{"role": "user", "content": "test"}]
        )
        logger.info("Anthropic client initialized, Claude 3.5 Sonnet accessible")
        return client
    except AnthropicError as e:
        logger.error(f"Failed to initialize Anthropic client: {e}")
        sys.exit(1)

# Process batch of logs with Claude 3.5
def analyze_logs(client: Anthropic, logs: List[Dict]) -> Dict:
    """Send log batch to Claude 3.5 for root cause analysis, return structured report"""
    # Compress logs to fit within 200k token context window (approx 150k words)
    log_batch = json.dumps(logs, indent=2)
    if len(log_batch) > 150000:
        logger.warning(f"Log batch size {len(log_batch)} exceeds recommended 150k chars, truncating...")
        log_batch = log_batch[:150000]

    prompt = f"""
    You are a senior site reliability engineer analyzing application logs.
    Analyze the following log batch, identify anomalies, root causes, and suggest remediation steps.
    Return your response as a JSON object with keys: incident_id, severity, root_cause, affected_components, remediation_steps.

    Log Batch:
    {log_batch}
    """

    try:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}]
        )
        # Parse Claude's response, handle non-JSON responses
        response_text = response.content[0].text
        # Extract JSON from response (Claude sometimes wraps in markdown)
        if "" in response_text:
            response_text = response_text.split("")[1].split("")[0].strip()
        report = json.loads(response_text)
        logger.info(f"Generated analysis report for incident {report.get('incident_id', 'unknown')}")
        return report
    except AnthropicError as e:
        logger.error(f"Claude API call failed: {e}")
        return {"error": str(e), "severity": "CRITICAL", "root_cause": "Claude API failure"}
    except json.JSONDecodeError as e:
        logger.error(f"Failed to parse Claude response as JSON: {e}")
        return {"error": "Invalid Claude response", "raw_response": response_text}

# gRPC server to receive logs from OpenTelemetry Collector
class LogAnalyzerServicer:
    def __init__(self, anthropic_client: Anthropic):
        self.anthropic_client = anthropic_client
        self.logger_provider = LoggerProvider()
        self.log_exporter = OTLPLogExporter(endpoint="otel-collector.otel-logs:4317")
        self.logger_provider.add_log_processor(BatchLogProcessor(self.log_exporter))
        self.logger = self.logger_provider.get_logger("claude-log-analyzer")

    def AnalyzeLogs(self, request, context):
        """gRPC method to receive log batches and return analysis reports"""
        try:
            log_batch = json.loads(request.log_batch)
            report = analyze_logs(self.anthropic_client, log_batch)
            # Log the report for audit
            self.logger.emit(
                body=f"Analysis report generated: {json.dumps(report)}",
                severity=plog.SeverityNumberInfo
            )
            return AnalyzeLogsResponse(report=json.dumps(report))
        except Exception as e:
            logger.error(f"Failed to process log batch: {e}")
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(str(e))
            return AnalyzeLogsResponse(report=json.dumps({"error": str(e)}))

def serve(port: int, anthropic_client: Anthropic):
    """Start gRPC server with graceful shutdown"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    add_LogAnalyzerServicer_to_server(LogAnalyzerServicer(anthropic_client), server)
    server.add_insecure_port(f"[::]:{port}")
    server.start()
    logger.info(f"Claude log analyzer service started on port {port}")

    # Graceful shutdown on SIGINT/SIGTERM
    signal.signal(signal.SIGINT, lambda *args: server.stop(5))
    signal.signal(signal.SIGTERM, lambda *args: server.stop(5))
    server.wait_for_termination()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Claude 3.5 Log Analysis Service")
    parser.add_argument("--port", type=int, default=50051, help="gRPC port to listen on")
    parser.add_argument("--anthropic-api-key", type=str, default=os.getenv("ANTHROPIC_API_KEY"), help="Anthropic API key")
    args = parser.parse_args()

    anthropic_client = init_anthropic_client(args.anthropic_api_key)
    serve(args.port, anthropic_client)

# filter-processor-config.yaml
processors:
  filter:
    logs:
      - record:
          - attributes["log.level"] == "DEBUG"
          - body matches "(?i)health check"
          - attributes["endpoint"] == "/health"

# sampler-config.yaml
processors:
  probabilistic_sampler:
    logs:
      - severity: INFO
        sampling_percentage: 10
      - severity: WARN
        sampling_percentage: 1
      - severity: ERROR
        sampling_percentage: 100
      - severity: CRITICAL
        sampling_percentage: 100

import tenacity

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_exponential(multiplier=1, min=1, max=4),
    retry=tenacity.retry_if_exception_type(AnthropicError),
    before_sleep=lambda retry_state: logger.warning(f"Retrying Claude API call, attempt {retry_state.attempt_number}")
)
def analyze_logs_with_retry(client: Anthropic, logs: List[Dict]) -> Dict:
    return analyze_logs(client, logs)

ai-log-analysis-otel-claude/
├── deploy/
│   ├── otel-collector/
│   │   ├── daemonset.yaml
│   │   ├── config.yaml
│   │   └── processor.yaml
│   ├── kubernetes/
│   │   ├── claude-analyzer-deployment.yaml
│   │   └── grafana-dashboard.json
│   └── standalone/
│       ├── otel-collector-config.yaml
│       └── run-analyzer.sh
├── src/
│   ├── go/
│   │   └── otel-log-processor/
│   │       ├── go.mod
│   │       ├── go.sum
│   │       └── otel_log_processor.go
│   └── python/
│       ├── claude-log-analyzer/
│       │   ├── requirements.txt
│       │   └── claude_log_analyzer.py
│       └── deploy-scripts/
│           └── deploy_otel_collector.py
├── benchmarks/
│   ├── triage-time-comparison.csv
│   └── cost-analysis.xlsx
└── README.md