1.11 — Autorremediação de Incidentes: Circuit Breaker e Rollback Progressivo (Auto-remediation)

ACOES_REMEDIACAO = {

    'oom_pod': AcaoRemediacao(

        tipo='aumentar_memory_limit',

        descricao='Aumentar memory limit do deployment em 50%',

        risco='baixo',

        reversivel=True,

        comando_dry_run='kubectl patch deployment {servico} ... --dry-run=client',

        comando_live='kubectl patch deployment {servico} ...',

    ),

    'cpu_alta': AcaoRemediacao(

        tipo='scale_out',

        descricao='Scale out: adicionar 2 replicas',

        risco='baixo',

        reversivel=True,

        ...

    ),

}

def circuit_breaker_remediacao(servico: str, historico_falhas: dict) -> bool:

    """Retorna True se o circuit breaker deve ser ativado (parar remediações)."""

    from datetime import datetime, timedelta

    

    falhas_do_servico = historico_falhas.get(servico, [])

    

    # Critério 1: mais de 3 falhas na última hora

    uma_hora_atras = datetime.utcnow() - timedelta(hours=1)

    falhas_recentes = [

        f for f in falhas_do_servico

        if datetime.fromisoformat(f['timestamp']) > uma_hora_atras

    ]

    if len(falhas_recentes) >= 3:

        print(f'[CIRCUIT BREAKER] ABERTO: {servico} teve {len(falhas_recentes)} falhas na última hora')

        return True

    

    # Critério 2: remediação revertida 2+ vezes

    reversoes = [f for f in falhas_do_servico if f.get('tipo') == 'revertido']

    if len(reversoes) >= 2:

        print(f'[CIRCUIT BREAKER] ABERTO: {servico} teve {len(reversoes)} reversões')

        return True

    

    return False  # circuit breaker fechado — pode continuar remediando

historico_falhas = {}  # dict: servico → lista de {timestamp, tipo, motivo}



# Antes de cada remediação

if circuit_breaker_remediacao(alerta.servico, historico_falhas):

    return escalar_para_humano(motivo='Circuit breaker ativado após múltiplas falhas')



# Após remediação falhar

historico_falhas.setdefault(alerta.servico, []).append({

    'timestamp': datetime.utcnow().isoformat(),

    'tipo': 'falha',

    'motivo': 'verificar_saude retornou False',

})

def canary_rollback(servico: str, versao_anterior: str, taxa_erro_atual: float) -> str:

    etapas = []

    

    # Condição de gatilho: taxa de erro > 5% após deploy

    if taxa_erro_atual < 5.0:

        return f'Taxa de erro {taxa_erro_atual}% está abaixo do threshold. Canary rollback não necessário.'

    

    # Etapa 1: Redirecionar 50% do tráfego para versão anterior

    etapas.append(f'[{datetime.utcnow().strftime("%H:%M:%S")}] Redirecionando 50% do tráfego para {servico}:{versao_anterior}')

    etapas.append(f'  [MOCK] kubectl patch service {servico} -p \'{"canary-weight": "50"}\'')

    

    # Etapa 2: Aguardar 2 minutos (simulado)

    etapas.append(f'  Aguardando 2 minutos para estabilização...')

    # time.sleep(120)  — comentado para não bloquear no lab

    

    # Etapa 3: Verificar se erro caiu

    import random

    erro_apos_canary = taxa_erro_atual * random.uniform(0.3, 0.7)  # simula melhora

    

    if erro_apos_canary < taxa_erro_atual * 0.5:

        # Erro caiu mais de 50% → rollback total

        etapas.append(f'  Taxa de erro caiu: {taxa_erro_atual:.1f}% → {erro_apos_canary:.1f}%')

        etapas.append(f'[{datetime.utcnow().strftime("%H:%M:%S")}] Redirecionando 100% para {versao_anterior}')

        etapas.append(f'  [MOCK] kubectl set image deployment/{servico} {servico}={servico}:{versao_anterior}')

        etapas.append(f'  ✅ Rollback completo para {versao_anterior}')

    else:

        # Erro não caiu → investigar versão anterior também

        etapas.append(f'  Taxa de erro após canary: {erro_apos_canary:.1f}% (esperado < {taxa_erro_atual * 0.5:.1f}%)')

        etapas.append(f'  ⚠️ Versão anterior pode ter o mesmo problema. Escalar para SRE.')

    

    return '\n'.join(etapas)

def executar_acao(acao: AcaoRemediacao, servico: str, modo: ModoExecucao) -> dict:

    if modo == ModoExecucao.DRY_RUN:

        cmd = acao.comando_dry_run.format(servico=servico)

        print(f'  [DRY-RUN] {cmd}')

        return {'status': 'simulado', 'comando': cmd, 'resultado': f'[DRY-RUN] simulado'}

    else:

        cmd = acao.comando_live.format(servico=servico)

        print(f'  [LIVE] {cmd}')

        return {'status': 'executado', 'resultado': 'sucesso (mock)'}

1. executar_remediacao(tipo='oom_pod', modo='dry_run')  → simular

2. Verificar que dry-run está correto

3. executar_remediacao(tipo='oom_pod', modo='live')     → aplicar

alerta recebido

    ↓

circuit_breaker_remediacao → aberto? → escalar_para_humano

    ↓ fechado

diagnosticar_alerta → mapear tipo → acao_recomendada

    ↓

executar_acao(dry_run) → verificar resultado esperado

    ↓

executar_acao(live)

    ↓

verificar_saude → resolvido?

    ↓ sim                    ↓ não

registrar_audit         registrar_falha → circuit_breaker?

end                             ↓ não → tentar outra ação

                                ↓ sim → escalar_para_humano

# Simular histórico de falhas

historico_falhas = {

    'checkout-service': [

        {'timestamp': (datetime.utcnow() - timedelta(minutes=30)).isoformat(), 'tipo': 'falha', 'motivo': 'memory ainda alta'},

        {'timestamp': (datetime.utcnow() - timedelta(minutes=20)).isoformat(), 'tipo': 'falha', 'motivo': 'memory ainda alta'},

        {'timestamp': (datetime.utcnow() - timedelta(minutes=10)).isoformat(), 'tipo': 'revertido', 'motivo': 'timeout'},

    ]

}



# Circuit breaker decide

bloqueado = circuit_breaker_remediacao('checkout-service', historico_falhas)

# → [CIRCUIT BREAKER] ABERTO: checkout-service teve 3 falhas na última hora

# → bloqueado = True



# Agente escala ao invés de remediar

if bloqueado:

    escalar_para_humano('Circuit breaker ativo após 3 falhas/1h', urgencia='P1')

resultado = canary_rollback('api-gateway', 'v2.0.5', taxa_erro_atual=12.5)

print(resultado)

# [14:35:01] Redirecionando 50% do tráfego para api-gateway:v2.0.5

#   [MOCK] kubectl patch service api-gateway -p '{"canary-weight": "50"}'

#   Aguardando 2 minutos para estabilização...

#   Taxa de erro caiu: 12.5% → 4.2%

# [14:35:01] Redirecionando 100% para v2.0.5

#   [MOCK] kubectl set image deployment/api-gateway api-gateway=api-gateway:v2.0.5

#   ✅ Rollback completo para v2.0.5

# Verificar falhas NA ÚLTIMA HORA — não falhas históricas totais

uma_hora_atras = datetime.utcnow() - timedelta(hours=1)

falhas_recentes = [f for f in falhas if datetime.fromisoformat(f['timestamp']) > uma_hora_atras]

@dataclass

class AcaoRemediacao:

    comando_dry_run: str  # sempre presente

    comando_live: str     # sempre presente

    # ambos são obrigatórios — não há ação sem dry-run

escalar_para_humano(

    motivo='3 remediações de OOMKilled falharam — memory continua aumentando',

    acoes_tentadas=['aumentar_memory_limit x3', 'escalar_hpa'],

    urgencia='P1',

)

# RUIM: conta falhas de todos os tempos

if len(historico_falhas[servico]) >= 3:

    return True  # pode bloquear por uma falha de semanas atrás



# BOM: só conta falhas recentes

falhas_recentes = [f for f in historico if datetime.fromisoformat(f['timestamp']) > uma_hora_atras]

# Sem timeout no canary, o agente fica esperando indefinidamente

# Simular timeout:

import asyncio



async def canary_com_timeout(servico, versao, erro, timeout_s=300):

    try:

        return await asyncio.wait_for(canary_rollback_async(servico, versao, erro), timeout=timeout_s)

    except asyncio.TimeoutError:

        return 'Canary timeout — escalar para humano'

# Sistema com policy que exige dry-run antes de live

def executar_acao(acao, servico, modo):

    if modo == ModoExecucao.LIVE:

        # Verificar que dry-run foi executado recentemente

        if not HISTORICO_DRY_RUN.get(f'{acao.tipo}_{servico}'):

            raise ValueError('Dry-run obrigatório antes de live')

def circuit_breaker_remediacao(servico: str, historico_falhas: dict) -> bool:

    from datetime import datetime, timedelta

    falhas = historico_falhas.get(servico, [])

    uma_hora_atras = datetime.utcnow() - timedelta(hours=1)

    recentes = [f for f in falhas if datetime.fromisoformat(f['timestamp']) > uma_hora_atras]

    if len(recentes) >= 3:

        print(f'[CB] ABERTO: {len(recentes)} falhas recentes em {servico}')

        return True

    reversoes = [f for f in falhas if f.get('tipo') == 'revertido']

    if len(reversoes) >= 2:

        print(f'[CB] ABERTO: {len(reversoes)} reversões em {servico}')

        return True

    return False