Blue-Green Deployment Testing: Complete Guide for DevOps Teams

#!/bin/bash
# smoke-test.sh - Basic health check for green environment

GREEN_URL="https://green.example.com"

# Check application is responding
if ! curl -f -s "${GREEN_URL}/health" > /dev/null; then
    echo "❌ Health endpoint not responding"
    exit 1
fi

# Verify database connectivity
if ! curl -f -s "${GREEN_URL}/api/db-check" | grep -q "OK"; then
    echo "❌ Database connection failed"
    exit 1
fi

# Check critical dependencies
for service in redis kafka elasticsearch; do
    if ! curl -f -s "${GREEN_URL}/api/check/${service}" | grep -q "healthy"; then
        echo "❌ ${service} dependency check failed"
        exit 1
    fi
done

echo "✅ All smoke tests passed"

# test_green_integration.py
import pytest
import requests

GREEN_BASE_URL = "https://green.example.com"

def test_user_registration_flow():
    """Test complete user registration workflow"""
    # Create user
    response = requests.post(f"{GREEN_BASE_URL}/api/users", json={
        "email": "test@example.com",
        "password": "SecurePass123!"
    })
    assert response.status_code == 201
    user_id = response.json()["id"]

    # Verify email sent
    email_check = requests.get(f"{GREEN_BASE_URL}/api/emails/{user_id}")
    assert email_check.json()["type"] == "verification"

    # Complete verification
    token = email_check.json()["token"]
    verify = requests.post(f"{GREEN_BASE_URL}/api/verify", json={"token": token})
    assert verify.status_code == 200

def test_payment_processing():
    """Verify payment gateway integration"""
    response = requests.post(f"{GREEN_BASE_URL}/api/payments", json={
        "amount": 1000,
        "currency": "USD",
        "method": "card"
    })
    assert response.status_code == 200
    assert response.json()["status"] == "processed"

-- validate_migration.sql
-- Run these checks before traffic switch

-- 1. Verify schema version
SELECT version FROM schema_migrations
ORDER BY version DESC LIMIT 1;
-- Expected: 20251102_latest_migration

-- 2. Check data consistency
SELECT
    (SELECT COUNT(*) FROM users) as total_users,
    (SELECT COUNT(*) FROM users WHERE created_at > NOW() - INTERVAL '1 hour') as recent_users;
-- Recent users should be 0 (green is new)

-- 3. Validate indexes
SELECT schemaname, tablename, indexname
FROM pg_indexes
WHERE schemaname = 'public'
AND tablename IN ('users', 'orders', 'products');
-- All expected indexes must exist

-- 4. Check foreign key constraints
SELECT COUNT(*) FROM information_schema.table_constraints
WHERE constraint_type = 'FOREIGN KEY'
AND table_schema = 'public';
-- Should match blue environment count

# prometheus-alerts.yml - Monitor green environment
groups:
  - name: blue_green_deployment
    interval: 30s
    rules:
      # Latency spike detection
      - alert: GreenLatencyHigh
        expr: histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{env="green"}[5m])) > 1.5
        for: 2m
        annotations:
          summary: "Green environment showing high latency"

      # Error rate increase
      - alert: GreenErrorRateHigh
        expr: rate(http_requests_total{env="green", status=~"5.."}[5m]) > 0.05
        for: 1m
        annotations:
          summary: "Green error rate exceeds 5%"

      # Traffic saturation
      - alert: GreenSaturation
        expr: rate(http_requests_total{env="green"}[1m]) > 10000
        for: 5m
        annotations:
          summary: "Green environment handling high load"

# parallel_test.py - Compare blue vs green responses
import asyncio
import aiohttp
import statistics

async def compare_endpoints(endpoint, iterations=100):
    """Compare response times and results between blue and green"""
    blue_times = []
    green_times = []
    discrepancies = []

    async with aiohttp.ClientSession() as session:
        for i in range(iterations):
            # Test blue
            start = asyncio.get_event_loop().time()
            async with session.get(f"https://blue.example.com{endpoint}") as resp:
                blue_result = await resp.json()
                blue_times.append(asyncio.get_event_loop().time() - start)

            # Test green
            start = asyncio.get_event_loop().time()
            async with session.get(f"https://green.example.com{endpoint}") as resp:
                green_result = await resp.json()
                green_times.append(asyncio.get_event_loop().time() - start)

            # Check for discrepancies
            if blue_result != green_result:
                discrepancies.append({
                    'iteration': i,
                    'blue': blue_result,
                    'green': green_result
                })

    return {
        'blue_avg': statistics.mean(blue_times),
        'green_avg': statistics.mean(green_times),
        'blue_p99': statistics.quantiles(blue_times, n=100)[98],
        'green_p99': statistics.quantiles(green_times, n=100)[98],
        'discrepancies': len(discrepancies),
        'discrepancy_rate': len(discrepancies) / iterations
    }

# Run comparison
results = asyncio.run(compare_endpoints('/api/products'))
print(f"Blue avg: {results['blue_avg']:.3f}s, Green avg: {results['green_avg']:.3f}s")
print(f"Discrepancy rate: {results['discrepancy_rate']*100:.2f}%")

# nginx.conf - Shadow traffic to green environment
upstream blue_backend {
    server blue.example.com:8080;
}

upstream green_backend {
    server green.example.com:8080;
}

server {
    listen 80;

    location / {
        # Primary traffic goes to blue
        proxy_pass http://blue_backend;

        # Mirror traffic to green (async, no response used)
        mirror /mirror;
        mirror_request_body on;
    }

    location /mirror {
        internal;
        proxy_pass http://green_backend$request_uri;
        proxy_set_header X-Shadow-Request "true";
    }
}

// synthetic-monitor.js - Datadog/New Relic style
const puppeteer = require('puppeteer');

async function runSyntheticTest(environment) {
    const browser = await puppeteer.launch();
    const page = await browser.newPage();

    try {
        // Monitor page load time
        const startTime = Date.now();
        await page.goto(`https://${environment}.example.com`);
        const loadTime = Date.now() - startTime;

        // Test critical user journey
        await page.click('#search-input');
        await page.type('#search-input', 'test product');
        await page.click('#search-button');
        await page.waitForSelector('.search-results');

        // Add to cart
        await page.click('.product-card:first-child .add-to-cart');
        await page.waitForSelector('.cart-notification');

        // Verify cart
        await page.click('#cart-icon');
        const cartItems = await page.$$('.cart-item');

        return {
            success: cartItems.length > 0,
            loadTime: loadTime,
            environment: environment,
            timestamp: new Date().toISOString()
        };
    } catch (error) {
        return {
            success: false,
            error: error.message,
            environment: environment
        };
    } finally {
        await browser.close();
    }
}

// Run every 5 minutes
setInterval(async () => {
    const greenResults = await runSyntheticTest('green');
    if (!greenResults.success) {
        // Alert on failure
        console.error('❌ Green synthetic test failed:', greenResults);
    }
}, 5 * 60 * 1000);

# db_validator.py - Compare database states
import psycopg2
from datetime import datetime, timedelta

def compare_databases(blue_conn, green_conn):
    """Compare critical database metrics between environments"""
    checks = []

    # 1. Row counts must match (with tolerance for recent writes)
    tables = ['users', 'orders', 'products', 'inventory']
    for table in tables:
        blue_count = execute_query(blue_conn, f"SELECT COUNT(*) FROM {table}")
        green_count = execute_query(green_conn, f"SELECT COUNT(*) FROM {table}")

        # Allow 1% difference for active writes
        tolerance = blue_count * 0.01
        if abs(blue_count - green_count) > tolerance:
            checks.append({
                'table': table,
                'status': 'FAIL',
                'blue_count': blue_count,
                'green_count': green_count,
                'difference': abs(blue_count - green_count)
            })
        else:
            checks.append({
                'table': table,
                'status': 'PASS'
            })

    # 2. Check recent data replication
    cutoff = datetime.now() - timedelta(hours=1)
    for table in ['orders', 'user_sessions']:
        query = f"SELECT COUNT(*) FROM {table} WHERE updated_at > %s"
        blue_recent = execute_query(blue_conn, query, (cutoff,))
        green_recent = execute_query(green_conn, query, (cutoff,))

        # Green should have similar or more recent data
        if green_recent < blue_recent * 0.95:
            checks.append({
                'check': f'{table}_recent_data',
                'status': 'FAIL',
                'message': 'Green missing recent updates'
            })

    return checks

def execute_query(conn, query, params=None):
    with conn.cursor() as cur:
        cur.execute(query, params)
        return cur.fetchone()[0]

# .ebextensions/blue-green-config.yml
option_settings:
  aws:elasticbeanstalk:command:
    DeploymentPolicy: Immutable
    Timeout: "600"

  # Health check configuration
  aws:elasticbeanstalk:healthreporting:system:
    SystemType: enhanced
    EnhancedHealthAuthEnabled: true

  # Rolling deployment settings
  aws:autoscaling:updatepolicy:rollingupdate:
    RollingUpdateEnabled: true
    MaxBatchSize: 1
    MinInstancesInService: 2
    PauseTime: "PT5M"  # 5 minute pause between batches

# Write to both old and new schema
def save_user(user_data):
    # Write to old schema (blue)
    old_db.users.insert({
        'name': user_data['name'],
        'email': user_data['email']
    })

    # Write to new schema (green)
    new_db.users.insert({
        'full_name': user_data['name'],
        'email_address': user_data['email'],
        'created_at': datetime.now()
    })

def get_user(user_id):
    # Read from new schema
    user = new_db.users.find_one({'_id': user_id})

    # Async validation against old schema
    asyncio.create_task(validate_data(user_id, user))

    return user

async def validate_data(user_id, new_data):
    old_data = old_db.users.find_one({'_id': user_id})
    if not data_matches(old_data, new_data):
        log_discrepancy(user_id, old_data, new_data)

# Key Performance Indicators (KPIs)
response_time:
  p50: < 100ms
  p95: < 300ms
  p99: < 1000ms

error_rate:
  warning: > 0.5%
  critical: > 1%

throughput:
  min_rps: 1000  # Should handle normal load
  max_rps: 5000  # Should handle peak

resource_usage:
  cpu: < 70%
  memory: < 80%
  disk: < 75%

dependencies:
  database_connections: < 80% of pool
  cache_hit_rate: > 90%
  queue_depth: < 1000 messages

# traffic_controller.py - Gradual traffic shift
import time

def gradual_traffic_shift(duration_minutes=60):
    """Shift traffic from blue to green over specified duration"""
    steps = [1, 5, 10, 25, 50, 75, 100]  # Percentage to green
    step_duration = duration_minutes * 60 / len(steps)

    for percentage in steps:
        print(f"Shifting {percentage}% traffic to green...")
        update_load_balancer(green_weight=percentage, blue_weight=100-percentage)

        # Monitor for issues
        time.sleep(step_duration)
        metrics = get_green_metrics()

        if metrics['error_rate'] > 0.01 or metrics['p99_latency'] > 1.5:
            print(f"❌ Metrics degraded at {percentage}%, rolling back")
            rollback_to_blue()
            return False

        print(f"✅ {percentage}% traffic handling well")

    return True

# auto_rollback.py
from prometheus_api_client import PrometheusConnect

prom = PrometheusConnect(url="http://prometheus:9090")

def check_rollback_conditions():
    """Check if automatic rollback should trigger"""

    # 1. Error rate spike
    error_rate_query = 'rate(http_requests_total{env="green",status=~"5.."}[5m])'
    error_rate = prom.custom_query(error_rate_query)[0]['value'][1]
    if float(error_rate) > 0.05:  # 5% error rate
        return True, "Error rate exceeded 5%"

    # 2. Latency degradation
    latency_query = 'histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{env="green"}[5m]))'
    p99_latency = prom.custom_query(latency_query)[0]['value'][1]
    if float(p99_latency) > 2.0:  # 2 second p99
        return True, "P99 latency exceeded 2 seconds"

    # 3. Resource exhaustion
    cpu_query = 'avg(rate(container_cpu_usage_seconds_total{env="green"}[5m]))'
    cpu_usage = prom.custom_query(cpu_query)[0]['value'][1]
    if float(cpu_usage) > 0.9:  # 90% CPU
        return True, "CPU usage exceeded 90%"

    return False, None

# Run every 30 seconds
while True:
    should_rollback, reason = check_rollback_conditions()
    if should_rollback:
        print(f"🚨 AUTOMATIC ROLLBACK TRIGGERED: {reason}")
        execute_rollback()
        send_alert(reason)
        break
    time.sleep(30)

# BAD - Breaking change
# Migration 1: Add NOT NULL column
ALTER TABLE users ADD COLUMN phone VARCHAR(20) NOT NULL;

# GOOD - Backwards compatible
# Migration 1: Add nullable column
ALTER TABLE users ADD COLUMN phone VARCHAR(20) NULL;

# Migration 2: Backfill data
UPDATE users SET phone = 'UNKNOWN' WHERE phone IS NULL;

# Migration 3: Add constraint (deploy after traffic fully on green)
ALTER TABLE users ALTER COLUMN phone SET NOT NULL;

# BAD - In-memory sessions (lost on environment switch)
from flask import Flask, session
app = Flask(__name__)
app.secret_key = 'secret'

@app.route('/login')
def login():
    session['user_id'] = 123  # Stored locally, lost on switch

# GOOD - Redis-backed sessions (persistent across environments)
from flask import Flask
from flask_session import Session
import redis

app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = redis.from_url('redis://shared-redis:6379')
Session(app)

@app.route('/login')
def login():
    session['user_id'] = 123  # Stored in Redis, survives switch

# rate_limit_manager.py
class EnvironmentAwareRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.env = os.getenv('ENVIRONMENT')  # 'blue' or 'green'

    def check_limit(self, api_name, limit_per_hour):
        """Check rate limit with environment-specific keys"""
        key = f"ratelimit:{self.env}:{api_name}:{datetime.now().hour}"
        current = self.redis.incr(key)
        self.redis.expire(key, 3600)  # 1 hour TTL

        return current <= limit_per_hour

    def use_quota(self, api_name):
        """Use quota from shared pool if blue environment"""
        if self.env == 'blue':
            # Use production quota
            return self.check_limit(api_name, 10000)
        else:
            # Use reduced quota for green testing
            return self.check_limit(api_name, 1000)

<!-- BAD - Same URL, cache may serve old version -->
<script src="/static/app.js"></script>

<!-- GOOD - Unique URL per build, no cache issues -->
<script src="/static/app.js?v=build-20251102-1534"></script>

<!-- BETTER - Content-based hashing -->
<script src="/static/app.a8f3d9e2.js"></script>

# blue-green.tf - Complete blue-green setup
resource "aws_lb_target_group" "blue" {
  name     = "app-blue-tg"
  port     = 8080
  protocol = "HTTP"
  vpc_id   = aws_vpc.main.id

  health_check {
    path                = "/health"
    interval            = 30
    timeout             = 5
    healthy_threshold   = 2
    unhealthy_threshold = 2
  }
}

resource "aws_lb_target_group" "green" {
  name     = "app-green-tg"
  port     = 8080
  protocol = "HTTP"
  vpc_id   = aws_vpc.main.id

  health_check {
    path                = "/health"
    interval            = 30
    timeout             = 5
    healthy_threshold   = 2
    unhealthy_threshold = 2
  }
}

resource "aws_lb_listener_rule" "production" {
  listener_arn = aws_lb_listener.main.arn
  priority     = 100

  action {
    type             = "forward"
    target_group_arn = var.active_environment == "blue" ?
                       aws_lb_target_group.blue.arn :
                       aws_lb_target_group.green.arn
  }

  condition {
    path_pattern {
      values = ["/*"]
    }
  }
}

# appspec.yml - CodeDeploy configuration
version: 0.0
Resources:
  - TargetService:
      Type: AWS::ECS::Service
      Properties:
        TaskDefinition: "arn:aws:ecs:us-east-1:123456:task-definition/app:2"
        LoadBalancerInfo:
          ContainerName: "app"
          ContainerPort: 8080
        PlatformVersion: "LATEST"

Hooks:
  - BeforeInstall: "scripts/pre-deployment-tests.sh"
  - AfterInstall: "scripts/smoke-tests.sh"
  - AfterAllowTestTraffic: "scripts/integration-tests.sh"
  - BeforeAllowTraffic: "scripts/validation.sh"
  - AfterAllowTraffic: "scripts/post-deployment-monitoring.sh"

# flagger-canary.yml
apiVersion: flagger.app/v1beta1
kind: Canary
metadata:
  name: app
spec:
  targetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: app
  service:
    port: 8080
  analysis:
    interval: 1m
    threshold: 5
    maxWeight: 50
    stepWeight: 10
    metrics:
    - name: request-success-rate
      thresholdRange:
        min: 99
      interval: 1m
    - name: request-duration
      thresholdRange:
        max: 500
      interval: 1m
    webhooks:
    - name: load-test
      url: http://flagger-loadtester/
      timeout: 5s
      metadata:
        cmd: "hey -z 1m -q 10 -c 2 http://app:8080/"