Skip to content

Metrics

LogMetricsHandler counts log records by level and computes a logs-per-second rate. Attach it to a logger with enable_metrics=True.

Quick Reference

Python
from pylogshield import get_logger

logger = get_logger("app", enable_metrics=True)

logger.info("Start")
logger.error("Something failed")

metrics = logger.get_metrics()
# {
#   'INFO': 0.5,      # logs/second for this level
#   'ERROR': 0.5,
#   'count': 2,       # total log count
#   'elapsed': 4.0,   # seconds since creation or last reset
#   'start': 12345.6  # monotonic timestamp of handler creation
# }

# Count only
if logger.metrics_handler:
    print(logger.metrics_handler.counts())
    # {'INFO': 1, 'ERROR': 1}

# Reset counters
    logger.metrics_handler.reset()

Examples

Health-check endpoint (FastAPI)

Expose log metrics through a /health endpoint so your monitoring stack can alert when error rates spike.

Python
from fastapi import FastAPI
from pylogshield import get_logger

app = FastAPI()
logger = get_logger("api", enable_metrics=True, enable_json=True)

@app.get("/health")
def health():
    metrics = logger.get_metrics()
    if metrics is None:
        return {"status": "ok"}
    error_rate = metrics.get("ERROR", 0.0)
    return {
        "status": "degraded" if error_rate > 5.0 else "ok",
        "log_counts": logger.metrics_handler.counts(),
        "error_rate_per_sec": round(error_rate, 3),
        "total_logs": metrics["count"],
        "uptime_seconds": round(metrics["elapsed"], 1),
    }

Periodic metrics emission

Reset counters every minute and log a summary line:

Python
import threading
from pylogshield import get_logger

logger = get_logger("worker", enable_metrics=True, enable_json=True)

def _emit_and_reset():
    while True:
        threading.Event().wait(60)  # wait 60 s
        metrics = logger.get_metrics()
        if metrics:
            logger.info({
                "event": "metrics",
                "counts": logger.metrics_handler.counts(),
                "error_rate": metrics.get("ERROR", 0.0),
                "elapsed": metrics["elapsed"],
            })
            logger.metrics_handler.reset()

threading.Thread(target=_emit_and_reset, daemon=True).start()

Thread Safety

All methods on LogMetricsHandler are thread-safe. emit(), counts(), logs_per_second(), and reset() are all protected by an internal threading.Lock.

API Reference

LogMetricsHandler()

Bases: Handler

Lightweight handler that counts emitted logs by level and estimates logs per second.

This handler is useful for monitoring log volume and identifying potential issues like excessive logging that could impact performance.

ATTRIBUTE DESCRIPTION
_counts

Counter tracking log counts by level name.

TYPE: Counter

_start

Monotonic timestamp when the handler was created or last reset.

TYPE: float

Examples:

Python Console Session
>>> logger = PyLogShield("app", enable_metrics=True)
>>> logger.info("test")
>>> metrics = logger.metrics_handler.logs_per_second()
>>> print(f"Total logs: {metrics['count']}")
METHOD DESCRIPTION
emit

Count the log record by its level name.
counts

Return counts by level.
logs_per_second

Return a metrics dictionary including per-level rates and totals.
reset

Reset counters and timing.
Source code in src/pylogshield/metrics.py
Python
31
32
33
34
35
def __init__(self) -> None:
    super().__init__()
    self._lock = threading.Lock()
    self._counts: Counter[str] = Counter()
    self._start = time.monotonic()

emit(record: logging.LogRecord) -> None

Count the log record by its level name.

PARAMETER DESCRIPTION

record

The log record to count.

TYPE: LogRecord

Source code in src/pylogshield/metrics.py
Python
37
38
39
40
41
42
43
44
45
46
def emit(self, record: logging.LogRecord) -> None:
    """Count the log record by its level name.

    Parameters
    ----------
    record : logging.LogRecord
        The log record to count.
    """
    with self._lock:
        self._counts[record.levelname] += 1

counts() -> Dict[str, int]

Return counts by level.

RETURNS DESCRIPTION
dict of str to int

Dictionary mapping level names (e.g., "INFO", "ERROR") to their counts.
Source code in src/pylogshield/metrics.py
Python
48
49
50
51
52
53
54
55
56
57
def counts(self) -> Dict[str, int]:
    """Return counts by level.

    Returns
    -------
    dict of str to int
        Dictionary mapping level names (e.g., "INFO", "ERROR") to their counts.
    """
    with self._lock:
        return dict(self._counts)

logs_per_second() -> Dict[str, Any]

Return a metrics dictionary including per-level rates and totals.

RETURNS DESCRIPTION
dict

Dictionary containing: - Per-level log rates (logs/second) - count: Total log count - elapsed: Elapsed time in seconds - start: Start timestamp (monotonic)
Source code in src/pylogshield/metrics.py
Python
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def logs_per_second(self) -> Dict[str, Any]:
    """Return a metrics dictionary including per-level rates and totals.

    Returns
    -------
    dict
        Dictionary containing:
        - Per-level log rates (logs/second)
        - ``count``: Total log count
        - ``elapsed``: Elapsed time in seconds
        - ``start``: Start timestamp (monotonic)
    """
    with self._lock:
        elapsed = max(0.001, time.monotonic() - self._start)
        rates: Dict[str, Any] = {
            lvl: cnt / elapsed for lvl, cnt in self._counts.items()
        }
        rates["count"] = sum(self._counts.values())
        rates["elapsed"] = elapsed
        rates["start"] = self._start
        return rates

reset() -> None

Reset counters and timing.

Clears all level counts and restarts the elapsed time measurement. Useful for testing or periodic metric snapshots.

Source code in src/pylogshield/metrics.py
Python
102
103
104
105
106
107
108
109
110
def reset(self) -> None:
    """Reset counters and timing.

    Clears all level counts and restarts the elapsed time measurement.
    Useful for testing or periodic metric snapshots.
    """
    with self._lock:
        self._counts.clear()
        self._start = time.monotonic()