API: Window

이 페이지는 사용 가능한 Window를 문서화합니다. Window는 최근 호출에 대한 Metric을 저장하고 제공하는 역할을 합니다.

Window 선택에 대한 개요는 Windows 컴포넌트 가이드를 참조하십시오.

---

fluxgate.windows.CountWindow

CountWindow(size: int)

Bases: Window

Count-based sliding window for tracking recent call metrics.

Maintains a fixed-size sliding window that keeps the most recent N calls. When the window is full, the oldest record is evicted when a new one arrives.

Per-threshold slow-call counters are populated from record.slow_at — the set of thresholds the call exceeded, decided by the producer (typically the circuit breaker). The window itself does not need to know which thresholds are active.

Examples:

>>> window = CountWindow(size=100)
>>> window.record(Record(success=True, duration=0.5, slow_at=()))
>>> metric = window.get_metric()

Parameters:

Name	Type	Description	Default
size	int	Maximum number of calls to track in the window	required

Source code in fluxgate/windows.py

def __init__(self, size: int) -> None:
    self._records: deque[Record] = deque(maxlen=size)
    self._metric = Metric.empty()

record

record(record: Record) -> None

Source code in fluxgate/windows.py

def record(self, record: Record) -> None:
    if len(self._records) == self._records.maxlen:
        evicted = self._records.popleft()
        self._metric = self._metric - Metric.from_record(evicted)
    self._records.append(record)
    self._metric = self._metric + Metric.from_record(record)

get_metric

get_metric() -> Metric

Source code in fluxgate/windows.py

def get_metric(self) -> Metric:
    return self._metric

reset

reset() -> None

Source code in fluxgate/windows.py

def reset(self) -> None:
    self._records.clear()
    self._metric = Metric.empty()

fluxgate.windows.TimeWindow

TimeWindow(size: int)

Bases: Window

Time-based sliding window for tracking metrics over a time period.

Divides time into fixed buckets (1 second each) and tracks metrics per bucket. When a bucket's time period expires, it is reset and reused for the new time.

Per-threshold slow-call counters are populated from record.slow_at, just like CountWindow.

Parameters:

Name	Type	Description	Default
size	int	Number of seconds to track (window size in seconds)	required

Note

Time precision is 1 second. All calls within the same second are grouped into the same bucket.

Source code in fluxgate/windows.py

def __init__(self, size: int) -> None:
    self._size = size
    self.reset()

record

record(record: Record) -> None

Source code in fluxgate/windows.py

def record(self, record: Record) -> None:
    now = int(record.timestamp)
    index = now % self._size

    if self._timestamps[index] != now:
        self._total = self._total - self._buckets[index]
        self._buckets[index] = Metric.empty()
        self._timestamps[index] = now

    contribution = Metric.from_record(record)
    self._buckets[index] = self._buckets[index] + contribution
    self._total = self._total + contribution

get_metric

get_metric() -> Metric

Source code in fluxgate/windows.py

def get_metric(self) -> Metric:
    return self._total

reset

reset() -> None

Source code in fluxgate/windows.py

def reset(self) -> None:
    self._buckets = [Metric.empty() for _ in range(self._size)]
    self._timestamps = [0 for _ in range(self._size)]
    self._total = Metric.empty()