feature-flags

功能标志

通过功能标志系统控制功能发布，并实现渐进式推出。

何时使用此技能

在以下情况下使用此技能：

• 实施渐进式功能推出
• 启用主干开发
• 运行 A/B 测试和实验
• 管理功能生命周期
• 为生产环境实施紧急关闭开关

前提条件

• 应用程序代码访问权限
• 功能标志服务或可自托管的解决方案
• 对部署模式的基本了解

功能标志类型

类型	用途	示例
发布	控制功能可见性	新的结账流程
实验	A/B 测试	按钮颜色测试
运维	运行时配置	速率限制
权限	用户访问控制	高级功能
紧急关闭开关	紧急禁用	第三方集成

LaunchDarkly

SDK 设置（Node.js）

const LaunchDarkly = require('launchdarkly-node-server-sdk');

const client = LaunchDarkly.init(process.env.LAUNCHDARKLY_SDK_KEY);

await client.waitForInitialization();

// Evaluate flag
const user = {
  key: 'user-123',
  email: '[email protected]',
  custom: {
    plan: 'premium',
    company: 'acme'
  }
};

const showNewFeature = await client.variation('new-checkout', user, false);

if (showNewFeature) {
  // New feature code
} else {
  // Existing code
}

React SDK

import { withLDProvider, useFlags, useLDClient } from 'launchdarkly-react-client-sdk';

// Provider setup
export default withLDProvider({
  clientSideID: 'your-client-side-id',
  user: {
    key: 'user-123',
    email: '[email protected]'
  }
})(App);

// Using flags in component
function FeatureComponent() {
  const { newCheckout, experimentVariant } = useFlags();
  const ldClient = useLDClient();

  // Track events
  const handleClick = () => {
    ldClient.track('checkout-started');
  };

  if (newCheckout) {
    return <NewCheckout onClick={handleClick} />;
  }
  return <OldCheckout onClick={handleClick} />;
}

定向规则

# LaunchDarkly targeting configuration
flag: new-checkout
targeting:
  # Individual users
  targets:
    - variation: true
      values: ['user-123', 'user-456']
  
  # Rules
  rules:
    # Beta users
    - variation: true
      clauses:
        - attribute: email
          op: endsWith
          values: ['@company.com']
    
    # Premium plan
    - variation: true
      clauses:
        - attribute: plan
          op: in
          values: ['premium', 'enterprise']
    
    # Percentage rollout
    - variation: true
      rollout:
        variations:
          - variation: true
            weight: 20000  # 20%
          - variation: false
            weight: 80000  # 80%
  
  # Default
  fallthrough:
    variation: false

Unleash

服务端设置

# docker-compose.yml
version: '3.8'

services:
  unleash:
    image: unleashorg/unleash-server:latest
    ports:
      - "4242:4242"
    environment:
      - DATABASE_URL=postgres://postgres:password@db/unleash
      - DATABASE_SSL=false
    depends_on:
      - db

  db:
    image: postgres:15
    environment:
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=unleash
    volumes:
      - postgres-data:/var/lib/postgresql/data

volumes:
  postgres-data:

SDK 设置（Node.js）

const { initialize } = require('unleash-client');

const unleash = initialize({
  url: 'http://localhost:4242/api',
  appName: 'my-app',
  customHeaders: {
    Authorization: 'your-api-token'
  }
});

unleash.on('ready', () => {
  // Check feature
  const isEnabled = unleash.isEnabled('new-checkout');
  
  // With context
  const context = {
    userId: 'user-123',
    properties: {
      plan: 'premium'
    }
  };
  
  const isEnabledForUser = unleash.isEnabled('new-checkout', context);
  
  // Get variant
  const variant = unleash.getVariant('experiment-flag', context);
  console.log(variant.name); // 'control' or 'treatment'
});

激活策略

# Standard strategies
strategies:
  - name: default
    # On/off for everyone
    
  - name: userWithId
    parameters:
      userIds: 'user-1,user-2,user-3'
    
  - name: gradualRolloutUserId
    parameters:
      percentage: 25
      groupId: 'new-feature'
    
  - name: gradualRolloutRandom
    parameters:
      percentage: 50
    
  - name: flexibleRollout
    parameters:
      rollout: 30
      stickiness: userId
      groupId: 'checkout-exp'

自定义实现

基于数据库的标志

# models.py
from django.db import models

class FeatureFlag(models.Model):
    name = models.CharField(max_length=100, unique=True)
    enabled = models.BooleanField(default=False)
    rollout_percentage = models.IntegerField(default=0)
    allowed_users = models.JSONField(default=list)
    rules = models.JSONField(default=dict)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

# service.py
import hashlib

class FeatureFlagService:
    def __init__(self):
        self._cache = {}
    
    def is_enabled(self, flag_name, user_id=None, context=None):
        flag = self._get_flag(flag_name)
        
        if not flag or not flag.enabled:
            return False
        
        # Check user allowlist
        if user_id and user_id in flag.allowed_users:
            return True
        
        # Check rules
        if context and self._evaluate_rules(flag.rules, context):
            return True
        
        # Check percentage rollout
        if flag.rollout_percentage > 0 and user_id:
            return self._is_in_rollout(flag_name, user_id, flag.rollout_percentage)
        
        return flag.rollout_percentage == 100
    
    def _is_in_rollout(self, flag_name, user_id, percentage):
        hash_input = f"{flag_name}:{user_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        return (hash_value % 100) < percentage
    
    def _evaluate_rules(self, rules, context):
        for rule in rules.get('rules', []):
            if self._evaluate_rule(rule, context):
                return True
        return False

基于 Redis 的标志

import redis
import json

class RedisFeatureFlags:
    def __init__(self, redis_url):
        self.redis = redis.from_url(redis_url)
        self.prefix = 'feature_flag:'
    
    def set_flag(self, name, config):
        key = f"{self.prefix}{name}"
        self.redis.set(key, json.dumps(config))
    
    def is_enabled(self, name, user_id=None):
        key = f"{self.prefix}{name}"
        data = self.redis.get(key)
        
        if not data:
            return False
        
        config = json.loads(data)
        
        if not config.get('enabled', False):
            return False
        
        # User allowlist
        if user_id in config.get('users', []):
            return True
        
        # Percentage rollout
        percentage = config.get('percentage', 0)
        if percentage == 100:
            return True
        
        if percentage > 0 and user_id:
            return self._hash_user(name, user_id) < percentage
        
        return False
    
    def _hash_user(self, flag, user_id):
        import hashlib
        hash_input = f"{flag}:{user_id}"
        return int(hashlib.sha256(hash_input.encode()).hexdigest(), 16) % 100

使用功能标志进行测试

单元测试

// Jest mocking
jest.mock('launchdarkly-node-server-sdk', () => ({
  init: jest.fn(() => ({
    waitForInitialization: jest.fn().mockResolvedValue(undefined),
    variation: jest.fn()
  }))
}));

describe('Checkout', () => {
  it('shows new checkout when flag enabled', async () => {
    const ldClient = require('launchdarkly-node-server-sdk').init();
    ldClient.variation.mockResolvedValue(true);
    
    const result = await renderCheckout(user);
    expect(result).toContain('NewCheckout');
  });
  
  it('shows old checkout when flag disabled', async () => {
    const ldClient = require('launchdarkly-node-server-sdk').init();
    ldClient.variation.mockResolvedValue(false);
    
    const result = await renderCheckout(user);
    expect(result).toContain('OldCheckout');
  });
});

集成测试

# pytest fixtures
import pytest

@pytest.fixture
def feature_flags():
    """Provide controllable feature flags for testing."""
    flags = {}
    
    class TestFlags:
        def set(self, name, value):
            flags[name] = value
        
        def is_enabled(self, name, **kwargs):
            return flags.get(name, False)
    
    return TestFlags()

def test_new_checkout(feature_flags):
    feature_flags.set('new-checkout', True)
    
    response = client.get('/checkout')
    assert 'new-checkout-form' in response.content

监控与分析

标志使用情况跟踪

// Track flag evaluations
const flagMetrics = {
  evaluations: new Map(),
  
  track(flagName, variation, user) {
    const key = `${flagName}:${variation}`;
    const count = this.evaluations.get(key) || 0;
    this.evaluations.set(key, count + 1);
    
    // Send to analytics
    analytics.track('feature_flag_evaluated', {
      flag: flagName,
      variation: variation,
      userId: user.key
    });
  }
};

过时标志检测

from datetime import datetime, timedelta

def detect_stale_flags():
    """Find flags that haven't been evaluated recently."""
    stale_threshold = timedelta(days=30)
    now = datetime.utcnow()
    
    stale_flags = []
    for flag in FeatureFlag.objects.all():
        if flag.last_evaluated:
            age = now - flag.last_evaluated
            if age > stale_threshold:
                stale_flags.append({
                    'name': flag.name,
                    'last_evaluated': flag.last_evaluated,
                    'age_days': age.days
                })
    
    return stale_flags

常见问题

问题：标志评估不一致

问题：同一用户看到不同的变体 解决方案：使用一致性哈希，检查缓存策略

问题：标志债务累积

问题：代码库中存在太多旧标志 解决方案：实施标志生命周期管理，定期安排清理冲刺

问题：性能影响

问题：标志评估导致请求变慢 解决方案：使用本地缓存，批量评估

最佳实践

• 使用一致的命名规范
• 记录标志的用途和负责人
• 为临时标志设置过期日期
• 实施标志生命周期管理
• 采用渐进式推出（而非从 0→100）
• 监控标志评估指标
• 定期清理旧标志
• 在 CI 中测试两种变体

相关技能

• 蓝绿部署 - 部署策略
• Git 工作流 - 主干开发
• 告警与值班 - 监控推出情况