Skip to main content

Feature Management: Flags, Rollouts, and Experimentation

Published: February 27, 2026 Updated: May 24, 2026 Larry Qu 14 min read

Feature flags decouple deployment from release. They let you ship code to production and control who sees what, when. This guide covers feature management strategies, implementation patterns, and building confidence in your releases.

What Are Feature Flags?

Feature flags are boolean controls that toggle features on/off without deploying new code:

# Simple feature flag
if feature_flags.is_enabled("new_checkout"):
    return render_new_checkout()
else:
    return render_old_checkout()

Architecture Overview

A feature flag system consists of several layers that work together:

┌─────────────────────────────────────────────────────────────┐
│                  Feature Flag System                           │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐    │
│  │   SDKs      │    │  Dashboard  │    │  Analytics  │    │
│  │  (Web,     │◄──▶│  (Manage,  │◄──▶│  (Track,   │    │
│  │   Mobile)  │    │   Monitor) │    │   Analyze) │    │
│  └──────┬──────┘    └──────┬──────┘    └──────┬──────┘    │
│         │                  │                   │           │
│         └──────────────────┼───────────────────┘           │
│                            ▼                                │
│                  ┌─────────────────────┐                   │
│                  │     Flag Service    │                   │
│                  │  (Evaluate, Rules)  │                   │
│                  └──────────┬──────────┘                   │
│                             │                               │
│         ┌───────────────────┼───────────────────┐         │
│         ▼                   ▼                   ▼         │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐   │
│  │   Redis     │    │  Database   │    │   Config    │   │
│  │   Cache     │    │   (Rules)   │    │   (Static)  │   │
│  └─────────────┘    └─────────────┘    └─────────────┘   │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Core Concepts and Terminology

Feature Flag: A switch that controls whether a feature is enabled or disabled. A boolean value (true/false) that determines if a code path should execute.

Variant: In A/B testing, variants are different versions of a feature. Instead of just true/false, variants allow multiple options. For example, a button color test might have variants: control (blue), variant_a (green), variant_b (red).

Context: Information about the current user or request that determines which variant they should see. This includes user ID, tenant, plan tier, location, device, and other relevant attributes.

{
  "user_id": "user_123",
  "tenant_id": "tenant_456",
  "plan": "enterprise",
  "country": "US",
  "device": "mobile"
}

Rollout Percentage: The percentage of users who should see a feature. Enables gradual rollouts starting at 1% and increasing over time.

Targeting Rules: Specific conditions that determine who sees a feature, going beyond simple percentages to include user attributes. For example: “Enable for enterprise users in the US with more than 100 employees.”

Evaluation: The process of determining which variant a user should see based on their context and the flag’s rules. Consistent hashing ensures the same user always gets the same variant.

Backing Out: The ability to quickly disable a feature without redeploying code. This is the emergency off-ramp that makes feature flags valuable.

Types of Feature Flags

Type Purpose Lifetime
Release Flags Hide unfinished features Short-term
Experiment Flags A/B testing Medium-term
Operational Flags Kill switches, config Long-term
Permission Flags User-specific features Long-term

Implementing Feature Flags

Simple In-Code Implementation

class FeatureFlags:
    def __init__(self):
        self._flags = {
            "new_checkout": False,
            "dark_mode": True,
            "ai_recommendations": False,
            "beta_dashboard": ["user-1", "user-2", "user-3"],  # Users list
            "percentage_rollout": 10,  # 10% rollout
        }
    
    def is_enabled(self, flag_name: str, user_id: str = None) -> bool:
        flag = self._flags.get(flag_name)
        
        if flag is None:
            return False
        
        if isinstance(flag, bool):
            return flag
        
        if isinstance(flag, list):
            return user_id in flag
        
        if isinstance(flag, int):
            # Percentage rollout
            user_hash = hash(f"{flag_name}:{user_id}") % 100
            return user_hash < flag
        
        return False

# Usage
flags = FeatureFlags()

if flags.is_enabled("new_checkout", user_id="user-123"):
    return NewCheckoutPage()
else:
    return OldCheckoutPage()

Using a Feature Flag Service

import requests

class LaunchDarklyClient:
    def __init__(self, api_key, project_key):
        self.api_key = api_key
        self.project_key = project_key
        self.base_url = "https://app.launchdarkly.com"
    
    def is_enabled(self, flag_key: str, user_key: str) -> bool:
        response = requests.get(
            f"{self.base_url}/flags/{self.project_key}/{flag_key}",
            headers={"Authorization": api_key}
        )
        # Evaluate targeting rules
        return evaluate_flag(response.json(), user_key)

# Or use the SDK
import launchdarkly_server_sdk

ld_client = launchdarkly_server_sdk.init("api-key")
user = {"key": "user-123", "email": "[email protected]"}

show_new_feature = ld_client.variation("new-checkout", user, False)

In Configuration

# config/features.yaml
features:
  new_checkout:
    enabled: false
    rollout_percentage: 0
  
  dark_mode:
    enabled: true
    rollout_percentage: 100
  
  ai_recommendations:
    enabled: true
    rollout_percentage: 10
    target:
      - email: "*@company.com"
      - plan: "premium"

  beta_dashboard:
    enabled: true
    target_users:
      - user-1
      - user-2

Class-Based Feature Flag Architecture

For enterprise systems, use a class hierarchy with clear separation of concerns:

class FeatureFlag:
    """Base class for feature flag evaluation."""
    
    def __init__(self, key: str, default: bool = False):
        self.key = key
        self.default = default
    
    def evaluate(self, context: dict) -> bool:
        raise NotImplementedError

class SimpleFeatureFlag(FeatureFlag):
    """Simple on/off flag."""
    
    def __init__(self, key: str, enabled: bool = False):
        super().__init__(key)
        self.enabled = enabled
    
    def evaluate(self, context: dict) -> bool:
        return self.enabled

class PercentageFeatureFlag(FeatureFlag):
    """Percentage-based rollout flag."""
    
    def __init__(self, key: str, percentage: int = 0):
        super().__init__(key)
        self.percentage = max(0, min(100, percentage))
    
    def evaluate(self, context: dict) -> bool:
        user_id = context.get('user_id', 'anonymous')
        hash_value = int(hashlib.md5(
            f"{self.key}:{user_id}".encode()
        ).hexdigest(), 16)
        bucket = hash_value % 100
        return bucket < self.percentage

class TargetingFeatureFlag(FeatureFlag):
    """Flag with targeted user/group rules."""
    
    def __init__(self, key: str, rules: dict):
        super().__init__(key)
        self.rules = rules
    
    def evaluate(self, context: dict) -> bool:
        # Check specific users
        if 'users' in self.rules:
            if context.get('user_id') in self.rules['users']:
                return True
        
        # Check user attributes
        for attr, values in self.rules.get('attributes', {}).items():
            if context.get(attr) in values:
                return True
        
        # Fall back to percentage-based rollout
        percentage = self.rules.get('percentage', 0)
        hash_value = int(hashlib.md5(
            f"{self.key}:{context.get('user_id', 'anonymous')}".encode()
        ).hexdigest(), 16)
        return (hash_value % 100) < percentage

Evaluation Engine with Caching

class FeatureEngine:
    """Feature flag evaluation engine with caching."""
    
    def __init__(self):
        self.flags = {}
        self.cache = {}
    
    def load_flags(self, config: dict):
        for key, config in config.items():
            flag_type = config.get('type', 'simple')
            
            if flag_type == 'simple':
                self.flags[key] = SimpleFeatureFlag(key, config.get('enabled', False))
            elif flag_type == 'percentage':
                self.flags[key] = PercentageFeatureFlag(key, config.get('percentage', 0))
            elif flag_type == 'targeting':
                self.flags[key] = TargetingFeatureFlag(key, config.get('rules', {}))
    
    def is_enabled(self, key: str, context: dict = None) -> bool:
        context = context or {}
        if key not in self.flags:
            return False
        return self.flags[key].evaluate(context)
    
    def get_variant(self, key: str, context: dict = None) -> str:
        """Get variant for A/B testing with caching."""
        variant_key = f"{key}:variant:{context.get('user_id', 'anonymous')}"
        
        if variant_key in self.cache:
            return self.cache[variant_key]
        
        hash_value = int(hashlib.md5(
            f"{key}:{context.get('user_id')}".encode()
        ).hexdigest(), 16)
        variants = ['control', 'variant_a']
        
        variant = variants[hash_value % len(variants)]
        self.cache[variant_key] = variant
        return variant

FastAPI Middleware Integration

from fastapi import Request

class FeatureMiddleware:
    """Inject feature flags into request scope."""
    
    def __init__(self, app, feature_engine):
        self.app = app
        self.engine = feature_engine
    
    async def __call__(self, scope, receive, send):
        if scope['type'] == 'http':
            context = {
                'user_id': dict(scope.get('headers', {})).get(b'x-user-id', b'anonymous').decode(),
                'plan': dict(scope.get('headers', {})).get(b'x-user-plan', b'free').decode()
            }
            scope['features'] = {
                key: self.engine.is_enabled(key, context)
                for key in self.engine.flags.keys()
            }
        
        await self.app(scope, receive, send)

# Usage
@app.get("/dashboard")
async def dashboard(request: Request):
    if request.app.state.features.get('new_dashboard'):
        return NewDashboard()
    return OldDashboard()

Canary Releases

Gradually roll out changes to a small percentage of users:

import hashlib

class CanaryRelease:
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.rollout_percentage = 10
    
    def should_route_to_canary(self, user_id: str = None) -> bool:
        if not user_id:
            # Use random for no-user requests
            import random
            return random.random() * 100 < self.rollout_percentage
        
        # Consistent hashing - same user always gets same result
        hash_value = int(hashlib.md5(
            f"{self.service_name}:{user_id}".encode()
        ).hexdigest(), 16)
        
        return (hash_value % 100) < self.rollout_percentage

# Usage with API Gateway
@app.route("/api/<path:endpoint>")
async def proxy(endpoint, request):
    user_id = get_user_id(request)
    
    canary = CanaryRelease("order-service")
    
    if canary.should_route_to_canary(user_id):
        # Route to canary version
        return await proxy_to("order-service-v2", request)
    else:
        # Route to stable version
        return await proxy_to("order-service-v1", request)

Kubernetes Canary

# Canary deployment with Istio
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    - route:
        - destination:
            host: order-service-v1
          weight: 90
        - destination:
            host: order-service-v2
          weight: 10

Flagger Canary

apiVersion: flagger.app/v1beta1
kind: Canary
metadata:
  name: order-service
spec:
  targetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: order-service
  service:
    port: 80
  canaryAnalysis:
    interval: 1m
    threshold: 10
    maxWeight: 50
    stepWeight: 10
    metrics:
      - name: request-success-rate
        threshold: 99
        interval: 1m
      - name: error-rate
        threshold: 5
        interval: 1m

A/B Testing

Test different variations with user segments:

class Experiment:
    def __init__(self, name: str, variations: dict):
        """
        variations: {
            "control": 50,  # 50% weight
            "variant_a": 25,
            "variant_b": 25
        }
        """
        self.name = name
        self.variations = variations
    
    def get_variant(self, user_id: str = None) -> str:
        if not user_id:
            import random
            r = random.random() * 100
        else:
            # Consistent hashing
            hash_value = int(hashlib.md5(
                f"{self.name}:{user_id}".encode()
            ).hexdigest(), 16)
            r = (hash_value % 10000) / 100
        
        cumulative = 0
        for variant, weight in self.variations.items():
            cumulative += weight
            if r < cumulative:
                return variant
        
        return "control"

# Usage
checkout_experiment = Experiment(
    name="new_checkout",
    variations={"control": 50, "variant_a": 25, "variant_b": 25}
)

variant = checkout_experiment.get_variant(user_id="user-123")

if variant == "control":
    return render_old_checkout()
elif variant == "variant_a":
    return render_checkout_with_visa()
else:
    return render_checkout_with_paypal()

Tracking Results

import analytics

class ExperimentTracker:
    def __init__(self, experiment_name: str):
        self.experiment_name = experiment_name
    
    def track_assignment(self, user_id: str, variant: str):
        analytics.track(
            user_id,
            "Experiment Assigned",
            {
                "experiment_name": self.experiment_name,
                "variant": variant
            }
        )
    
    def track_conversion(self, user_id: str, event_name: str, properties: dict = None):
        analytics.track(
            user_id,
            event_name,
            {
                "experiment_name": self.experiment_name,
                **(properties or {})
            }
        )

# Usage
tracker = ExperimentTracker("new_checkout")

@app.route("/checkout")
def checkout():
    variant = checkout_experiment.get_variant(user_id)
    tracker.track_assignment(user_id, variant)
    
    if variant == "control":
        return render_checkout_control()
    return render_checkout_variant()

@app.route("/purchase")
def purchase():
    # Track conversion
    tracker.track_conversion(user_id, "purchase_completed", {
        "total": 99.99,
        "variant": checkout_experiment.get_variant(user_id)
    })

Kill Switches

Quickly disable problematic features:

class KillSwitch:
    def __init__(self):
        self._switches = {}
    
    def enable(self, feature: str):
        self._switches[feature] = True
    
    def disable(self, feature: str):
        self._switches[feature] = False
    
    def is_active(self, feature: str) -> bool:
        return self._switches.get(feature, True)
    
    async def wrap(self, feature: str, func, *args, **kwargs):
        if not self.is_active(feature):
            return {"error": f"Feature {feature} is disabled"}
        
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            # Auto-disable on error
            logger.error(f"Feature {feature} failed, disabling", error=str(e))
            self.disable(feature)
            raise

# Usage
killswitch = KillSwitch()

@killswitch.wrap("ai_recommendations", ai_service.get_recommendations, user_id)
async def get_recommendations(user_id):
    return await ai_service.get_recommendations(user_id)

Gradual Rollout

Increase rollout based on metrics:

class GradualRollout:
    def __init__(self, feature_name: str, initial_percentage: int = 1):
        self.feature_name = feature_name
        self.current_percentage = initial_percentage
        self.metrics = {"errors": 0, "success": 0, "p50": [], "p99": []}
    
    def record_success(self, latency_ms: int):
        self.metrics["success"] += 1
        self.metrics["p50"].append(latency_ms)
        self.metrics["p99"].append(latency_ms)
    
    def record_error(self):
        self.metrics["errors"] += 1
    
    def should_increase(self) -> bool:
        total = self.metrics["success"] + self.metrics["errors"]
        if total < 1000:
            return False
        
        error_rate = self.metrics["errors"] / total
        p99_latency = sorted(self.metrics["p99"])[int(len(self.metrics["p99"]) * 0.99)]
        
        # Increase if error rate < 1% and p99 < 500ms
        return error_rate < 0.01 and p99_latency < 500
    
    def promote(self):
        if self.should_increase() and self.current_percentage < 100:
            self.current_percentage = min(100, self.current_percentage * 2)
            logger.info(f"Promoted {self.feature_name} to {self.current_percentage}%")
            
            # Reset metrics for next phase
            self.metrics = {"errors": 0, "success": 0, "p50": [], "p99": []}

Sample Size Calculation for Experiments

Statistical significance depends on having enough data. Calculate required sample sizes before running experiments:

import math
from scipy import stats

class SampleSizeCalculator:
    """Calculate required sample size for A/B tests."""
    
    def calculate_sample_size(self,
                             baseline_rate: float,
                             min_detectable_effect: float,
                             alpha: float = 0.05,
                             power: float = 0.8) -> int:
        """Calculate required sample size per variant."""
        z_alpha = stats.norm.ppf(1 - alpha/2)
        z_beta = stats.norm.ppf(power)
        
        control_rate = baseline_rate
        variant_rate = baseline_rate * (1 + min_detectable_effect)
        pooled_rate = (control_rate + variant_rate) / 2
        
        sample_size = (
            (z_alpha + z_beta) ** 2 *
            (control_rate * (1 - control_rate) + variant_rate * (1 - variant_rate))
        ) / (variant_rate - control_rate) ** 2
        
        return math.ceil(sample_size)
    
    def calculate_duration(self, sample_size: int, daily_traffic: int) -> int:
        """Calculate test duration in days."""
        return math.ceil(sample_size / daily_traffic)

# Usage
calculator = SampleSizeCalculator()
n = calculator.calculate_sample_size(
    baseline_rate=0.05,  # 5% baseline conversion
    min_detectable_effect=0.1,  # Want to detect 10% relative change
    alpha=0.05,  # 95% confidence
    power=0.8    # 80% power
)
print(f"Need {n} users per variant")

S-Curve Rollout Strategy

An S-curve rollout starts slowly, accelerates through the middle, and slows at the end — matching how adoption naturally occurs:

from datetime import datetime
import math

class SCurveRollout:
    """S-curve rollout with logistic function."""
    
    def __init__(self, start_percentage: int = 1,
                 end_percentage: int = 100,
                 duration_hours: int = 72):
        self.start_percentage = start_percentage
        self.end_percentage = end_percentage
        self.duration_hours = duration_hours
        self.start_time = datetime.utcnow()
    
    def get_current_percentage(self) -> int:
        elapsed = (datetime.utcnow() - self.start_time).total_seconds() / 3600
        if elapsed >= self.duration_hours:
            return self.end_percentage
        
        progress = elapsed / self.duration_hours
        k = 10  # Steepness factor
        
        # Logistic function for S-curve
        percentage = self.start_percentage + (
            (self.end_percentage - self.start_percentage) /
            (1 + math.exp(-k * (progress - 0.5)))
        )
        return int(percentage)

Canary Rollout with Health Checks

class CanaryRollout:
    """Canary rollout with automated health verification."""
    
    def __init__(self, initial_percentage: int = 1,
                 increment: int = 5,
                 health_interval: int = 300):
        self.current_percentage = initial_percentage
        self.increment = increment
        self.health_interval = health_interval
        self.errors = []
    
    def should_enable(self, user_id: str) -> bool:
        hash_value = int(hashlib.md5(f"canary:{user_id}".encode()).hexdigest(), 16)
        return (hash_value % 100) < self.current_percentage
    
    def record_error(self, error: str):
        self.errors.append({
            'timestamp': datetime.utcnow(),
            'error': error
        })
    
    def should_increase(self) -> bool:
        if len(self.errors) < 10:
            return True
        
        recent = [e for e in self.errors
                  if (datetime.utcnow() - e['timestamp']).total_seconds() < 3600]
        error_rate = len(recent) / len(self.errors)
        return error_rate < 0.01  # Less than 1% error rate

Feature Flag Management UI

Build a simple admin interface:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI()

class FeatureFlag(BaseModel):
    name: str
    enabled: bool
    rollout_percentage: int = 100
    target_users: Optional[List[str]] = None

flags_db = {}

@app.get("/flags")
def list_flags():
    return flags_db

@app.post("/flags")
def create_flag(flag: FeatureFlag):
    flags_db[flag.name] = flag
    return flag

@app.put("/flags/{flag_name}")
def update_flag(flag_name: str, flag: FeatureFlag):
    if flag_name not in flags_db:
        raise HTTPException(status_code=404, detail="Flag not found")
    flags_db[flag_name] = flag
    return flag

@app.delete("/flags/{flag_name}")
def delete_flag(flag_name: str):
    if flag_name in flags_db:
        del flags_db[flag_name]
    return {"status": "deleted"}

Tools Comparison

Tool Type Features Pricing
LaunchDarkly Managed Full-featured $$$
Split.io Managed A/B testing $$$
Unleash Open-source Self-hosted option Free/$
FeatureFlags.io Open-source Simple Free
Cloudflare Workers Built-in Edge $
Custom Build your own Flexible Free

Best Practices

Do

  • Keep flags short-lived
  • Clean up after rollout
  • Monitor flag metrics
  • Have rollback plan
  • Use meaningful names

Don’t

  • Don’t use for every small change
  • Don’t nest flags deeply
  • Don’t forget to clean up
  • Don’t ignore failure modes

Anti-Patterns to Avoid

1. Flag Sprawl: Creating flags for every minor change leads to a tangled mess. Use flags for significant features only and clean up after rollout complete.

2. Flag Debt: Leaving flags in code for years creates dead code paths and confusion. Set expiration dates and remove flags after successful rollout.

3. No Monitoring: Flags that aren’t monitored for performance impact create blind spots. Track error rates, latency, and user feedback for every active flag.

4. Complex Flag Logic: Deeply nested flag conditions make code impossible to reason about. Keep flag logic simple and use targeting rules instead of conditional nesting.

5. Bypassing Flags in Tests: Test both flag states explicitly. Bugs often hide in the disabled code path that nobody tests.

// Test both flag states
it('should show new dashboard when flag enabled', () => {
  flags.enable('new-dashboard');
  const result = renderDashboard();
  expect(result).toContain('New Feature');
});

it('should show old dashboard when flag disabled', () => {
  flags.disable('new-dashboard');
  const result = renderDashboard();
  expect(result).toContain('Legacy Dashboard');
});

Postgres-Based Flags

# Self-hosted feature flags
class PostgresFeatureFlags:
    
    def __init__(self, db_pool):
        self.db = db_pool
    
    def get_flags(self, user_id):
        """Get all flags for user."""
        query = """
            SELECT f.name, f.enabled, f.rollout_percentage
            FROM feature_flags f
            WHERE f.enabled = true
        """
        flags = self.db.execute(query)
        
        return {
            f["name"]: self._evaluate_flag(f, user_id)
            for f in flags
        }
    
    def _evaluate_flag(self, flag, user_id):
        if flag["rollout_percentage"] is None:
            return True
        
        return hash(f"{flag['name']}:{user_id}") % 100 < flag["rollout_percentage"]

Unleash Feature Flags

# Unleash client
from unleash_client import UnleashClient

client = UnleashClient(
    url="https://unleash.example.com/api",
    app_name="payment-service",
    environment="prod"
)

client.start()

# Check flag
if client.is_enabled("new-payment-flow"):
    enable_new_flow()

Experimentation

A/B Testing Setup

# Simple A/B test implementation
class Experiment:
    
    def __init__(self, name, variants):
        """
        variants: {"control": 50, "treatment": 50}
        """
        self.name = name
        self.variants = variants
    
    def assign_variant(self, user_id):
        """Assign user to variant."""
        # Consistent assignment
        bucket = hash(f"{self.name}:{user_id}") % 100
        
        cumulative = 0
        for variant, percentage in self.variants.items():
            cumulative += percentage
            if bucket < cumulative:
                return variant
        
        return "control"

# Usage
checkout_experiment = Experiment(
    "new_checkout",
    {"control": 50, "treatment": 50}
)

variant = checkout_experiment.assign_variant(current_user.id)

if variant == "treatment":
    show_new_checkout()
    track_event("checkout_viewed", {"variant": "treatment"})
else:
    show_old_checkout()
    track_event("checkout_viewed", {"variant": "control"})

Statistical Analysis

# Calculate experiment results
import math
from dataclasses import dataclass

@dataclass
class ExperimentResults:
    control_conversions: int
    treatment_conversions: int
    control_size: int
    treatment_size: int
    
    @property
    def control_rate(self):
        return self.control_conversions / self.control_size
    
    @property
    def treatment_rate(self):
        return self.treatment_conversions / self.treatment_size
    
    def confidence_level(self):
        """Calculate statistical significance."""
        n1, n2 = self.control_size, self.treatment_size
        p1, p2 = self.control_rate, self.treatment_rate
        
        # Pooled proportion
        p = (self.control_conversions + self.treatment_conversions) / (n1 + n2)
        
        # Standard error
        se = math.sqrt(p * (1 - p) * (1/n1 + 1/n2))
        
        # Z-score
        z = (p2 - p1) / se
        
        # Convert to confidence
        if z > 1.96:
            return "95% confident - significant"
        elif z > 1.645:
            return "90% confident"
        else:
            return "Not significant"

Kubernetes Canary Deployments

# Kubernetes canary with feature flags
apiVersion: v1
kind: Service
metadata:
  name: payment-service
spec:
  selector:
    app: payment-service
  ports:
    - port: 80
      targetPort: 8080
---
# Stable deployment (90%)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: payment-service-stable
spec:
  replicas: 9
  selector:
    matchLabels:
      app: payment-service
      version: stable
  template:
    spec:
      containers:
        - name: payment
          image: payment-service:v2.0
          env:
            - name: FEATURE_NEW_CHECKOUT
              value: "false"
---
# Canary deployment (10%)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: payment-service-canary
spec:
  replicas: 1
  selector:
    matchLabels:
      app: payment-service
      version: canary
  template:
    spec:
      containers:
        - name: payment
          image: payment-service:v2.0
          env:
            - name: FEATURE_NEW_CHECKOUT
              value: "true"

Progressive Rollout

# Automated progressive rollout
async def progressive_rollout(flag_name, target_percentage, step=10):
    """Gradually increase rollout percentage."""
    current = 0
    
    while current < target_percentage:
        current += step
        
        # Update flag
        await update_flag_percentage(flag_name, current)
        
        # Wait and monitor
        await wait(1 hour)
        
        # Check metrics
        error_rate = await get_error_rate(flag_name)
        
        if error_rate > 0.01:  # 1% threshold
            print(f"Error rate {error_rate} exceeded threshold, rolling back")
            await update_flag_percentage(flag_name, 0)
            break
        
        print(f"Rollout: {current}% - error rate: {error_rate}")

Progressive Rollout

# Automated progressive rollout
async def progressive_rollout(flag_name, target_percentage, step=10):
    """Gradually increase rollout percentage."""
    current = 0
    
    while current < target_percentage:
        current += step
        
        # Update flag
        await update_flag_percentage(flag_name, current)
        
        # Wait and monitor
        await wait(1 hour)
        
        # Check metrics
        error_rate = await get_error_rate(flag_name)
        
        if error_rate > 0.01:  # 1% threshold
            print(f"Error rate {error_rate} exceeded threshold, rolling back")
            await update_flag_percentage(flag_name, 0)
            break
        
        print(f"Rollout: {current}% - error rate: {error_rate}")

Conclusion

Feature flags transform how you ship software:

  • Separate deployment from release
  • Enable canary and gradual rollouts
  • Run A/B experiments safely
  • Kill switches for quick response
  • Build your own or use a service

Start with simple flags, add experimentation, then build full progressive delivery.

External Resources

Comments

👍 Was this article helpful?