import logging
from abc import ABC, abstractmethod
from typing import Optional
from olc.alg.admission.admission_alg import AdmissionAbstractAlgorithm
from olc.alg.alg_context import AlgContextBuilder
from olc.alg.alg_manager import OlcAlgManager
from olc.bean.match_wrapper import MatchWrapper
from olc.bean.olc_config_rule import AdmissionPolicy, AlgInstance
from olc.collector.collector_manager import OlcCollectorManager
from olc.exception.olc_dynamic_exception import OLCDynamicException
log = logging.getLogger(__name__)
class AbsAdmissionController(ABC):
def __init__(self):
super().__init__()
@abstractmethod
def refresh_status(self):
pass
"""
判断是否可以通过
return 返回是否可以通过 true 拒绝 false 通过
"""
@abstractmethod
def get_result(self) -> bool:
pass
class AllowController(AbsAdmissionController):
def refresh_status(self):
pass
def get_result(self) -> bool:
return False
class DenyController(AbsAdmissionController):
def refresh_status(self):
pass
def get_result(self) -> bool:
return True
class DynamicController(AbsAdmissionController):
def __init__(self, matchWrapper: MatchWrapper[AdmissionPolicy]):
super().__init__()
self._tag_group = matchWrapper.tag_group
self._policy = matchWrapper.policy
alg = matchWrapper.policy.calculate_alg
self._cal_ins: AlgInstance = alg
self._cal_alg: Optional[AdmissionAbstractAlgorithm] = (
OlcAlgManager.get_instance().get_admission_alg(alg.alg_name)
)
self._pre_status: bool = False
self._status: bool = False
def run_alg(self):
try:
if self._cal_alg is None:
log.warning(f"not found admission alg {self._cal_ins.alg_name}")
return False
instance_paras = self._cal_ins.alg_param.copy()
for key, value in instance_paras.items():
alg_param = self._cal_alg.para_names.get(key)
if alg_param is not None:
if alg_param.dynamic:
true_value = OlcCollectorManager.get_instance().get_dyn_key_true_value(value)
if true_value is None:
log.warning(f"get dynamic value err, dynamic key {alg_param} value is null")
raise OLCDynamicException(f"Dynamic parameter conversion error")
instance_paras[key] = true_value
builder = AlgContextBuilder[bool]().new_builder()
builder.set_group(self._tag_group)
builder.set_policy(self._policy)
builder.set_parameter(instance_paras)
builder.set_previous(self._pre_status)
builder.set_init_result(self._status)
return self._cal_alg.make_decision(builder.build())
except Exception as e:
log.warning(f"admission dynamic algorithm executed error. err :{repr(e)}")
return False
def refresh_status(self):
if self._cal_alg is None:
return
if not self._cal_alg.every_request:
alg = self.run_alg()
self.update_controller(alg)
def get_result(self) -> bool:
if self._cal_alg.every_request:
return self.run_alg()
else:
return self._status
def update_controller(self, status: bool):
self._pre_status = self._status
self._status = status
def __repr__(self):
return f"DynamicController(tag_group = {self._tag_group}, policy = {self._policy})"
