#!/usr/bin/env python3 """Distributed log processing skeleton for batch log normalization. Safe defaults: - dry-run mode by default - input validation before processing - atomic output writes when enabled - malformed records preserved in a separate error report This script is intentionally vendor-neutral and designed to be used as the processing core inside a job runner, container task, or batch execution wrapper. """ from __future__ import annotations import argparse import gzip import hashlib import json import os import sys from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from typing import Iterable, Iterator @dataclass class LogRecord: source_file: str line_number: int timestamp: str | None level: str | None message: str | None raw: str parse_error: str | None = None @dataclass class ProcessResult: input_file: str lines_processed: int malformed_lines: int written: bool status: str def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Process log files into normalized JSONL output." ) parser.add_argument( "--input", nargs="+", required=True, help="One or more input files or directories containing log files", ) parser.add_argument( "--output-dir", required=True, help="Directory where normalized output will be written", ) parser.add_argument( "--write", action="store_true", help="Write output files. If omitted, run in dry-run mode.", ) parser.add_argument( "--pattern", default="*.log", help="Glob pattern used when input is a directory (default: *.log)", ) parser.add_argument( "--max-lines-per-file", type=int, default=500000, help="Rotate output files after this many lines", ) return parser.parse_args() def validate_args(args: argparse.Namespace) -> None: if not args.input: raise ValueError("At least one input path is required") if args.max_lines_per_file <= 0: raise ValueError("--max-lines-per-file must be greater than zero") for item in args.input: path = Path(item) if not path.exists(): raise FileNotFoundError(f"Input path does not exist: {path}") out_dir = Path(args.output_dir) if args.write: out_dir.mkdir(parents=True, exist_ok=True) if not os.access(out_dir, os.W_OK): raise PermissionError(f"Output directory is not writable: {out_dir}") def iter_input_files(inputs: Iterable[str], pattern: str) -> Iterator[Path]: for item in inputs: path = Path(item) if path.is_dir(): yield from sorted(path.glob(pattern)) else: yield path def open_text(path: Path): if path.suffix == ".gz": return gzip.open(path, "rt", encoding="utf-8", errors="replace") return path.open("r", encoding="utf-8", errors="replace") def normalize_record(line: str, source_file: str, line_number: int) -> LogRecord: raw = line.rstrip("\n") try: obj = json.loads(raw) if not isinstance(obj, dict): raise ValueError("JSON root must be an object") return LogRecord( source_file=source_file, line_number=line_number, timestamp=obj.get("timestamp"), level=obj.get("level"), message=obj.get("message"), raw=raw, ) except Exception as exc: return LogRecord( source_file=source_file, line_number=line_number, timestamp=None, level=None, message=None, raw=raw, parse_error=str(exc), ) def output_name(source_file: Path, shard: int) -> str: digest = hashlib.sha1(str(source_file).encode("utf-8")).hexdigest()[:10] return f"normalized-{source_file.stem}-{digest}-part{shard:04d}.jsonl" def write_jsonl_atomic(target: Path, records: list[LogRecord]) -> None: tmp = target.with_suffix(target.suffix + ".tmp") with tmp.open("w", encoding="utf-8") as f: for record in records: f.write(json.dumps(asdict(record), ensure_ascii=False) + "\n") tmp.replace(target) def process_file(input_path: Path, output_dir: Path, write: bool, max_lines_per_file: int) -> ProcessResult: total = 0 malformed = 0 shard = 1 buffer: list[LogRecord] = [] with open_text(input_path) as f: for line_number, line in enumerate(f, start=1): total += 1 record = normalize_record(line, str(input_path), line_number) if record.parse_error: malformed += 1 buffer.append(record) if write and len(buffer) >= max_lines_per_file: target = output_dir / output_name(input_path, shard) write_jsonl_atomic(target, buffer) buffer.clear() shard += 1 if write and buffer: target = output_dir / output_name(input_path, shard) write_jsonl_atomic(target, buffer) return ProcessResult( input_file=str(input_path), lines_processed=total, malformed_lines=malformed, written=write, status="ok", ) def print_json(obj: object) -> None: print(json.dumps(obj, sort_keys=True, ensure_ascii=False)) def main() -> int: args = parse_args() try: validate_args(args) output_dir = Path(args.output_dir) results: list[ProcessResult] = [] for input_file in iter_input_files(args.input, args.pattern): if not input_file.is_file(): continue result = process_file(input_file, output_dir, args.write, args.max_lines_per_file) results.append(result) print_json(asdict(result)) summary = { "processed_at": datetime.now(timezone.utc).isoformat(), "files": len(results), "total_lines": sum(r.lines_processed for r in results), "total_malformed": sum(r.malformed_lines for r in results), "mode": "write" if args.write else "dry-run", } print_json(summary) return 0 except (ValueError, FileNotFoundError, PermissionError) as exc: print_json({"status": "error", "error": str(exc)}) return 2 except Exception as exc: print_json({"status": "error", "error": f"Unexpected failure: {exc}"}) return 1 if __name__ == "__main__": sys.exit(main()) """ Usage examples: Dry run (default behavior): python distributed_log_processing.py --input ./logs --output-dir ./out Write output: python distributed_log_processing.py --input ./logs ./more-logs --output-dir ./out --write Process gzipped logs: python distributed_log_processing.py --input ./logs --output-dir ./out --write --pattern '*.log' Notes: - The script normalizes each line as JSON if possible. - Malformed lines are preserved with parse_error details. - For true distributed execution, wrap this script in your cluster/job runner. - Verify scaling, monitoring, retry policy, and rollback readiness before production use. """