PacedFilter

Rate-limit DEBUG records from the chatty vendored stacks by template.

Records whose logger name starts with one of PACED_PREFIXES pass the first time their message template is seen and then once per period seconds; a record that passes after held-back repeats says how many it stands for. Every other record, and every record at INFO or above, passes untouched.

Attributes

attributeperiod

= period

Functions

constructor__init__(period=PACE_PERIOD_S) -> None

Source Code

def __init__(self, period: float = PACE_PERIOD_S) -> None:
    super().__init__()
    self.period = period
    self._seen: Dict[Tuple[str, str], Tuple[float, int]] = {}

paramperiodfloat

= PACE_PERIOD_S

Returns

None

funcfilter(record) -> bool

Source Code

def filter(self, record: logging.LogRecord) -> bool:
    if record.levelno >= logging.INFO or not record.name.startswith(PACED_PREFIXES):
        return True
    key = (record.name, str(record.msg))
    now = time.monotonic()
    last, held = self._seen.get(key, (None, 0))
    if last is not None and now - last < self.period:
        self._seen[key] = (last, held + 1)
        return False
    self._seen[key] = (now, 0)
    if held:
        record.msg = f"{record.msg} (+{held} alike in the last {self.period:.0f}s)"
    return True

paramrecordlogging.LogRecord

Returns

bool

Attributes

Functions

On this page