⚡ PulseLog

A non-blocking Python logging library with a real-time browser dashboard and durable checkpoint store.

Built for ML training, data pipelines, backend services, experiments, and long-running workloads where logging should stay out of the critical path.

---

✨ Why PulseLog?

Traditional logging can become surprisingly expensive when it is called inside:

• ML training loops
• ETL/data pipelines
• inference workloads
• batch processing
• concurrent worker systems
• long-running experiments

PulseLog is designed around a simple principle:

Logging should not become the bottleneck of the application.

Log records are handled asynchronously so application code does not need to wait for dashboard delivery or other downstream processing.

---

📦 Installation

pip install pulselog

Then:

from pulselog import Logger

log = Logger("my-app")
log.info("application started")

---

🚀 Quick Start

from pulselog import Logger

log = Logger("training")

log.info("training started", epoch=1)
log.warning("learning rate is high", learning_rate=0.1)

log.shutdown()

When the dashboard is enabled, PulseLog provides a browser-based view of the log stream.

Default dashboard:

http://localhost:5678

---

💡 Examples

Basic Logging

from pulselog import Logger

log = Logger("my-app")

log.debug("debugging info")           # DEBUG level
log.info("something happened")        # INFO level
log.warning("something seems off")    # WARNING level
log.error("something went wrong")     # ERROR level
log.critical("system failure")        # CRITICAL level

log.shutdown()

With Structured Data

from pulselog import Logger

log = Logger("api-server")

log.info(
    "request handled",
    method="GET",
    path="/users/42",
    status=200,
    latency_ms=12,
)

log.warning(
    "slow request",
    path="/reports",
    latency_ms=2500,
)

log.shutdown()

Error Handling

from pulselog import Logger

log = Logger("data-processor")

try:
    result = process_data(raw_input)
except ValueError as e:
    log.error("invalid input format", error=str(e))
except Exception:
    log.exception("unexpected error during processing")

log.shutdown()

Without Dashboard (Production)

from pulselog import Logger

log = Logger(
    "production-worker",
    dashboard=False,              # Disable browser dashboard
    queue_size=50000,             # Larger queue for high throughput
    worker_interval=0.005,        # Faster processing
)

for item in work_queue:
    log.info("processing", item_id=item.id)

log.shutdown()

ML Training Loop

from pulselog import Logger

log = Logger("training", checkpoint_path="training.db")

for epoch in range(10):
    train_loss = train_one_epoch()
    val_accuracy = validate()

    log.info(
        f"epoch {epoch + 1} complete",
        loss=round(train_loss, 4),
        accuracy=round(val_accuracy, 4),
    )

    log.save(
        f"epoch-{epoch + 1}",
        {"loss": train_loss, "accuracy": val_accuracy},
        status="DONE",
        progress=(epoch + 1) * 10,
    )

log.shutdown()

Web Request Handler

from pulselog import Logger

log = Logger("web-server")

def handle_request(request):
    log.info("request received", method=request.method, path=request.path)

    try:
        response = process(request)
        log.info("request complete", status=response.status_code)
        return response

    except AuthError:
        log.warning("authentication failed", ip=request.ip)
        raise

    except Exception:
        log.exception("request failed")
        raise

log.shutdown()

Background Tasks

from concurrent.futures import ThreadPoolExecutor
from pulselog import Logger

log = Logger("task-runner")

def run_task(task):
    log.info("task started", task_id=task.id)
    result = task.execute()
    log.info("task finished", task_id=task.id, duration_ms=result.duration)
    return result

with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(run_task, tasks))

log.info("all tasks complete", total=len(results))
log.shutdown()

Automatic Function Logging (Decorators)

PulseLog can instrument functions automatically — timing and exceptions are logged without touching the function body.

from pulselog import Logger

log = Logger("services")

@log
def fetch_user(user_id: int):
    return database.get_user(user_id)

fetch_user(42)

Exceptions are captured automatically — the exception is logged with its traceback and then re-raised, so control flow is never changed:

@log
def risky_operation(config: dict):
    ...
# If the function raises, PulseLog logs it before propagating.

Nesting support: decorated functions compose safely. Deeply nested calls are logged at every level with linear cost:

@log
def pipeline():          # level 1
    stage_one()          # level 2
    stage_two()          # level 2

@log
def stage_one():
    load_data()          # level 3

pipeline()
# Logs all levels with correct nesting and per-function timings

Useful for:

• tracing request flow through service layers
• profiling slow functions in pipelines
• debugging nested call chains
• auditing entry/exit of critical code paths

Measured overhead: ~4 µs per decoration level, verified linear up to 250-deep nesting (see Performance).

⚠️ Adjust any example output above to match your library's actual log record format before publishing.

Using Tags

from pulselog import Logger

log = Logger("pipeline")

log.tag("ingestion")
log.info("loading data", source="database")
log.info("loaded rows", count=15000)

log.tag("transform")
log.info("applying transforms")
log.info("transforms complete")

log.tag("export")
log.info("writing output", destination="s3://bucket/data")

log.shutdown()

Context Manager for Tags

from pulselog import Logger

log = Logger("pipeline")

with log.context(tag="extract"):
    log.info("connecting to source")
    data = extract()
    log.info("extraction complete", rows=len(data))

with log.context(tag="transform"):
    log.info("cleaning data")
    clean = transform(data)
    log.info("transform complete")

with log.context(tag="load"):
    log.info("writing to warehouse")
    load(clean)
    log.info("load complete")

log.shutdown()

Checkpoints for Resumable Work

from pulselog import Logger

log = Logger("batch-job", checkpoint_path="batch.db")

# Check if we already completed this step
if log.load("step-2"):
    log.info("step-2 already done, skipping")
else:
    log.info("starting step-2")
    process_step_2()
    log.save("step-2", {"status": "complete"}, status="DONE", progress=50)

# Batch saves — one transaction, ~5x faster than individual saves
log.save_many([
    ("step-3a", {"rows": 1200}),
    ("step-3b", {"rows": 3400}),
], status="DONE")

log.shutdown()

Standard Library Integration

import logging
from pulselog.handler import PulseHandler

# Route standard logging to PulseLog
handler = PulseHandler("my-app")
logging.getLogger().addHandler(handler)
logging.getLogger().setLevel(logging.INFO)

logging.info("application started")
logging.warning("disk space low")
logging.error("connection failed")

try:
    risky_operation()
except Exception:
    logging.exception("operation failed")  # Includes traceback

handler.shutdown()

Monitoring Drops

from pulselog import Logger

log = Logger("high-throughput", queue_size=1000)

for i in range(1_000_000):
    log.info("processing", index=i)

log.flush()

stats = log.stats()
print(f"Processed: {stats.get('records_processed', 'N/A')}")
print(f"Dropped:   {stats.get('records_dropped', 'N/A')}")

if stats.get("records_dropped", 0) > 0:
    print("Consider increasing queue_size or reducing log volume")

log.shutdown()

Custom Worker Interval

from pulselog import Logger

log = Logger("realtime", worker_interval=0.001)   # Low latency (real-time dashboards)
log = Logger("batch", worker_interval=0.1)        # Low CPU (background jobs)
log = Logger("default", worker_interval=0.01)     # Good balance (10ms)

log.shutdown()

---

🖥️ Real-Time Dashboard

PulseLog includes a browser-based dashboard designed for real-time visibility into application logs.

It provides:

• real-time log updates
• severity filtering
• full-text search
• structured metadata
• automatic scrolling
• session export

Typical levels:

DEBUG
INFO
WARNING
ERROR
CRITICAL

Dashboard overhead when enabled: ~2% on median log-call latency (measured).

---

📊 Structured Logging

PulseLog supports structured metadata without requiring you to build formatted log strings manually.

log.info(
    "request completed",
    request_id="abc123",
    user_id=42,
    latency_ms=18,
    status_code=200,
)

Structured fields are useful for data pipelines, ML experiments, API services, batch jobs, model evaluation, debugging, and operational monitoring.

Adding 50 structured fields costs ~+2 µs per call.

---

🧵 Concurrent Logging

PulseLog is designed for applications where multiple threads produce logs concurrently.

from concurrent.futures import ThreadPoolExecutor
from pulselog import Logger

log = Logger("worker")

def process(i):
    log.info("processing item", item=i)

with ThreadPoolExecutor(max_workers=8) as executor:
    list(executor.map(process, range(10_000)))

log.shutdown()

---

💾 Checkpoints

PulseLog includes an embedded SQLite-backed (WAL mode) checkpoint store for long-running workflows.

Useful for:

• ML training
• ETL jobs
• experiments
• batch processing
• resumable workflows

log.save(
    name="epoch-5",
    data={
        "loss": 0.31,
        "accuracy": 0.94,
    },
    status="DONE",
    note="best model so far",
    progress=50,
)

Supported statuses:

DONE
IN_PROGRESS
FAILED
SKIPPED

result = log.load("epoch-5")              # Load one checkpoint
names = log.checkpoints()                 # List all names
log.delete_checkpoint("epoch-3")          # Delete

For tests or short-lived workloads:

log = Logger("test", checkpoint_path=":memory:")

Scaling guarantee: checkpoint loads are O(1) regardless of store size — verified flat at 1,000 / 100,000 / 500,000 / 1,000,000 entries (~0.7 µs at every size).

---

➗ Dividers

log.divider("epoch boundary")

---

🔄 Flush and Shutdown

Flush pending records:

success = log.flush(timeout=2.0)

Shutdown cleanly:

log.shutdown()

Explicit shutdown is recommended for long-running applications so pending records can be processed before termination. Shutdown during active concurrent writes is safe (verified under race testing).

---

📈 Runtime Statistics

PulseLog can expose runtime statistics through:

stats = log.stats()

Depending on configuration, statistics can include:

• records processed
• records dropped
• queue size / capacity / utilisation
• checkpoint activity
• dashboard clients
• uptime

---

⚙️ Configuration

Configuration can be supplied through:

1. Logger() arguments
2. environment variables
3. pulselog.toml
4. built-in defaults

Example:

log = Logger(
    name="my-app",
    host="localhost",
    port=5678,
    auto_open=True,
    dashboard=True,
    checkpoint_path=".pulselog/checkpoints.db",
    level="DEBUG",
    worker_interval=0.01,
)

Environment variables

export PULSELOG_DASHBOARD=false
export PULSELOG_HOST=0.0.0.0
export PULSELOG_PORT=8080
export PULSELOG_AUTO_OPEN=false
export PULSELOG_CHECKPOINT_PATH=/data/checkpoints.db
export PULSELOG_LEVEL=INFO
export PULSELOG_WORKER_INTERVAL=0.01

pulselog.toml

[pulselog]

host = "0.0.0.0"
port = 8080

auto_open = false
dashboard = true

level = "INFO"
worker_interval = 0.01
checkpoint_path = ".pulselog/checkpoints.db"

---

🏭 Production Usage

from pulselog import Logger

log = Logger(
    "production",
    dashboard=False,
    checkpoint_path="/data/checkpoints.db",
)

For CI:

export PULSELOG_DASHBOARD=false
export PULSELOG_AUTO_OPEN=false

---

⚡ Performance

All numbers below are reproducible via the included benchmark suite (python -m pulselog.benchmark) and were measured on Apple Silicon, Python 3.10, macOS. Run benchmarks in your own environment for exact figures.

At a Glance

Metric	Value	Notes
Hot-path latency (median)	~1.6 µs	Non-blocking enqueue; mean ≈ 2 µs incl. outliers
Sustained throughput	~520–540k logs/sec	Verified over continuous 60-second runs (31M+ messages)
Multi-thread throughput	~570k logs/sec	Flat from 1 → 64 threads
Burst absorption	1M messages / 1.7 s	+13 MB RSS — memory stays bounded
Decorator overhead	~4 µs per level	Linear to 250-deep nesting
Context manager cost	~2.5–4 µs	Per nested context level
Checkpoint load	~0.7 µs	O(1) at 1,000,000 entries (131 MB store)
Checkpoint save	~50 µs	WAL mode, synchronous=NORMAL
Batch checkpoint save	~5× faster than individual	Single transaction

Hot-Path Latency

The key performance property: logging does not block your application.

log.info("message")  # Returns in ~2 µs (message queued, not delivered)

End-to-end latency (queue → handler) depends on worker_interval:

worker_interval	P50 delivery latency	P99 delivery latency
0.001s (1ms)	~0.6ms	~1ms
0.01s (10ms)	~6ms	~7ms
0.1s (100ms)	~60ms	~70ms

Throughput & Concurrency

Producer throughput (messages enqueued per second):

Single thread:    ~450,000 – 600,000 ops/sec
Multi-thread:     ~570,000 ops/sec aggregate (flat up to 64 threads)

Because PulseLog's enqueue path holds no coarse locks, adding producer threads causes no meaningful contention: aggregate throughput stays within ~20% of single-thread peak even at 64 concurrent producers. Note that CPython's GIL caps pure-Python enqueue throughput at roughly one core — multi-core scale comes from running multiple processes, each with its own Logger.

Memory

PulseLog uses bounded memory by design:

• Queue size is configurable (default: 10,000)
• No unbounded growth — a 1M-message instantaneous burst added only ~13 MB RSS
• Zero memory leaks detected across repeated leak checks (< 1 MB growth over millions of operations)
• LogRecord overhead: ~80 bytes per message

Backpressure Behavior

When the queue is full, new messages are dropped and counted:

Queue capacity: 1,000
Messages sent:  100,000
Result:         ~38,000 processed, ~62,000 dropped (counted, not lost silently)

This is by design — it prevents logging from causing out-of-memory errors in your application. Monitor stats()["records_dropped"] to track drops.

To reduce drops under high load:

• Increase queue_size (trades memory for drop tolerance)
• Decrease worker_interval (trades CPU for faster drain)
• Reduce message volume or batch logs

Exception Logging

Exception logging (with traceback) is slower due to Python's traceback.format_exc():

Plain log.info():           ~450,000 ops/sec
log.exception():            ~75,000 ops/sec  (6x slower)

This is expected and acceptable since exception logging should be rare in production.

What Affects Performance

Factor	Impact
worker_interval	Lower = lower latency, slightly higher CPU
queue_size	Larger = more buffering, more memory
Message size	Minimal impact (messages are references, not copied)
Structured fields	Minimal (+~2 µs for 50 fields)
Number of handlers	Linear impact per handler
CPU-bound competitors	Significant (GIL contention)

Reliability Under Stress

Verified behaviors from adversarial limit testing:

• ✅ Sustained load: 31M+ messages over 60 s without failure
• ✅ Shutdown-during-writes: zero errors, zero hung threads across concurrent producers
• ✅ Corrupt database files raise explicit sqlite3.DatabaseError (no silent corruption)
• ✅ Non-serializable payloads raise TypeError immediately (fail-fast, not silent loss)
• ✅ Read-only working directories fall back to a writable temp location
• ✅ Multi-process checkpoint writers supported via SQLite WAL + busy-timeout

Running Benchmarks

# Quick benchmark
python -m pulselog.benchmark

# Limit tests (sustained, saturation, failure modes)
python -m pulselog.benchmark --stress

Benchmark results depend on Python version, OS, CPU, and configuration. Always run benchmarks in your own environment for accurate numbers.

---

🔒 Reliability

PulseLog is designed around:

• bounded buffering
• asynchronous processing
• graceful shutdown
• structured records
• checkpoint persistence (SQLite WAL)
• concurrent producer support
• configurable runtime behaviour

Applications should still treat logging as an auxiliary system and avoid placing critical business state exclusively in logs.

---

🧩 Design Goals

Goal	Description
Low application overhead	Keep logging work away from the main application path
Non-blocking operation	Avoid waiting for dashboard consumers
Bounded memory	Prevent an unlimited logging backlog
Structured data	Preserve useful metadata
Concurrency	Support multiple producer threads without lock degradation
Batch processing	Process pending records efficiently
Real-time visibility	Make application behaviour visible in a browser
Resumable workflows	Provide O(1) checkpoint support at any store size
Python-first API	Keep the public API simple

---

📦 Requirements

• Python 3.8+
• websockets >= 11.0

For Python versions below 3.11, PulseLog uses tomli for TOML configuration support.

---

🧪 Development

git clone <repository-url>
cd pulselog

pip install -e ".[dev]"
pytest

---

📊 Benchmarking

The project includes two suites:

Core benchmarks (python -m pulselog.benchmark):

Latency · Throughput · Concurrency · CPU · Memory
Queue pressure · Drops · Shutdown · Sustained load

Limit tests (--stress) — adversarial scenarios:

60-second saturation      Thread scaling 1→64
Payload sizes 10B→1MB     Nesting depth 1→250
1M-message bursts         1M-row checkpoint stores
Multi-process WAL writes  Failure modes (deleted/corrupt DB, races)

Every regression-sensitive claim in this README (checkpoint O(1) loads, save latency, drop accounting) is enforced as an assertion in the suite — if a future change breaks it, the benchmark fails rather than the docs going stale.

Recommended concurrency levels: 1 2 4 8 16 32 64 Recommended message sizes: 32 B 128 B 512 B 1 KB 4 KB 16 KB 64 KB Recommended structured-field counts: 0 1 5 10 25 50 100

Benchmark results should always include the machine and Python environment used for the measurement.

---

🧭 Roadmap

Potential areas of development include:

• OpenTelemetry integration
• additional output handlers (syslog, Kafka, OTLP)
• distributed/multi-node checkpoint coordination
• richer backpressure controls (spill-to-disk)
• dashboard improvements (live metrics charts)
• persistent log storage options
• further performance improvements
• long-duration stability testing

---

📌 Project Status

Current version:

2.0.1

Development status:

Alpha

PulseLog is actively evolving and the public API may change during early releases.

For reproducible deployments:

pip install "pulselog==2.0.1"

---

📄 License

MIT License.

---