import logging
import threading
from typing import List, Optional
from olc.bean.match_wrapper import MatchWrapper
from olc.bean.olc_config_rule import FlowPolicy, PolicyType, FlowControlMode
from olc.cache.safe__ttl_cache import SafeTTLCache
from olc.event.sny_event import Event
from olc.limit.abs_dynamic_limiter import AbsDynamicLimiter
from olc.limit.limiter import Limit
from olc.limit.node.dynamic.dynamic_concurrent_limiter import DynamicConcurrentLimiter
from olc.limit.node.static.concurrent_limiter import ConcurrentLimiter
from olc.limit.node.static.qps_limiter import QpsLimiter
from olc.limit.node.static.quota_limiter import QuotaLimiter
from olc.utils.str_utils import StrUtils
logger = logging.getLogger(__name__)
class OlcRateLimiterFactory:
"""
限速器工厂类
"""
_instance = None
_lock = threading.RLock()
def __init__(self):
self._cache = SafeTTLCache[Limit](maxsize=100000, ttl=60)
self._cache_lock = threading.RLock()
Event.get_instance().olc_clean_rule.connect(self.subscribe_clean)
@classmethod
def get_instance(cls):
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = cls()
return cls._instance
def get_rate_limiters(
self, wrappers: List[MatchWrapper[FlowPolicy]]
) -> List[Limit]:
limiters = []
for wrapper in wrappers:
limiter = self.get_rate_limiter(wrapper)
if limiter is not None:
limiters.append(limiter)
else:
logger.warning("no limit for %s", wrapper.policy.name)
return limiters
def get_rate_limiter(self, wrapper: MatchWrapper[FlowPolicy]) -> Optional[Limit]:
policy = wrapper.policy
tag_group = wrapper.tag_group
ttl = policy.time_unit_to_seconds()
if ttl < 60:
ttl = 60
limiter = self._cache.get(key=tag_group.name, ttl=ttl)
if limiter is None:
with self._cache_lock:
limiter = self._cache.get(tag_group.name)
if limiter is None:
limiter = self._to_limit(wrapper)
if limiter is not None:
self._cache.put(tag_group.name, limiter, ttl)
return limiter
def _to_limit(self, wrapper: MatchWrapper[FlowPolicy]) -> Optional[Limit]:
if StrUtils.equals_ignore_case(
wrapper.policy.policy_type, PolicyType.CLUSTER.value
):
logger.debug("create cluster limit for %s", wrapper.policy.name)
return self._get_cluster_rate_limiter(wrapper)
else:
logger.debug("create node limit %s", wrapper.policy.name)
return self._get_node_rate_limiter(wrapper)
def _get_node_rate_limiter(
self, wrapper: MatchWrapper[FlowPolicy]
) -> Optional[Limit]:
policy = wrapper.policy
if policy.calculate_alg is not None:
return self._get_dynamic_limiter(wrapper)
else:
return self._get_static_limiter(wrapper)
@staticmethod
def _get_cluster_rate_limiter(wrapper: MatchWrapper[FlowPolicy]):
policy = wrapper.policy
control_mode = policy.flow_control_mode
limiter = None
if (StrUtils.equals_ignore_case(control_mode, FlowControlMode.QUOTA.value)
or StrUtils.equals_ignore_case(control_mode, FlowControlMode.RESPONSE_QUOTA.value)):
from olc.limit.cluster.quota_cluster_limiter import QuotaClusterLimiter
logger.debug("create cluster quota limit for %s", wrapper.policy.name)
limiter = QuotaClusterLimiter(wrapper)
elif StrUtils.equals_ignore_case(control_mode, FlowControlMode.QPS.value):
logger.debug("create cluster qps limit for %s", wrapper.policy.name)
from olc.limit.cluster.qps_cluster_limiter import QpsClusterLimiter
limiter = QpsClusterLimiter(wrapper)
else:
logger.warning("unknown flow control mode %s for %s", control_mode, policy.name)
return limiter
@staticmethod
def _get_dynamic_limiter(
wrapper: MatchWrapper[FlowPolicy]
) -> Optional[Limit]:
policy = wrapper.policy
mode = policy.flow_control_mode
limiter = None
if StrUtils.equals_ignore_case(mode, FlowControlMode.CONCURRENT.value):
logger.debug("create dynamic concurrent limit for %s", wrapper.policy.name)
limiter = DynamicConcurrentLimiter(wrapper)
else:
logger.warning("dynamic limit not implemented for %s", wrapper.policy.name)
return limiter
@staticmethod
def _get_static_limiter(wrapper: MatchWrapper[FlowPolicy]) -> Optional[Limit]:
policy = wrapper.policy
mode = policy.flow_control_mode
limiter = None
if StrUtils.equals_ignore_case(mode, FlowControlMode.QPS.value):
logger.debug("create qps limit for %s", policy)
limiter = QpsLimiter(wrapper)
elif StrUtils.equals_ignore_case(
mode, FlowControlMode.QUOTA.value
) or StrUtils.equals_ignore_case(mode, FlowControlMode.RESPONSE_QUOTA.value):
logger.debug("create quota limit for %s", policy)
limiter = QuotaLimiter(wrapper)
elif StrUtils.equals_ignore_case(mode, FlowControlMode.CONCURRENT.value):
limiter = ConcurrentLimiter(wrapper)
logger.debug("create concurrent limit for %s", policy)
else:
logger.warning("unknown flow control mode %s for %s", mode, policy.name)
return limiter
def get_dyna_limiters(self) -> List[AbsDynamicLimiter]:
result = []
for value in self._cache.get_all_values():
if isinstance(value, AbsDynamicLimiter):
result.append(value)
return result
def subscribe_clean(self, sender, **kwargs):
logger.info("clean limit factory cache")
self._cache.clear()