🔍 Kafka Log Analyzer

AI-powered Kafka log analysis — intelligent diagnostics for Kafka clusters with Claude Code

中文文档

🎯 The Problem It Solves

| Scenario | Traditional Debugging | Kafka Log Analyzer | |----------|----------------------|-------------------| | Log Analysis | Manually grep thousands of lines | AI extracts events, detects anomalies, prioritizes issues | | Consumer Lag | Check multiple dashboards | Single command shows lag across all consumer groups | | Error Diagnosis | Correlate logs with source code manually | AI links errors to configuration and implementation | | Incident Response | Copy-paste logs to Slack/Feishu | Auto-generate formatted reports ready to share |

Core insight: Kafka troubleshooting requires understanding log semantics, not just keyword matching.

✨ Core Features

📊 Intelligent Log Analysis (v0.1.0)

Structured parsing with zero configuration:

• Multi-format Support — Parse both text and JSON Kafka logs
• 11 Event Types — send_success, send_failure, consumer_lag, rebalance, commit_failure, buffer_exhausted, leader_change, offset_out_of_range, serialization_error, network_error, auth_error
• Anomaly Detection — 7 built-in detectors for error spikes, rebalance storms, lag spikes, leader instability, replica lag, serialization issues, network problems

🔎 Enhanced Filtering & Reports (v0.2.0)

Parameterized control over analysis output:

• --focus Filter — Filter by component (producer/consumer/broker) or semantics (lag/error)
• --priority Filter — Show only anomalies at specified severity (P0-P3)
• --timeline Analysis — Time-windowed event distribution with component breakdown
• --report Formats — Unified markdown/json/slack output for CLI and MCP
• --debug Mode — Verbose logging for troubleshooting

📡 Prometheus & Data Source Integration (v0.3.0)

Real-time metrics from Prometheus/Kafka Exporter:

• Prometheus Connector — Query Kafka Exporter metrics via PromQL (queryInstant, queryRange)
• Consumer Lag Monitoring — Real-time kafka_consumergroup_lag from Prometheus with group/topic/partition filtering
• Exporter Source Analysis — analyze_log --source exporter fetches live metrics and detects anomalies from lag data
• Loki Log Source — analyze_log --source loki queries Kafka logs via LogQL and reuses the Python analysis pipeline
• MCP Resources — Subscribe to live metric streams (kafka://metrics/{cluster}, kafka://lag/{cluster})
• Graceful Degradation — All data source tools return empty data + friendly warnings when Prometheus/Loki are unreachable
• Config Validators — isPrometheusConfigured() / isLokiConfigured() / validateSourceConfig() helpers

🔔 Hook Automation & Push Integration (v0.4.0)

Automatic alert triggering and team notifications:

• Grafana Alert Hook — Receive Grafana webhook payloads, auto-trigger analysis
• PagerDuty Alert Hook — Parse PagerDuty Events API v2 alerts into normalized analysis triggers
• Hook HTTP Server — Lightweight server with POST /hooks/grafana, POST /hooks/pagerduty, GET /hooks/health
• Hook Dedup Engine — Fingerprint-based in-memory dedup with configurable TTL window and merge count
• Feishu Webhook Push — Send analysis results as interactive cards to Feishu groups
• Slack Webhook Push — Send results as Block Kit messages to Slack channels
• JIRA Integration — Auto-create JIRA issues from analysis with priority mapping
• Push Dispatcher — Parallel push to all configured targets (Feishu/Slack/JIRA)
• CLI: hooks — npx kafka-log-analyze hooks [--port P] to start the hook server
• CLI: push-test — npx kafka-log-analyze push-test [--target feishu|slack|all] to test push

📜 Historical Trends & Analysis (v0.5.0)

Persistent storage and trend comparison:

• SQLite Persistence — Automatic database creation with WAL mode, storing all analysis records, anomaly details, lag snapshots, and baselines
• Event-Driven Architecture — Decoupled pipeline: analysis results trigger persistence and automatic baseline updates via EventEmitter
• Historical Query — query_history tool filters past analysis by time range, source, and cluster
• Trend Comparison — compare_trend tool compares metrics (error rate, lag, anomaly count) across configurable time windows (1h/6h/1d/7d) with previous/last_week/last_month comparison
• Anomaly Multiplier Detection — Automatically annotates anomalies when values are ≥ 2× baseline (e.g. "比基线多 3.5 倍")
• Baseline Management — Auto baselines from 7-day rolling averages; manual overrides via set_baseline / list_baselines tools
• Data Cleanup — Configurable retention policy (default 30 days) with cleanup_storage tool and automatic daily cleanup

🔒 Security Hardening (v1.0.0)

Production-ready security and reliability:

• Webhook URL Validation — SSRF protection: enforces HTTPS and rejects private IP ranges for Feishu and Slack integrations
• Path Traversal Prevention — sanitizePath() validates file paths, blocks ../ and null byte injection
• SQLite Parameter Safety — Full audit confirms all queries use prepared statements (no SQL injection)
• Secret Redaction in Logs — redact() function automatically masks sensitive values (token, secret, password, key, auth, credential)
• Test Coverage ≥85% — MCP server integration, event listeners, CLI, dedup, get_lag, JIRA, security, benchmark tests
• Stdin Piping — Paste mode pipes content via child.stdin instead of temp file I/O for better performance

🌏 User Experience Optimization (v1.1.0)

Internationalization, visualization, and smart diagnostics:

• i18n Support (EN/ZH) — Full internationalization with --locale / -l flag or LOCALE env var (~180 keys each). All user-visible strings use t() function.
• In-Memory Analysis Cache — TTL-based cache with per-source expiry (paste/file: 24h, exporter: 5m, loki: 15m). Prevents redundant re-analysis of identical inputs.
• ASCII Timeline Chart — Stacked bar chart with severity-segmented bars (█ ERROR, · WARN, ░ INFO) and auto-generated legend. Sparkline mini-charts in push notifications.
• Shell Completion — npx kafka-log-analyze completion <bash|zsh|fish> generates shell completion scripts with enum suggestions for flags.
• MCP Schema Enhancement — Dynamic analyze_log inputSchema with cluster/consumer_group/topic enums auto-populated from history.
• Folded Markdown Reports — New --report folded-markdown format with collapsible <details>/<summary> blocks.
• Interactive TUI Browser — --interactive / -i flag for keyboard-driven section navigation (↑/↓, Enter/Space, e, c, q). Zero additional dependencies.
• Diagnostic Templates — 5 built-in templates (lag-diagnosis, rebalance-storm, producer-errors, broker-health, full-audit). --template <id> flag and templates CLI command.
• Smart Template Recommendation — Templates ranked by historical anomaly patterns via recommendTemplates(). New MCP tools diagnose and list_templates.

🎯 Priority Classification

Automatic severity grading (P0-P3):

| Priority | Severity | Examples | |----------|----------|----------| | 🟢 P0 (Critical) | Cluster down, data loss risk | Complete broker failure, unrecoverable corruption | | 🟡 P1 (High) | Consumer lag > 10K, frequent rebalances | Sustained high lag, rebalance storms | | 🟠 P2 (Medium) | Leader changes, transient errors | Leader election, temporary network issues | | 🔴 P3 (Low) | Warnings, informational events | Configuration warnings, debug logs |

🔄 Multiple Output Formats

• Markdown — Detailed report with sections and recommendations
• JSON — Structured data for programmatic processing
• Slack — Compact format optimized for team channels
• Folded Markdown — Collapsible sections with <details>/<summary> for GitHub/Slack

🖥️ Claude Code Integration

• MCP Tools — Call analysis from Claude Code or any MCP client
• Slash Commands — /kafka-analyze for quick analysis
• Source Code Linking — AI correlates errors with your Kafka configuration

🚀 Getting Started

Option 1: Claude Code Plugin (Recommended)

This project is a Claude Code Plugin. Install via marketplace for one-click setup.

Method A: Plugin Marketplace (Recommended)

# In Claude Code, run:
/plugin marketplace add saqqdy/kafka-log-analyzer
/plugin install kafka-log-analyzer

Method B: Local Install

# 1. Go to your project
cd your-project

# 2. Install npm package
pnpm add -D kafka-log-analyzer

# 3. Copy plugin files
mkdir -p .claude/skills
cp -r node_modules/kafka-log-analyzer/.claude/skills/kafka-log-analyzer .claude/skills/

Available Commands

Type these commands in Claude Code:

| Command | Description | Example | |---------|-------------|---------| | /kafka-analyze | Analyze Kafka logs | /kafka-analyze --source file --path server.log |

Output Example

/kafka-analyze --source file --path server.log

📊 Analyzing server.log...

📋 Summary:
  Total Events: 847
  P0 (Critical): 3
  P1 (High): 12

⚠️ Anomalies:
  🔴 send_failure_spike (P0)
     → Error rate 15% exceeds threshold (5%)
     → Check broker availability and network connectivity

  🟠 consumer_lag (P1)
     → Consumer group order-processor lag: 15,000 messages
     → Consider scaling consumer instances

💡 Recommendations:
  1. Check broker availability
  2. Review consumer group configuration

Option 2: Programmatic Usage

pnpm add kafka-log-analyzer

import { analyzeLog } from 'kafka-log-analyzer'

const result = await analyzeLog({
  source: 'file',
  path: '/var/log/kafka/server.log',
  focus: ['producer', 'error'],
  timeline: '1h'
})

console.log(`Total Events: ${result.summary.total}`)
console.log(`P0: ${result.summary.byPriority.P0}`)

Option 3: CLI (Zero-Install)

# Analyze log file
npx kafka-log-analyze analyze /var/log/kafka/server.log

# Paste log content via stdin
cat error.log | npx kafka-log-analyze analyze-paste

# Fetch consumer lag from Prometheus
npx kafka-log-analyze lag --cluster production --group order-processor

# Analyze from Kafka Exporter via Prometheus
npx kafka-log-analyze analyze-exporter --cluster production

# Analyze from Loki log system
npx kafka-log-analyze analyze-loki --cluster production --limit 500

# Start hook server for receiving Grafana/PagerDuty alerts
npx kafka-log-analyze hooks --port 3100

# Test push integration
npx kafka-log-analyze push-test --target feishu

# Generate shell completion script
npx kafka-log-analyze completion bash > /etc/bash_completion.d/kafka-log-analyze

# List diagnostic templates
npx kafka-log-analyze templates

# Analyze with a diagnostic template
npx kafka-log-analyze analyze server.log --template lag-diagnosis

# Analyze with interactive TUI browser
npx kafka-log-analyze analyze server.log --interactive

# Use Chinese locale
npx kafka-log-analyze analyze server.log --locale zh

Option 4: Clone and Run Examples

git clone https://github.com/saqqdy/kafka-log-analyzer.git
cd kafka-log-analyzer
pnpm install

# Run examples
node dist/cli.js --source file --path tests/fixtures/sample-kafka-log.txt

📊 Basic Usage

1. Analyze Logs from File

# Analyze a Kafka log file
node dist/commands/kafka-analyze.js --source file --path /var/log/kafka/server.log

# With timeline analysis
node dist/commands/kafka-analyze.js --source file --path server.log --timeline 1h

# Focus on specific areas
node dist/commands/kafka-analyze.js --source file --path server.log --focus producer,consumer

2. Analyze Pasted Logs

# Paste logs via stdin
cat kafka-error.log | node dist/commands/kafka-analyze.js --source paste

# Or use heredoc
node dist/commands/kafka-analyze.js --source paste <<EOF
[2024-01-15 10:00:01] ERROR [producer] Failed to send record to topic orders
[2024-01-15 10:00:02] WARN  [consumer] lag exceeded threshold (5000 messages)
EOF

3. Use MCP Tools in Claude Code

// In Claude Code, call the analyze_log tool
{
  "source": "paste",
  "content": "[2024-01-15 10:00:01] ERROR [producer] Failed to send record...",
  "focus": ["producer", "error"],
  "timeline": "1h",
  "priority": ["P0", "P1"],
  "report": "markdown"
}

📋 Version Roadmap

| Version | Codename | Theme | Status | |---------|----------|-------|--------| | v0.1.0 | Sentinel | Core analysis engine + MCP tools | ✅ Released | | v0.2.0 | Watchtower | Enhanced commands + CLI improvements | ✅ Released | | v0.2.1 | — | Bug fixes + CI workflow hardening | ✅ Released | | v0.3.0 | Guardian | Prometheus + Kafka Exporter integration | ✅ Released | | v0.4.0 | Dispatcher | Grafana alerts + Feishu/Slack/JIRA push | ✅ Released | | v0.5.0 | Historian | Historical trends + baseline comparison | ✅ Released | | v1.0.0 | Architect | Production-ready release | ✅ Released | | v1.1.0 | Polyglot | User experience optimization (i18n, cache, charts, TUI, templates) | ✅ Released |

🗂️ Project Structure

kafka-log-analyzer/
├── .claude-plugin/
│   └── plugin.json              # Plugin manifest (required)
├── .claude/skills/
│   └── kafka-log-analyzer/      # Skill prompts (core product)
│       └── skill.md             # Commands + execution flow
├── src/
│   ├── index.ts                 # Public API exports
│   ├── cli.ts                   # CLI entry point
│   ├── types.ts                 # Core types
│   ├── mcp-server/              # MCP Server implementation
│   │   ├── index.ts             # Server entry
│   │   └── tools/
│   │       ├── analyze_log.ts   # Log analysis tool
│   │       └── get_lag.ts       # Consumer lag tool
│   ├── i18n/                   # Internationalization
│   │   ├── index.ts            # t(), setLocale(), getLocale()
│   │   ├── en.ts               # English dictionary
│   │   └── zh.ts               # Chinese dictionary
│   ├── cache/                  # In-memory analysis cache
│   │   └── index.ts            # MemoryCache + computeCacheKey()
│   ├── chart/                  # ASCII visualization
│   │   └── timeline.ts         # renderTimeline(), renderSparkline()
│   ├── completion/             # Shell completion & MCP schema
│   │   ├── shell.ts            # generateCompletion(shell)
│   │   └── schema.ts           # getEnhancedAnalyzeSchema()
│   ├── report/                 # Report formatters
│   │   ├── markdown.ts         # formatFoldedReport(), buildSections()
│   │   └── tui.ts              # launchTUI() interactive browser
│   ├── templates/              # Diagnostic templates
│   │   ├── presets.ts          # BUILT_IN_TEMPLATES, getTemplate()
│   │   └── recommend.ts        # recommendTemplates()
│   ├── sources/                 # Data source connectors
│   │   ├── prometheus.ts        # Prometheus API client
│   │   └── loki.ts              # Loki API client
│   ├── hooks/                  # Hook receivers & dedup
│   │   ├── grafana.ts          # Grafana alert parser
│   │   ├── pagerduty.ts        # PagerDuty alert parser
│   │   ├── dedup.ts            # Fingerprint dedup engine
│   │   └── server.ts           # Hook HTTP server
│   ├── push/                   # Push notification targets
│   │   ├── feishu.ts           # Feishu webhook push
│   │   ├── slack.ts            # Slack webhook push
│   │   ├── jira.ts             # JIRA issue creation
│   │   └── dispatch.ts         # Push dispatcher
│   ├── storage/                 # SQLite persistence & cleanup
│   │   ├── db.ts                # Database initialization
│   │   ├── repository.ts        # Data access layer
│   │   └── cleanup.ts           # Retention & auto-cleanup
│   ├── analysis/                # Trend & multiplier detection
│   │   ├── trend.ts             # Trend comparison logic
│   │   └── multiplier.ts        # Baseline multiplier detection
│   ├── events/                  # Event-driven architecture
│   │   ├── bus.ts               # EventEmitter event bus
│   │   └── listeners/           # Persistence & baseline listeners
│   └── utils/                   # Utilities
│       ├── config.ts            # Config management
│       └── format.ts            # Output formatting
├── scripts/                     # Python analysis scripts
│   └── parse_kafka_log.py       # Log parser
├── references/                  # Kafka reference docs
│   ├── kafka-patterns.md        # Common patterns
│   ├── error-codes.md           # Error code reference
│   └── tuning-guide.md          # Performance tuning
├── tests/                       # Test suites
│   └ fixtures/                  # Test data
└   └ analyze_log.test.ts        # Unit tests
└── docs/                        # VitePress docs

🛠️ Development

pnpm install          # Install dependencies
pnpm run lint         # ESLint + auto-fix
pnpm run typecheck    # TypeScript check
pnpm run test         # Run tests (vitest)
pnpm run build        # Build (ESM + CJS)
pnpm run docs:dev     # Start docs server

🆚 Comparison

vs Traditional Log Analysis

| Dimension | grep/awk | Kafka Log Analyzer | |-----------|----------|-------------------| | Output | Raw text lines | Structured Event[] + anomalies | | Anomalies | ❌ Manual spotting | ✅ 7 built-in detectors | | Priority | ❌ No | ✅ P0-P3 automatic grading | | Timeline | ❌ No | ✅ Time-window distribution | | Recommendations | ❌ No | ✅ Actionable fix suggestions | | Formats | ❌ Text only | ✅ Markdown, JSON, Slack |

🔧 Configuration

Create .env file from template:

cp .env.example .env

Environment Variables:

| Variable | Description | Default | |----------|-------------|---------| | PROMETHEUS_URL | Prometheus endpoint | http://localhost:9090 | | KAFKA_EXPORTER_URL | Kafka Exporter endpoint | http://localhost:9308 | | LOKI_URL | Loki log endpoint | http://localhost:3100 | | GRAFANA_WEBHOOK_PORT | Hook server port | 3100 | | FEISHU_WEBHOOK_URL | Feishu webhook URL | - | | SLACK_WEBHOOK_URL | Slack webhook URL | - | | JIRA_URL | JIRA base URL | - | | JIRA_TOKEN | JIRA API token | - | | JIRA_PROJECT_KEY | JIRA project key | - | | HOOK_DEDUP_WINDOW_MS | Dedup TTL (ms) | 300000 | | SQLITE_PATH | SQLite database path | ./storage/kafka-analyzer.db | | STORAGE_RETENTION_DAYS | Data retention in days | 30 | | STORAGE_AUTO_CLEANUP | Enable automatic daily cleanup | true | | LOG_LEVEL | Logging level | info |

🤝 Contributing

Contributions welcome! See Contributing Guide.

npm run dev          # Watch mode
npm test             # Run tests
npm run build        # Build

📚 Documentation

| Document | Description | |----------|-------------| | Architecture | Technical architecture | | API Reference | MCP Tools API | | Deployment | Deployment guide | | Contributing | Development guide |

📄 License

MIT © saqqdy

📚 Resources

• Kafka Documentation
• MCP Protocol
• Claude Code Plugins