codesnips
python 161 lines · 4 tabs

Size-and-Time Batching Metrics Buffer With a Background Flush Loop

Shared by codesnips Aug 2026
4 tabs 
import time
import threading
from dataclasses import dataclass, field
from typing import Callable, Dict, List


@dataclass
class MetricPoint:
    name: str
    value: float
    timestamp: float = field(default_factory=time.time)
    tags: Dict[str, str] = field(default_factory=dict)

    def as_dict(self) -> dict:
        return {
            "name": self.name,
            "value": self.value,
            "ts": self.timestamp,
            "tags": self.tags,
        }


class MetricBuffer:
    def __init__(self, sink: Callable[[List[MetricPoint]], None], max_size: int = 500):
        self._sink = sink
        self._max_size = max_size
        self._lock = threading.Lock()
        self._points: List[MetricPoint] = []

    def record(self, point: MetricPoint) -> bool:
        with self._lock:
            self._points.append(point)
            return len(self._points) >= self._max_size

    def flush(self) -> int:
        with self._lock:
            batch, self._points = self._points, []
        if not batch:
            return 0
        return self._deliver(batch)

    def _deliver(self, batch: List[MetricPoint]) -> int:
        try:
            self._sink(batch)
            return len(batch)
        except Exception:
            # Re-queue at the front for at-least-once delivery.
            with self._lock:
                self._points = batch + self._points
            raise
4 files · python Explain with highlit

This snippet shows a common building block in telemetry pipelines: a client-side buffer that accumulates metric points in memory and ships them to a remote sink in batches, flushing whenever the buffer reaches a size threshold OR a time interval elapses — whichever comes first. Batching amortizes the cost of network round-trips and reduces load on the sink, while the time trigger bounds how long a point can sit unsent, keeping data reasonably fresh even under low traffic.

In MetricPoint, a lightweight dataclass captures the shape of a single sample and a default_tags factory avoids the classic mutable-default-argument bug. It exposes as_dict so the sink layer decides on wire encoding rather than the buffer.

MetricBuffer is the core. record appends under a threading.Lock and returns whether the buffer is now full; the caller uses that signal to trigger an out-of-band flush without waiting for the timer. flush swaps the internal list for a fresh one while holding the lock (a cheap drain-and-release), then does the slow network call outside the lock so recording threads are never blocked on I/O. This drain-then-send ordering is the key concurrency trade-off: it keeps the critical section tiny at the cost of briefly holding a detached batch that must not be lost. To protect against that, _flush_locked re-queues the batch back onto the front of the buffer if the sink raises, giving simple at-least-once retry semantics; duplicates are accepted as the price of durability.

BatchingMetricsClient in the same tab wires timing to the buffer. A daemon Thread runs _run, sleeping on a threading.Event so close can wake it immediately for a clean shutdown. Each flush_interval it flushes; gauge and increment call record and eagerly flush when the size trigger fires. close sets _stop, joins the thread, and performs a final flush so buffered points are not dropped on exit.

HttpMetricSink shows a realistic sink: it posts the batch as JSON with a short timeout and raises on non-2xx so the buffer's retry path engages. The test_batching_flush tab demonstrates both triggers — filling to capacity forces an immediate send, and advancing the fake clock exercises the interval path — and asserts the sink received exactly one batch of the expected size.

Related snips

go 
package api

import (
  "net/http"
  "runtime/debug"
)

Expose build metadata for debugging deploys

go observability build
by Leah Thompson 1 tab
rust 
use crossbeam::channel::unbounded;
use std::thread;

fn main() {
    let (tx, rx) = unbounded();

Crossbeam for advanced concurrent data structures

rust concurrency lock-free
by Marcus Chen 1 tab
javascript 
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
  const success = true;

  setTimeout(() => {
    if (success) {

Promises and async/await patterns for asynchronous JavaScript

javascript promises async-await
by Alex Chang 1 tab
rust 
use tracing::{info, instrument};

#[instrument]
fn process_request(user_id: u64) {
    info!(user_id, "Processing request");
    // Work happens here

tracing for structured logging and distributed tracing

rust observability tracing
by Marcus Chen 1 tab
rust 
use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

Channels (mpsc) for message passing between threads

rust concurrency channels
by Marcus Chen 1 tab
typescript 
export type Settled<R> =
  | { status: 'fulfilled'; value: R }
  | { status: 'rejected'; reason: unknown };

export interface ConcurrencyOptions {
  limit: number;

Simple concurrency limiter for batch operations

node concurrency async
by codesnips 2 tabs

Share this code

Here's the card — post it anywhere.

Size-and-Time Batching Metrics Buffer With a Background Flush Loop — share card
Link copied