← Back to all products
$29
Log Aggregator
Centralized log aggregation with JSON parsing, search, filtering, and log rotation.
MarkdownPythonNginx
📄 Product Preview
Try the interactive reader and demo tools below, or get the full product with all content unlocked.
📖 Interactive Reader (Free Preview) ⚙ Try Demo Tools 📦 Download Free Sample📁 File Structure 11 files
log-aggregator/
├── LICENSE
├── README.md
├── examples/
│ ├── access.log
│ └── app.log
├── free-sample.zip
├── guide/
│ ├── 01_features.md
│ ├── 02_quick-start.md
│ ├── 03_cli-reference.md
│ └── 04_license.md
├── index.html
└── src/
└── log_aggregator.py
📖 Documentation Preview README excerpt
Log Aggregator
Part of the Deploy Kit by CodeVault
Collect, parse, filter, and search log files from multiple sources. Supports JSON, Apache, nginx, and syslog formats with pattern matching, alerts, and structured output for analysis.
Features
- Multi-file tailing: Monitor multiple log files simultaneously
- Format parsing: Auto-detect and parse JSON, Apache Combined, nginx, and syslog formats
- Filtering: Include/exclude lines by log level, pattern, or time range
- Search: Regex search across log files with context lines
- Alerting: Trigger alerts when error patterns are detected
- Statistics: Count log levels, top error messages, and request rates
- Output: Structured JSON output for feeding into dashboards
- Handles log rotation (detects file truncation/recreation)
- Python stdlib only — zero dependencies
Quick Start
# Parse and display a log file
python src/log_aggregator.py parse --files examples/app.log
# Search for error patterns across multiple files
python src/log_aggregator.py search --files examples/app.log,examples/access.log --pattern "error|ERROR|500"
# Filter by log level
python src/log_aggregator.py parse --files examples/app.log --level error
# Generate statistics report
python src/log_aggregator.py stats --files examples/app.log --format json
# Parse with specific format
python src/log_aggregator.py parse --files examples/access.log --log-format apache
# Watch files in real-time (tail mode)
python src/log_aggregator.py tail --files examples/app.log --follow
CLI Reference
| Command | Description |
|---|---|
parse | Parse and display log entries |
search | Search logs with regex patterns |
stats | Generate log statistics report |
tail | Watch log files in real-time |
| Flag | Description |
|---|---|
--files | Comma-separated list of log file paths |
--pattern | Regex pattern for search |
--level | Filter by log level: debug, info, warning, error, critical |
--log-format | Force log format: auto, json, apache, nginx, syslog |
--format, -f | Output format: text or json |
--output, -o | Write output to file |
--context, -C | Lines of context around search matches |
--follow | Keep watching files for new content (tail -f mode) |
--verbose, -v | Enable verbose logging |
... continues with setup instructions, usage examples, and more.
📄 Code Sample .py preview
src/log_aggregator.py
#!/usr/bin/env python3
"""
Log Aggregator — Deploy Kit by DataNest
Collect, parse, filter, search, and analyze log files from multiple sources.
Supports JSON, Apache Combined, nginx, and syslog log formats with pattern
matching, level filtering, and structured output.
Why this exists:
When something goes wrong in production, you need to search across
multiple log files quickly. grep works, but it doesn't understand log
formats, can't filter by level, and doesn't give you statistics.
This tool parses structured logs, lets you filter by level or time,
and produces stats that help you understand what's happening.
Usage:
python log_aggregator.py parse --files app.log
python log_aggregator.py search --files app.log,access.log --pattern "error|500"
python log_aggregator.py stats --files app.log --format json
python log_aggregator.py tail --files app.log --follow
License: MIT
"""
from __future__ import annotations
import argparse
import json
import logging
import os
import re
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
LOG = logging.getLogger("log-aggregator")
LEVEL_PRIORITY = {
"debug": 0,
"info": 1,
"warning": 2,
"warn": 2,
"error": 3,
# ... 655 more lines ...