OLC 扩展开发指南

本文档指导开发者如何为 OLC 添加新的扩展。OLC 提供了四个主要扩展点，每个扩展点遵循相同的模式：定义抽象 → 实现类 → 注册到 Manager/Factory。

扩展点总览

扩展点	抽象基类	注册位置	适用场景
框架适配器	BaseHTTPMiddleware	app.add_middleware()	支持新的 Web 框架
限流器	Limit	OlcRateLimiterFactory	新的限流算法或策略
算法	FlowAbstractAlgorithm	OlcAlgManager	新的自适应算法
配置源	Stub	OlcConfigPluginManager	新的配置获取方式

选择决策树：

• 需要拦截 HTTP 请求？→ 框架适配器扩展
• 需要新的限流策略？→ 限流器扩展
• 需要动态调整限流阈值？→ 算法扩展
• 需要从新来源获取规则？→ 配置源扩展

---

框架适配器扩展

适配器负责：拦截请求 → 提取标签 → 调用 OLC.process() → 处理 block → 调用 OLC.clean()

步骤

1. 创建 olc/adapters/<framework>/ 目录
2. 实现 adapter.py：参考 OlcFastAPIAdapter
3. 实现 adapter_config.py：定义配置类
4. 在 pyproject.toml 添加 optional-dependencies
5. 编写测试

示例：Flask 适配器

olc/adapters/flask/adapter_config.py：

from typing import Callable, Dict, Optional, Union, Awaitable

from flask import Request, Response


TagExtractor = Callable[[Request], Union[Dict[str, str], Awaitable[Dict[str, str]]]]

BlockResponseBuilder = Callable[[Request, str], Union[Response, Awaitable[Response]]]


class OlcFlaskAdapterConfig:
    def __init__(
        self,
        tag_extractor: Optional[TagExtractor] = None,
        block_msg: str = "Request blocked by rate limiter",
        block_response_builder: Optional[BlockResponseBuilder] = None,
        token_number: int = 1,
    ):
        self.tag_extractor = tag_extractor
        self.block_msg = block_msg
        self.block_response_builder = block_response_builder
        self.token_number = token_number

olc/adapters/flask/adapter.py：

import logging
from typing import List, Optional

from flask import Request, Response, jsonify

from olc.adapters.flask.adapter_config import OlcFlaskAdapterConfig
from olc.bean.dimension import Method, URL
from olc.bean.olc_control_request import OlcControlRequest
from olc.bean.tag_group import TagGroup
from olc.config.properties_loader import OlcConfigManager
from olc.control.context.context import OlcContext
from olc.olc import OLC

logger = logging.getLogger(__name__)


class OlcFlaskAdapter:
    def __init__(
        self,
        app,
        config: Optional[OlcFlaskAdapterConfig] = None,
        groups: Optional[List[TagGroup]] = None,
    ):
        self.app = app
        self.config = config or OlcFlaskAdapterConfig()
        self.groups = groups
        app.before_request(self._before_request)
        app.after_request(self._after_request)

    def _extract_tags(self, request: Request) -> dict:
        if self.config.tag_extractor is None:
            return {
                f"{URL}": request.path,
                f"{Method}": request.method,
            }
        return self.config.tag_extractor(request)

    def _before_request(self):
        from flask import request as flask_request

        if not OlcConfigManager.get_instance().get_sdk_config().switch:
            return None

        try:
            tags = self._extract_tags(flask_request)
            olc_request = OlcControlRequest(
                tags=tags,
                token_number=self.config.token_number,
            )
            context = OLC.get_instance().process(olc_request, self.groups)
            flask_request.olc_context = context

            if context.result.block:
                block_msg = context.result.block_msg or self.config.block_msg
                if self.config.block_response_builder is not None:
                    return self.config.block_response_builder(flask_request, block_msg)
                return jsonify({
                    "error": "Too Many Requests",
                    "message": block_msg,
                    "path": flask_request.path,
                }), 429
        except Exception as e:
            logger.error(f"OLC process error: {e}", exc_info=True)

        return None

    def _after_request(self, response: Response):
        from flask import request as flask_request

        context = getattr(flask_request, "olc_context", None)
        if context is not None:
            try:
                OLC.get_instance().clean(context)
            except Exception as e:
                logger.error(f"OLC cleanup error: {e}", exc_info=True)
        return response

pyproject.toml 新增：

[project.optional-dependencies]
flask = [
    "flask>=2.0.0",
]

---

限流器扩展

限流器负责：实现 Limit 抽象接口的 5 个方法。

步骤

1. 在 olc/limit/node/ 或 olc/limit/cluster/ 下创建新限流器类
2. 继承 Limit 抽象类，实现所有抽象方法
3. 如需新的 FlowControlMode，在 FlowControlMode 枚举中添加
4. 在 OlcRateLimiterFactory 对应方法中注册新分支
5. 编写测试

示例：滑动窗口限流器

olc/limit/node/static/sliding_window_limiter.py：

import threading
import time
from collections import deque

from olc.bean.match_wrapper import MatchWrapper
from olc.bean.olc_config_rule import FlowPolicy
from olc.limit.limiter import Limit


class SlidingWindowLimiter(Limit):
    def __init__(self, wrapper: MatchWrapper[FlowPolicy]):
        self._wrapper = wrapper
        self._window_seconds = wrapper.policy.time_unit_to_seconds()
        self._rate_limit = wrapper.policy.rate_limit
        self._lock = threading.Lock()
        self._timestamps: deque = deque()

    def get_match_wrapper(self) -> MatchWrapper[FlowPolicy]:
        return self._wrapper

    def _cleanup(self, now: float):
        cutoff = now - self._window_seconds
        while self._timestamps and self._timestamps[0] <= cutoff:
            self._timestamps.popleft()

    def pre_check(self, token_number: int) -> bool:
        now = time.time()
        with self._lock:
            self._cleanup(now)
            return len(self._timestamps) + token_number <= self._rate_limit

    def try_acquire(self, token_number: int) -> bool:
        now = time.time()
        with self._lock:
            self._cleanup(now)
            if len(self._timestamps) + token_number <= self._rate_limit:
                for _ in range(token_number):
                    self._timestamps.append(now)
                return True
            return False

    def roll_back(self, token_number: int):
        with self._lock:
            for _ in range(min(token_number, len(self._timestamps))):
                self._timestamps.pop()

    def dec_token(self, token_number: int):
        pass

注册到工厂 — 修改 OlcRateLimiterFactory._get_static_limiter：

def _get_static_limiter(self, wrapper: MatchWrapper[FlowPolicy]) -> Optional[Limit]:
    policy = wrapper.policy
    mode = policy.flow_control_mode
    limiter = None
    if StrUtils.equals_ignore_case(mode, FlowControlMode.QPS.value):
        limiter = QpsLimiter(wrapper)
    elif StrUtils.equals_ignore_case(
        mode, FlowControlMode.QUOTA.value
    ) or StrUtils.equals_ignore_case(mode, FlowControlMode.RESPONSE_QUOTA.value):
        limiter = QuotaLimiter(wrapper)
    elif StrUtils.equals_ignore_case(mode, FlowControlMode.CONCURRENT.value):
        limiter = ConcurrentLimiter(wrapper)
    elif StrUtils.equals_ignore_case(mode, FlowControlMode.SLIDING_WINDOW.value):
        limiter = SlidingWindowLimiter(wrapper)
    else:
        logger.warning(f"unknown flow control mode {mode} for {policy.name}")
    return limiter

添加 FlowControlMode 枚举值 — 修改 olc/bean/olc_config_rule.py：

class FlowControlMode(Enum):
    QPS = "qps"
    QUOTA = "quota"
    RESPONSE_QUOTA = "response_quota"
    CONCURRENT = "concurrent"
    SLIDING_WINDOW = "sliding_window"

---

算法扩展

算法负责：实现 FlowAbstractAlgorithm.make_decision 方法，根据资源使用率动态调整限流阈值。

步骤

1. 在 olc/alg/flow/ 下创建新算法类
2. 继承 FlowAbstractAlgorithm，定义参数（ParaName），实现 make_decision
3. 在 OlcAlgManager.__init__ 中注册
4. 编写测试

示例：AIMD 算法

AIMD (Additive Increase Multiplicative Decrease) 是一种经典的拥塞控制算法：正常时线性增加限流阈值，过载时乘性减少。

olc/alg/flow/aimd_algorithm.py：

import logging
from typing import Any

from olc.alg.alg_context import AlgContext
from olc.alg.algorithm import T, ParaName
from olc.alg.flow.flow_alg import (
    FlowAbstractAlgorithm,
    MIN_WAL,
    MAX_WAL,
    DELTA,
    CURRENT_USAGE,
)
from olc.event.sny_event import Event

logger = logging.getLogger(__name__)

ADDITIVE_STEP = "additiveStep"
MULTIPLICATIVE_FACTOR = "multiplicativeFactor"
OVERLOAD_THRESHOLD = "overloadThreshold"


class AimdAlgorithm(FlowAbstractAlgorithm):
    NAME = "AIMD"

    def __init__(self, auto: bool = False):
        super().__init__(auto)
        self.para_names[MIN_WAL] = ParaName(MIN_WAL, 0.5)
        self.para_names[MAX_WAL] = ParaName(MAX_WAL, 2.0)
        self.para_names[DELTA] = ParaName(DELTA, 5)

        self.para_names[ADDITIVE_STEP] = ParaName(ADDITIVE_STEP, 10)
        self.para_names[MULTIPLICATIVE_FACTOR] = ParaName(
            MULTIPLICATIVE_FACTOR, 0.5
        )
        self.para_names[OVERLOAD_THRESHOLD] = ParaName(
            OVERLOAD_THRESHOLD, 0.8
        )

        self.para_names[CURRENT_USAGE] = ParaName(
            CURRENT_USAGE, "localresource.memory", True
        )

        self._states = {}
        Event.get_instance().OLC_CLEAN_RULE.connect(self._clean_states)

    def make_decision(self, alg_context: AlgContext[T]) -> T:
        previous = alg_context.previous
        init_result = alg_context.init_result
        group_name = alg_context.group.name if alg_context.group else "default"

        current_usage = alg_context.get_float_value_by_dynamic(CURRENT_USAGE)
        current_usage = max(0.0, min(1.0, current_usage))

        additive_step = alg_context.get_float_value_by_param(
            ADDITIVE_STEP, self.para_names[ADDITIVE_STEP].default_value
        )
        multiplicative_factor = alg_context.get_float_value_by_param(
            MULTIPLICATIVE_FACTOR,
            self.para_names[MULTIPLICATIVE_FACTOR].default_value,
        )
        overload_threshold = alg_context.get_float_value_by_param(
            OVERLOAD_THRESHOLD,
            self.para_names[OVERLOAD_THRESHOLD].default_value,
        )

        min_wal = alg_context.get_float_value_by_param(
            MIN_WAL, self.para_names[MIN_WAL].default_value
        )
        max_wal = alg_context.get_float_value_by_param(
            MAX_WAL, self.para_names[MAX_WAL].default_value
        )

        if current_usage > overload_threshold:
            new_threshold = int(previous * multiplicative_factor)
        else:
            new_threshold = previous + int(additive_step)

        new_threshold = max(
            int(init_result * min_wal),
            min(int(init_result * max_wal), new_threshold),
        )

        return new_threshold

    def check(self, params: dict[str, Any]) -> bool:
        if not super().check(params):
            return False
        if not self.valid_min_max_param(MIN_WAL, MAX_WAL, params):
            return False
        factor = params.get(MULTIPLICATIVE_FACTOR)
        if factor is not None and (float(factor) <= 0 or float(factor) >= 1):
            logger.warning(
                f"multiplicativeFactor must be in (0, 1), got {factor}"
            )
            return False
        return True

    def _clean_states(self, sender, **kwargs):
        self._states.clear()

注册到算法管理器 — 修改 OlcAlgManager.__init__：

def __init__(self):
    self._flow_alg_map: Dict[str, Optional[FlowAbstractAlgorithm]] = {}
    self.add_flow_alg(PidAlgorithm.NAME, PidAlgorithm())
    self.add_flow_alg(AimdAlgorithm.NAME, AimdAlgorithm())

---

配置源扩展

配置源负责：实现 Stub 抽象类的 get_olc_config_rules 方法，返回 JSON 字符串。

步骤

1. 在 olc/stub/ 下创建新配置源类
2. 继承 Stub，实现 get_olc_config_rules
3. 在 OlcConfigPluginManager 中注册
4. 在 overload-config.properties 中配置 olc.sdk.config.stub=<name>
5. 编写测试

示例：HTTP 远程配置源

olc/stub/http_config_stub.py：

import logging
from typing import Optional

import requests

from olc.stub.config_stub_manager import Stub
from olc.utils.str_utils import StrUtils

logger = logging.getLogger(__name__)


class HttpConfigStub(Stub):
    NAME = "http"

    def __init__(self):
        from olc.config.properties_loader import OlcConfigManager

        sdk_config = OlcConfigManager.get_instance().get_sdk_config()
        self._url = getattr(sdk_config, "config_http_url", "")

    def closeStub(self):
        pass

    def get_stub_name(self) -> str:
        return HttpConfigStub.NAME

    def get_olc_config_rules(self) -> Optional[str]:
        if StrUtils.is_empty(self._url):
            logger.error("HTTP config URL is empty")
            return None

        try:
            response = requests.get(self._url, timeout=5)
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            logger.error(f"Failed to fetch config from {self._url}: {e}")
            return None

注册到插件管理器 — 修改 OlcConfigPluginManager.get_instance：

@classmethod
def get_instance(cls) -> "OlcConfigPluginManager":
    if cls._instance is None:
        with cls._lock:
            if cls._instance is None:
                cls._instance = OlcConfigPluginManager()
                cls._instance.register_stub(JsonFileConfigStub.NAME, JsonFileConfigStub)
                cls._instance.register_stub(HttpConfigStub.NAME, HttpConfigStub)
    return cls._instance

配置使用 — 在 overload-config.properties 中：

olc.sdk.config.stub=http
olc.sdk.config.http.url=http://config-server:8080/api/olc/rules