Estrategias de Despliegue para Equipos QA: Blue-Green, Canary y Rollouts Progresivos

# kubernetes/blue-green-deployment.yaml
apiVersion: v1
kind: Service
metadata:
  name: app-service
spec:
  selector:
    app: myapp
    version: blue  # Cambiar entre blue/green
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8080
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: app-blue
  labels:
    version: blue
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
      version: blue
  template:
    metadata:
      labels:
        app: myapp
        version: blue
    spec:
      containers:
      - name: app
        image: myapp:v1.2.0
        ports:
        - containerPort: 8080
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: app-green
  labels:
    version: green
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
      version: green
  template:
    metadata:
      labels:
        app: myapp
        version: green
    spec:
      containers:
      - name: app
        image: myapp:v1.3.0
        ports:
        - containerPort: 8080
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5

# tests/blue_green_deployment_test.py
import pytest
import requests
import time
from kubernetes import client, config
from typing import Dict, List

class ProbadorDespliegueBlueGreen:
    def __init__(self, namespace: str = "production"):
        config.load_kube_config()
        self.apps_v1 = client.AppsV1Api()
        self.core_v1 = client.CoreV1Api()
        self.namespace = namespace

    def test_salud_entorno_green(self):
        """Probar entorno green antes de cambiar tráfico"""
        # Obtener despliegue green
        green_deployment = self.apps_v1.read_namespaced_deployment(
            name="app-green",
            namespace=self.namespace
        )

        # Verificar que todas las réplicas están listas
        assert green_deployment.status.ready_replicas == green_deployment.spec.replicas
        assert green_deployment.status.available_replicas == green_deployment.spec.replicas

        # Obtener pods green
        green_pods = self.core_v1.list_namespaced_pod(
            namespace=self.namespace,
            label_selector="app=myapp,version=green"
        )

        for pod in green_pods.items:
            assert pod.status.phase == "Running"

            # Verificar salud del contenedor
            for container_status in pod.status.container_statuses:
                assert container_status.ready == True
                assert container_status.state.running is not None

    def test_conectividad_entorno_green(self):
        """Probar conectividad interna a pods green"""
        green_pods = self.core_v1.list_namespaced_pod(
            namespace=self.namespace,
            label_selector="app=myapp,version=green"
        )

        for pod in green_pods.items:
            pod_ip = pod.status.pod_ip

            # Probar endpoint de salud
            response = requests.get(f"http://{pod_ip}:8080/health", timeout=5)
            assert response.status_code == 200

            # Probar endpoint de readiness
            response = requests.get(f"http://{pod_ip}:8080/ready", timeout=5)
            assert response.status_code == 200

    def test_pruebas_humo_en_green(self):
        """Ejecutar pruebas de humo contra entorno green"""
        # Obtener endpoint del servicio green (servicio temporal de prueba)
        green_service_url = self._obtener_url_servicio_green()

        # Pruebas de humo de ruta crítica
        pruebas_humo = [
            {"endpoint": "/api/users", "method": "GET", "expected_status": 200},
            {"endpoint": "/api/health", "method": "GET", "expected_status": 200},
            {"endpoint": "/api/version", "method": "GET", "expected_status": 200},
        ]

        for test in pruebas_humo:
            response = requests.request(
                method=test["method"],
                url=f"{green_service_url}{test['endpoint']}",
                timeout=10
            )
            assert response.status_code == test["expected_status"]

    def realizar_cambio_trafico(self):
        """Cambiar tráfico de blue a green"""
        service = self.core_v1.read_namespaced_service(
            name="app-service",
            namespace=self.namespace
        )

        # Actualizar selector del servicio para apuntar a green
        service.spec.selector = {
            "app": "myapp",
            "version": "green"
        }

        self.core_v1.patch_namespaced_service(
            name="app-service",
            namespace=self.namespace,
            body=service
        )

        # Esperar a que el servicio se propague
        time.sleep(5)

    def test_validacion_post_cambio(self):
        """Validar servicio después del cambio de tráfico"""
        # Obtener servicio actual
        service = self.core_v1.read_namespaced_service(
            name="app-service",
            namespace=self.namespace
        )

        # Verificar que el servicio apunta a green
        assert service.spec.selector["version"] == "green"

        # Probar endpoint del servicio
        service_url = self._obtener_url_servicio()

        # Ejecutar pruebas de validación
        for _ in range(10):  # Probar múltiples veces para asegurar consistencia
            response = requests.get(f"{service_url}/api/version", timeout=5)
            assert response.status_code == 200

            version_data = response.json()
            assert "v1.3.0" in version_data["version"]  # Nueva versión

            time.sleep(1)

    def monitorear_tasas_error_post_despliegue(self, duracion_minutos: int = 5):
        """Monitorear tasas de error después del despliegue"""
        tiempo_inicio = time.time()
        conteo_errores = 0
        total_peticiones = 0

        service_url = self._obtener_url_servicio()

        while time.time() - tiempo_inicio < duracion_minutos * 60:
            try:
                response = requests.get(f"{service_url}/api/health", timeout=5)
                total_peticiones += 1

                if response.status_code >= 500:
                    conteo_errores += 1

            except requests.exceptions.RequestException:
                conteo_errores += 1
                total_peticiones += 1

            time.sleep(1)

        tasa_error = (conteo_errores / total_peticiones) * 100 if total_peticiones > 0 else 0

        # Aseverar que la tasa de error está por debajo del umbral
        assert tasa_error < 1.0, f"Tasa de error {tasa_error}% excede el umbral"

        return {
            "total_peticiones": total_peticiones,
            "errores": conteo_errores,
            "tasa_error": tasa_error
        }

    def rollback_a_blue(self):
        """Revertir tráfico al entorno blue"""
        service = self.core_v1.read_namespaced_service(
            name="app-service",
            namespace=self.namespace
        )

        service.spec.selector = {
            "app": "myapp",
            "version": "blue"
        }

        self.core_v1.patch_namespaced_service(
            name="app-service",
            namespace=self.namespace,
            body=service
        )

@pytest.fixture
def probador_despliegue():
    return ProbadorDespliegueBlueGreen(namespace="production")

def test_despliegue_blue_green_completo(probador_despliegue):
    """Prueba end-to-end de despliegue blue-green"""
    # Fase 1: Validación pre-despliegue
    probador_despliegue.test_salud_entorno_green()
    probador_despliegue.test_conectividad_entorno_green()
    probador_despliegue.test_pruebas_humo_en_green()

    # Fase 2: Cambio de tráfico
    probador_despliegue.realizar_cambio_trafico()

    # Fase 3: Validación post-despliegue
    probador_despliegue.test_validacion_post_cambio()

    # Fase 4: Monitorear despliegue
    metricas = probador_despliegue.monitorear_tasas_error_post_despliegue(duracion_minutos=5)

    print(f"Despliegue exitoso. Métricas: {metricas}")

# istio/canary-virtualservice.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: app-canary
spec:
  hosts:
  - app.example.com
  http:
  - match:
    - headers:
        x-canary:
          exact: "true"
    route:
    - destination:
        host: app-service
        subset: canary
      weight: 100
  - route:
    - destination:
        host: app-service
        subset: stable
      weight: 90
    - destination:
        host: app-service
        subset: canary
      weight: 10
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: app-destination
spec:
  host: app-service
  subsets:
  - name: stable
    labels:
      version: v1.0.0
  - name: canary
    labels:
      version: v1.1.0
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        http2MaxRequests: 100

# tests/canary_analysis.py
import time
import requests
from dataclasses import dataclass
from typing import List, Dict
import prometheus_api_client

@dataclass
class MetricasCanary:
    tasa_error: float
    latencia_p50: float
    latencia_p95: float
    latencia_p99: float
    tasa_exito: float
    conteo_peticiones: int

class AnalizadorCanary:
    def __init__(self, url_prometheus: str, nombre_servicio: str):
        self.prom = prometheus_api_client.PrometheusConnect(url=url_prometheus)
        self.nombre_servicio = nombre_servicio

    def obtener_metricas(self, version: str, duracion_minutos: int = 5) -> MetricasCanary:
        """Obtener métricas para una versión específica"""

        # Consulta de tasa de error
        consulta_tasa_error = f'''
        sum(rate(http_requests_total{{
            service="{self.nombre_servicio}",
            version="{version}",
            status=~"5.."
        }}[{duracion_minutos}m])) /
        sum(rate(http_requests_total{{
            service="{self.nombre_servicio}",
            version="{version}"
        }}[{duracion_minutos}m])) * 100
        '''
        tasa_error = self._consultar_metrica(consulta_tasa_error)

        # Consultas de latencia
        consulta_latencia_p50 = f'''
        histogram_quantile(0.50, sum(rate(http_request_duration_seconds_bucket{{
            service="{self.nombre_servicio}",
            version="{version}"
        }}[{duracion_minutos}m])) by (le))
        '''
        latencia_p50 = self._consultar_metrica(consulta_latencia_p50)

        consulta_latencia_p95 = f'''
        histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket{{
            service="{self.nombre_servicio}",
            version="{version}"
        }}[{duracion_minutos}m])) by (le))
        '''
        latencia_p95 = self._consultar_metrica(consulta_latencia_p95)

        consulta_latencia_p99 = f'''
        histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{{
            service="{self.nombre_servicio}",
            version="{version}"
        }}[{duracion_minutos}m])) by (le))
        '''
        latencia_p99 = self._consultar_metrica(consulta_latencia_p99)

        # Tasa de éxito
        consulta_tasa_exito = f'''
        sum(rate(http_requests_total{{
            service="{self.nombre_servicio}",
            version="{version}",
            status=~"2..|3.."
        }}[{duracion_minutos}m])) /
        sum(rate(http_requests_total{{
            service="{self.nombre_servicio}",
            version="{version}"
        }}[{duracion_minutos}m])) * 100
        '''
        tasa_exito = self._consultar_metrica(consulta_tasa_exito)

        # Conteo de peticiones
        consulta_conteo_peticiones = f'''
        sum(increase(http_requests_total{{
            service="{self.nombre_servicio}",
            version="{version}"
        }}[{duracion_minutos}m]))
        '''
        conteo_peticiones = self._consultar_metrica(consulta_conteo_peticiones)

        return MetricasCanary(
            tasa_error=tasa_error or 0.0,
            latencia_p50=latencia_p50 or 0.0,
            latencia_p95=latencia_p95 or 0.0,
            latencia_p99=latencia_p99 or 0.0,
            tasa_exito=tasa_exito or 0.0,
            conteo_peticiones=int(conteo_peticiones or 0)
        )

    def _consultar_metrica(self, consulta: str) -> float:
        """Ejecutar consulta Prometheus y retornar resultado escalar"""
        resultado = self.prom.custom_query(query=consulta)
        if resultado and len(resultado) > 0:
            return float(resultado[0]['value'][1])
        return 0.0

    def comparar_versiones(self, version_estable: str, version_canary: str) -> Dict:
        """Comparar canary contra línea base estable"""
        metricas_estable = self.obtener_metricas(version_estable)
        metricas_canary = self.obtener_metricas(version_canary)

        # Calcular deltas
        delta_tasa_error = metricas_canary.tasa_error - metricas_estable.tasa_error
        delta_latencia_p95 = metricas_canary.latencia_p95 - metricas_estable.latencia_p95
        delta_tasa_exito = metricas_canary.tasa_exito - metricas_estable.tasa_exito

        # Umbrales de decisión
        umbrales = {
            "max_incremento_tasa_error": 1.0,  # 1% de incremento
            "max_incremento_latencia_p95": 0.1,  # 100ms de incremento
            "min_tasa_exito": 99.0,  # 99% tasa de éxito
            "min_conteo_peticiones": 100  # Peticiones mínimas para significancia estadística
        }

        # Analizar resultados
        paso = True
        fallos = []

        if metricas_canary.conteo_peticiones < umbrales["min_conteo_peticiones"]:
            paso = False
            fallos.append(f"Peticiones insuficientes: {metricas_canary.conteo_peticiones}")

        if delta_tasa_error > umbrales["max_incremento_tasa_error"]:
            paso = False
            fallos.append(f"Tasa de error incrementó en {delta_tasa_error:.2f}%")

        if delta_latencia_p95 > umbrales["max_incremento_latencia_p95"]:
            paso = False
            fallos.append(f"Latencia P95 incrementó en {delta_latencia_p95*1000:.0f}ms")

        if metricas_canary.tasa_exito < umbrales["min_tasa_exito"]:
            paso = False
            fallos.append(f"Tasa de éxito {metricas_canary.tasa_exito:.2f}% por debajo del umbral")

        return {
            "paso": paso,
            "fallos": fallos,
            "metricas_estable": metricas_estable,
            "metricas_canary": metricas_canary,
            "deltas": {
                "tasa_error": delta_tasa_error,
                "latencia_p95": delta_latencia_p95,
                "tasa_exito": delta_tasa_exito
            }
        }

class DespliegueCanaryProgresivo:
    def __init__(self, analizador: AnalizadorCanary, namespace: str = "production"):
        self.analizador = analizador
        self.namespace = namespace
        self.etapas_trafico = [10, 25, 50, 75, 100]  # Etapas de porcentaje de tráfico

    def ejecutar_rollout_progresivo(self, version_estable: str, version_canary: str):
        """Ejecutar rollout canary progresivo con análisis automatizado"""

        for etapa in self.etapas_trafico:
            print(f"\n=== Etapa: {etapa}% de tráfico a canary ===")

            # Actualizar división de tráfico
            self._actualizar_division_trafico(peso_canary=etapa)

            # Esperar a que las métricas se estabilicen
            tiempo_estabilizacion = 5  # minutos
            print(f"Esperando {tiempo_estabilizacion} minutos para que las métricas se estabilicen...")
            time.sleep(tiempo_estabilizacion * 60)

            # Analizar rendimiento canary
            analisis = self.analizador.comparar_versiones(version_estable, version_canary)

            print(f"Resultados del Análisis:")
            print(f"  Tasa Error Canary: {analisis['metricas_canary'].tasa_error:.2f}%")
            print(f"  Latencia P95 Canary: {analisis['metricas_canary'].latencia_p95*1000:.0f}ms")
            print(f"  Tasa Éxito Canary: {analisis['metricas_canary'].tasa_exito:.2f}%")
            print(f"  Delta Tasa Error: {analisis['deltas']['tasa_error']:.2f}%")

            if not analisis["paso"]:
                print(f"\n❌ Canary falló en {etapa}% de tráfico!")
                print("Fallos:")
                for fallo in analisis["fallos"]:
                    print(f"  - {fallo}")

                print("\nIniciando rollback...")
                self._rollback()
                return False

            print(f"✓ Canary pasó en {etapa}% de tráfico")

        print("\n✓ Despliegue canary exitoso!")
        return True

    def _actualizar_division_trafico(self, peso_canary: int):
        """Actualizar VirtualService de Istio con nueva división de tráfico"""
        # La implementación usaría la API de Kubernetes para actualizar VirtualService
        pass

    def _rollback(self):
        """Revertir a 100% de tráfico estable"""
        self._actualizar_division_trafico(peso_canary=0)

# tests/feature_flag_testing.py
import pytest
from launchdarkly import Context
from typing import Dict, List

class ProbadorFeatureFlags:
    def __init__(self, cliente_ld, entorno: str):
        self.cliente = cliente_ld
        self.entorno = entorno

    def test_porcentajes_rollout_feature_flag(self, clave_flag: str):
        """Probar que el feature flag respeta los porcentajes de rollout"""
        tamaño_muestra = 1000
        conteo_habilitado = 0

        for i in range(tamaño_muestra):
            contexto = Context.builder(f"user-{i}").build()
            if self.cliente.variation(clave_flag, contexto, default=False):
                conteo_habilitado += 1

        porcentaje_real = (conteo_habilitado / tamaño_muestra) * 100
        porcentaje_esperado = self._obtener_porcentaje_rollout_flag(clave_flag)

        # Permitir variación del 5% debido al muestreo
        assert abs(porcentaje_real - porcentaje_esperado) < 5.0

    def test_reglas_targeting_feature_flag(self, clave_flag: str):
        """Probar que las reglas de targeting se aplican correctamente"""
        casos_prueba = [
            {
                "contexto": Context.builder("beta-user-1")
                    .set("beta_tester", True)
                    .build(),
                "esperado": True
            },
            {
                "contexto": Context.builder("regular-user-1")
                    .set("beta_tester", False)
                    .build(),
                "esperado": False
            },
            {
                "contexto": Context.builder("premium-user-1")
                    .set("plan", "premium")
                    .build(),
                "esperado": True
            }
        ]

        for caso_prueba in casos_prueba:
            resultado = self.cliente.variation(
                clave_flag,
                caso_prueba["contexto"],
                default=False
            )
            assert resultado == caso_prueba["esperado"]

    def test_valores_predeterminados_feature_flags(self):
        """Probar que los feature flags tienen valores predeterminados apropiados"""
        flags_criticos = [
            "procesamiento-pagos",
            "autenticacion-usuario",
            "encriptacion-datos"
        ]

        contexto_anonimo = Context.builder("anonimo").build()

        for clave_flag in flags_criticos:
            # Los flags críticos deben tener valores predeterminados seguros/conservadores
            resultado = self.cliente.variation(clave_flag, contexto_anonimo, default=True)
            assert resultado == True  # Las características críticas deben estar habilitadas por defecto

    def test_progresion_rollout_gradual(self, clave_flag: str):
        """Probar rollout progresivo a lo largo del tiempo"""
        calendario_rollout = [
            {"porcentaje": 10, "duracion_horas": 2},
            {"porcentaje": 25, "duracion_horas": 4},
            {"porcentaje": 50, "duracion_horas": 8},
            {"porcentaje": 100, "duracion_horas": 24}
        ]

        for etapa in calendario_rollout:
            # Actualizar flag al porcentaje de la etapa
            self._actualizar_porcentaje_flag(clave_flag, etapa["porcentaje"])

            # Verificar que el porcentaje es correcto
            self.test_porcentajes_rollout_feature_flag(clave_flag)

            # Monitorear métricas durante esta etapa
            metricas = self._monitorear_metricas(
                duracion_horas=etapa["duracion_horas"]
            )

            # Validar que las métricas cumplen los umbrales
            assert metricas["tasa_error"] < 1.0
            assert metricas["latencia_p95"] < 500  # ms

            print(f"Etapa {etapa['porcentaje']}% exitosa")