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.
Python code
36 linesimport 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
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
- Use `aiofiles` to asynchronously read a live log file in chunks.
- 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
More from Observability & SRE
- Adding a Correlation ID to Log Context in Python medium
- Check if a Timestamp Falls in a Daily Maintenance Window in Python easy
- Export Metrics with OTLP Mock in Python medium
- Generate Mock CPU and Memory Metrics in Python easy
- Generate Prometheus Text Exposition Format in Python easy
- Generate Synthetic CPU Utilization Metrics in Python easy
Keep learning
Related tutorials and quizzes for this topic.