bufferlog 0.1.0

BufferLog — Buffer logs in memory, flush only on errors. Save 90%+ on APM costs.

bufferlog

Buffer logs in memory per-request. Flush only on errors. Save 90%+ on APM costs.

Table of Contents

• Install
• Usage
  • Flask
  • FastAPI
  • Django
• Configuration
• Control Plane
• Metrics
• Shutdown
• License

Install

pip install bufferlog

With framework extras:

pip install bufferlog[flask]
pip install bufferlog[fastapi]
pip install bufferlog[django]

Usage

Flask

import logging
from flask import Flask
from bufferlog import BufferLog
from bufferlog.adapters import StdOutAdapter

app = Flask(__name__)
bl = BufferLog(adapters=[StdOutAdapter(pretty=True)])
bl.init_flask(app)

logger = logging.getLogger("myapp")
logger.addHandler(bl.logging_handler())
logger.setLevel(logging.DEBUG)

@app.route("/")
def index():
    logger.info("Processing request")   # Buffered (discarded on 200)
    logger.debug("SQL: SELECT * ...")    # Buffered (discarded on 200)
    return "OK"                          # 200 → logs discarded ($0)

@app.route("/fail")
def fail():
    logger.info("Starting transaction")
    logger.error("Connection lost")      # Triggers flush
    return "Error", 500                  # 500 → context sent to APM

FastAPI

import logging
from fastapi import FastAPI
from bufferlog import BufferLog
from bufferlog.adapters import StdOutAdapter
from bufferlog.middleware.fastapi_mw import BufferLogMiddleware

app = FastAPI()
bl = BufferLog(adapters=[StdOutAdapter()])
app.add_middleware(BufferLogMiddleware, **bl.asgi_kwargs())

logger = logging.getLogger("myapp")
logger.addHandler(bl.logging_handler())
logger.setLevel(logging.DEBUG)

@app.get("/")
async def index():
    logger.info("Handling request")
    return {"ok": True}

Django

# settings.py
MIDDLEWARE = [
    "bufferlog.middleware.django_mw.BufferLogDjangoMiddleware",
    # ... other middleware
]

# apps.py or wsgi.py
from bufferlog import BufferLog
from bufferlog.adapters import StdOutAdapter

bl = BufferLog(adapters=[StdOutAdapter()])
bl.init_django()

Configuration

from bufferlog import BufferLog, BufferLogConfig
from bufferlog.adapters import DatadogAdapter, SplunkAdapter, StdOutAdapter

bl = BufferLog(config=BufferLogConfig(
    # Max logs per request buffer. Oldest overwritten when full.
    buffer_capacity=100,

    # Log levels that trigger immediate flush.
    flush_on_levels=["error", "critical"],

    # HTTP status codes that trigger flush.
    flush_on_status_codes=[500, 501, 502, 503, 504],

    # Downstream targets for flushed logs.
    adapters=[
        DatadogAdapter(api_key="your-key"),
        SplunkAdapter(token="your-token", url="https://splunk.example.com"),
        StdOutAdapter(),
    ],

    # Enable/disable buffering.
    enabled=True,

    # If True, logs fall through to stderr on adapter failure.
    fail_open=True,

    # Scrub PII before logs enter the buffer.
    scrubber=lambda msg, meta: (msg.replace("password", "***"), meta),
))

Control Plane

Connect to the BufferLog Control Plane for remote policy management and usage tracking.

from bufferlog import BufferLog, ControlPlaneConfig

bl = BufferLog(
    control_plane=ControlPlaneConfig(
        url="https://control.bufferlog.io",
        api_key="bl_sk_your_key",
        poll_interval_s=60,
        telemetry_interval_s=60,
    )
)

When connected:

• Poll for policy updates — buffer capacity, sampling rate, flush triggers, and bypass rules can all be changed remotely.
• Push usage metrics — counters (never log data) are sent to power the ROI dashboard.

If the control plane is unreachable, the SDK continues with its last known configuration.

Metrics

metrics = bl.get_metrics()
# {
#   "buffers":  {"created": 1000, "discarded": 995, "flushed": 5, "active": 0},
#   "flash":    {"flush_count": 5, "events_flushed": 45, "adapter_errors": 0}
# }

Shutdown

Stop background threads and send a final telemetry report.

import atexit

atexit.register(bl.shutdown)

License

MIT