⚡ 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_checkpoint(
        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: ~2.5 µs per decoration level, verified linear up to 250-deep nesting (see Performance).

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_checkpoint("step-2"):
    log.info("step-2 already done, skipping")
else:
    log.info("starting step-2")
    process_step_2()
    log.save_checkpoint("step-2", {"status": "complete"}, status="DONE", progress=50)

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

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: ~17% on median log-call latency (measured). Enable only for development/debugging; disable in production for maximum performance.

---

📊 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_checkpoint(
    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_checkpoint("epoch-5")              # Load one checkpoint
names = log.list_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.6-1.4 µ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).

Post-shutdown behavior: After shutdown(), subsequent log calls are silently ignored (safe no-op). This is a documented contract to prevent crashes during application shutdown.

---

📈 Runtime Statistics

PulseLog can expose runtime statistics through:

stats = log.stats()

Depending on configuration, statistics can include:

Key	Description
records_logged	Total records accepted by the logger
records_dropped	Records dropped due to queue saturation
queue_size	Current number of queued records
queue_capacity	Maximum queue capacity
queue_fill_pct	Queue utilization percentage
checkpoints_saved	Number of checkpoints persisted
dashboard_clients	Connected dashboard clients
uptime_seconds	Logger uptime in seconds

---

⚙️ 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,           # Disable dashboard in production
    checkpoint_path="/data/checkpoints.db",
    queue_size=500_000,        # Larger queue for burst tolerance
    overflow="drop",           # Drop when full (monitor stats)
)

# Monitor drops in production
stats = log.stats()
if stats.get("records_dropped", 0) > 0:
    alert("Logs are being dropped! Increase queue_size or reduce volume.")

log.shutdown()

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)	~0.7-1.0 µs	Non-blocking enqueue; mean ≈ 1-2 µs incl. outliers
Hot-path latency (p99)	~1.0-2.0 µs	Reliable tail latency
Producer throughput (single thread)	~400-600k logs/sec	Raw enqueue rate
Producer throughput (multi-thread)	~1.2M logs/sec	8 threads, aggregate
End-to-end sustained throughput	90-230k logs/sec	Verified up to 10M records, zero drops
Burst absorption	1M messages / 4.4s	Memory stable, zero drops
Decorator overhead	~2.5 µs per level	Linear to 250-deep nesting
Context manager cost	~2-4 µs	Per nested context level
Checkpoint load	~0.6-1.4 µs	O(1) at 1,000,000 entries
Checkpoint save	~30-70 µs	WAL mode, synchronous=NORMAL
Batch checkpoint save	~5× faster than individual	Single transaction
Memory stability	<1 MB growth	Verified over 200k+ operations

Understanding Throughput Numbers

PulseLog performance is measured at different layers:

1. Producer enqueue rate (microbenchmark):

   • info() with kwargs: ~400-600k ops/sec
   • info_fast() (no kwargs): ~800k-1M ops/sec

2. End-to-end sustained throughput (production workload):

   • With handlers and multi-threading: 90-230k records/sec
   • This is the realistic throughput for production use
   • Verified up to 10 million records with zero drops

3. Why the difference?

   • End-to-end includes: producer enqueue → worker drain → handler processing
   • Handlers (dashboard, callbacks) add overhead
   • CPython GIL limits pure-Python throughput to roughly one core

Hot-Path Latency

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

log.info("message")  # Returns in ~1 µ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:    ~400,000 – 600,000 ops/sec
Multi-thread:     ~1,200,000 ops/sec aggregate (8 threads)

Because PulseLog's enqueue path holds no coarse locks, adding producer threads causes no meaningful contention: aggregate throughput scales well with thread count. 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.

End-to-End Sustained Throughput (Production Verified)

Records	Duration	Throughput	Drops	Status
1,000,000	4.4s	228k/sec	0	✅
5,000,000	44.8s	112k/sec	0	✅
10,000,000	111.5s	90k/sec	0	✅

Zero data loss at all scales. Memory stable. Verified with 8 producers and handlers.

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:         ~100 processed, ~99,900 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
Dashboard enabled	~17% overhead on median latency
CPU-bound competitors	Significant (GIL contention)

Reliability Under Stress

Verified behaviors from adversarial limit testing:

Test	Result
Sustained load (10M+ messages)	✅ Zero drops, stable memory
Shutdown-during-writes	✅ Zero errors, no hung threads
Corrupt database files	✅ Raises explicit sqlite3.DatabaseError
Non-serializable payloads	✅ Raises TypeError immediately
Read-only directories	✅ Falls back to writable temp location
Multi-process checkpoint writers	✅ Supported via SQLite WAL + busy-timeout
Memory leak checks	✅ <1 MB growth over millions of operations

Running Benchmarks

# Quick benchmark
python -m pulselog.benchmark

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

# Run sustained throughput tests
PULSELOG_SUSTAINED=1 python test_pulselog_production.py

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
• zero data loss when queue not saturated
• explicit drop accounting via stats()

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
Zero silent data loss	Every dropped record is counted and reported

---

📦 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)

Production validation (test_pulselog_production.py):

Core logging performance   Dashboard overhead
Decorator overhead         Multi-thread throughput
Queue saturation (drop/raise/block)
Flush synchronization      Concurrent operations
Memory leak checks         Checkpoint performance
Shutdown behavior          Worker failure handling

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.

---

📌 Project Status

Current version:

2.1.1

Development status:

Production Candidate ✅

PulseLog has been validated through extensive testing including:

• ✅ 10M+ record sustained throughput tests
• ✅ Zero data loss verification
• ✅ Memory leak testing
• ✅ Concurrent producer testing (8+ threads)
• ✅ Shutdown safety testing
• ✅ Checkpoint performance validation
• ✅ All benchmarks passing

The public API is stabilizing and will be maintained through future releases.

For reproducible deployments:

pip install "pulselog==2.1.1"

---

📄 License

MIT License.

---