Calculate Error Rate from Log Stream in Python

Parses a mock log stream to count errors and compute the error percentage using a rolling window of recent entries.

Easy Python 3.9+ Aug 9, 2026 Observability & SRE 15 views 0 copies

Python code

36 lines
Python 3.9+
import re
from collections import deque

def error_rate_from_log_stream(message):
    log_pattern = r'^\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\] (ERROR|INFO|DEBUG): (.*)$'
    recent_entries = deque(maxlen=100)
    error_count = 0
    total_count = 0

    for line in message.strip().split('\n'):
        match = re.match(log_pattern, line)
        if match:
            total_count += 1
            recent_entries.append((match.group(1), match.group(2), match.group(3)))
            if match.group(2) == 'ERROR':
                error_count += 1

    current_error_rate = (error_count / total_count * 100) if total_count else 0.0
    return current_error_rate, error_count, total_count, list(recent_entries)

if __name__ == "__main__":
    mock_logs = """\
[2025-03-15 10:30:01] INFO: Application started
[2025-03-15 10:30:02] ERROR: Database connection failed
[2025-03-15 10:30:03] INFO: Retrying connection
[2025-03-15 10:30:04] DEBUG: Cache refreshed
[2025-03-15 10:30:05] ERROR: Timeout on API call
[2025-03-15 10:30:06] INFO: Request completed
"""
    rate, errors, total, entries = error_rate_from_log_stream(mock_logs)
    print(f"Total log entries: {total}")
    print(f"Error count: {errors}")
    print(f"Error rate: {rate:.2f}%")
    print(f"\nRecent entries (last {len(entries)}):")
    for ts, level, msg in entries:
        print(f"  [{ts}] {level}: {msg}")

Output

stdout
Total log entries: 6
Error count: 2
Error rate: 33.33%

Recent entries (last 6):
  [2025-03-15 10:30:01] INFO: Application started
  [2025-03-15 10:30:02] ERROR: Database connection failed
  [2025-03-15 10:30:03] INFO: Retrying connection
  [2025-03-15 10:30:04] DEBUG: Cache refreshed
  [2025-03-15 10:30:05] ERROR: Timeout on API call
  [2025-03-15 10:30:06] INFO: Request completed

How it works

The regex extracts the timestamp, level, and message from each log line. A deque with maxlen=100 keeps only the most recent 100 entries, mimicking a rolling buffer in a live stream. The error rate is simply errors / total * 100, with a guard against division by zero. This pattern is useful for calculating real-time error rates from log aggregators or streaming pipelines.

Common mistakes

  • Using `re.findall` instead of `re.match` can accidentally match partial lines.
  • Forgetting the zero-division check when `total_count` is 0.
  • Assuming the log format is fixed without handling malformed lines gracefully.

Variations

  1. Use `aiofiles` to asynchronously read a live log file in chunks.
  2. Implement a time-windowed error rate (e.g., last 60 seconds) using timestamps from the log.

Real-world use cases

  • Monitoring API error rates in real time to trigger alerts when they exceed a threshold.
  • Generating SLO/SLI error budgets from service logs in a production environment.
  • Analyzing delivery failure rates in a message queue consumer by parsing producer logs.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Observability & SRE

Related tutorials and quizzes for this topic.