Log Analytics with Rugo

gistfile1.txt

#!/usr/bin/env python
"""
Benchmark: Searching and aggregating web traffic logs.

Use case: Monitor opteryx.app by calculating rolling 1-hour average
          response times for the /api/data endpoint.

Compares three approaches on a 1GB+ JSONL log file:
  1. grep + Python (fast filter, manual parsing for stats)
  2. pandas (loads everything into memory)
  3. Rugo (selective streaming with column projection + predicates)

Measures: time, peak memory usage, and correctness.
"""

import json
import os
import random
import resource
import subprocess
import sys
import time
from datetime import datetime, timedelta

sys.path.insert(1, os.path.join(sys.path[0], ".."))

# ============================================================
# Configuration
# ============================================================

LOG_FILE = "/tmp/rugo_benchmark_logs.jsonl"
TARGET_SIZE = 1_100_000_000     # ~1.1 GB

# Realistic URL distribution for opteryx.app
# /api/data gets ~15% of traffic - selective but not contrived
URLS = {
       "/api/data": 0.15,
       "/api/query": 0.20,
       "/api/health": 0.10,
       "/api/status": 0.05,
       "/dashboard": 0.10,
       "/static/main.js": 0.10,
       "/static/styles.css": 0.05,
       "/blog/performance": 0.05,
       "/blog/rugo": 0.05,
       "/docs/install": 0.05,
       "/api/login": 0.05,
       "/api/register": 0.05,
}

METHODS = ["GET", "POST", "PUT"]
METHOD_WEIGHTS = [70, 20, 10]
STATUS_CODES = [200, 200, 200, 200, 200, 304, 404, 500]

USER_AGENTS = [
       "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
       "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
       "curl/7.88.1",
       "python-requests/2.31.0",
]

# ============================================================
# Helpers
# ============================================================

def peak_memory_mb():
     """Return peak RSS in MB (macOS/Linux)."""
     usage = resource.getrusage(resource.RUSAGE_SELF)
     # ru_maxrss is KB on Linux, bytes on macOS
     val = usage.ru_maxrss
     if sys.platform == "darwin":
         return val / (1024 * 1024)     # bytes -> MB
     return val / 1024   # KB -> MB

def rolling_1hr_avg(timestamps_iso, values):
     """Sliding-window 1-hour average. O(n)."""
    timestamps = [datetime.fromisoformat(ts) for ts in timestamps_iso]
    
    avgs = []
    window_start = 0
    window_sum = 0.0
    
    for i in range(len(timestamps)):
        window_sum += values[i]
        
        while timestamps[i] - timestamps[window_start] > timedelta(hours=1):
            window_sum -= values[window_start]
            window_start += 1
        
        window_len = i - window_start + 1
        if window_len > 0:
            avgs.append(window_sum / window_len)
    
    return avgs

# ============================================================
# 1. Generate synthetic log data
# ============================================================

def generate_logs():
    print("Generating synthetic log file (~1.1 GB)...")
    t0 = time.time()

    random.seed(42)
    base_time = datetime(2026, 6, 1)
    count = 0
    size = 0

    with open(LOG_FILE, "w") as f:
        ts = base_time
        while size < TARGET_SIZE:
            ts += timedelta(seconds=random.randint(1, 30))
            
            url = random.choices(list(URLS.keys()), weights=list(URLS.values()))[0]
            method = random.choices(METHODS, weights=METHOD_WEIGHTS)[0]
            status = random.choice(STATUS_CODES)
            
            if status == 500:
                rt = round(random.uniform(5000, 15000), 2)
            elif "/api/" in url:
                rt = round(random.uniform(50, 500), 2)
            else:
                rt = round(random.uniform(10, 100), 2)
            
            entry = json.dumps({
                   "timestamp": ts.isoformat(),
                   "url": url,
                   "method": method,
                   "status": status,
                   "response_ms": rt,
                   "user_agent": random.choice(USER_AGENTS),
                   "bytes_sent": random.randint(500, 100000),
               }) + "\n"
            
            f.write(entry)
            size += len(entry)
            count += 1
            
            if count % 1_000_000 == 0:
                print(f"    {count:,} entries, {size / 1024 / 1024:.0f} MB")

    elapsed = time.time() - t0
    print(f"    {count:,} entries in {size / 1024 / 1024:.0f} MB ({elapsed:.0f}s)")
    return count

# ============================================================
# 2. Approach: grep + Python
# ============================================================

def run_grep():
    print("\n--- grep + Python (filter then parse) ---")
    
    base_mem = peak_memory_mb()
    t0 = time.time()
    
    result = subprocess.run(
           ["grep", "/api/data", LOG_FILE],
        capture_output=True,
      )
    
    entries = []
    for line in result.stdout.decode().split("\n"):
        if line:
            entry = json.loads(line)
            if entry["url"] == "/api/data":
                entries.append(entry)
    
    timestamps = [e["timestamp"] for e in entries]
    values = [e["response_ms"] for e in entries]
    
    avgs = rolling_1hr_avg(timestamps, values)
    elapsed = time.time() - t0
    
    peak = peak_memory_mb() - base_mem
    overall = sum(avgs) / len(avgs) if avgs else 0
    print(f"   Matched: {len(entries):,} entries")
    print(f"   Peak memory: {peak:.0f} MB")
    print(f"   Overall avg response time: {overall:.2f} ms   ({elapsed:.2f}s)")
    return {"time": elapsed, "entries": len(entries), "overall_avg": overall, "memory_mb": peak}

# ============================================================
# 3. Approach: pandas (load all)
# ============================================================

def run_pandas():
    print("\n--- pandas (load all into memory) ---")
    
    base_mem = peak_memory_mb()
    t0 = time.time()
    
    import pandas as pd
    
    df = pd.read_json(LOG_FILE, lines=True)
    df_api = df[df["url"] == "/api/data"]
    df_api = df_api.sort_values("timestamp")
    
    df_api["ts"] = pd.to_datetime(df_api["timestamp"])
    df_api = df_api.set_index("ts")
    df_api["rolling_avg"] = df_api["response_ms"].rolling("1h").mean()
    
    avgs = df_api["rolling_avg"].dropna().tolist()
    elapsed = time.time() - t0
    
    peak = peak_memory_mb() - base_mem
    overall = sum(avgs) / len(avgs) if avgs else 0
    print(f"   Matched: {len(df_api):,} entries")
    print(f"   Peak memory: {peak:.0f} MB")
    print(f"   Overall avg response time: {overall:.2f} ms   ({elapsed:.2f}s)")
    return {"time": elapsed, "entries": len(df_api), "overall_avg": overall, "memory_mb": peak}

# ============================================================
# 4. Approach: Rugo (selective streaming)
# ============================================================

def run_rugo():
    print("\n--- Rugo (selective streaming) ---")
    
    base_mem = peak_memory_mb()
    t0 = time.time()
    
    from rugo import jsonl
    
    timestamps = []
    values = []
    
    with jsonl.read_jsonl(
           LOG_FILE,
        columns=["url", "response_ms", "timestamp"],
        predicates=[("url", "==", "/api/data")],
      ) as reader:
        for morsel in reader:
            timestamps.extend(morsel.column("timestamp").to_pylist())
            values.extend(morsel.column("response_ms").to_pylist())
    
    avgs = rolling_1hr_avg(timestamps, values)
    elapsed = time.time() - t0
    
    peak = peak_memory_mb() - base_mem
    overall = sum(avgs) / len(avgs) if avgs else 0
    print(f"   Matched: {len(timestamps):,} entries")
    print(f"   Peak memory: {peak:.0f} MB")
    print(f"   Overall avg response time: {overall:.2f} ms   ({elapsed:.2f}s)")
    return {"time": elapsed, "entries": len(timestamps), "overall_avg": overall, "memory_mb": peak}

# ============================================================
# Main
# ============================================================

if __name__ == "__main__":
    total = generate_logs()
    
    expected_api_data = total * 0.15
    print(f"\nExpected /api/data entries: ~{expected_api_data:,.0f} ({expected_api_data/total*100:.0f}%)")
    
    results = {}
    
    try:
        results["grep"] = run_grep()
    except Exception as e:
        print(f"   FAILED: {e}")
        import traceback
        traceback.print_exc()
    
    try:
        results["pandas"] = run_pandas()
    except Exception as e:
        print(f"   FAILED: {e}")
        import traceback
        traceback.print_exc()
    
    try:
        results["rugo"] = run_rugo()
    except Exception as e:
        print(f"   FAILED: {e}")
        import traceback
        traceback.print_exc()
    
    # Summary table
    print("\n" + "=" * 80)
    print("RESULTS")
    print("=" * 80)
    print(f"{'Method':<12} {'Time (s)':<12} {'Memory (MB)':<14} {'Entries':<15} {'Avg (ms)':<12}")
    print("-" * 80)
    for name in ["grep", "pandas", "rugo"]:
        r = results[name]
        print(f"{name:<12} {r['time']:<12.2f} {r['memory_mb']:<14.0f} {r['entries']:<15,} {r['overall_avg']:<12.2f}")
    
    # Verify consistency
    if len(results) >= 2:
        avgs = [r["overall_avg"] for r in results.values()]
        print(f"\nAll overall averages within {max(avgs) - min(avgs):.2f} ms")
    
    print(f"\nCleaning up {LOG_FILE}...")
    os.remove(LOG_FILE)
    print("Done.")