OLC 模块详解
本文档详细说明 OLC 各模块的内部实现和模块间协作关系。
bean — 数据模型层
数据模型层定义了 OLC 中所有核心数据结构，是各模块间通信的基础。
OlcControlRequest
请求控制模型，由适配器构建并传入 OLC.process()。
class OlcControlRequest:
tags: Dict[str, str] # 请求标签，如 {"URL": "/api/chat", "Method": "POST"}
token_number: int # 请求消耗的令牌数，默认 1
tags的 key 来自Dimension中注册的维度名（URL、Method、Model 等）token_number用于配额限流场景，表示本次请求消耗的配额量
OlcControlResult
控制结果模型，由责任链各 Handler 填充，最终由适配器读取。
class OlcControlResult:
block: bool = False # 是否拦截
block_msg: str = "" # 拦截消息
block_rule: MatchWrapper = None # 触发拦截的规则
tokens_to_deduct: int = 0 # 响应配额模式下需调整的令牌数（负数为补充）
规则模型体系
OlcConfigRule # 顶层配置
├── domain: str # 域名
└── rules: List[OlcRule] # 规则列表
└── OlcRule # 单条规则
├── group: Group # 匹配分组
├── flow: FlowPolicy # 流控策略（可选）
└── admission: AdmissionPolicy # 准入策略（可选）
Group 定义了请求匹配规则：
class Group:
priority: int # 优先级，默认 500
enabled: bool # 是否启用
name: str # 分组名称（全局唯一）
type: str # "normal" 根据示所有标签匹配的请求 或 "global" 根据请求标签匹配任意标签的请求。 会根据标签拆分成多个组，每个组包含一个标签。
tags: List[OlcMultiTag] # 匹配标签列表
class OlcMultiTag:
tag: str # 维度名 对应 `Dimension` 中注册的维度名
match: str # 匹配方式："equal" / "regex" / "default"
values: List[str] # 匹配值列表
share: Optional[bool] # 是否共享（share=True 的标签不创建子组）
FlowPolicy 定义了流控策略：
class FlowPolicy(BasePolicy):
policy_type: str # "NODE" 或 "CLUSTER"
flow_control_mode: str # "qps" / "quota" / "response_quota" / "concurrent"
time_unit: str # "second" / "minute" / "hour" / "day"
time_interval: int # 时间间隔
rate_limit: int # 限流阈值
burst_limit: int # 突发上限
max_wait_time_ms: int # 最大等待时间
calculate_alg: AlgInstance # 计算算法（可选，用于动态限流）
assign_alg: AlgInstance # 分配算法（集群模式可选）
AdmissionPolicy 定义了准入策略：
class AdmissionPolicy(BasePolicy):
type: str # "allow" / "deny" / "dynamic"
calculate_alg: AlgInstance # 动态准入算法（可选）
TagGroup / SubGroup / OlcTag
标签分组模型用于运行时匹配和子组生成。
TagGroup 是规则 Group 在运行时的表示：
class TagGroup:
domain: str
name: str # 分组名（全局唯一标识）
priority: int
enabled: bool
tags: List[OlcTag] # 运行时标签
OlcTag 与配置中的 OlcMultiTag 对应，但增加了 value 和 share 属性：
class OlcTag:
match: str # 匹配方式："equal" / "regex" / "default"
tag: str # 维度名
value: str # 匹配值
share: bool # 是否共享（share=True 的标签不创建子组）
share 机制：当 share=True 时，该维度的标签不会创建子组（即不会按请求的实际值拆分），适用于需要跨请求共享限流配额的场景。share=False 的标签会根据请求中的实际值创建 SubGroup，实现更细粒度的限流。
SubGroup 继承自 TagGroup，根据请求的实际标签值动态生成：
class SubGroup(TagGroup):
parent: TagGroup # 父组引用
# reset_tags: 将 share=False 的标签替换为请求中的实际值
# reset_name: 重新生成名称，格式为 "parentName$$tag1@value1&&tag2@value2"
MatchWrapper
策略与分组的绑定包装，泛型类：
class MatchWrapper(Generic[T]):
policy: T # FlowPolicy 或 AdmissionPolicy
tag_group: TagGroup # 匹配到的分组
Dimension
维度注册表，单例模式，定义了可用的标签维度：
# 内置维度
URL = "URL" # 请求路径
Method = "Method" # HTTP 方法
Model = "Model" # 模型名（AI 场景）
IP = "IP" # IP 地址
ServiceName = "ServiceName"
InterfaceName = "InterfaceName"
Tenant = "tenant"
TenantLevel = "TenantLevel"
Priority = "Priority"
UUID = "UUID"
Resource = "Resource"
ENV = "ENV"
UserPriority = "UserPriority"
config — 配置管理
PropertiesManager
配置文件管理器，支持热加载、解密和变更回调。
核心功能：
- 加载：从 Properties 文件读取键值对，映射到嵌套 dataclass
- 热加载：后台线程定期检查文件修改时间，变更时两阶段更新（先在临时对象上更新，成功后原子替换）
- 解密：支持
ENC(...)格式的加密值，通过Encryptor解密 - 回调：配置变更时触发
on_change回调
class PropertiesManager:
def __init__(self, file_path, interval=5, encryptor=None, on_change=None): ...
def get(self) -> OlcConfig: ... # 获取配置快照（线程安全）
def stop(self): ... # 停止热加载线程
OlcConfigManager
全局配置管理单例，封装 PropertiesManager。
class OlcConfigManager:
@classmethod
def get_instance(cls) -> "OlcConfigManager": ...
def get_sdk_config(self) -> SdkConfig: ...
def get_redis_config(self) -> RedisConfig: ...
配置文件路径通过环境变量 OLC_CONFIG_PATH 指定，默认加载 overload-config.properties。
配置数据类
@dataclass
class SdkConfig:
domain: str = "" # 域名
switch: bool = True # 全局开关
dynamic_switch: bool = False # 动态开关
collector_active: str = "" # 激活的采集器（分号分隔）
collector_capacity: int = 10 # 采集器存储容量
collector_sample: int = 10 # 采集器采样数
collector_interval: int = 1 # 采集间隔（秒）
config_stub: str = "jsonfile" # 配置源类型
policy_update_period: int = 6 # 规则刷新间隔（秒）
@dataclass
class RedisConfig:
mode: str = "alone" # Redis 模式（alone/cluster/sentinel）
node_list: str = "" # 节点列表
password: str = None # 密码
timeout: int = 2000 # 超时（毫秒）
max_connections: int = 64 # 最大连接数
service_name: str = "" # Sentinel 服务名
retry_times: int = 3 # 重试次数
retry_interval: int = 1000 # 重试间隔（毫秒）
socket_timeout: int = 2000 # Socket 超时（毫秒）
control — 控制逻辑
OlcContext
请求上下文，贯穿整个责任链，承载请求、结果和中间状态。
class OlcContext(Context):
request: OlcControlRequest # 原始请求
result: OlcControlResult # 控制结果
match_wrapper: List[MatchWrapper[BasePolicy]] # 匹配到的策略列表
statistics: dict[TagGroup, IStatistic] # 统计对象映射
stop: bool = False # 短路标志
request_id: str = "" # 请求 ID
start_time: float = 0 # 开始时间
is_cancelled: bool = False # 是否取消
OlcHandlerChain
责任链的静态构建和入口。
class OlcHandlerChain:
chain = DefaultBoxHandlerChain()
chain.add_last_box(GroupMatcherHandler())
chain.add_last_box(AdmissionHandler())
chain.add_last_box(FlowHandler())
chain.add_last_box(StatisticHandler())
调用方式：
- 前置处理：
OlcHandlerChain.chain.in_coming(context, groups) - 后置处理：
OlcHandlerChain.chain.out_coming(context)
GroupMatcherHandler
标签匹配 Handler，责任链第一个节点。
in_coming 逻辑：
- 调用
OlcMatcherProvider.get_group_matcher().match_tag_groups(request)匹配请求对应的 TagGroup 列表 - 为每个匹配到的 TagGroup 创建统计对象
- 将 TagGroup 列表传递给下一个 Handler
匹配过程（GroupMatcher.match_tag_groups）：
- 先查缓存（
SafeTTLCache，以 request 为 key） - 缓存未命中时，遍历
OlcRuleManager中的所有 TagGroup - 对每个 TagGroup，使用
OperationMatcher逐个匹配标签 - 如果 TagGroup 有
share=False的标签，创建SubGroup - 匹配结果写入缓存
AdmissionHandler
准入控制 Handler。
in_coming 逻辑：
- 如果
context.stop为 True，跳过 - 调用
OlcMatcherProvider.get_admission_matcher().match(request, groups)获取匹配的准入策略 - 对每个匹配的策略，从
AdmissionFactory获取控制器 - 调用
controller.get_result()判断是否放行 - 任一控制器拒绝则设置
context.stop = True和context.result.block = True
FlowHandler
流控 Handler，核心限流逻辑。
in_coming 逻辑：
- 如果
context.stop为 True，跳过核心逻辑 - 调用
OlcMatcherProvider.get_flow_matcher().match(request, groups)获取匹配的流控策略 - 调用
FlowProcessor.process(context, match)执行限流判定 - 判定不通过则设置
context.result.block = True和context.stop = True
FlowProcessor.process 逻辑：
- 从
OlcRateLimiterFactory获取所有限流器实例 - 依次对每个限流器执行
pre_check→try_acquire - 任一限流器获取令牌失败，回滚已成功的限流器（
roll_back） - 全部成功返回 True
out_coming 逻辑：
- 如果请求未被拦截且
tokens_to_deduct != 0，处理响应配额模式的令牌调整
StatisticHandler
统计 Handler，管理并发计数。
in_coming：对所有匹配的统计对象调用 inc_thread_num()
out_coming：对所有匹配的统计对象调用 dec_thread_num()
Matcher 体系
OlcMatcherProvider（静态提供者）
├── GroupMatcher # 标签分组匹配
├── FlowMatcher # 流控策略匹配
└── AdmissionMatcher # 准入策略匹配
OperationMatcherFactory（操作匹配器工厂）
├── EqualMatcher # 精确匹配
├── DefaultMatcher # 默认匹配（通配 *）
└── RegexMatcher # 正则匹配
FlowMatcher 和 AdmissionMatcher 继承自 AbstractPolicyMatcher，通过 OlcRuleManager 查询策略并与 TagGroup 绑定为 MatchWrapper。
limit — 限流器
Limit 抽象接口
class Limit(ABC):
def get_match_wrapper(self) -> MatchWrapper[FlowPolicy]: ...
def try_acquire(self, token_number: int) -> bool: ... # 尝试获取令牌
def pre_check(self, token_number: int) -> bool: ... # 预检查
def roll_back(self, token_number: int): ... # 回滚令牌
def dec_token(self, token_number: int): ... # 强制扣减/补充令牌
单机限流器
| 限流器 | 流控模式 | 实现原理 |
|---|---|---|
QpsLimiter |
qps | 基于 QpsTokenBucket 令牌桶 |
ConcurrentLimiter |
concurrent | 基于 IStatistic 并发计数 |
QuotaLimiter |
quota / response_quota | 基于 QuotaBucket 配额桶 |
动态限流器
DynamicConcurrentLimiter：并发限流器的动态版本，限流阈值由自适应算法动态调整。
集群限流器
| 限流器 | 流控模式 | 实现原理 |
|---|---|---|
QpsClusterLimiter |
qps | Redis + Lua 脚本（qps_token_bucket.lua） |
QuotaClusterLimiter |
quota / response_quota | Redis + Lua 脚本（quota_token_bucket.lua） |
集群限流器通过 OlcClusterLimiterRefresh 定期从 Redis 获取令牌补充本地缓存。
OlcRateLimiterFactory
限流器工厂，根据策略配置创建对应的限流器实例，内部使用 SafeTTLCache 缓存。
创建逻辑：
FlowPolicy
├── policy_type == "CLUSTER"
│ ├── flowControlMode == "qps" → QpsClusterLimiter
│ └── flowControlMode == "quota"/"response_quota" → QuotaClusterLimiter
└── policy_type == "NODE"（默认）
├── calculate_alg 存在 → DynamicConcurrentLimiter
└── calculate_alg 为空
├── flowControlMode == "qps" → QpsLimiter
├── flowControlMode == "quota"/"response_quota" → QuotaLimiter
└── flowControlMode == "concurrent" → ConcurrentLimiter
alg — 算法模块
Algorithm 抽象基类
class Algorithm(ABC, Generic[T]):
alg_type: AlgType # FLOWDECISION / ADMISSION
auto: bool # 是否自动执行
para_names: dict[str, ParaName] # 参数定义
def make_decision(self, alg_context: AlgContext[T]) -> T: ...
def check(self, params: dict[str, Any]) -> bool: ...
def get_dynamic_param_key(self) -> list[str]: ...
ParaName 定义算法参数：
class ParaName:
key: str # 参数名
default_value: Any # 默认值
dynamic: bool = False # 是否为动态参数（从采集器获取）
FlowAbstractAlgorithm
流控算法抽象基类，预定义了三个通用参数：
minWal：流控阈值最小比例（默认 0.5）maxWal：流控阈值最大比例（默认 2.0）delta：流控预制最小调整大小（默认 2.0）
PidAlgorithm
PID 自适应算法，根据资源使用率动态调整限流阈值。
参数：
| 参数 | 默认值 | 说明 |
|---|---|---|
| Kp | 0.5 | 比例系数 |
| Ki | 0 | 积分系数 |
| kd | 0.5 | 微分系数 |
| minResource | 0.5 | 资源使用率下限 |
| maxResource | 0.5 | 资源使用率上限 |
| currentUsage | "localresource.memory" | 动态参数，当前资源使用率 |
决策逻辑：
- 获取当前资源使用率
current_usage - 若低于
minResource，方向为增加（direction=1） - 若高于
maxResource，方向为减少（direction=-1） - 在区间内则保持不变
- 计算 PID 输出，基于初始值调整限流阈值
- 将结果限制在
[init_result * minWal, init_result * maxWal]范围内
OlcAlgManager
算法注册表单例，管理所有流控算法实例。
class OlcAlgManager:
def get_flow_alg(self, name: str) -> Optional[FlowAbstractAlgorithm]: ...
def add_flow_alg(self, name: str, alg: FlowAbstractAlgorithm): ...
OlcAlgRefreshManager
定时刷新动态限流器的算法参数，调用算法的 make_decision 更新限流阈值。
admission — 准入控制
准入控制器
| 控制器 | 类型 | 行为 |
|---|---|---|
AllowController |
allow | 始终放行 |
DenyController |
deny | 始终拒绝 |
DynamicController |
dynamic | 动态判定（待实现） |
AdmissionFactory
准入控制器工厂单例，根据 AdmissionPolicy.type 创建对应的控制器实例，内部使用 SafeTTLCache 缓存。
rule — 规则管理
OlcRuleConvert
规则转换器，负责将 JSON 配置转换为内部规则缓存。
核心流程：
- 从
Stub配置源获取 JSON 字符串 - 解析为
OlcConfigRule - 校验 domain 和规则
- 将
Group转换为TagGroup（支持 normal 和 global 两种类型） - 校验算法实例
- 更新
OlcRuleManager的规则缓存 - 发送
OLC_CLEAN_RULE事件清理各组件缓存
normal 类型：标签间做笛卡尔积，生成多个 TagGroup global 类型：每个标签值独立生成一个 TagGroup
OlcRuleManager / OlcRuleCache
规则缓存管理器，存储 TagGroup、FlowPolicy、AdmissionPolicy 的映射关系。
class OlcRuleCache:
tag_groups: Dict[str, TagGroup] # 分组名 → TagGroup
bind_flow_relation: Dict[str, FlowPolicy] # 分组名 → 流控策略
bind_admission_relation: Dict[str, AdmissionPolicy] # 分组名 → 准入策略
sub_groups: SafeTTLCache[SubGroup] # 子组缓存
collector — 资源采集
AbsCollector
采集器抽象基类，定义了采集器的生命周期和接口。
class AbsCollector(ABC):
def start(self): ... # 启动采集（注册定时任务）
def stop(self): ... # 停止采集
def do_measure(self): ... # 执行一次采集
def get_name(self) -> str: ... # 采集器名称
def get_storage(self) -> List[Measurable]: ... # 获取存储
def get_average_property(self, property_name, sample_size) -> Optional[float]: ...
采集器通过 OlcScheduleManager 注册定时任务，按 collector_interval 间隔执行采集。
VLLMResourceCollector
vLLM 资源采集器，采集 vLLM 服务的资源指标。
OlcCollectorManager
采集器管理单例，负责注册和激活采集器。
class OlcCollectorManager:
def add_collector(self, name, collector): ... # 注册采集器类型
def get_collector(self, name): ... # 获取激活的采集器实例
def valid(self, param): ... # 校验动态参数
激活的采集器通过 olc.sdk.collector.active 配置项指定（分号分隔）。
stub — 配置源插件
Stub 抽象
class Stub(ABC):
def closeStub(self): ...
def get_stub_name(self) -> str: ...
def get_olc_config_rules(self) -> Optional[str]: ... # 返回 JSON 字符串
JsonFileConfigStub
从本地 JSON 文件读取规则配置，文件路径为 {OLC_CONFIG_PATH}/olc.json。
OlcConfigPluginManager
配置源插件管理单例，支持注册和查询配置源类型。
class OlcConfigPluginManager:
def register_stub(self, name, stub): ... # 注册插件
def get_stub_plugin(self, name): ... # 获取插件类
其他辅助模块
cache — SafeTTLCache
线程安全的 TTL 缓存，基于 cacheout 库封装，增加了线程锁保护。
event — 事件总线
基于 blinker 库的信号机制，目前只有一个事件 OLC_CLEAN_RULE。
encryptor — 加密/解密
配置值加密/解密工具，支持 ENC(...) 格式的加密值。
utils — 工具函数
| 模块 | 功能 |
|---|---|
atomic_num |
原子数值操作 |
empty_checker |
空值检查 |
env_config |
环境变量读取 |
file_utils |
文件读取（含 Properties 解析） |
num_utils |
数值类型转换 |
path_utils |
路径处理（配置路径解析） |
str_utils |
字符串工具（类型转换、比较） |
time_utils |
时间工具 |
task_schedule — 定时任务
基于 schedule 库的定时任务管理器，使用线程池执行任务。
remote — HTTP 客户端
基于 requests 库的 HTTP 客户端封装。
resource/redis/lua_scripts — Lua 脚本
集群限流使用的 Redis Lua 脚本：
qps_token_bucket.lua：QPS 令牌桶操作quota_token_bucket.lua：配额令牌桶操作quota_token_bucket_rollback.lua：配额令牌桶回滚操作