API Replay Tester from Traffic Logs

API Replay Tester from Traffic Logs

Problem Statement

When you migrate an API to a new environment, refactor a backend service, or introduce a reverse proxy, you need confidence that the new stack responds identically to the old one. Unit and integration tests only cover the paths you thought to write — real production traffic covers the paths your users actually hit. Replaying recorded HTTP requests against a staging or canary target is one of the cheapest, highest-signal regression checks you can run.

This script parses an nginx/Apache-style access log (or a JSON-lines export), extracts the HTTP method, path, query string, and status code from each request, then re-issues those requests against a target host. It compares the new status code against the recorded one and reports drift, so you can catch broken routes, changed auth behavior, or 5xx regressions before they reach production.

How It Works

The architecture is a straightforward pipeline: parse → filter → replay → compare → report. Each stage is isolated so it can be swapped or extended independently.

1. Parser — A regex-based tokenizer handles the Combined Log Format ($remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$referer" "$http_user_agent"). Lines that don't match are skipped silently (they're often health-check noise or malformed entries).
2. Filter — Requests are filtered by HTTP method (default: GET/POST only), by status class (default: skip 5xx from the source since they're usually transient), and by an optional path prefix.
3. Replayer — A ThreadPoolExecutor fans requests out concurrently. Each worker uses a shared requests.Session per thread for connection reuse. Timeouts and a max body read cap keep runaway responses from blowing up memory.
4. Comparator — Status codes are compared exactly. Optionally, response body hashes can be compared (disabled by default since dynamic content like timestamps will always differ).
5. Reporter — Results are printed as a human-readable table plus a summary line. An optional JSON report is written for CI consumption.

Key design decisions: no external deps beyond requests; everything tunable via CLI flags or env vars; deterministic exit code (0 = clean, 1 = drift detected) so it drops into CI without wrappers.

Full Script

#!/usr/bin/env python3
"""
API Replay Tester from Traffic Logs

Parses an HTTP access log (nginx Combined Log Format or JSON-lines),
replays each request against a target host, and reports status-code drift.

Usage:
    python3 replay_test.py --traffic-log nginx_access.log --target http://staging:3000
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import sys
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Iterable, Iterator, Optional
from urllib.parse import urljoin, urlparse

import requests

# ---------------------------------------------------------------------------
# Configuration (env vars provide defaults; CLI flags override them)
# ---------------------------------------------------------------------------
DEFAULT_TARGET = os.environ.get("REPLAY_TARGET", "http://localhost:3000")
DEFAULT_TIMEOUT = float(os.environ.get("REPLAY_TIMEOUT", "10"))
DEFAULT_CONCURRENCY = int(os.environ.get("REPLAY_CONCURRENCY", "8"))
DEFAULT_METHODS = os.environ.get("REPLAY_METHODS", "GET,POST").split(",")
DEFAULT_MAX_BODY = int(os.environ.get("REPLAY_MAX_BODY", str(64 * 1024)))  # 64 KiB
USER_AGENT = os.environ.get("REPLAY_USER_AGENT", "api-replay-tester/1.0")

# nginx Combined Log Format:
# 127.0.0.1 - - [10/Oct/2024:13:55:36 +0000] "GET /api/v1/users?page=2 HTTP/1.1" 200 1234 "-" "curl/8.0"
COMBINED_RE = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[(?P<ts>[^\]]+)\] '
    r'"(?P<method>[A-Z]+) (?P<path>\S+) (?P<proto>[^"]+)" '
    r'(?P<status>\d{3}) (?P<bytes>\S+)'
)


@dataclass(frozen=True)
class LogEntry:
    """A single parsed request from the traffic log."""
    method: str
    path: str          # includes query string
    status: int        # status code recorded in the log
    source: str = "log"


@dataclass
class ReplayResult:
    """Outcome of replaying a single LogEntry."""
    entry: LogEntry
    new_status: Optional[int] = None
    error: Optional[str] = None
    body_hash: Optional[str] = None
    elapsed_ms: float = 0.0

    @property
    def drifted(self) -> bool:
        """True if the replayed status differs from the recorded one."""
        return self.error is not None or self.new_status != self.entry.status


# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------
def parse_combined_line(line: str) -> Optional[LogEntry]:
    """Parse one nginx Combined Log Format line. Returns None if unparseable."""
    m = COMBINED_RE.match(line.strip())
    if not m:
        return None
    try:
        return LogEntry(
            method=m.group("method"),
            path=m.group("path"),
            status=int(m.group("status")),
        )
    except ValueError:
        return None


def parse_jsonl_line(line: str) -> Optional[LogEntry]:
    """Parse one JSON-lines record. Expects keys: method, path, status."""
    try:
        obj = json.loads(line)
        return LogEntry(
            method=str(obj["method"]).upper(),
            path=str(obj["path"]),
            status=int(obj["status"]),
        )
    except (json.JSONDecodeError, KeyError, ValueError, TypeError):
        return None


def iter_log(path: str, fmt: str) -> Iterator[LogEntry]:
    """Yield LogEntry objects from a file, sniffing format if requested."""
    parser = parse_jsonl_line if fmt == "jsonl" else parse_combined_line
    with open(path, "r", encoding="utf-8", errors="replace") as fh:
        for line in fh:
            entry = parser(line)
            if entry is not None:
                yield entry


def filter_entries(
    entries: Iterable[LogEntry],
    methods: set[str],
    skip_status_5xx: bool,
    path_prefix: Optional[str],
    limit: Optional[int],
) -> list[LogEntry]:
    """Apply method/status/prefix filters and optionally cap the list."""
    out: list[LogEntry] = []
    for e in entries:
        if e.method not in methods:
            continue
        if skip_status_5xx and 500 <= e.status < 600:
            # Source-side 5xx are usually transient — replaying them
            # mostly produces noise. Skip unless explicitly kept.
            continue
        if path_prefix and not e.path.startswith(path_prefix):
            continue
        out.append(e)
        if limit is not None and len(out) >= limit:
            break
    return out


# ---------------------------------------------------------------------------
# Replay
# ---------------------------------------------------------------------------
def replay_one(
    session: requests.Session,
    target: str,
    entry: LogEntry,
    timeout: float,
    max_body: int,
) -> ReplayResult:
    """Re-issue one request and capture status/error/body hash."""
    # urljoin handles the case where target has a path prefix like /v2.
    url = urljoin(target.rstrip("/") + "/", entry.path.lstrip("/"))
    result = ReplayResult(entry=entry)
    try:
        # POST bodies aren't in the access log, so we send an empty body.
        # This is usually fine for read-only endpoints but will fail for
        # strict validators — filter those out via --methods GET.
        resp = session.request(
            method=entry.method,
            url=url,
            timeout=timeout,
            allow_redirects=False,
            headers={"User-Agent": USER_AGENT},
            data=b"" if entry.method in ("POST", "PUT", "PATCH") else None,
        )
        result.new_status = resp.status_code
        result.elapsed_ms = resp.elapsed.total_seconds() * 1000.0
        # Read a bounded chunk so huge downloads don't OOM the process.
        chunk = resp.raw.read(max_body, decode_content=True) or b""
        result.body_hash = hashlib.sha256(chunk).hexdigest()[:16]
    except requests.Timeout:
        result.error = "timeout"
    except requests.ConnectionError as exc:
        result.error = f"connection_error: {exc.__class__.__name__}"
    except requests.RequestException as exc:
        result.error = f"request_error: {exc.__class__.__name__}"
    return result


def replay_all(
    entries: list[LogEntry],
    target: str,
    timeout: float,
    concurrency: int,
    max_body: int,
) -> list[ReplayResult]:
    """Replay all entries using a thread pool. Order is not preserved."""
    results: list[ReplayResult] = []
    # One Session per worker thread is safe; requests.Session is not
    # documented as thread-safe, so we create one per task via a thread-local.
    import threading
    local = threading.local()

    def worker(entry: LogEntry) -> ReplayResult:
        if not hasattr(local, "session"):
            local.session = requests.Session()
        return replay_one(local.session, target, entry, timeout, max_body)

    with ThreadPoolExecutor(max_workers=max(1, concurrency)) as pool:
        futures = {pool.submit(worker, e): e for e in entries}
        for fut in as_completed(futures):
            try:
                results.append(fut.result())
            except Exception as exc:  # defensive: worker should never raise
                entry = futures[fut]
                results.append(ReplayResult(entry=entry, error=f"worker_crash: {exc}"))
    return results


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------
def print_report(results: list[ReplayResult], show_ok: bool) -> None:
    """Print a compact per-request report and a summary."""
    drift_count = 0
    for r in sorted(results, key=lambda x: (x.entry.path, x.entry.method)):
        if r.drifted:
            drift_count += 1
            if r.error:
                print(f"  [ERR] {r.entry.method:6s} {r.entry.path} "
                      f"expected={r.entry.status} -> {r.error}")
            else:
                print(f"  [DRIFT] {r.entry.method:6s} {r.entry.path} "
                      f"expected={r.entry.status} got={r.new_status} "
                      f"({r.elapsed_ms:.0f}ms)")
        elif show_ok:
            print(f"  [ OK ] {r.entry.method:6s} {r.entry.path} "
                  f"status={r.new_status} ({r.elapsed_ms:.0f}ms)")

    status_counts = Counter(
        r.new_status if r.new_status is not None else "error" for r in results
    )
    print()
    print(f"Replayed: {len(results)}  Drift: {drift_count}  "
          f"Clean: {len(results) - drift_count}")
    print("Status distribution: " +
          ", ".join(f"{k}={v}" for k, v in sorted(status_counts.items(), key=str)))


def write_json_report(results: list[ReplayResult], path: str) -> None:
    """Write machine-readable results for CI artifact consumption."""
    payload = {
        "total": len(results),
        "drift": sum(1 for r in results if r.drifted),
        "results": [
            {
                "method": r.entry.method,
                "path": r.entry.path,
                "expected_status": r.entry.status,
                "actual_status": r.new_status,
                "error": r.error,
                "elapsed_ms": round(r.elapsed_ms, 2),
                "body_hash": r.body_hash,
            }
            for r in results
        ],
    }
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Replay HTTP requests from an access log against a target host.",
    )
    p.add_argument("--traffic-log", required=True,
                   help="Path to nginx Combined Log or JSON-lines file.")
    p.add_argument("--target", default=DEFAULT_TARGET,
                   help=f"Base URL of the target (default: {DEFAULT_TARGET}).")
    p.add_argument("--format", choices=("combined", "jsonl"), default="combined",
                   help="Log format (default: combined).")
    p.add_argument("--methods", default=",".join(DEFAULT_METHODS),
                   help="Comma-separated HTTP methods to replay.")
    p.add_argument("--path-prefix", default=None,
                   help="Only replay paths starting with this prefix.")
    p.add_argument("--limit", type=int, default=None,
                   help="Cap the number of requests replayed.")
    p.add_argument("--concurrency", type=int, default=DEFAULT_CONCURRENCY,
                   help=f"Parallel workers (default: {DEFAULT_CONCURRENCY}).")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
                   help=f"Per-request timeout in seconds (default: {DEFAULT_TIMEOUT}).")
    p.add_argument("--max-body", type=int, default=DEFAULT_MAX_BODY,
                   help=f"Max bytes to read per response (default: {DEFAULT_MAX_BODY}).")
    p.add_argument("--keep-5xx", action="store_true",
                   help="Do not skip source-side 5xx requests.")
    p.add_argument("--show-ok", action="store_true",
                   help="Print successful replays too (verbose).")
    p.add_argument("--json-report", default=None,
                   help="Write a JSON report to this path.")
    return p


def main(argv: Optional[list[str]] = None) -> int:
    args = build_parser().parse_args(argv)

    if not os.path.isfile(args.traffic_log):
        print(f"error: traffic log not found: {args.traffic_log}", file=sys.stderr)
        return 2

    target_parsed = urlparse(args.target)
    if target_parsed.scheme not in ("http", "https"):
        print(f"error: --target must be http(s), got {args.target!r}", file=sys.stderr)
        return 2

    methods = {m.strip().upper() for m in args.methods.split(",") if m.strip()}
    if not methods:
        print("error: --methods cannot be empty", file=sys.stderr)
        return 2

    print(f"Parsing {args.traffic_log} ({args.format}) ...")
    try:
        raw = list(iter_log(args.traffic_log, args.format))
    except OSError as exc:
        print(f"error: cannot read log: {exc}", file=sys.stderr)
        return 2

    entries = filter_entries(
        raw,
        methods=methods,
        skip_status_5xx=not args.keep_5xx,
        path_prefix=args.path_prefix,
        limit=args.limit,
    )
    print(f"Parsed {len(raw)} lines, selected {len(entries)} for replay.")
    if not entries:
        print("Nothing to replay — check --methods / --path-prefix filters.")
        return 0

    print(f"Replaying against {args.target} "
          f"(concurrency={args.concurrency}, timeout={args.timeout}s) ...")
    results = replay_all(
        entries,
        target=args.target,
        timeout=args.timeout,
        concurrency=args.concurrency,
        max_body=args.max_body,
    )

    print_report(results, show_ok=args.show_ok)

    if args.json_report:
        try:
            write_json_report(results, args.json_report)
            print(f"JSON report written to {args.json_report}")
        except OSError as exc:
            print(f"warning: could not write JSON report: {exc}", file=sys.stderr)

    return 1 if any(r.drifted for r in results) else 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

All knobs are settable via CLI flags; the env vars below provide defaults so you can bake them into a container image or CI job without changing the invocation.

CLI flags that don't have env equivalents: --path-prefix, --limit, --keep-5xx, --show-ok, --json-report, --format.

Usage Examples

1. Basic replay against staging, GETs only:

$ python3 replay_test.py --traffic-log nginx_access.log \
    --target http://staging:3000 --methods GET
Parsing nginx_access.log (combined) ...
Parsed 14823 lines, selected 612 for replay.
Replaying against http://staging:3000 (concurrency=8, timeout=10.0s) ...
  [DRIFT] GET    /api/v1/users/42 expected=200 got=404 (12ms)
  [DRIFT] GET    /api/v1/orders?since=2024-10-01 expected=200 got=500 (88ms)
  [ERR]   GET    /api/v1/reports/export expected=200 -> timeout

Replayed: 612  Drift: 3  Clean: 609
Status distribution: 200=580, 404=17, 500=15

2. Focused run with a JSON report for CI:

$ python3 replay_test.py --traffic-log traffic.jsonl --format jsonl \
    --target https://canary.internal --path-prefix /api/v2 \
    --limit 200 --json-report replay.json
...
Replayed: 200  Drift: 0  Clean: 200
JSON report written to replay.json
$ echo $?
0

3. Verbose run against a local dev server:

$ python3 replay_test.py --traffic-log access.log \
    --target http://localhost:8000 --show-ok --concurrency 4
  [ OK ] GET    /healthz status=200 (3ms)
  [ OK ] GET    /api/items status=200 (14ms)
  ...
Replayed: 87  Drift: 0  Clean: 87

Error Handling

• Missing log file → exits with code 2 and a clear message rather than a traceback.
• Malformed lines → skipped silently by the parser. This is deliberate: real access logs contain health checks, scanner probes, and partial writes that would otherwise abort the run.
• Non-HTTP target → validated up front via urlparse; a ftp:// or empty scheme is rejected before any traffic is sent.
• Network failures → requests.Timeout, ConnectionError, and other RequestException subclasses are caught per-request and recorded as error on the ReplayResult. One dead endpoint never aborts the batch.
• Worker crash → the as_completed loop wraps fut.result() defensively so an unexpected exception in a worker becomes a single drift entry instead of killing the pool.
• Oversized responses → resp.raw.read(max_body, ...) caps memory. The body hash will differ from a full-body hash, but for drift detection that's fine.
• Unwritable JSON report → warning to stderr, exit code still reflects drift status.
• Exit codes — 0 means clean, 1 means at least one drift or error, 2 means a configuration/IO problem. CI can treat 1 as a soft failure and 2 as a hard failure.

Extending Ideas

• Header and body replay — Extend the parser to read a HAR file (Chrome DevTools, mitmproxy) which contains full request headers and bodies. This lets you replay authenticated and stateful endpoints accurately, at the cost of a har parsing dependency.
• Semantic response diffing — Instead of comparing status codes only, capture a baseline response from the "old" target with --record and diff JSON bodies structurally (ignoring volatile fields like id, timestamp, request_id). A small recursive diff function is enough.
• Rate limiting and backoff — Wrap replay_one with a token-bucket limiter and retry on 429 / 503 with exponential backoff. Useful when replaying against a production-shaped canary that has real rate limits.
• Distributed replay — For logs with millions of entries, shard the parsed list across workers (e.g., --shard 3/8) and run each shard as a separate process or Kubernetes Job, then merge the JSON reports. The --json-report format is already merge-friendly.