app.py

#!/usr/bin/env python3
"""
HTTP Event Collector
====================
A lightweight HTTP server that receives events via POST requests
and saves them to a JSONL file (one JSON object per line).

Usage:
    python http_event_collector.py [--host HOST] [--port PORT] [--output FILE] [--token TOKEN]

Endpoints:
    POST /events          — Submit a single event (JSON body)
    POST /events/batch    — Submit multiple events (JSON array body)
    GET  /health          — Health check
    GET  /stats           — Collection statistics

Example (send an event):
    curl -X POST http://localhost:8088/events \
         -H "Content-Type: application/json" \
         -H "Authorization: Bearer mytoken" \
         -d '{"source": "app", "event": "user_login", "user": "alice"}'
"""

import argparse
import json
import logging
import os
import signal
import sys
import threading
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import Any

# ── Configuration ────────────────────────────────────────────────────────────

DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 8088
DEFAULT_OUTPUT = "events.jsonl"
DEFAULT_TOKEN = None          # Set to a string to require bearer-token auth

# ── Globals ──────────────────────────────────────────────────────────────────

stats = {"received": 0, "saved": 0, "errors": 0}
stats_lock = threading.Lock()
file_lock = threading.Lock()
output_path: Path = Path(DEFAULT_OUTPUT)
auth_token: str | None = DEFAULT_TOKEN

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
log = logging.getLogger("hec")


# ── Helpers ──────────────────────────────────────────────────────────────────

def enrich(event: dict[str, Any], remote_addr: str) -> dict[str, Any]:
    """Add server-side metadata to an event."""
    if "timestamp" not in event:
        event["timestamp"] = datetime.now(timezone.utc).isoformat()
    event.setdefault("_collector", {
        "remote_addr": remote_addr,
        "collected_at": datetime.now(timezone.utc).isoformat(),
    })
    return event


def save_events(events: list[dict[str, Any]]) -> int:
    """Append events to the JSONL output file. Returns number saved."""
    saved = 0
    with file_lock:
        with output_path.open("a", encoding="utf-8") as f:
            for ev in events:
                f.write(json.dumps(ev, ensure_ascii=False) + "\n")
                saved += 1
    return saved


def check_auth(handler: "EventHandler") -> bool:
    """Return True if auth passes (or no auth configured)."""
    if auth_token is None:
        return True
    header = handler.headers.get("Authorization", "")
    if header.startswith("Bearer ") and header[7:] == auth_token:
        return True
    handler.send_error_json(401, "Unauthorized — invalid or missing Bearer token")
    return False


# ── Request Handler ──────────────────────────────────────────────────────────

class EventHandler(BaseHTTPRequestHandler):

    # ── Helpers ──────────────────────────────────────────────────────────────

    def send_json(self, code: int, payload: dict) -> None:
        body = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def send_error_json(self, code: int, message: str) -> None:
        self.send_json(code, {"status": "error", "message": message})

    def read_json_body(self) -> tuple[Any, str | None]:
        """Read + parse JSON body. Returns (parsed, error_message)."""
        length = int(self.headers.get("Content-Length", 0))
        if length == 0:
            return None, "Empty body"
        raw = self.rfile.read(length)
        try:
            return json.loads(raw), None
        except json.JSONDecodeError as exc:
            return None, f"Invalid JSON: {exc}"

    def log_message(self, fmt, *args):  # silence default access log
        pass

    # ── Routing ──────────────────────────────────────────────────────────────

    def do_GET(self):
        if self.path == "/health":
            self.handle_health()
        elif self.path == "/stats":
            self.handle_stats()
        else:
            self.send_error_json(404, f"Not found: {self.path}")

    def do_POST(self):
        if self.path in ("/events", "/events/"):
            self.handle_single_event()
        elif self.path in ("/events/batch", "/events/batch/"):
            self.handle_batch_events()
        else:
            self.send_error_json(404, f"Not found: {self.path}")

    # ── Endpoint handlers ────────────────────────────────────────────────────

    def handle_health(self):
        self.send_json(200, {
            "status": "ok",
            "output_file": str(output_path.resolve()),
            "uptime": "running",
        })

    def handle_stats(self):
        with stats_lock:
            snap = dict(stats)
        snap["output_file"] = str(output_path.resolve())
        snap["output_size_bytes"] = output_path.stat().st_size if output_path.exists() else 0
        self.send_json(200, snap)

    def handle_single_event(self):
        if not check_auth(self):
            return
        data, err = self.read_json_body()
        if err:
            self.send_error_json(400, err)
            return
        if not isinstance(data, dict):
            self.send_error_json(400, "Body must be a JSON object")
            return

        event = enrich(data, self.client_address[0])
        saved = save_events([event])

        with stats_lock:
            stats["received"] += 1
            stats["saved"] += saved

        log.info("Event saved | source=%s event=%s",
                 event.get("source", "-"), event.get("event", "-"))
        self.send_json(200, {"status": "ok", "saved": saved})

    def handle_batch_events(self):
        if not check_auth(self):
            return
        data, err = self.read_json_body()
        if err:
            self.send_error_json(400, err)
            return
        if not isinstance(data, list):
            self.send_error_json(400, "Body must be a JSON array of objects")
            return

        events = []
        for i, item in enumerate(data):
            if not isinstance(item, dict):
                self.send_error_json(400, f"Item {i} is not a JSON object")
                return
            events.append(enrich(item, self.client_address[0]))

        saved = save_events(events)
        with stats_lock:
            stats["received"] += len(events)
            stats["saved"] += saved

        log.info("Batch saved | count=%d", saved)
        self.send_json(200, {"status": "ok", "saved": saved})


# ── Entry point ──────────────────────────────────────────────────────────────

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="HTTP Event Collector")
    p.add_argument("--host", default=DEFAULT_HOST, help=f"Bind host (default: {DEFAULT_HOST})")
    p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"Port (default: {DEFAULT_PORT})")
    p.add_argument("--output", default=DEFAULT_OUTPUT, help=f"Output JSONL file (default: {DEFAULT_OUTPUT})")
    p.add_argument("--token", default=None, help="Optional Bearer token for auth")
    return p.parse_args()


def main():
    global output_path, auth_token
    args = parse_args()
    output_path = Path(args.output)
    auth_token = args.token

    # Ensure output directory exists
    output_path.parent.mkdir(parents=True, exist_ok=True)

    server = HTTPServer((args.host, args.port), EventHandler)

    def shutdown(sig, frame):
        log.info("Shutting down…")
        with stats_lock:
            log.info("Final stats: %s", stats)
        server.shutdown()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    log.info("HTTP Event Collector started")
    log.info("  Listening : http://%s:%d", args.host, args.port)
    log.info("  Output    : %s", output_path.resolve())
    log.info("  Auth      : %s", "Bearer token required" if auth_token else "disabled")
    log.info("  Endpoints : POST /events  |  POST /events/batch  |  GET /health  |  GET /stats")

    server.serve_forever()


if __name__ == "__main__":
    main()