The Problem

When running background services or backend processes, unhandled exceptions and critical errors inevitably show up in system log files or process standard output. Detecting these errors promptly is essential, but developers often face two persistent challenges when monitoring log streams directly on a server:

  1. Missing context around the error: A simple alert containing only the isolated line that triggered an alert is rarely enough to diagnose what went wrong. For example, knowing that a service threw a generic database connection failure does not tell you which query was executing or what user action immediately preceded it. You typically need the preceding log lines to understand the root cause and subsequent log lines to see if recovery routines succeeded.
  2. Notification flooding from repeated errors: When an outage occurs, an application may emit hundreds of identical error lines within a few seconds. If an alerting script triggers on every matched line, external alert targets become flooded with duplicate notifications. This alert fatigue can hide other distinct failures happening on the system.

Many log aggregation stacks require running dedicated background agents, external databases, and complex query processors just to watch a single log stream. When all you need is regex-based pattern matching, deduplication cooldowns, surrounding context collection, and lightweight notification delivery, managing an entire monitoring cluster introduces unnecessary operational overhead.

Install & Usage

The tool is packaged as a standalone Python script with no external third-party dependencies. To use it, download the single file, save it as logtrip_pattern_alerter.py, and run it with Python 3.

Because the tool relies exclusively on Python standard library modules—such as argparse, re, smtplib, urllib.request, and subprocess—no third-party packages need to be installed via pip.

To inspect the built-in command-line arguments and check that the script is executable, run:

python logtrip_pattern_alerter.py --help

You can point the monitor to tail a physical log file or watch output streams produced by system utilities. When matching log streams, the tool reads incoming lines, compares each line against configured rules, accumulates context lines, and routes events to configured notification targets such as stdout, webhooks, or email.

How It Works

The codebase is organized into modular components responsible for stream reading, rule evaluation, deduplication tracking, context buffering, and notification dispatching.

1. Rule Definition and Event Formatting

Rules configure the target regex pattern, alert severity, cooldown duration, before/after context line counts, and output channels. When a rule triggers, an AlertEvent object constructs a formatted text block containing the exact matched line, the rule metadata, and the captured context lines.

Here is the context formatting method from AlertEvent:

def format_text(self) -> str: t_str = time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime(self.timestamp)) lines = [ f"[{self.rule.severity}] LogTrip Alert: {self.rule.name} (ID: {self.id})", f"Timestamp: {t_str}", f"Source: {self.source}", f"Rule Regex: {self.rule.raw_pattern}", "", "--- Context Trace ---", ] for l in self.before_lines: lines.append(f" [ctx] {l}") lines.append(f"> [FIRE] {self.matched_line}") for l in self.after_lines: lines.append(f" [ctx] {l}") lines.append("----------------------") return "\n".join(lines)

Each event generates a unique identifier using a SHA-256 hash of the rule name combined with the matched line. This identifier helps correlate repeated alerts across delivery channels.

2. Cooldown and Deduplication

To prevent alert flooding when a service outputs the same failure continuously, the CooldownTracker class tracks when each pattern last triggered.

class CooldownTracker: """Maintains deduplication windows per rule and trigger fingerprint.""" def __init__(self): self._last_fired: Dict[str, float] = {} def should_fire(self, rule_name: str, fingerprint: str, cooldown_sec: float, now: Optional[float] = None) -> bool: ts = now if now is not None else time.time() key = f"{rule_name}::{fingerprint}" last = self._last_fired.get(key, 0.0) if (ts - last) >= cooldown_sec: self._last_fired[key] = ts return True return False def clear(self) -> None: self._last_fired.clear()

If a line matching the same rule is encountered before cooldown_sec has elapsed, the trigger is suppressed, preventing repetitive notification spam while the underlying issue persists.

3. File Tail and Rotation Handling

Reading log files continuously on Unix-like systems requires handling log rotation routines, where the current file is renamed or truncated and a new file is opened under the original path.

The FileStreamReader tracks both the inode number and the file size to detect file rotations automatically:

# End of current stream reached: inspect possible rotation time.sleep(self.poll_interval) try: st = os.stat(self.file_path) current_size = self._current_file.tell() # File rotated if inode changed or truncated  if st.st_ino != self._current_inode or st.st_size < current_size: sys.stderr.write(f"[INFO] Log rotation detected on {self.file_path}. Reopening...\n") self._current_file.close() self._current_file, self._current_inode = self._open_file() except FileNotFoundError: # File temporarily missing during rename/re-create  continue except Exception as e: sys.stderr.write(f"[WARN] Stat error on {self.file_path}: {e}\n")

If the operating system changes the file's inode or if the file size drops below the current stream offset, the reader closes the stale descriptor and re-opens the path to seamlessly continue following new lines.

4. Process Streams

In addition to static files, the tool includes a SubprocessStreamReader that spawns an external monitoring command (such as system log followers) via subprocess.Popen and reads lines directly from standard output line by line.

Limitations

Before deploying the script in production environments, consider the following technical constraints present in the implementation:

  • In-Memory Cooldown State: The CooldownTracker maintains timestamps in a standard Python dictionary in memory. If the script restarts, all deduplication history is cleared, meaning any recurring errors will immediately trigger new notifications until the cooldown window is rebuilt.
  • Synchronous Network Calls: The NotificationDispatcher sends outgoing HTTP requests using urllib.request.urlopen and email notifications via smtplib.SMTP directly within the main loop. If an external notification endpoint is slow or times out (with the default 10-second timeout), log consumption pauses until the network call finishes or fails.
  • Single-Line Pattern Evaluation: Rules match against single lines using re.search. The tool cannot match a regular expression that spans across multiple lines simultaneously. Instead, you must match a specific anchor line and rely on before_context and after_context to capture surrounding multiline stack traces.
  • File Polling Latency: Rather than using kernel-level event APIs, FileStreamReader relies on a polling loop with a configurable sleep interval (poll_interval, defaulting to 0.5 seconds). Log entries are processed as they are polled rather than instantaneously via file descriptor notifications.

FAQ

Does this script require installing external packages with pip?

No. The implementation relies entirely on Python standard library modules. You do not need to run pip install for any third-party dependencies.

How does the log follower handle log rotations?

The reader inspects the inode via os.fstat and checks the file size against the current read offset. If the inode changes or the file size becomes smaller than the previous read position, the file is identified as rotated, closed, and reopened.

Can I verify my alert rules without sending actual notifications?

Yes. The notification dispatcher includes a dry-run mode. When active, it writes formatted messages to standard output describing the request that would have been dispatched without making actual network requests to webhooks or mail servers.

Full version and the rest of the toolkit: LogTrip pattern alerter on PyForge. Product page: https://py-forge.com/products/logtrip-pattern-alerter