Filters

Two logging.Filter implementations tame a logger that says the same thing too often. Both attach to any stdlib logger, so they work with every GREL_LOG_BACKEND.

For dropping Kubernetes probe lines from the access log, see Quieting health probes.

Deduplicating Noisy Logs

DuplicateFilter silences repeated log records.

from logging import getLogger

from grelmicro.log import DuplicateFilter

logger = getLogger("grelmicro.health")
logger.addFilter(DuplicateFilter())

After 5 identical records, the filter silently drops any further occurrences. It tracks up to 100 distinct keys in an LRU cache.

key_mode="template" (default) uses the raw format string as the key, so %-style calls with different arguments share one counter. It is also about 3 times faster than rendered keying. DuplicateFilter and RateLimitFilter share the same five key_mode values:

key_mode	Counter scope	Good for
"logger"	One counter per logger name	Collapse every record from a noisy logger
"level"	One counter per log level	Collapse all records at a level
"global"	One shared counter	Collapse everything into one budget
"template" (default)	One counter per (logger, level, str(record.msg))	Shares across arg values of the same template
"rendered"	One counter per (logger, level, record.getMessage())	Distinguishes fully-rendered messages

Use key_mode="rendered" to track each rendered message separately, or pass key= for a custom fingerprint:

logger.addFilter(DuplicateFilter(key_mode="rendered"))
logger.addFilter(DuplicateFilter(key=lambda r: (r.name, r.exc_info)))

Set ttl to re-emit a burst of allowed_repetitions records every window during sustained floods, so operators continue to receive periodic reminders:

logger.addFilter(DuplicateFilter(allowed_repetitions=5, ttl=300))

State is in-process only. There is no cross-process sharing and no explicit reset API: construct a new filter if you need to wipe counters.

Tip

For code using from loguru import logger or structlog.get_logger() directly, use those libraries' native filtering.

Rate-Limiting Noisy Logs

RateLimitFilter drops records when a token bucket is empty. It allows bursts: up to capacity records can pass through at once, and the bucket then refills at refill_rate records per second.

from logging import getLogger

from grelmicro.log import RateLimitFilter

# Allow a burst of 10 records per logger, then 1 record/sec sustained.
logger = getLogger("grelmicro.ingest")
logger.addFilter(RateLimitFilter(capacity=10, refill_rate=1))

By default the filter buckets per logger: each logger has its own burst budget. Swap key_mode for different grouping:

key_mode	Bucket scope	Good for
"logger" (default)	One bucket per logger name	Noisy third-party libraries that flood a single logger
"level"	One bucket per log level	Throttle all WARNING/ERROR across the app
"global"	One shared bucket	App-wide safety net on the root handler
"template"	One bucket per (logger, level, str(record.msg))	Shares across arg values of the same template
"rendered"	One bucket per (logger, level, record.getMessage())	Distinguishes fully-rendered messages

from logging import getLogger

from grelmicro.log import RateLimitFilter

# One shared bucket across every record the handler sees.
# Useful on the root or app-level handler as a global safety net.
root = getLogger()
root.addFilter(RateLimitFilter(capacity=100, refill_rate=10, key_mode="global"))

Pass a custom key= callable for any other grouping:

logger.addFilter(
    RateLimitFilter(
        capacity=20,
        refill_rate=2,
        key=lambda r: f"{r.name}|{r.exc_info is not None}",
    )
)

Use cost= when a record should spend multiple tokens (e.g. on a verbose-level handler):

logger.addFilter(RateLimitFilter(capacity=100, refill_rate=10, cost=2))

State is in-process only, backed by MemoryTokenBucket. Call filter.reset(key) to clear one key, or construct a new filter to wipe all state.

Tip

RateLimitFilter and DuplicateFilter compose well: attach the dedup filter first to collapse true duplicates, then the rate-limit filter to cap the sustained flow.