
Бизнес‑ценность проста: вы выпускаете изменения раньше и безопаснее. Вместо «всё или ничего» — включение новой логики для 1–5% аудитории, сбор метрик, и только потом — для всех. Если что-то пошло не так, переключатель откатывает фичу за секунды, без тревожного ночного деплоя.
Практические выгоды:
Фича‑флаги — это правила на сервере или в приложении, которые решают: кому включить новую логику.
Основные типы:
Стратегии включения:
Важно: процентное включение должно быть детерминированным — один и тот же пользователь всегда попадает в один и тот же «бакет», иначе будет «мигание» интерфейса.
Ниже — простой, но рабочий набор: таблица флагов, серверная проверка (FastAPI), клиентская проверка в интерфейсе и «кнопка‑убийца».
CREATE TABLE IF NOT EXISTS feature_flags (
key TEXT PRIMARY KEY,
enabled BOOLEAN NOT NULL DEFAULT FALSE,
rollout INTEGER NOT NULL DEFAULT 0 CHECK (rollout >= 0 AND rollout <= 100),
description TEXT NOT NULL,
owner TEXT NOT NULL,
expires_at TIMESTAMP NULL,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
-- Пример начальных данных
INSERT INTO feature_flags (key, enabled, rollout, description, owner)
VALUES
('checkout_new_ui', FALSE, 0, 'Новый интерфейс корзины', 'team-commerce'),
('fraud_kill_switch', TRUE, 100, 'Кнопка-убийца антифрода', 'team-risk')
ON CONFLICT(key) DO NOTHING;
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
import hashlib
import os
import sqlite3
from typing import Optional
DB_PATH = os.getenv('FF_DB', 'flags.db')
SALT = os.getenv('FF_SALT', 'keep-it-secret')
app = FastAPI(title='Feature Flags Service')
# Инициализация БД
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
conn.execute('''CREATE TABLE IF NOT EXISTS feature_flags (
key TEXT PRIMARY KEY,
enabled INTEGER NOT NULL DEFAULT 0,
rollout INTEGER NOT NULL DEFAULT 0,
description TEXT NOT NULL,
owner TEXT NOT NULL,
expires_at TEXT NULL,
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
)''')
conn.commit()
class FlagUpsert(BaseModel):
key: str
enabled: bool
rollout: int = 0
description: str
owner: str
expires_at: Optional[str] = None
class EvalResponse(BaseModel):n key: str
user_id: str
enabled: bool
reason: str
def get_flag(key: str):
cur = conn.execute('SELECT key, enabled, rollout FROM feature_flags WHERE key = ?', (key,))
row = cur.fetchone()
if not row:
return None
return {'key': row[0], 'enabled': bool(row[1]), 'rollout': int(row[2])}
def bucket(user_id: str, key: str) -> int:
raw = f'{user_id}:{key}:{SALT}'.encode('utf-8')
h = hashlib.sha256(raw).hexdigest()
# Возьмём первые 8 символов, чтобы превратить в число
num = int(h[:8], 16)
return num % 100
@app.get('/eval/{key}', response_model=EvalResponse)
def eval_flag(key: str, user_id: str = Query(...)):
flag = get_flag(key)
if not flag:
raise HTTPException(status_code=404, detail='flag not found')
if not flag['enabled']:
return EvalResponse(key=key, user_id=user_id, enabled=False, reason='disabled')
if flag['rollout'] >= 100:
return EvalResponse(key=key, user_id=user_id, enabled=True, reason='100%')
b = bucket(user_id, key)
enabled = b < flag['rollout']
return EvalResponse(key=key, user_id=user_id, enabled=enabled, reason=f'bucket={b}')
@app.post('/flags')
def upsert_flag(f: FlagUpsert):
if f.rollout < 0 or f.rollout > 100:
raise HTTPException(400, 'rollout must be 0..100')
conn.execute(
'INSERT INTO feature_flags (key, enabled, rollout, description, owner, expires_at, updated_at)\n VALUES (?, ?, ?, ?, ?, ?, datetime(\'now\'))\n ON CONFLICT(key) DO UPDATE SET enabled=excluded.enabled, rollout=excluded.rollout,\n description=excluded.description, owner=excluded.owner, expires_at=excluded.expires_at,\n updated_at=datetime(\'now\')',
(f.key, int(f.enabled), f.rollout, f.description, f.owner, f.expires_at)
)
conn.commit()
return {'status': 'ok'}
Использование:
# Включить фичу на 5%
curl -X POST http://localhost:8000/flags \
-H 'Content-Type: application/json' \
-d '{"key":"checkout_new_ui","enabled":true,"rollout":5,"description":"Новый UI корзины","owner":"team-commerce"}'
# Проверить для пользователя 42
curl 'http://localhost:8000/eval/checkout_new_ui?user_id=42'
Для чувствительных вещей решение должно приниматься на сервере. Но для «безопасного» UI можно проверить ответ флаг‑сервиса на клиенте.
// Детерминированный хэш для процента (если используете клиентский вариант)
function hashToBucket(str) {
// Простая DJB2
let h = 5381;
for (let i = 0; i < str.length; i++) h = ((h << 5) + h) + str.charCodeAt(i);
// Превратим в 0..99
return Math.abs(h) % 100;
}
function isEnabled(flag, userId) {
if (!flag.enabled) return false;
if (flag.rollout >= 100) return true;
const b = hashToBucket(`${userId}:${flag.key}`);
return b < flag.rollout;
}
// Пример
const flag = { key: 'checkout_new_ui', enabled: true, rollout: 25 };
const userId = 'u-123';
if (isEnabled(flag, userId)) {
// Показать новый UI
} else {
// Оставить старый
}
Храним флаг типа fraud_kill_switch и всегда проверяем его до вызова рискованной интеграции. В случае проблем — выключаем флаг, трафик переезжает на безопасный путь.
# Пример в обработчике заказов
from fastapi import HTTPException
@app.post('/pay')
def pay(order_id: str):
k = get_flag('fraud_kill_switch')
if k and not k['enabled']:
# Блокируем внешний антифрод и идём по резервной ветке
return {'status': 'accepted_without_fraud_check'}
# Основной путь
# ... логика с внешним сервисом
return {'status': 'accepted'}
Если у вас две версии сервиса (old и canary), можно разруливать трафик на уровне балансировщика. В Nginx это делается без внешних платформ.
# 5% трафика отправим на canary по детерминированному разрезанию
split_clients "$request_id" $canary_bucket {
5% "canary";
* "stable";
}
upstream app_stable {
server 10.0.0.1:8000;
}
upstream app_canary {
server 10.0.0.2:8000;
}
map $cookie_canary $force_canary {
~^1$ 1; # если есть cookie, принудительно на canary
default 0;
}
server {
listen 80;
location / {
if ($force_canary) {
proxy_pass http://app_canary;
break;
}
if ($canary_bucket = canary) {
proxy_pass http://app_canary;
}
if ($canary_bucket = stable) {
proxy_pass http://app_stable;
}
proxy_set_header Host $host;
proxy_set_header X-Request-ID $request_id;
}
}
Как управлять шагами:
После успешных тестов и деплоя — пошаговое увеличение процента. В GitHub Actions можно сделать ручной запуск с параметром «этап».
name: Progressive rollout
on:
workflow_dispatch:
inputs:
stage:
description: 'Этап раскатки (5,25,50,100)'
required: true
default: '5'
jobs:
rollout:
runs-on: ubuntu-latest
steps:
- name: Проверить доступность флаг-сервиса
run: curl -sSf http://flags.internal/health || exit 1
- name: Установить процент
env:
PERCENT: ${{ inputs.stage }}
run: |
curl -sS -X POST http://flags.internal/flags \
-H 'Content-Type: application/json' \
-d '{"key":"checkout_new_ui","enabled":true,"rollout':"'"$PERCENT"'",'"description"':"'Новый UI'",'"owner"':"'team-commerce'"}'
- name: Проверки после выкрута
run: |
# Замените на реальные health/probe/метрики
curl -sSf https://your.site/health
Лучше добавить «защитные ворота»: не повышать процент, пока метрики ошибок и времени отклика не в норме.
Минимальный набор:
Пример интеграции с Prometheus в нашем FastAPI (счётчик экспозиций):
from prometheus_client import Counter, generate_latest
from fastapi import Response
exposures = Counter('flag_exposures', 'Flag exposures', ['key', 'variant'])
@app.get('/eval/{key}', response_model=EvalResponse)
def eval_flag(key: str, user_id: str = Query(...)):
flag = get_flag(key)
if not flag:
raise HTTPException(status_code=404, detail='flag not found')
if not flag['enabled']:
exposures.labels(key=key, variant='off').inc()
return EvalResponse(key=key, user_id=user_id, enabled=False, reason='disabled')
if flag['rollout'] >= 100:
exposures.labels(key=key, variant='on').inc()
return EvalResponse(key=key, user_id=user_id, enabled=True, reason='100%')
b = bucket(user_id, key)
enabled = b < flag['rollout']
exposures.labels(key=key, variant='on' if enabled else 'off').inc()
return EvalResponse(key=key, user_id=user_id, enabled=enabled, reason=f'bucket={b}')
@app.get('/metrics')
def metrics():
return Response(generate_latest(), media_type='text/plain; version=0.0.4; charset=utf-8')
Экономия складывается из двух частей:
Снижение стоимости инцидентов. Если раньше критический баг ловили на 100% трафика, а теперь — на 5%, то прямые потери и косвенные расходы (поддержка, откаты) падают примерно в 20 раз.
Ускорение вывода фич. Разделив «развёртывание» и «включение», команда выпускает чаще, но меньшими порциями. Это снижает время до первой ценности — вы начинаете получать данные и деньги раньше.
Упрощённая формула для оценки:
Подставьте свои числа: даже 1–2 крупных инцидента, пойманных на 5% трафика, окупают внедрение.
День 1:
День 2:
День 3:
День 4:
День 5:
День 6:
День 7:
Фича‑флаги и канареечные релизы — это страховка и ускоритель для продукта. Вы разделяете «задеплоили» и «включили», берёте под контроль риски и начинаете учиться на реальных пользователях раньше конкурентов. Начните с минимальной реализации: одна таблица флагов, простой сервис принятия решений, канареечная балансировка и базовые метрики. Уже этого достаточно, чтобы снизить стоимость инцидентов и ускорить вывод фич в 2–3 раза. Дальше — автоматизация и удобные панели, но фундамент закладывается за неделю.