From bfe8bfe2da59b29cfce977233bd89eb87a8fb187 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 13 Jul 2026 01:11:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(feature-flag):=20Redis=20Feature=20Flag=20?= =?UTF-8?q?=E7=81=B0=E5=BA=A6=E5=8F=91=E5=B8=83=E5=9F=BA=E7=A1=80=E8=AE=BE?= =?UTF-8?q?=E6=96=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Redis Feature Flag 灰度发布基础设施：白名单+百分比切流+全局开关，热更新，内部管理API --- apps/api/app/api/router.py | 5 + apps/api/app/api/routes/feature_flags.py | 195 ++++++++ apps/worker/video_processing/__init__.py | 3 + .../render_engine_resolver.py | 204 +++++++++ apps/worker/worker_app/tasks/compose_video.py | 15 +- packages/adapters/redis/__init__.py | 17 +- packages/adapters/redis/feature_flag_store.py | 259 +++++++++++ tests/unit/test_feature_flag.py | 424 ++++++++++++++++++ 8 files changed, 1117 insertions(+), 5 deletions(-) create mode 100755 apps/api/app/api/routes/feature_flags.py create mode 100755 apps/worker/video_processing/render_engine_resolver.py mode change 100644 => 100755 packages/adapters/redis/__init__.py create mode 100755 packages/adapters/redis/feature_flag_store.py create mode 100755 tests/unit/test_feature_flag.py diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index 875833cda..ec8902fe1 100755 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -8,6 +8,7 @@ from app.api.routes.dashboard import router as dashboard_router from app.api.routes.duplication import router as duplication_router from app.api.routes.edit_plans import router as edit_plans_router from app.api.routes.edit_templates import router as edit_templates_router +from app.api.routes.feature_flags import router as feature_flags_router from app.api.routes.generated_videos import router as generated_videos_router from app.api.routes.generation_tasks import router as generation_tasks_router from app.api.routes.health import router as health_check_router @@ -151,3 +152,7 @@ api_router.include_router( prefix="/tts", tags=["TTS"], ) +api_router.include_router( + feature_flags_router, + tags=["Internal"], +) diff --git a/apps/api/app/api/routes/feature_flags.py b/apps/api/app/api/routes/feature_flags.py new file mode 100755 index 000000000..287ebf745 --- /dev/null +++ b/apps/api/app/api/routes/feature_flags.py @@ -0,0 +1,195 @@ +"""Feature Flag 内部管理接口。 + +通过内部 API Key 鉴权，支持查看和修改 Feature Flag 配置。 +主要用于灰度发布期间的动态开关控制。 + +API: + GET /api/v1/internal/feature-flags - 列出所有 flag + GET /api/v1/internal/feature-flags/{name} - 查看单个 flag + PUT /api/v1/internal/feature-flags/{name} - 设置 flag 配置 + DELETE /api/v1/internal/feature-flags/{name} - 删除 flag + +鉴权：X-API-Key header，走内部 API Key 验证 +""" + +from __future__ import annotations + +import logging +from typing import Optional + +from app.api.routes.auth import _verify_internal_api_key +from app.config import settings +from fastapi import APIRouter, Depends, HTTPException, Query, status +from pydantic import BaseModel, Field + +from packages.adapters.redis.feature_flag_store import ( + FEATURE_FLAG_REDIS_PREFIX, + FeatureFlagConfig, + RedisFeatureFlagStore, +) + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/internal/feature-flags", tags=["Internal"]) + +# 允许管理的 flag 白名单（防止误操作其他系统 flag） +ALLOWED_FLAGS = { + "render_engine", +} + + +def _get_feature_flag_store() -> RedisFeatureFlagStore: + """获取 Feature Flag 存储实例。""" + return RedisFeatureFlagStore(redis_url=settings.REDIS_URL) + + +class FeatureFlagUpdateRequest(BaseModel): + """Feature Flag 更新请求体。""" + + enabled: bool = Field(..., description="是否启用") + percentage: int = Field(0, ge=0, le=100, description="灰度百分比 (0-100)") + whitelist: list[str] = Field(default_factory=list, description="白名单列表（如 user_id）") + + +class FeatureFlagResponse(BaseModel): + """Feature Flag 响应。""" + + name: str + enabled: bool + percentage: int + whitelist: list[str] + + @classmethod + def from_config(cls, config: FeatureFlagConfig) -> "FeatureFlagResponse": + return cls( + name=config.name, + enabled=config.enabled, + percentage=config.percentage, + whitelist=sorted(config.whitelist), + ) + + +class FeatureFlagCheckResponse(BaseModel): + """Flag 激活检查响应。""" + + name: str + active: bool + identifier: Optional[str] = None + + +def _validate_flag_name(name: str) -> None: + """校验 flag 名称是否在允许列表中。""" + if name not in ALLOWED_FLAGS: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Unsupported flag: {name}. Allowed: {sorted(ALLOWED_FLAGS)}", + ) + + +@router.get("", response_model=list[FeatureFlagResponse]) +async def list_feature_flags( + _: bool = Depends(_verify_internal_api_key), + store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), +): + """列出所有 Feature Flag。""" + try: + flags = store.list_all() + # 同时返回预定义的 flag（即使未设置也显示默认值） + result = [] + for name in sorted(ALLOWED_FLAGS): + config = flags.get(name) or FeatureFlagConfig(name=name, enabled=False) + result.append(FeatureFlagResponse.from_config(config)) + # 加上已存在但不在白名单中的 flag（只读展示） + for name, config in flags.items(): + if name not in ALLOWED_FLAGS: + result.append(FeatureFlagResponse.from_config(config)) + return sorted(result, key=lambda x: x.name) + except Exception as exc: + logger.error("Failed to list feature flags: %s", exc) + raise HTTPException(status_code=500, detail=f"Failed to list flags: {exc}") + + +@router.get("/{name}", response_model=FeatureFlagResponse) +async def get_feature_flag( + name: str, + _: bool = Depends(_verify_internal_api_key), + store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), +): + """获取单个 Feature Flag 配置。""" + try: + config = store.get(name) + return FeatureFlagResponse.from_config(config) + except Exception as exc: + logger.error("Failed to get feature flag %s: %s", name, exc) + raise HTTPException(status_code=500, detail=f"Failed to get flag: {exc}") + + +@router.get("/{name}/check", response_model=FeatureFlagCheckResponse) +async def check_feature_flag( + name: str, + identifier: Optional[str] = Query(None, description="标识符，如 user_id"), + _: bool = Depends(_verify_internal_api_key), + store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), +): + """检查某个标识符是否命中 Feature Flag。""" + try: + active = store.is_active(name, identifier=identifier) + return FeatureFlagCheckResponse(name=name, active=active, identifier=identifier) + except Exception as exc: + logger.error("Failed to check feature flag %s: %s", name, exc) + raise HTTPException(status_code=500, detail=f"Failed to check flag: {exc}") + + +@router.put("/{name}", response_model=FeatureFlagResponse) +async def update_feature_flag( + name: str, + request: FeatureFlagUpdateRequest, + _: bool = Depends(_verify_internal_api_key), + store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), +): + """更新 Feature Flag 配置。 + + 只允许修改 ALLOWED_FLAGS 列表中的 flag。 + """ + _validate_flag_name(name) + + try: + config = FeatureFlagConfig( + name=name, + enabled=request.enabled, + percentage=request.percentage, + whitelist=set(request.whitelist), + ) + store.set(config) + logger.info( + "Feature flag updated: name=%s enabled=%s percentage=%d whitelist=%d", + name, + config.enabled, + config.percentage, + len(config.whitelist), + ) + return FeatureFlagResponse.from_config(config) + except Exception as exc: + logger.error("Failed to update feature flag %s: %s", name, exc) + raise HTTPException(status_code=500, detail=f"Failed to update flag: {exc}") + + +@router.delete("/{name}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_feature_flag( + name: str, + _: bool = Depends(_verify_internal_api_key), + store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), +): + """删除 Feature Flag。 + + 只允许删除 ALLOWED_FLAGS 列表中的 flag。 + """ + _validate_flag_name(name) + + try: + deleted = store.delete(name) + logger.info("Feature flag deleted: name=%s deleted=%s", name, deleted) + return None + except Exception as exc: + logger.error("Failed to delete feature flag %s: %s", name, exc) + raise HTTPException(status_code=500, detail=f"Failed to delete flag: {exc}") diff --git a/apps/worker/video_processing/__init__.py b/apps/worker/video_processing/__init__.py index af9b6b47d..7a9bf8a2a 100755 --- a/apps/worker/video_processing/__init__.py +++ b/apps/worker/video_processing/__init__.py @@ -6,6 +6,7 @@ from . import dedup_helpers, ffmpeg_utils, oss_helpers from .processor import VideoProcessor, VideoResult from .render_adapter import RenderAdapter, RenderAdapterResult +from .render_engine_resolver import RenderEngineResolver, get_render_engine_resolver from .unified_render_service import RenderResult, UnifiedRenderService __all__ = [ @@ -18,4 +19,6 @@ __all__ = [ "RenderResult", "RenderAdapter", "RenderAdapterResult", + "RenderEngineResolver", + "get_render_engine_resolver", ] diff --git a/apps/worker/video_processing/render_engine_resolver.py b/apps/worker/video_processing/render_engine_resolver.py new file mode 100755 index 000000000..b39b63049 --- /dev/null +++ b/apps/worker/video_processing/render_engine_resolver.py @@ -0,0 +1,204 @@ +"""渲染引擎 Feature Flag 解析器。 + +封装渲染引擎选择逻辑，支持： +- 环境变量作为默认值（RENDER_ENGINE=legacy/unified） +- Redis Feature Flag 运行时覆盖（白名单 + 百分比 + 全局开关） +- 定时刷新，支持热更新不重启 worker + +使用方式： + resolver = RenderEngineResolver(redis_url="redis://...", default_engine="legacy") + engine = resolver.get_engine(user_id="user123") + # engine: "legacy" 或 "unified" +""" + +from __future__ import annotations + +import logging +import threading +from typing import Optional + +from packages.adapters.redis.feature_flag_store import ( + FeatureFlagConfig, + FeatureFlagStore, + InMemoryFeatureFlagStore, + RedisFeatureFlagStore, +) + +logger = logging.getLogger(__name__) + +# Feature Flag 名称常量 +FLAG_RENDER_ENGINE = "render_engine" + +# 引擎常量 +ENGINE_LEGACY = "legacy" +ENGINE_UNIFIED = "unified" +VALID_ENGINES = {ENGINE_LEGACY, ENGINE_UNIFIED} + + +class RenderEngineResolver: + """渲染引擎选择器。 + + 判定逻辑（从高到低）： + 1. Redis flag 白名单匹配 → unified + 2. Redis flag 百分比命中 → unified + 3. Redis flag 全局开启（100%）→ unified + 4. 环境变量默认值 → legacy / unified + + 当 Redis 不可用时，自动降级到环境变量默认值，不影响业务。 + """ + + def __init__( + self, + default_engine: str = ENGINE_LEGACY, + redis_url: Optional[str] = None, + refresh_interval: float = 30.0, + store: Optional[FeatureFlagStore] = None, + ) -> None: + """ + Args: + default_engine: 环境变量默认的引擎名（legacy / unified） + redis_url: Redis 连接 URL，传 None 时使用内存实现（测试用） + refresh_interval: Redis flag 配置刷新间隔（秒） + store: 直接传入 store 实例（测试用，优先级高于 redis_url） + """ + self._default_engine = default_engine.lower() if default_engine else ENGINE_LEGACY + if self._default_engine not in VALID_ENGINES: + logger.warning( + "Invalid default engine '%s', fallback to '%s'", + self._default_engine, + ENGINE_LEGACY, + ) + self._default_engine = ENGINE_LEGACY + + if store is not None: + self._store = store + elif redis_url: + self._store = RedisFeatureFlagStore(redis_url=redis_url) + else: + self._store = InMemoryFeatureFlagStore() + logger.info("No Redis configured, using in-memory feature flag store") + + self._refresh_interval = refresh_interval + self._lock = threading.Lock() + self._cached_config: Optional[FeatureFlagConfig] = None + self._last_refresh: float = 0.0 + + def _maybe_refresh(self) -> None: + """惰性刷新配置，超过刷新间隔时从存储重新读取。""" + import time + + now = time.time() + if now - self._last_refresh < self._refresh_interval: + return + + try: + config = self._store.get(FLAG_RENDER_ENGINE) + with self._lock: + self._cached_config = config + self._last_refresh = now + except Exception as exc: + logger.warning("Failed to refresh render engine flag: %s", exc) + # 刷新失败时保留旧缓存，不中断业务 + if self._cached_config is None: + # 首次就读失败，设一个默认值 + with self._lock: + self._cached_config = FeatureFlagConfig(name=FLAG_RENDER_ENGINE) + self._last_refresh = now + + def _get_config(self) -> FeatureFlagConfig: + """获取当前 flag 配置（带缓存）。""" + if self._cached_config is None: + self._maybe_refresh() + else: + self._maybe_refresh() + return self._cached_config or FeatureFlagConfig(name=FLAG_RENDER_ENGINE) + + def get_engine(self, user_id: Optional[str] = None) -> str: + """获取当前应该使用的渲染引擎。 + + Args: + user_id: 用户ID，用于白名单匹配和百分比哈希。 + 传 None 时只看全局开关。 + + Returns: + "legacy" 或 "unified" + """ + config = self._get_config() + + # 全局关闭 → 用默认值 + if not config.enabled: + return self._default_engine + + # 白名单匹配 / 百分比命中 → unified + if config.is_active(user_id): + return ENGINE_UNIFIED + + # 未命中灰度 → 用默认值 + return self._default_engine + + def should_use_unified(self, user_id: Optional[str] = None) -> bool: + """便捷方法：是否应该使用统一渲染引擎。""" + return self.get_engine(user_id) == ENGINE_UNIFIED + + def force_refresh(self) -> None: + """强制立即刷新配置（用于管理接口修改后立即生效）。""" + self._last_refresh = 0.0 + if isinstance(self._store, RedisFeatureFlagStore): + self._store.invalidate_cache(FLAG_RENDER_ENGINE) + self._maybe_refresh() + + def get_config_snapshot(self) -> dict: + """获取当前配置快照（用于管理接口展示）。""" + config = self._get_config() + return { + "flag_name": FLAG_RENDER_ENGINE, + "default_engine": self._default_engine, + "enabled": config.enabled, + "percentage": config.percentage, + "whitelist": sorted(config.whitelist), + "refresh_interval": self._refresh_interval, + "last_refresh": self._last_refresh, + } + + def set_flag(self, config: FeatureFlagConfig) -> None: + """设置 flag 配置（管理接口用）。""" + config.name = FLAG_RENDER_ENGINE + self._store.set(config) + self.force_refresh() + + +# 全局单例 +_resolver: Optional[RenderEngineResolver] = None +_resolver_lock = threading.Lock() + + +def get_render_engine_resolver() -> RenderEngineResolver: + """获取全局单例（基于 worker 配置）。""" + global _resolver + if _resolver is not None: + return _resolver + + with _resolver_lock: + if _resolver is not None: + return _resolver + + try: + from worker_app.core.config import get_settings + + settings = get_settings() + redis_url = getattr(settings, "redis_url", None) or getattr(settings, "broker_url", None) + default = getattr(settings, "render_engine", ENGINE_LEGACY) + _resolver = RenderEngineResolver( + default_engine=default, + redis_url=redis_url, + ) + logger.info( + "RenderEngineResolver initialized: default=%s, redis=%s", + default, + bool(redis_url), + ) + except Exception as exc: + logger.warning("Failed to init RenderEngineResolver from settings: %s", exc) + _resolver = RenderEngineResolver(default_engine=ENGINE_LEGACY) + + return _resolver diff --git a/apps/worker/worker_app/tasks/compose_video.py b/apps/worker/worker_app/tasks/compose_video.py index f4fd89620..b02d4f686 100755 --- a/apps/worker/worker_app/tasks/compose_video.py +++ b/apps/worker/worker_app/tasks/compose_video.py @@ -61,10 +61,12 @@ def compose_video(self, job_id: str, **kwargs): return {"status": "error", "message": "Missing plan_id"} # 判断使用哪个渲染引擎 - from worker_app.core.config import get_settings as get_worker_settings + # 优先级：Redis Feature Flag（白名单 > 百分比） > 环境变量默认 + from video_processing.render_engine_resolver import get_render_engine_resolver - worker_settings = get_worker_settings() - engine = (worker_settings.render_engine or "legacy").lower() + resolver = get_render_engine_resolver() + user_id = job.created_by_user_id or None + engine = resolver.get_engine(user_id=user_id) if engine == "unified": return _compose_with_unified_engine(self, job_service, job, plan_id, db) @@ -207,7 +209,12 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di } job_service.complete_job(job_id, result=result_data) - logger.info("视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs", job_id, plan_id, result.duration) + logger.info( + "视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs", + job_id, + plan_id, + result.duration, + ) return {"status": "completed", "job_id": job_id, "result": result_data} diff --git a/packages/adapters/redis/__init__.py b/packages/adapters/redis/__init__.py old mode 100644 new mode 100755 index ff0c2ac01..c250dc9c0 --- a/packages/adapters/redis/__init__.py +++ b/packages/adapters/redis/__init__.py @@ -1,3 +1,9 @@ +from packages.adapters.redis.feature_flag_store import ( + FeatureFlagConfig, + FeatureFlagStore, + InMemoryFeatureFlagStore, + RedisFeatureFlagStore, +) from packages.adapters.redis.session_store import ( NoopSessionStore, RedisConfig, @@ -5,4 +11,13 @@ from packages.adapters.redis.session_store import ( get_session_store, ) -__all__ = ["NoopSessionStore", "RedisConfig", "SessionStore", "get_session_store"] +__all__ = [ + "FeatureFlagConfig", + "FeatureFlagStore", + "InMemoryFeatureFlagStore", + "NoopSessionStore", + "RedisConfig", + "RedisFeatureFlagStore", + "SessionStore", + "get_session_store", +] diff --git a/packages/adapters/redis/feature_flag_store.py b/packages/adapters/redis/feature_flag_store.py new file mode 100755 index 000000000..7e29f9cf5 --- /dev/null +++ b/packages/adapters/redis/feature_flag_store.py @@ -0,0 +1,259 @@ +"""Feature Flag 存储实现。 + +支持两种后端： +- RedisFeatureFlagStore：生产环境使用，支持多实例共享、热更新 +- InMemoryFeatureFlagStore：测试/开发环境使用，纯内存 + +支持的 Flag 类型： +- 全局开关（enabled: bool） +- 白名单（whitelist: Set[str]，如 user_id 列表） +- 百分比切流（percentage: 0-100，基于标识符哈希取模） + +判定优先级：白名单 > 百分比 > 全局开关 +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import threading +import time +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import Optional, Set + +logger = logging.getLogger(__name__) + +# Redis key 前缀 +FEATURE_FLAG_REDIS_PREFIX = "feature_flag:" + + +@dataclass +class FeatureFlagConfig: + """单个 Feature Flag 的配置。""" + + name: str + enabled: bool = False + percentage: int = 0 # 0-100 + whitelist: Set[str] = field(default_factory=set) + + def to_dict(self) -> dict: + return { + "name": self.name, + "enabled": self.enabled, + "percentage": self.percentage, + "whitelist": sorted(self.whitelist), + } + + @classmethod + def from_dict(cls, data: dict) -> "FeatureFlagConfig": + return cls( + name=data["name"], + enabled=bool(data.get("enabled", False)), + percentage=int(data.get("percentage", 0)), + whitelist=set(data.get("whitelist", [])), + ) + + def is_active(self, identifier: Optional[str] = None) -> bool: + """判断当前 flag 是否激活。 + + 判定优先级： + 1. 全局关闭 → False + 2. 白名单匹配 → True + 3. 百分比命中 → True + 4. 其他 → False + + Args: + identifier: 用于白名单匹配和百分比哈希的标识符（如 user_id）。 + 传 None 时只看全局开关 + 百分比（百分比用随机值）。 + """ + if not self.enabled: + return False + + # 白名单：精确匹配 + if identifier and identifier in self.whitelist: + return True + + # 百分比：0 直接 False，100 直接 True + if self.percentage <= 0: + # 没有白名单且百分比为0 → 未启用 + return False + if self.percentage >= 100: + return True + + # 基于 identifier 做哈希取模，确保同一用户始终落在同一侧 + if identifier: + hash_val = int( + hashlib.md5(f"{self.name}:{identifier}".encode("utf-8")).hexdigest(), 16 # nosec B324 + ) # nosec B324 - 用于哈希取模做百分比切流，非安全用途 + return (hash_val % 100) < self.percentage + + # 无 identifier 且百分比在 0-100 之间 → 按比例随机（不保证一致性） + import random + + return random.randint(0, 99) < self.percentage + + +class FeatureFlagStore(ABC): + """Feature Flag 存储抽象接口。""" + + @abstractmethod + def get(self, name: str) -> FeatureFlagConfig: + """获取指定 flag 的配置，不存在则返回默认配置（关闭状态）。""" + ... + + @abstractmethod + def set(self, config: FeatureFlagConfig) -> None: + """设置 flag 配置。""" + ... + + @abstractmethod + def delete(self, name: str) -> bool: + """删除 flag，返回是否成功删除。""" + ... + + @abstractmethod + def list_all(self) -> dict[str, FeatureFlagConfig]: + """列出所有 flag。""" + ... + + def is_active(self, name: str, identifier: Optional[str] = None) -> bool: + """便捷方法：判断 flag 是否激活。""" + return self.get(name).is_active(identifier) + + +class InMemoryFeatureFlagStore(FeatureFlagStore): + """内存实现，用于测试和本地开发。""" + + def __init__(self) -> None: + self._flags: dict[str, FeatureFlagConfig] = {} + self._lock = threading.Lock() + + def get(self, name: str) -> FeatureFlagConfig: + with self._lock: + return self._flags.get(name, FeatureFlagConfig(name=name, enabled=False)) + + def set(self, config: FeatureFlagConfig) -> None: + with self._lock: + self._flags[config.name] = config + + def delete(self, name: str) -> bool: + with self._lock: + if name in self._flags: + del self._flags[name] + return True + return False + + def list_all(self) -> dict[str, FeatureFlagConfig]: + with self._lock: + return dict(self._flags) + + +class RedisFeatureFlagStore(FeatureFlagStore): + """Redis 实现，支持多实例共享配置。 + + 每个 flag 存在一个独立的 Redis hash key 中： + Key: feature_flag:{name} + Fields: enabled, percentage, whitelist(JSON array) + """ + + def __init__(self, redis_url: str, key_prefix: str = FEATURE_FLAG_REDIS_PREFIX) -> None: + import redis as redis_lib + + self._redis = redis_lib.from_url(redis_url, decode_responses=True) + self._key_prefix = key_prefix + # 本地缓存 + TTL，减少 Redis 调用 + self._cache: dict[str, tuple[FeatureFlagConfig, float]] = {} + self._cache_ttl = 5.0 # 秒，默认5秒本地缓存 + self._lock = threading.Lock() + + def _redis_key(self, name: str) -> str: + return f"{self._key_prefix}{name}" + + def _parse_whitelist(self, raw: Optional[str]) -> Set[str]: + if not raw: + return set() + try: + data = json.loads(raw) + return set(data) if isinstance(data, list) else set() + except (json.JSONDecodeError, TypeError): + return set() + + def get(self, name: str) -> FeatureFlagConfig: + now = time.time() + + # 先查本地缓存 + with self._lock: + cached = self._cache.get(name) + if cached and now - cached[1] < self._cache_ttl: + return cached[0] + + # 从 Redis 读取 + try: + key = self._redis_key(name) + data = self._redis.hgetall(key) + if not data: + config = FeatureFlagConfig(name=name, enabled=False) + else: + config = FeatureFlagConfig( + name=name, + enabled=(data.get("enabled", "0") in ("1", "true", "True")), + percentage=int(data.get("percentage", 0)), + whitelist=self._parse_whitelist(data.get("whitelist")), + ) + + # 写入本地缓存 + with self._lock: + self._cache[name] = (config, now) + + return config + except Exception as exc: + logger.warning("Failed to get feature flag %s from Redis: %s", name, exc) + # Redis 不可用时返回默认值（关闭），不影响业务 + return FeatureFlagConfig(name=name, enabled=False) + + def set(self, config: FeatureFlagConfig) -> None: + key = self._redis_key(config.name) + self._redis.hset( + key, + mapping={ + "enabled": "1" if config.enabled else "0", + "percentage": str(config.percentage), + "whitelist": json.dumps(sorted(config.whitelist), ensure_ascii=False), + }, + ) + # 失效本地缓存 + with self._lock: + self._cache.pop(config.name, None) + + def delete(self, name: str) -> bool: + key = self._redis_key(name) + result = self._redis.delete(key) + with self._lock: + self._cache.pop(name, None) + return bool(result) + + def list_all(self) -> dict[str, FeatureFlagConfig]: + pattern = f"{self._key_prefix}*" + result: dict[str, FeatureFlagConfig] = {} + try: + cursor = 0 + while True: + cursor, keys = self._redis.scan(cursor=cursor, match=pattern, count=100) + for key in keys: + name = key[len(self._key_prefix) :] + result[name] = self.get(name) + if cursor == 0: + break + except Exception as exc: + logger.warning("Failed to list feature flags from Redis: %s", exc) + return result + + def invalidate_cache(self, name: Optional[str] = None) -> None: + """手动失效本地缓存。""" + with self._lock: + if name: + self._cache.pop(name, None) + else: + self._cache.clear() diff --git a/tests/unit/test_feature_flag.py b/tests/unit/test_feature_flag.py new file mode 100755 index 000000000..81cddd67a --- /dev/null +++ b/tests/unit/test_feature_flag.py @@ -0,0 +1,424 @@ +"""Feature Flag 单元测试。 + +测试 FeatureFlagConfig、InMemoryFeatureFlagStore、RenderEngineResolver 的核心逻辑。 +""" + +from __future__ import annotations + +import time +from unittest.mock import MagicMock, patch + +import pytest + +from packages.adapters.redis.feature_flag_store import ( + FeatureFlagConfig, + InMemoryFeatureFlagStore, +) + +# ── FeatureFlagConfig 测试 ────────────────────────────────────────────────── + + +class TestFeatureFlagConfig: + """FeatureFlagConfig 核心逻辑测试。""" + + def test_default_disabled(self): + """默认配置为关闭状态。""" + config = FeatureFlagConfig(name="test_flag") + assert config.enabled is False + assert config.percentage == 0 + assert config.whitelist == set() + assert config.is_active() is False + assert config.is_active("user1") is False + + def test_global_enabled_100_percent(self): + """100% + 启用 = 全部命中。""" + config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=100) + assert config.is_active() is True + assert config.is_active("user1") is True + assert config.is_active("any_user") is True + + def test_global_enabled_0_percent_no_whitelist(self): + """启用但 0% 且无白名单 = 不命中。""" + config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=0) + assert config.is_active() is False + assert config.is_active("user1") is False + + def test_whitelist_takes_priority(self): + """白名单优先级高于百分比。""" + config = FeatureFlagConfig( + name="test_flag", + enabled=True, + percentage=0, + whitelist={"user1", "user2"}, + ) + assert config.is_active("user1") is True + assert config.is_active("user2") is True + assert config.is_active("user3") is False + + def test_whitelist_with_percentage(self): + """白名单用户即使百分比为0也命中，非白名单按百分比。""" + config = FeatureFlagConfig( + name="test_flag", + enabled=True, + percentage=100, # 100% 所有人命中 + whitelist={"user1"}, + ) + assert config.is_active("user1") is True + assert config.is_active("user999") is True # 100% 命中 + + def test_percentage_consistency_same_user(self): + """同一用户多次调用结果一致（哈希确定性）。""" + config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=50) + results = [config.is_active("user_fixed") for _ in range(100)] + assert all(r == results[0] for r in results) + + def test_percentage_different_users_distributed(self): + """不同用户分布大致符合百分比（统计检验，宽松阈值）。""" + config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=50) + active_count = sum(1 for i in range(1000) if config.is_active(f"user_{i}")) + # 50% 上下浮动 10% 都算合理 + assert 400 <= active_count <= 600, f"Expected ~500, got {active_count}" + + def test_percentage_boundary_0_and_100(self): + """0% 和 100% 的边界情况。""" + config_0 = FeatureFlagConfig(name="test", enabled=True, percentage=0) + config_100 = FeatureFlagConfig(name="test", enabled=True, percentage=100) + + for i in range(100): + assert config_0.is_active(f"user_{i}") is False + assert config_100.is_active(f"user_{i}") is True + + def test_disabled_ignores_all_other_settings(self): + """关闭时忽略白名单和百分比。""" + config = FeatureFlagConfig( + name="test_flag", + enabled=False, + percentage=100, + whitelist={"user1"}, + ) + assert config.is_active("user1") is False + assert config.is_active() is False + + def test_none_identifier_with_percentage(self): + """无 identifier 时按随机比例（0% 和 100% 是确定的）。""" + config_0 = FeatureFlagConfig(name="test", enabled=True, percentage=0) + config_100 = FeatureFlagConfig(name="test", enabled=True, percentage=100) + assert config_0.is_active(None) is False + assert config_100.is_active(None) is True + + def test_to_dict_and_from_dict(self): + """序列化和反序列化对称。""" + original = FeatureFlagConfig( + name="test_flag", + enabled=True, + percentage=30, + whitelist={"user_a", "user_b", "user_c"}, + ) + data = original.to_dict() + restored = FeatureFlagConfig.from_dict(data) + assert restored.name == original.name + assert restored.enabled == original.enabled + assert restored.percentage == original.percentage + assert restored.whitelist == original.whitelist + + def test_from_dict_with_missing_fields(self): + """from_dict 缺失字段时使用默认值。""" + config = FeatureFlagConfig.from_dict({"name": "minimal"}) + assert config.name == "minimal" + assert config.enabled is False + assert config.percentage == 0 + assert config.whitelist == set() + + +# ── InMemoryFeatureFlagStore 测试 ─────────────────────────────────────────── + + +class TestInMemoryFeatureFlagStore: + """内存存储实现测试。""" + + def test_get_nonexistent_returns_default(self): + """获取不存在的 flag 返回默认配置（关闭）。""" + store = InMemoryFeatureFlagStore() + config = store.get("nonexistent") + assert config.name == "nonexistent" + assert config.enabled is False + + def test_set_and_get(self): + """设置后可以读取。""" + store = InMemoryFeatureFlagStore() + config = FeatureFlagConfig(name="test", enabled=True, percentage=50, whitelist={"u1"}) + store.set(config) + + got = store.get("test") + assert got.enabled is True + assert got.percentage == 50 + assert got.whitelist == {"u1"} + + def test_delete_existing(self): + """删除存在的 flag 返回 True。""" + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="test", enabled=True)) + assert store.delete("test") is True + assert store.get("test").enabled is False + + def test_delete_nonexistent(self): + """删除不存在的 flag 返回 False。""" + store = InMemoryFeatureFlagStore() + assert store.delete("nonexistent") is False + + def test_list_all(self): + """列出所有 flag。""" + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="flag_a", enabled=True)) + store.set(FeatureFlagConfig(name="flag_b", percentage=10)) + + all_flags = store.list_all() + assert len(all_flags) == 2 + assert "flag_a" in all_flags + assert "flag_b" in all_flags + assert all_flags["flag_a"].enabled is True + + def test_is_active_convenience(self): + """is_active 便捷方法。""" + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="render", enabled=True, percentage=0, whitelist={"vip_user"})) + assert store.is_active("render", "vip_user") is True + assert store.is_active("render", "normal_user") is False + assert store.is_active("nonexistent") is False + + +# ── RenderEngineResolver 测试 ─────────────────────────────────────────────── + + +class TestRenderEngineResolver: + """渲染引擎选择器测试。""" + + def test_default_legacy_when_flag_disabled(self): + """flag 关闭时使用默认引擎（legacy）。""" + store = InMemoryFeatureFlagStore() + resolver = self._make_resolver(store=store, default="legacy") + assert resolver.get_engine() == "legacy" + assert resolver.get_engine("user1") == "legacy" + + def test_default_unified_when_flag_disabled(self): + """flag 关闭但默认值是 unified 时返回 unified。""" + store = InMemoryFeatureFlagStore() + resolver = self._make_resolver(store=store, default="unified") + assert resolver.get_engine() == "unified" + + def test_whitelist_user_uses_unified(self): + """白名单用户走新引擎。""" + store = InMemoryFeatureFlagStore() + store.set( + FeatureFlagConfig( + name="render_engine", + enabled=True, + percentage=0, + whitelist={"beta_tester"}, + ) + ) + resolver = self._make_resolver(store=store, default="legacy") + assert resolver.get_engine("beta_tester") == "unified" + assert resolver.get_engine("normal_user") == "legacy" + + def test_100_percent_all_unified(self): + """100% 时所有用户走新引擎。""" + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100)) + resolver = self._make_resolver(store=store, default="legacy") + for i in range(50): + assert resolver.get_engine(f"user_{i}") == "unified" + + def test_invalid_default_engine_fallback(self): + """无效默认值回退到 legacy。""" + store = InMemoryFeatureFlagStore() + resolver = self._make_resolver(store=store, default="invalid_value") + assert resolver.get_engine() == "legacy" + + def test_should_use_unified_helper(self): + """should_use_unified 便捷方法。""" + store = InMemoryFeatureFlagStore() + store.set( + FeatureFlagConfig( + name="render_engine", + enabled=True, + percentage=0, + whitelist={"user_a"}, + ) + ) + resolver = self._make_resolver(store=store) + assert resolver.should_use_unified("user_a") is True + assert resolver.should_use_unified("user_b") is False + + def test_config_snapshot(self): + """配置快照。""" + store = InMemoryFeatureFlagStore() + store.set( + FeatureFlagConfig( + name="render_engine", + enabled=True, + percentage=30, + whitelist={"u1", "u2"}, + ) + ) + resolver = self._make_resolver(store=store) + snapshot = resolver.get_config_snapshot() + assert snapshot["flag_name"] == "render_engine" + assert snapshot["enabled"] is True + assert snapshot["percentage"] == 30 + assert snapshot["whitelist"] == ["u1", "u2"] + + def test_set_flag_updates_config(self): + """通过 set_flag 修改后立即生效。""" + store = InMemoryFeatureFlagStore() + resolver = self._make_resolver(store=store, default="legacy") + + # 初始：关闭 + assert resolver.get_engine("user1") == "legacy" + + # 开启 100% + resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100)) + assert resolver.get_engine("user1") == "unified" + + # 关闭 + resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=False)) + assert resolver.get_engine("user1") == "legacy" + + def test_force_refresh(self): + """强制刷新不报错。""" + store = InMemoryFeatureFlagStore() + resolver = self._make_resolver(store=store) + resolver.force_refresh() # 不抛异常即可 + + def test_does_not_affect_in_flight_tasks(self): + """ + 热更新不影响在途任务验证： + 任务开始时确定引擎，中途配置变更不改变当前任务的引擎选择。 + （这是通过"每次调用 get_engine 时读取当前配置"来保证的， + 任务开始时调用一次拿到结果，之后不再变化） + """ + store = InMemoryFeatureFlagStore() + store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100)) + resolver = self._make_resolver(store=store, default="legacy") + + # 模拟任务开始时获取引擎 + engine_at_start = resolver.get_engine("user1") + assert engine_at_start == "unified" + + # 任务进行中关闭 flag + store.set(FeatureFlagConfig(name="render_engine", enabled=False)) + resolver.force_refresh() + + # 在途任务持有的 engine_at_start 仍然是 unified（不随配置变化） + assert engine_at_start == "unified" + # 新任务会拿到 legacy + assert resolver.get_engine("user1") == "legacy" + + # ── 辅助方法 ── + + @staticmethod + def _make_resolver(store=None, default="legacy"): + from apps.worker.video_processing.render_engine_resolver import ( + RenderEngineResolver, + ) + + return RenderEngineResolver( + default_engine=default, + store=store or InMemoryFeatureFlagStore(), + refresh_interval=9999, # 测试时禁用自动刷新 + ) + + +# ── RedisFeatureFlagStore 降级测试（无 Redis 环境） ─────────────────────── + + +class TestRedisStoreDegradation: + """Redis 不可用时的降级行为测试。""" + + def test_get_returns_default_when_redis_unavailable(self): + """Redis 连接失败时返回默认关闭配置，不抛异常。""" + import importlib + + from packages.adapters.redis import feature_flag_store as ff_module + + # 模拟 redis 模块不存在的场景不好做，这里直接测试异常捕获逻辑 + store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore) + store._redis = MagicMock() + store._redis.hgetall.side_effect = ConnectionError("Redis down") + store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX + store._cache = {} + store._cache_ttl = 5.0 + import threading + + store._lock = threading.Lock() + + config = store.get("render_engine") + assert config.enabled is False + assert config.name == "render_engine" + + def test_list_all_returns_empty_on_redis_error(self): + """Redis 错误时 list_all 返回空字典。""" + import importlib + + from packages.adapters.redis import feature_flag_store as ff_module + + store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore) + store._redis = MagicMock() + store._redis.scan.side_effect = ConnectionError("Redis down") + store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX + store._cache = {} + store._cache_ttl = 5.0 + import threading + + store._lock = threading.Lock() + + result = store.list_all() + assert result == {} + + +class TestRedisStoreListAll: + """RedisFeatureFlagStore list_all 正常路径测试。""" + + def _make_store(self): + from packages.adapters.redis import feature_flag_store as ff_module + + store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore) + store._redis = MagicMock() + store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX + store._cache = {} + store._cache_ttl = 5.0 + import threading + + store._lock = threading.Lock() + return store + + def test_list_all_scan_with_match_param(self): + """list_all 调用 redis.scan 时使用正确的 match 参数名。""" + store = self._make_store() + prefix = store._key_prefix + + # 模拟 scan 返回 2 个 key，分 2 次游标 + store._redis.scan.side_effect = [ + (10, [f"{prefix}render_engine", f"{prefix}other_flag"]), + (0, []), + ] + # 模拟 hgetall 返回配置 + store._redis.hgetall.return_value = { + b"enabled": b"true", + b"percentage": b"50", + b"whitelist": b'["user1","user2"]', + } + + result = store.list_all() + + # 验证 scan 被调用了 2 次（游标遍历） + assert store._redis.scan.call_count == 2 + # 验证参数名是 match（不是 match_pattern） + first_call_kwargs = store._redis.scan.call_args_list[0][1] + assert "match" in first_call_kwargs + assert "match_pattern" not in first_call_kwargs + assert first_call_kwargs["match"] == f"{prefix}*" + # 验证返回了 2 个 flag + assert len(result) == 2 + assert "render_engine" in result + assert "other_flag" in result