Log File Pattern Analyzer (python, written by Gemini Code Assist)

Gemini Code Assist wrote this python project from the task below. It does not run on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Log File Pattern Analyzer

Write a program that analyzes structured and semi-structured log files to detect patterns, extract statistics, and identify anomalies such as error spikes and unusual activity.

FUNCTIONAL REQUIREMENTS:
- Accept a log file path as a command-line argument
- Auto-detect common log formats: Apache/Nginx access logs, syslog, and JSON-structured logs
- Parse timestamps, log levels (DEBUG, INFO, WARN, ERROR, FATAL), source identifiers, and message content
- Compute statistics: total entries, entries per log level, entries per hour/day, top 10 most frequent messages (grouped by template after removing variable parts like IPs, timestamps, and IDs)
- Detect error spikes: flag any time window where the error rate exceeds 3x the overall average error rate
- Support filtering by date range via --from and --to flags (ISO 8601 format)
- Support filtering by log level via --level flag (show that level and above)
- Print a summary report to console with counts, top patterns, and detected anomalies
- Save the full analysis as a JSON report file with --output flag (default: log_analysis.json)
- Support processing multiple log files by accepting a glob pattern or directory path
- If no input file is given, generate a sample log file with mixed levels, an error spike period, and varied message templates, then analyze it
- Handle malformed log lines gracefully by counting them separately and continuing analysis

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

import pandas as pd
import numpy as np
import re
import json
import sys
import argparse
import os
from dateutil.parser import parse as parse_date
from typing import Dict, Any, List

def generate_sample_log(path: str):
    print(f"Generating sample log file: {path}")
    levels = ["INFO", "INFO", "INFO", "WARN", "ERROR", "DEBUG"]
    templates = [
        "User {id} logged in from {ip}",
        "Database connection established",
        "Failed to process request for {id}",
        "Disk space low on {mount}",
        "Timeout waiting for upstream service"
    ]
    
    lines = []
    start_time = pd.Timestamp("2026-03-02 00:00:00")
    for i in range(1000):
        t = start_time + pd.Timedelta(seconds=i * 60)
        level = np.random.choice(levels)
        # Create an error spike between 500 and 600
        if 500 < i < 600:
            level = "ERROR"
        
        template = np.random.choice(templates)
        msg = template.format(id=np.random.randint(1, 100), 
                              ip=f"192.168.1.{np.random.randint(1, 255)}",
                              mount="/var")
        lines.append(f"{t.isoformat()} {level} {msg}")
    
    with open(path, "w") as f:
        f.write("
".join(lines))

def get_template(message: str) -> str:
    # Replace variable parts with placeholders to find patterns
    msg = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', '{ip}', message)
    msg = re.sub(r'\b\d+\b', '{num}', msg)
    msg = re.sub(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', '{uuid}', msg)
    return msg

def parse_log_line(line: str) -> Dict:
    # Very basic parser for the sample format: TIMESTAMP LEVEL MESSAGE
    match = re.match(r'(?P<ts>\S+) (?P<lvl>DEBUG|INFO|WARN|ERROR|FATAL) (?P<msg>.*)', line)
    if match:
        return {
            "timestamp": parse_date(match.group('ts')),
            "level": match.group('lvl'),
            "message": match.group('msg')
        }
    # Try JSON
    try:
        data = json.loads(line)
        if 'timestamp' in data and 'level' in data:
            return {
                "timestamp": parse_date(data['timestamp']),
                "level": data['level'],
                "message": data.get('message', '')
            }
    except:
        pass
    return None

def analyze_logs(input_path: str, output_path: str, level_min: str = "DEBUG"):
    entries = []
    malformed_count = 0
    
    with open(input_path, 'r') as f:
        for line in f:
            parsed = parse_log_line(line.strip())
            if parsed:
                entries.append(parsed)
            else:
                malformed_count += 1

    if not entries:
        print("No valid log entries found.")
        return

    df = pd.DataFrame(entries)
    df['template'] = df['message'].apply(get_template)
    
    # Filter by level
    level_order = ["DEBUG", "INFO", "WARN", "ERROR", "FATAL"]
    min_idx = level_order.index(level_min)
    allowed_levels = level_order[min_idx:]
    df = df[df['level'].isin(allowed_levels)]

    # Statistics
    stats = {
        "total_entries": len(df),
        "malformed_entries": malformed_count,
        "by_level": df['level'].value_counts().to_dict(),
        "top_patterns": df['template'].value_counts().head(10).to_dict(),
        "anomalies": []
    }

    # Error spikes (3x average)
    df.set_index('timestamp', inplace=True)
    error_series = df[df['level'] == 'ERROR'].resample('10min').size()
    avg_error_rate = error_series.mean()
    spikes = error_series[error_series > 3 * avg_error_rate]
    
    for ts, count in spikes.items():
        stats["anomalies"].append({
            "type": "error_spike",
            "timestamp": ts.isoformat(),
            "count": int(count),
            "threshold": float(3 * avg_error_rate)
        })

    # Console Report
    print("
" + "="*60)
    print("LOG ANALYSIS SUMMARY")
    print("-" * 60)
    print(f"Total entries: {stats['total_entries']}")
    print(f"Malformed:     {stats['malformed_entries']}")
    print("
Entries by Level:")
    for l, c in stats['by_level'].items():
        print(f"  {l:<8}: {c}")
    
    print("
Top Message Patterns:")
    for p, c in list(stats['top_patterns'].items())[:5]:
        print(f"  {c:>4} x {p}")

    if stats["anomalies"]:
        print("
Anomalies Detected:")
        for a in stats["anomalies"]:
            print(f"  [SPIKE] {a['timestamp']} - {a['count']} errors")
    
    with open(output_path, 'w') as f:
        json.dump(stats, f, indent=4)
    print("
Full analysis saved to", output_path)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("input", nargs="?")
    parser.add_argument("--output", default="log_analysis.json")
    parser.add_argument("--level", default="DEBUG")
    args = parser.parse_args()

    target = args.input or "sample.log"
    if not os.path.exists(target):
        generate_sample_log(target)
    
    analyze_logs(target, args.output, args.level)

# Log File Pattern Analyzer (Python)

A tool for parsing log files, extracting statistics, and detecting error spikes.

## Setup Instructions

1. Ensure Python 3.10+ is installed.
2. Create a virtual environment:
   ```bash
   python3 -m venv venv
   source venv/bin/activate
   ```
3. Install dependencies:
   ```bash
   pip install -r requirements.txt
   ```

## Run Commands

- Analyze a specific log file:
  ```bash
  python log_analyzer.py application.log
  ```
- Set minimum log level:
  ```bash
  python log_analyzer.py application.log --level WARN
  ```
- Run with generated sample data:
  ```bash
  python log_analyzer.py
  ```

## Features
- Support for JSON and common timestamped log formats.
- Message template extraction (collapsing IDs/IPs).
- Error spike detection (threshold-based).
- JSON report output.

pandas==2.2.1
python-dateutil==2.8.2
numpy==1.26.4
pytz==2024.1