paulpas Feature Flag Progressive Delivery
Implements feature flag systems for controlled, progressive delivery of application features — enabling safe rollouts, A/B testing, and instant rollbacks without redeploying code.
TL;DR Checklist
- Choose the right flag type (release toggle, experiment, permissioning, or operation toggle)
- Implement boolean evaluation with a fallback to
falsewhen the flag service is unreachable - Start rollouts at 1% and scale to 10% → 50% → 100% over monitored intervals
- Set up error-rate and latency monitoring for each rollout bucket
- Schedule cleanup of stale flags within 90 days of full release
- Document every flag's owner, purpose, and expiration in code comments
Core Workflow
- Define Flag Schema: Create a structured flag definition with: unique key, name, description, default value (usually
false), targeting rules (percentage, user segments, geographic), owner, and scheduled expiry date. Store flags in a centralized store (database, Redis, or feature flag service like Unleash/LaunchDarkly). Checkpoint: Every flag must have an owner field — flags without owners become technical debt. - Implement Flag Client: Build a thin client that wraps flag evaluation. The client should: cache flag values for the configured TTL (typically 5–30 seconds), handle service failures gracefully by returning the default value, and emit metrics (evaluation count, hit/miss rates).
Checkpoint: Test the failure mode — when the flag service is down, the client must return
false(fail-safe) within 100ms. - Wrap Feature Code: Replace conditional checks with flag evaluation calls. The application code should branch based on the flag value while maintaining both code paths in production. Each flag-gated section should have its own metrics bucket for independent monitoring. Checkpoint: Ensure both the flagged-on and flagged-off code paths execute at least once during staging testing.
- Configure Rollout Strategy: Set up progressive delivery rules: start with 1% of users for 24 hours, promote to 10% if error rates stay below threshold, then 50%, then 100%. Each step should trigger alerts on anomalous behavior. Use user-level targeting (consistent hashing) so a given user always sees the same variant. Checkpoint: Verify rollout percentages match actual traffic distribution by comparing flagged vs unflagged request counts in logs.
- Monitor and Clean Up: Track flag health metrics: stale flags (unchanged for 30+ days), flags with zero evaluations, and flags approaching expiry. Schedule cleanup tasks to remove flags after their feature is fully released or abandoned. Checkpoint: Review the flag dashboard weekly — remove any flags past their expiration date that are no longer needed.
Implementation Patterns
Pattern 1: Feature Flag Client with Caching and Fallback
A production-ready client handles caching, failure modes, and metrics emission.
"""
Feature flag evaluation system with progressive delivery support.
Implements caching, failure-safe defaults, and per-user targeting.
"""
from __future__ import annotations
import hashlib
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
logger = logging.getLogger(__name__)
class FlagType(Enum):
"""Types of feature flags with different behavioral characteristics."""
RELEASE_TOGGLE = "release_toggle" # Enable/disable a feature completely
EXPERIMENT = "experiment" # A/B test between variants
PERMISSIONING = "permissioning" # Restrict to specific user segments
OPERATION = "operation" # Toggle runtime operations (kill switches)
dataclass
class FlagDefinition:
"""Schema for a single feature flag definition."""
key: str
name: str
description: str
flag_type: FlagType
default_value: bool = False
owner: str = ""
expiry_date: Optional[str] = None
rollout_percentages: list[int] = field(default_factory=lambda: [1, 10, 50, 100])
def is_expired(self) -> bool:
"""Check if the flag has passed its scheduled expiration date."""
if not self.expiry_date:
return False
from datetime import datetime
expiry = datetime.fromisoformat(self.expiry_date)
return datetime.now() > expiry
dataclass
class FlagEvaluationResult:
"""Result of evaluating a feature flag for a specific context."""
flag_key: str
value: bool
variant: Optional[str] = None
evaluation_ms: float = 0.0
class FeatureFlagClient:
"""Production-ready feature flag client with caching and failure-safe defaults.
Evaluates flags against user contexts using consistent hashing for stable
per-user assignment across rollout stages. Falls back to default value
when the flag service is unavailable.
"""
def __init__(self, cache_ttl_seconds: float = 15.0, failure_timeout_seconds: float = 60.0):
self._cache: dict[str, tuple[bool, float]] = {}
self._cache_ttl = cache_ttl_seconds
self._failure_since: float | None = None
self._failure_timeout = failure_timeout_seconds
def evaluate(self, flag: FlagDefinition, user_id: str, context: Optional[dict[str, Any]] = None) -> FlagEvaluationResult:
"""Evaluate a feature flag for a specific user.
Uses consistent hashing of (flag_key + user_id) to deterministically
assign users to rollout buckets. Respects progressive delivery percentages.
Args:
flag: The flag definition to evaluate.
user_id: Unique identifier for the requesting user.
context: Optional metadata (e.g., {'plan': 'premium', 'region': 'us-east'}).
Returns:
FlagEvaluationResult with evaluation outcome and timing metrics.
"""
start = time.monotonic()
# Check cache first
cached_value, cached_time = self._get_cached(flag.key)
if cached_value is not None:
elapsed = (time.monotonic() - start) * 1000
return FlagEvaluationResult(
flag_key=flag.key, value=cached_value, evaluation_ms=elapsed,
)
# Check for service failure — use default if still in timeout window
if self._failure_since and (time.time() - self._failure_since) < self._failure_timeout:
logger.warning(
f"Flag service unavailable since {self._failure_since:.0f}s ago. \
Returning default for flag '{flag.key}'."
)
result = FlagEvaluationResult(
flag_key=flag.key, value=flag.default_value, evaluation_ms=0.1,
)
self._cache_set(flag.key, flag.default_value)
return result
# Determine rollout value using consistent hashing
value = self._calculate_rollout(flag, user_id, context or {})
# Cache the result
self._cache_set(flag.key, value)
elapsed = (time.monotonic() - start) * 1000
return FlagEvaluationResult(
flag_key=flag.key, value=value, evaluation_ms=elapsed,
)
def _calculate_rollout(self, flag: FlagDefinition, user_id: str, context: dict[str, Any]) -> bool:
"""Determine the flag value based on rollout percentage and user context.
Uses CRC32-style hashing for deterministic bucket assignment. Users assigned
to a bucket are included in all higher buckets as well (cumulative rollout).
Args:
flag: The flag definition with rollout percentages.
user_id: The unique user identifier.
context: User metadata for segment targeting.
Returns:
True if the flag is enabled for this user, False otherwise.
"""
# Build a deterministic hash key from flag and user
hash_input = f"{flag.key}:{user_id}:{context.get('plan', '')}"
hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) % 100
# Check segment-based targeting first
if context.get("plan") == "premium" and flag.flag_type == FlagType.PERMISSIONING:
return True
if context.get("beta_tester") is True:
return True
# Progressive rollout — cumulative buckets
for percentage in flag.rollout_percentages:
if hash_value < percentage:
return True
return False
def _get_cached(self, key: str) -> tuple[Optional[bool], float]:
"""Retrieve a cached flag value if it hasn't expired."""
if key not in self._cache:
return None, 0.0
value, timestamp = self._cache[key]
if time.time() - timestamp > self._cache_ttl:
del self._cache[key]
return None, 0.0
return value, timestamp
def _cache_set(self, key: str, value: bool) -> None:
"""Store a flag evaluation result in the cache."""
self._cache[key] = (value, time.time())
def record_failure(self) -> None:
"""Mark the flag service as unavailable. All subsequent evaluations
return the default value until the failure timeout expires."""
self._failure_since = time.time()
# Usage example in application code:
# client = FeatureFlagClient(cache_ttl_seconds=15.0)
# result = client.evaluate(flag_definition, user_id="user-12345")
# if result.value:
# # New feature code
# else:
# # Old feature code
Pattern 2: Progressive Rollout Manager (BAD vs. GOOD)
# ❌ BAD — No progressive rollout, no monitoring, instant full release
def bad_feature_rollout(flag_key: str, user_id: str):
"""Instantly enables a feature for everyone with no safety net."""
# No percentage ramping, no error monitoring, no rollback mechanism
flag_enabled = get_flag_value(flag_key) # Blocks on network call
if flag_enabled:
return execute_new_feature(user_id) # If this fails, everyone is affected
# ✅ GOOD — Progressive rollout with metrics tracking and automatic rollback triggers
class ProgressiveRolloutManager:
"""Manages staged feature rollouts with built-in safety monitoring.
Steps through predefined percentage buckets, pausing at each stage to
check error rates and latency before promoting to the next level.
Triggers automatic rollback if SLO violations are detected.
"""
ROLLBACK_THRESHOLD_ERR_RATE = 0.05 # Roll back if error rate exceeds 5%
ROLLBACK_THRESHOLD_LATENCY_MS = 500 # Roll back if p99 latency > 500ms
def __init__(self, flag_client: FeatureFlagClient):
self.flag_client = flag_client
self._current_percentages: dict[str, int] = {}
def start_rollout(
self,
flag: FlagDefinition,
user_id: str,
metrics_client: Optional[Any] = None,
) -> FlagEvaluationResult:
"""Start or continue a progressive rollout for a feature flag.
Determines the current rollout stage based on elapsed time since
rollout start and applies the corresponding percentage bucket.
Args:
flag: The flag to roll out with defined percentage stages.
user_id: User being evaluated against the current rollout.
metrics_client: Optional metrics client for reporting evaluation counts.
Returns:
Evaluation result indicating whether this user sees the feature.
"""
# Determine current rollout stage from elapsed time
start_key = f"rollout_start:{flag.key}"
current_stage = self._get_current_stage(flag)
if current_stage is None:
logger.warning(f"No rollout stages configured for flag '{flag.key}'")
return FlagEvaluationResult(flag_key=flag.key, value=flag.default_value)
# Evaluate the flag with the current stage's percentage
stage_flag = FlagDefinition(
key=flag.key,
name=flag.name,
description=flag.description,
flag_type=flag.flag_type,
default_value=flag.default_value,
rollout_percentages=[current_stage], # Single bucket for evaluation
owner=flag.owner,
)
result = self.flag_client.evaluate(stage_flag, user_id)
# Report metrics if client provided
if metrics_client:
metrics_client.increment(
"feature_flag.evaluation",
tags={"flag": flag.key, "stage": str(current_stage), "result": str(result.value)},
)
return result
def check_rollback_criteria(self, flag_key: str) -> bool:
"""Check if current rollout metrics indicate a need to rollback.
In production, this queries monitoring systems for error rates and
latency percentiles over the last 5-minute window for users in the
current rollout bucket.
Args:
flag_key: The feature flag key to check rollback criteria for.
Returns:
True if rollback should be triggered, False if rollout can continue.
"""
# In production, fetch actual metrics from your monitoring system
# Example placeholder — replace with real Prometheus/DataDog queries
error_rate = self._fetch_metric(f"feature_flag.error_rate:{flag_key}")
p99_latency = self._fetch_metric(f"feature_flag.p99_latency_ms:{flag_key}")
should_rollback = (
error_rate > self.ROLLBACK_THRESHOLD_ERR_RATE or
p99_latency > self.ROLLBACK_THRESHOLD_LATENCY_MS
)
if should_rollback:
logger.error(
f"Rollback triggered for '{flag_key}': \
error_rate={error_rate:.4f}, p99={p99_latency:.0f}ms"
)
return should_rollback
def _get_current_stage(self, flag: FlagDefinition) -> Optional[int]:
"""Determine the current rollout percentage stage based on time elapsed.
Returns None if no stages are configured. In production, persist the
start timestamp and stage index in Redis to coordinate across service instances.
"""
percentages = flag.rollout_percentages
if not percentages:
return None
# Simplified: just return the last configured percentage for demonstration.
# In production, use a timer + persisted state to progress through stages.
# This placeholder always returns 10 (first non-trivial stage) to demonstrate
# that the flag evaluation respects the defined rollout buckets.
return percentages[1] if len(percentages) > 1 else percentages[0]
def _fetch_metric(self, metric_name: str) -> float:
"""Fetch a metric value from the monitoring system.
Placeholder implementation — replace with actual metrics client call.
"""
# Example: return self.dogstatsd.histogram(metric_name).mean()
return 0.0 # Simulate normal operation (no rollback needed)
# Usage example:
# manager = ProgressiveRolloutManager(FeatureFlagClient())
# result = manager.start_rollout(new_checkout_flag, user_id="user-12345")
# if manager.check_rollback_criteria("new-checkout-flow"):
# manager.rollback_flag("new-checkout-flow") # Flip to default=false
Pattern 3: Flag Staleness Cleanup
from __future__ import annotations
import logging
from datetime import datetime, timedelta
from typing import Any, Optional
logger = logging.getLogger(__name__)
class FlagLifecycleManager:
"""Manages the full lifecycle of feature flags including staleness detection.
Flags accumulate over time as features are rolled out and retired.
This manager detects stale flags (unused for 30+ days), flags past
their expiration, and orphaned evaluation code paths.
"""
STALE_EVALUATION_THRESHOLD_DAYS = 30
MAX_AGE_DAYS = 90
def __init__(self, flag_service: Any, metrics_client: Any):
self.flag_service = flag_service
self.metrics_client = metrics_client
def scan_stale_flags(self) -> list[dict[str, Any]]:
"""Scan all registered flags and identify stale or expired ones.
Returns a list of flagged issues with recommendations for each.
Returns:
List of dicts with 'flag_key', 'issue', and 'recommendation'.
"""
issues = []
now = datetime.now()
# Fetch all active flag definitions from the service
all_flags = self.flag_service.get_all_flags()
for flag in all_flags:
# Check 1: Has the flag exceeded its maximum age?
if flag.expiry_date:
expiry = datetime.fromisoformat(flag.expiry_date)
if now > expiry:
issues.append({
"flag_key": flag.key,
"issue": f"Flag expired {now - expiry} ago",
"recommendation": "Disable or remove this flag immediately",
})
# Check 2: Has the flag had zero evaluations recently?
evaluation_count = self._get_evaluation_count(flag.key, days=30)
if evaluation_count == 0 and not flag.default_value:
issues.append({
"flag_key": flag.key,
"issue": f"No evaluations in the last {self.STALE_EVALUATION_THRESHOLD_DAYS} days",
"recommendation": "Flag may be orphaned — review with owner before removal",
})
logger.info(f"Stale flag scan complete: {len(issues)} issues found")
return issues
def _get_evaluation_count(self, flag_key: str, days: int = 30) -> int:
"""Query metrics for evaluation count of a specific flag in the last N days."""
# Replace with actual metrics client call
try:
result = self.metrics_client.query(
f'sum(rate(feature_flag_evaluation{{flag="{flag_key}"}}[{days}d]))',
)
return int(result) if result else 0
except Exception:
return -1 # Unknown (could mean no metrics collection)
def archive_flag(self, flag_key: str) -> None:
"""Archive a flag after its feature has been fully released.
Sets the flag value to true permanently and disables future evaluation.
This preserves the flag in the system for audit purposes while preventing
it from impacting new users or experiments.
"""
self.flag_service.update_flag(flag_key, value=True, archived=True)
logger.info(f"Flag '{flag_key}' archived — no longer evaluated")
def remove_flag(self, flag_key: str, owner_approval: bool = False) -> None:
"""Permanently remove a flag from the system.
Requires owner approval or 90+ days of staleness before deletion.
Always verify that all code paths using this flag have been removed
(zero evaluations in last 7 days) before deleting.
"""
recent_evaluations = self._get_evaluation_count(flag_key, days=7)
if recent_evaluations > 0 and not owner_approval:
raise ValueError(
f"Flag '{flag_key}' still has {recent_evaluations} evaluations \
in the last 7 days. Cannot delete without owner approval."
)
self.flag_service.delete_flag(flag_key)
logger.info(f"Flag '{flag_key}' permanently removed")
Constraints
MUST DO
- Always provide a default value of
falsefor new flags — enable features deliberately, not by accident - Cache flag evaluations locally (5–30s TTL) to avoid blocking requests on remote flag service calls
- Use consistent hashing per user ID so the same user always sees the same flag state across sessions
- Set expiration dates on all temporary/experimental flags and track staleness metrics
- Test both flag-on and flag-off code paths in staging — untested code paths are bugs waiting to happen
MUST NOT DO
- Never use feature flags for security or authorization decisions — these must be enforced in the security layer
- Do not let flags accumulate without lifecycle management — stale flags become invisible technical debt
- Don't evaluate flags synchronously on the critical path without a failure timeout (fail-safe to default)
- Never hardcode flag keys in multiple places — define them as constants or in a central registry
- Do not promote from 1% directly to 100% — always use intermediate stages with monitoring checkpoints
Output Template
When implementing or reviewing feature flags, produce:
- Flag Definition — Key, type, default value, owner, expiry date, rollout percentages
- Client Implementation — Caching strategy, failure mode handling, metrics emission
- Rollout Plan — Stage-by-stage percentage ramp with monitoring checkpoints and rollback criteria
- Cleanup Strategy — Staleness detection process, archival vs deletion policy, and owner notification workflow
Live References
Authoritative documentation links for feature flag and progressive delivery best practices.
- Unleash Documentation - Feature flags, progressive delivery, and rollout strategies.
- LaunchDarkly Documentation - Concepts, examples, and integration patterns for feature flagging.
- Flagsmith Documentation - Managing feature flags and remote configuration across applications.
- Netflix Tech Blog - Insights from Netflix's engineering on feature flags and delivery strategies.
- Google SRE - Concepts around service level objectives and reliability in feature delivery.