import threading
class MinuteRingBuffer:
def __init__(self, window_minutes=15):
if window_minutes < 1:
raise ValueError("window_minutes must be >= 1")
self.size = window_minutes
self._counts = [0] * self.size
self._last_minute = None
self._lock = threading.Lock()
def _advance(self, minute):
if self._last_minute is None:
self._last_minute = minute
return
gap = minute - self._last_minute
if gap <= 0:
return
# Zero out every bucket we skip so a previous lap can't leak in.
for step in range(1, min(gap, self.size) + 1):
self._counts[(self._last_minute + step) % self.size] = 0
self._last_minute = minute
def add(self, epoch_seconds, weight=1):
minute = int(epoch_seconds) // 60
with self._lock:
if self._last_minute is not None and minute < self._last_minute - self.size:
return # too old for the current window
self._advance(minute)
self._counts[minute % self.size] += weight
def snapshot(self):
with self._lock:
if self._last_minute is None:
return []
base = self._last_minute - self.size + 1
out = []
for offset in range(self.size):
minute = base + offset
out.append((minute, self._counts[minute % self.size]))
return out
import re
from collections import Counter
from datetime import datetime, timezone
from ring_buffer import MinuteRingBuffer
LINE_RE = re.compile(
r"^(?P<ts>\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})\s+(?P<level>[A-Z]+)\b"
)
class LogAggregator:
def __init__(self, window_minutes=15):
self.ring = MinuteRingBuffer(window_minutes)
self.levels = Counter()
self.dropped = 0
def _parse_epoch(self, raw):
stamp = raw.replace("T", " ")
dt = datetime.strptime(stamp, "%Y-%m-%d %H:%M:%S")
return dt.replace(tzinfo=timezone.utc).timestamp()
def ingest(self, line):
match = LINE_RE.match(line.strip())
if not match:
self.dropped += 1
return
try:
epoch = self._parse_epoch(match.group("ts"))
except ValueError:
self.dropped += 1
return
self.ring.add(epoch)
self.levels[match.group("level")] += 1
def report(self):
return {
"per_minute": self.ring.snapshot(),
"by_level": dict(self.levels),
"dropped": self.dropped,
}
import asyncio
import sys
from aggregator import LogAggregator
async def ingest_lines(agg, reader):
loop = asyncio.get_event_loop()
while True:
line = await loop.run_in_executor(None, reader.readline)
if not line:
await asyncio.sleep(0.2)
continue
agg.ingest(line)
async def print_snapshot(agg, interval=5.0):
while True:
await asyncio.sleep(interval)
report = agg.report()
recent = [c for _, c in report["per_minute"]]
print(
"window={} last_minute_count={} by_level={} dropped={}".format(
recent, recent[-1] if recent else 0,
report["by_level"], report["dropped"],
)
)
async def tail_and_report(reader=sys.stdin, window_minutes=15):
agg = LogAggregator(window_minutes)
await asyncio.gather(
ingest_lines(agg, reader),
print_snapshot(agg),
)
if __name__ == "__main__":
asyncio.run(tail_and_report())
This snippet shows how to turn a stream of raw log lines into a compact rolling window of per-minute counts, the kind of primitive that backs an in-process rate widget or a /metrics endpoint. The core idea is a fixed-size ring buffer indexed by minute: instead of retaining every log line or an unbounded dict of timestamps, only N integer buckets are kept, and old buckets are lazily zeroed as time advances. This gives constant memory and O(1) increments regardless of throughput.
In MinuteRingBuffer, the buffer is a plain list of size window_minutes, and each event's epoch second is mapped to an absolute minute via integer division. The trick is _advance: when the current minute is ahead of the last-seen minute, the code walks forward and resets every bucket it passes so stale counts from a previous lap of the ring never leak into the new window. The modulo minute % self.size selects the slot, and a threading.Lock guards both the advance and the increment so concurrent producers and a reader thread stay consistent. snapshot returns an ordered list of (minute, count) pairs, oldest first, which is exactly the shape a sparkline or Prometheus gauge wants.
The pitfall this design accepts on purpose is that events older than the window are dropped rather than counted, and clock skew or events far in the future would fast-forward the ring, discarding the current window. That trade-off is what keeps it bounded.
In LogAggregator, a compiled regex extracts an ISO-ish timestamp and log level from each line, converting it to an epoch second. Unparseable lines are counted separately under dropped instead of raising, since log tailers must tolerate garbage. A separate ring per level would be a natural extension; here a single ring tracks total volume while a small Counter tracks level breakdown.
In tail_and_report, the aggregator is driven by an async tailer that reads lines as they arrive and prints a snapshot on a fixed cadence using asyncio.gather, showing how the synchronous, lock-protected buffer composes cleanly with an async I/O loop. Because the buffer's public methods are cheap and self-contained, the reporting coroutine never blocks the ingest coroutine for long. This pattern is a good fit whenever a rolling rate is needed without pulling in a full time-series database.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
