Testing de Despliegue Blue-Green: Guía Completa para Equipos DevOps

#!/bin/bash
# smoke-test.sh - Health check básico para entorno green

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

# Verificar que la aplicación responde
if ! curl -f -s "${GREEN_URL}/health" > /dev/null; then
    echo "❌ Endpoint de salud no responde"
    exit 1
fi

# Verificar conectividad de base de datos
if ! curl -f -s "${GREEN_URL}/api/db-check" | grep -q "OK"; then
    echo "❌ Conexión a base de datos falló"
    exit 1
fi

# Verificar dependencias críticas
for service in redis kafka elasticsearch; do
    if ! curl -f -s "${GREEN_URL}/api/check/${service}" | grep -q "healthy"; then
        echo "❌ Verificación de dependencia ${service} falló"
        exit 1
    fi
done

echo "✅ Todos los smoke tests pasaron"

# test_green_integration.py
import pytest
import requests

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

def test_user_registration_flow():
    """Prueba flujo completo de registro de usuario"""
    # Crear usuario
    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"]

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

    # Completar verificación
    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():
    """Verificar integración con pasarela de pagos"""
    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
-- Ejecutar estas verificaciones antes del cambio de tráfico

-- 1. Verificar versión de schema
SELECT version FROM schema_migrations
ORDER BY version DESC LIMIT 1;
-- Esperado: 20251102_latest_migration

-- 2. Verificar consistencia de datos
SELECT
    (SELECT COUNT(*) FROM users) as total_users,
    (SELECT COUNT(*) FROM users WHERE created_at > NOW() - INTERVAL '1 hour') as recent_users;
-- recent_users debería ser 0 (green es nuevo)

-- 3. Validar índices
SELECT schemaname, tablename, indexname
FROM pg_indexes
WHERE schemaname = 'public'
AND tablename IN ('users', 'orders', 'products');
-- Todos los índices esperados deben existir

-- 4. Verificar restricciones de foreign key
SELECT COUNT(*) FROM information_schema.table_constraints
WHERE constraint_type = 'FOREIGN KEY'
AND table_schema = 'public';
-- Debe coincidir con el conteo del entorno blue

# prometheus-alerts.yml - Monitorear entorno green
groups:

  - name: blue_green_deployment
    interval: 30s
    rules:
      # Detección de picos de latencia
      - alert: GreenLatencyHigh
        expr: histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{env="green"}[5m])) > 1.5
        for: 2m
        annotations:
          summary: "Entorno green mostrando alta latencia"

      # Incremento de tasa de errores
      - alert: GreenErrorRateHigh
        expr: rate(http_requests_total{env="green", status=~"5.."}[5m]) > 0.05
        for: 1m
        annotations:
          summary: "Tasa de errores de green excede 5%"

      # Saturación de tráfico
      - alert: GreenSaturation
        expr: rate(http_requests_total{env="green"}[1m]) > 10000
        for: 5m
        annotations:
          summary: "Entorno green manejando carga alta"

# parallel_test.py - Comparar respuestas de blue vs green
import asyncio
import aiohttp
import statistics

async def compare_endpoints(endpoint, iterations=100):
    """Comparar tiempos de respuesta y resultados entre blue y green"""
    blue_times = []
    green_times = []
    discrepancies = []

    async with aiohttp.ClientSession() as session:
        for i in range(iterations):
            # Probar 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)

            # Probar 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)

            # Verificar discrepancias
            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
    }

# Ejecutar comparación
results = asyncio.run(compare_endpoints('/api/products'))
print(f"Blue promedio: {results['blue_avg']:.3f}s, Green promedio: {results['green_avg']:.3f}s")
print(f"Tasa de discrepancia: {results['discrepancy_rate']*100:.2f}%")

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

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

server {
    listen 80;

    location / {
        # Tráfico primario va a blue
        proxy_pass http://blue_backend;

        # Duplicar tráfico a green (asíncrono, respuesta no se usa)
        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 - Estilo Datadog/New Relic
const puppeteer = require('puppeteer');

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

    try {
        // Monitorear tiempo de carga de página
        const startTime = Date.now();
        await page.goto(`https://${environment}.example.com`);
        const loadTime = Date.now() - startTime;

        // Probar journey crítico de usuario
        await page.click('#search-input');
        await page.type('#search-input', 'test product');
        await page.click('#search-button');
        await page.waitForSelector('.search-results');

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

        // Verificar carrito
        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();
    }
}

// Ejecutar cada 5 minutos
setInterval(async () => {
    const greenResults = await runSyntheticTest('green');
    if (!greenResults.success) {
        // Alertar en caso de fallo
        console.error('❌ Test sintético de green falló:', greenResults);
    }
}, 5 * 60 * 1000);

# db_validator.py - Comparar estados de bases de datos
import psycopg2
from datetime import datetime, timedelta

def compare_databases(blue_conn, green_conn):
    """Comparar métricas críticas de base de datos entre entornos"""
    checks = []

    # 1. Conteos de filas deben coincidir (con tolerancia para escrituras recientes)
    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}")

        # Permitir 1% de diferencia para escrituras activas
        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. Verificar replicación de datos recientes
    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 debería tener datos similares o más recientes
        if green_recent < blue_recent * 0.95:
            checks.append({
                'check': f'{table}_recent_data',
                'status': 'FAIL',
                'message': 'Green falta actualizaciones recientes'
            })

    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"

  # Configuración de health check
  aws:elasticbeanstalk:healthreporting:system:
    SystemType: enhanced
    EnhancedHealthAuthEnabled: true

  # Configuración de despliegue rolling
  aws:autoscaling:updatepolicy:rollingupdate:
    RollingUpdateEnabled: true
    MaxBatchSize: 1
    MinInstancesInService: 2
    PauseTime: "PT5M"  # Pausa de 5 minutos entre lotes

# Escribir a ambos schemas antiguo y nuevo
def save_user(user_data):
    # Escribir a schema antiguo (blue)
    old_db.users.insert({
        'name': user_data['name'],
        'email': user_data['email']
    })

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

def get_user(user_id):
    # Leer del schema nuevo
    user = new_db.users.find_one({'_id': user_id})

    # Validación asíncrona contra schema antiguo
    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)

# Indicadores Clave de Rendimiento (KPIs)
response_time:
  p50: < 100ms
  p95: < 300ms
  p99: < 1000ms

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

throughput:
  min_rps: 1000  # Debe manejar carga normal
  max_rps: 5000  # Debe manejar pico

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

dependencies:
  database_connections: < 80% del pool
  cache_hit_rate: > 90%
  queue_depth: < 1000 mensajes

# traffic_controller.py - Cambio gradual de tráfico
import time

def gradual_traffic_shift(duration_minutes=60):
    """Cambiar tráfico de blue a green durante duración especificada"""
    steps = [1, 5, 10, 25, 50, 75, 100]  # Porcentaje a green
    step_duration = duration_minutes * 60 / len(steps)

    for percentage in steps:
        print(f"Cambiando {percentage}% tráfico a green...")
        update_load_balancer(green_weight=percentage, blue_weight=100-percentage)

        # Monitorear problemas
        time.sleep(step_duration)
        metrics = get_green_metrics()

        if metrics['error_rate'] > 0.01 or metrics['p99_latency'] > 1.5:
            print(f"❌ Métricas degradadas al {percentage}%, haciendo rollback")
            rollback_to_blue()
            return False

        print(f"✅ {percentage}% del tráfico manejándose bien")

    return True

# auto_rollback.py
from prometheus_api_client import PrometheusConnect

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

def check_rollback_conditions():
    """Verificar si el rollback automático debería activarse"""

    # 1. Pico en tasa de errores
    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% de tasa de errores
        return True, "Tasa de errores excedió 5%"

    # 2. Degradación de latencia
    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 segundos p99
        return True, "Latencia P99 excedió 2 segundos"

    # 3. Agotamiento de recursos
    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, "Uso de CPU excedió 90%"

    return False, None

# Ejecutar cada 30 segundos
while True:
    should_rollback, reason = check_rollback_conditions()
    if should_rollback:
        print(f"🚨 ROLLBACK AUTOMÁTICO ACTIVADO: {reason}")
        execute_rollback()
        send_alert(reason)
        break
    time.sleep(30)

# MAL - Cambio que rompe compatibilidad
# Migración 1: Agregar columna NOT NULL
ALTER TABLE users ADD COLUMN phone VARCHAR(20) NOT NULL;

# BIEN - Backwards compatible
# Migración 1: Agregar columna nullable
ALTER TABLE users ADD COLUMN phone VARCHAR(20) NULL;

# Migración 2: Rellenar datos
UPDATE users SET phone = 'UNKNOWN' WHERE phone IS NULL;

# Migración 3: Agregar restricción (desplegar después de tráfico totalmente en green)
ALTER TABLE users ALTER COLUMN phone SET NOT NULL;

# MAL - Sesiones en memoria (perdidas en cambio de entorno)
from flask import Flask, session
app = Flask(__name__)
app.secret_key = 'secret'

@app.route('/login')
def login():
    session['user_id'] = 123  # Almacenado localmente, perdido en cambio

# BIEN - Sesiones respaldadas por Redis (persistentes entre entornos)
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  # Almacenado en Redis, sobrevive al cambio

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

    def check_limit(self, api_name, limit_per_hour):
        """Verificar límite de tasa con claves específicas por entorno"""
        key = f"ratelimit:{self.env}:{api_name}:{datetime.now().hour}"
        current = self.redis.incr(key)
        self.redis.expire(key, 3600)  # TTL de 1 hora

        return current <= limit_per_hour

    def use_quota(self, api_name):
        """Usar cuota del pool compartido si es entorno blue"""
        if self.env == 'blue':
            # Usar cuota de producción
            return self.check_limit(api_name, 10000)
        else:
            # Usar cuota reducida para testing de green
            return self.check_limit(api_name, 1000)

<!-- MAL - Misma URL, caché puede servir versión antigua -->
<script src="/static/app.js"></script>

<!-- BIEN - URL única por build, sin problemas de caché -->
<script src="/static/app.js?v=build-20251102-1534"></script>

<!-- MEJOR - Hashing basado en contenido -->
<script src="/static/app.a8f3d9e2.js"></script>

# blue-green.tf - Configuración completa blue-green
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 - Configuración de CodeDeploy
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/"