API: Window

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

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

---

fluxgate.windows.CountWindow

CountWindow(size: int)

Bases: IWindow

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.

Examples:

>>> window = CountWindow(size=100)  # Track last 100 calls
>>>
>>> window.record(Record(success=True, duration=0.5))
>>>
>>> metric = window.get_metric()
>>> print(metric.total_count)
1

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._max_size = size
    self._records: deque[Record] = deque(maxlen=size)
    self._total_count = 0
    self._total_failure_count = 0
    self._total_duration = 0.0
    self._slow_call_count = 0

record

record(record: Record) -> None

Add a call record to the window.

If window is full, evicts the oldest record before adding the new one.

Parameters:

Name	Type	Description	Default
record	Record	Call result to record	required

Source code in fluxgate/windows.py

def record(self, record: Record) -> None:
    """Add a call record to the window.

    If window is full, evicts the oldest record before adding the new one.

    Args:
        record: Call result to record
    """
    if len(self._records) == self._max_size:
        evicted = self._records.popleft()
        self._evict(evicted)
    self._admit(record)

get_metric

get_metric() -> Metric

Get aggregated metrics for all records in the window.

Returns:

Type	Description
Metric	Aggregated metrics (counts, durations)

Source code in fluxgate/windows.py

def get_metric(self) -> Metric:
    """Get aggregated metrics for all records in the window.

    Returns:
        Aggregated metrics (counts, durations)
    """
    return Metric(
        total_count=self._total_count,
        failure_count=self._total_failure_count,
        total_duration=self._total_duration,
        slow_count=self._slow_call_count,
    )

reset

reset() -> None

Clear all records and reset metrics to zero.

Source code in fluxgate/windows.py

def reset(self) -> None:
    """Clear all records and reset metrics to zero."""
    self._records.clear()
    self._total_count = 0
    self._total_failure_count = 0
    self._total_duration = 0.0
    self._slow_call_count = 0

fluxgate.windows.TimeWindow

TimeWindow(size: int)

Bases: IWindow

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.

Examples:

>>> window = TimeWindow(size=60)  # Track last 60 seconds
>>>
>>> window.record(Record(success=True, duration=0.5))
>>>
>>> metric = window.get_metric()

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._buckets = [self.Bucket() for _ in range(size)]
    self._timestamps = [0 for _ in range(size)]
    self._total_count = 0
    self._total_failure_count = 0
    self._total_duration = 0.0
    self._slow_call_count = 0

Bucket dataclass

Bucket(
    sec_count: int = 0,
    sec_failure_count: int = 0,
    sec_total_duration: float = 0.0,
    sec_slow_call_count: int = 0,
)

sec_count class-attribute instance-attribute

sec_count: int = field(default=0)

sec_failure_count class-attribute instance-attribute

sec_failure_count: int = field(default=0)

sec_total_duration class-attribute instance-attribute

sec_total_duration: float = field(default=0.0)

sec_slow_call_count class-attribute instance-attribute

sec_slow_call_count: int = field(default=0)

admit

admit(record: Record) -> None

Source code in fluxgate/windows.py

def admit(self, record: Record) -> None:
    self.sec_count += 1
    self.sec_failure_count += 1 if not record.success else 0
    self.sec_total_duration += record.duration
    self.sec_slow_call_count += 1 if record.is_slow else 0

reset

reset() -> None

Source code in fluxgate/windows.py

def reset(self) -> None:
    self.sec_count = 0
    self.sec_failure_count = 0
    self.sec_total_duration = 0.0
    self.sec_slow_call_count = 0

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
    bucket = self._buckets[index]

    if self._timestamps[index] != now:
        evicted = self._buckets[index]
        self._evict(evicted)
        self._timestamps[index] = now

    self._admit(record, bucket)

get_metric

get_metric() -> Metric

Source code in fluxgate/windows.py

def get_metric(self) -> Metric:
    return Metric(
        total_count=self._total_count,
        failure_count=self._total_failure_count,
        total_duration=self._total_duration,
        slow_count=self._slow_call_count,
    )

reset

reset() -> None

Source code in fluxgate/windows.py

def reset(self) -> None:
    for bucket in self._buckets:
        bucket.reset()
    self._timestamps = [0 for _ in range(self._size)]
    self._total_count = 0
    self._total_failure_count = 0
    self._total_duration = 0.0
    self._slow_call_count = 0