from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from olc.config.properties_loader import OlcConfigManager
from olc.statistic.request_stat_window import RequestStatSlidingWindow
from olc.utils.atomic_num import AtomicNum
@dataclass
class WindowSnapshot:
total_request_num: int = 0
pass_request_num: int = 0
block_request_num: int = 0
pass_avg_latency_ms: float = 0.0
pass_max_latency_ms: float = 0.0
pass_min_latency_ms: float = 0.0
@dataclass
class StatisticSnapshot:
cur_thread_num: int = 0
second_snap: WindowSnapshot = field(default_factory=WindowSnapshot)
custom_snap: WindowSnapshot = field(default_factory=WindowSnapshot)
@property
def total_request_num_pre_second(self) -> int:
return self.second_snap.total_request_num
@property
def pass_request_num_pre_second(self) -> int:
return self.second_snap.pass_request_num
@property
def block_request_num_pre_second(self) -> int:
return self.second_snap.block_request_num
@property
def pass_avg_latency_ms_pre_second(self) -> float:
return self.second_snap.pass_avg_latency_ms
@property
def pass_max_latency_ms_pre_second(self) -> float:
return self.second_snap.pass_max_latency_ms
@property
def pass_min_latency_ms_pre_second(self) -> float:
return self.second_snap.pass_min_latency_ms
class IStatistic(ABC):
@abstractmethod
def inc_thread_num(self) -> None:
"""
增加当前并发数
:return:
"""
@abstractmethod
def dec_thread_num(self) -> None:
"""
减少当前并发数，最低限制为 0
:return:
"""
@abstractmethod
def cur_thread_num(self) -> int:
"""
获取当前并发数
:return: 当前并发数
"""
@abstractmethod
def total_request_num(self) -> int:
"""
获取秒级滑动窗口内的总请求数（通过数 + 拒绝数）
:return: 最近 1 秒内的总请求数
"""
@abstractmethod
def inc_pass_request_num(self) -> None:
"""
增加通过请求计数（秒级和分钟级滑动窗口同时记录）
:return:
"""
@abstractmethod
def pass_request_num(self) -> int:
"""
获取秒级滑动窗口内的通过请求数
:return: 最近 1 秒内的通过请求数
"""
@abstractmethod
def inc_block_request_num(self) -> None:
"""
增加拒绝请求计数（秒级和分钟级滑动窗口同时记录）
:return:
"""
@abstractmethod
def block_request_num(self) -> int:
"""
获取秒级滑动窗口内的拒绝请求数
:return: 最近 1 秒内的拒绝请求数
"""
@abstractmethod
def record_pass_latency_ms(self, latency_ms: float) -> None:
"""
记录单次通过请求延迟（秒级和分钟级滑动窗口同时记录）
:param latency_ms: 单次请求延迟（毫秒），负数会被忽略
:return:
"""
@abstractmethod
def pass_avg_latency_ms(self) -> float:
"""
获取秒级滑动窗口内的通过请求平均延迟
:return: 最近 1 秒内的平均延迟（毫秒），无记录时返回 0.0
"""
@abstractmethod
def pass_max_latency_ms(self) -> float:
"""
获取秒级滑动窗口内的通过请求最大延迟
:return: 最近 1 秒内的最大延迟（毫秒）
"""
@abstractmethod
def pass_min_latency_ms(self) -> float:
"""
获取秒级滑动窗口内的通过请求最小延迟
:return: 最近 1 秒内的最小延迟（毫秒）
"""
@abstractmethod
def snapshot(self) -> StatisticSnapshot:
"""
获取当前统计快照，包含秒级和分钟级双粒度数据
:return: 统计快照对象
"""
class OlcStatistic(IStatistic):
def __init__(self):
window_interval_in_sec = OlcConfigManager.get_instance().get_sdk_config().window_interval_in_sec
window_duration_ms = OlcConfigManager.get_instance().get_sdk_config().window_length_in_ms
self._cur_thread_num = AtomicNum(0)
self._second_window = RequestStatSlidingWindow(total_window_ms=1000, window_duration_ms=100)
self._custom_window = RequestStatSlidingWindow(total_window_ms=window_interval_in_sec * 1000,
window_duration_ms=window_duration_ms)
def inc_thread_num(self) -> None:
self._cur_thread_num.add(1)
def dec_thread_num(self) -> None:
self._cur_thread_num.dec_and_limit(1, 0)
def cur_thread_num(self) -> int:
return int(self._cur_thread_num.get())
def total_request_num(self) -> int:
buckets = self._second_window.get_all_buckets()
total: int = 0
for bucket in buckets:
total += bucket.get_pass_num() + bucket.get_block_num()
return total
def inc_pass_request_num(self) -> None:
self._second_window.record_pass_num()
self._custom_window.record_pass_num()
def pass_request_num(self) -> int:
return self._second_window.get_pass_num_sum()
def inc_block_request_num(self) -> None:
self._second_window.record_block_num()
self._custom_window.record_block_num()
def block_request_num(self) -> int:
return self._second_window.get_block_num_sum()
def record_pass_latency_ms(self, latency_ms: float) -> None:
if latency_ms < 0:
return
self._second_window.record_pass_rt(latency_ms)
self._custom_window.record_pass_rt(latency_ms)
def pass_avg_latency_ms(self) -> float:
return self._second_window.get_pass_avg_rt()
def pass_max_latency_ms(self) -> float:
max_val = self._second_window.get_pass_rt_max()
return 0.0 if max_val is None else max_val
def pass_min_latency_ms(self) -> float:
min_val = self._second_window.get_pass_rt_min()
return 0.0 if min_val is None else min_val
def _build_window_snapshot(self, window: RequestStatSlidingWindow) -> WindowSnapshot:
buckets = window.get_all_buckets()
pass_num: int = 0
block_num: int = 0
pass_rt_sum: float = 0.0
pass_rt_min: float | None = None
pass_rt_max: float | None = None
for bucket in buckets:
pass_num += bucket.get_pass_num()
block_num += bucket.get_block_num()
pass_rt_sum += bucket.get_pass_rt()
bucket_min = bucket.get_min_pass_rt()
if bucket_min is not None:
if pass_rt_min is None or bucket_min < pass_rt_min:
pass_rt_min = bucket_min
bucket_max = bucket.get_max_pass_rt()
if bucket_max is not None:
if pass_rt_max is None or bucket_max > pass_rt_max:
pass_rt_max = bucket_max
latency_min = 0.0 if pass_rt_min is None else pass_rt_min
latency_max = 0.0 if pass_rt_max is None else pass_rt_max
latency_avg = pass_rt_sum / pass_num if pass_num > 0 else 0.0
return WindowSnapshot(
total_request_num=pass_num + block_num,
pass_request_num=pass_num,
block_request_num=block_num,
pass_avg_latency_ms=latency_avg,
pass_max_latency_ms=latency_max,
pass_min_latency_ms=latency_min,
)
def snapshot(self) -> StatisticSnapshot:
return StatisticSnapshot(
cur_thread_num=int(self._cur_thread_num.get()),
second_snap=self._build_window_snapshot(self._second_window),
custom_snap=self._build_window_snapshot(self._custom_window),
)