1.5 — AIOps: Consultas Métricas Assistidas e Detecção de Anomalias (Observabilidade)

Objetivo da Aula

Ao concluir esta aula, você será capaz de:

Converter descrições em linguagem natural para PromQL válido usando Claude

Implementar detecção de anomalias por Z-score com série temporal simulada

Descrever como Prophet e IsolationForest funcionam para anomaly detection

Gerar regras de alerta Prometheus (YAML) via NL

Completar os TODOs: entender detectar_anomalias_prophet e detectar_anomalias_isolation_forest

Por que isso importa

Prometheus é onipresente em ambientes Kubernetes, mas PromQL tem curva de aprendizado alta. histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{job="api"}[5m])) é difícil de lembrar e mais difícil ainda de adaptar para variações.

LLMs que convertem NL → PromQL democratizam Prometheus: qualquer dev pode consultar métricas sem memorizar a sintaxe. E detecção de anomalias automática vai além de thresholds fixos — um Z-score identifica valores que são estatisticamente anormais para aquela série, não para uma régua universal.

Conceitos Fundamentais

Gerador NL → PromQL

def nl_para_promql(descricao: str) -> str:

    response = client.messages.create(

        model='claude-haiku-4-5-20251001',

        max_tokens=300,

        system='Você é especialista em Prometheus/PromQL. Converta a descrição em uma query PromQL válida. Retorne APENAS a query, sem explicação.',

        messages=[{'role': 'user', 'content': descricao}],

    )

    return response.content[0].text.strip()



# Exemplos de conversão

nl_para_promql('taxa de erro HTTP 5xx do api-gateway nos últimos 5 minutos')

# → rate(http_requests_total{job="api-gateway",status=~"5.."}[5m])



nl_para_promql('p99 de latência do checkout-service')

# → histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{job="checkout-service"}[5m]))



nl_para_promql('uso de CPU acima de 80% por namespace')

# → sum(rate(container_cpu_usage_seconds_total[5m])) by (namespace) > 0.8

O system prompt “retorne APENAS a query” é crucial — sem ele, o LLM adiciona explicações que quebram o copy-paste para o Prometheus.

Série Temporal Simulada

O starter gera uma série temporal realista com padrão diário (seno de 24h) e anomalias injetadas:

def gerar_serie_temporal(nome: str, horas: int = 24) -> list[dict]:

    agora = datetime.utcnow()

    serie = []

    

    for i in range(horas * 12):  # ponto a cada 5 min

        ts = agora - timedelta(minutes=5 * (horas * 12 - i))

        hora = ts.hour

        base = 100 + 50 * math.sin(2 * math.pi * hora / 24)  # padrão diário

        ruido = random.gauss(0, 5)

        

        anomalia = 0

        if 30 <= i <= 35:   # spike

            anomalia = random.uniform(80, 120)

        if 100 <= i <= 102: # drop

            anomalia = -random.uniform(60, 90)

        

        serie.append({'timestamp': ts.isoformat(), 'value': max(0, base + ruido + anomalia)})

    

    return serie

A math.sin(2 * math.pi * hora / 24) cria um ciclo perfeito de 24h — tráfego alto no dia, baixo à noite, simulando um sistema real.

Fundamento: Métodos de Detecção de Anomalia

Threshold fixo (se CPU > 80%, alerta) falha em cargas sazonais — o que é "normal" muda ao longo do dia e da semana. Três alternativas: Z-score mede desvios-padrão da média histórica recente — alerta quando o valor está N desvios acima do "normal para esse horário". Prophet modela a série temporal com tendência + sazonalidade + feriados e alerta quando o valor observado se desvia do valor previsto. Isolation Forest treina um modelo que aprende a isolar pontos anômalos — pontos que precisam de menos divisões aleatórias para ser isolados são outliers. Use Z-score para séries estacionárias simples; Prophet quando há sazonalidade clara; Isolation Forest para padrões multivariados complexos.

Detecção por Z-Score (implementado)

def detectar_anomalias_zscore(serie: list[dict], threshold: float = 2.5) -> list[dict]:

    valores = [p['value'] for p in serie]

    media = sum(valores) / len(valores)

    variancia = sum((v - media) ** 2 for v in valores) / len(valores)

    desvio = math.sqrt(variancia)

    

    anomalias = []

    for ponto in serie:

        # z-score = quantos desvios padrão acima/abaixo da média

        z = abs(ponto['value'] - media) / desvio if desvio > 0 else 0

        if z > threshold:

            anomalias.append({

                **ponto,

                'z_score': round(z, 2),

                'tipo': 'spike' if ponto['value'] > media else 'drop',

            })

    

    return anomalias

Z-score 2.5 significa: valores a mais de 2.5 desvios padrão da média são anomalias. Em distribuição normal, isso captura ~1.2% dos valores — os extremos reais.

TODO: Prophet para Forecasting + Anomaly Detection

Prophet (Meta) é uma biblioteca de forecasting para séries temporais com sazonalidade:

def detectar_anomalias_prophet(serie: list[dict]) -> list[dict]:

    # pip install prophet

    from prophet import Prophet

    import pandas as pd

    

    # 1. Converter para DataFrame com colunas ds (datetime) e y (valor)

    df = pd.DataFrame(serie).rename(columns={'timestamp': 'ds', 'value': 'y'})

    df['ds'] = pd.to_datetime(df['ds'])

    

    # 2. Treinar modelo com sazonalidade diária

    model = Prophet(daily_seasonality=True, weekly_seasonality=False)

    model.fit(df)

    

    # 3. Gerar forecast (predict nos mesmos timestamps)

    forecast = model.predict(df)

    

    # 4. Comparar valor real com previsto — anomalia = desvio grande

    df = df.merge(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']], on='ds')

    anomalias_df = df[(df['y'] > df['yhat_upper']) | (df['y'] < df['yhat_lower'])]

    

    return anomalias_df.to_dict('records')

Vantagem sobre Z-score: Prophet entende sazonalidade. Tráfego alto às 14h é normal — Z-score marcaria como anomalia, Prophet não.

TODO: IsolationForest para Detecção Não-Supervisionada

IsolationForest (scikit-learn) detecta anomalias sem precisar de distribuição gaussiana:

def detectar_anomalias_isolation_forest(serie: list[dict]) -> list[dict]:

    # pip install scikit-learn

    from sklearn.ensemble import IsolationForest

    import numpy as np

    

    # 1. Preparar dados como matriz

    X = np.array([p['value'] for p in serie]).reshape(-1, 1)

    

    # 2. Treinar modelo (contamination = fração esperada de anomalias)

    model = IsolationForest(contamination=0.05, random_state=42)

    labels = model.fit_predict(X)

    

    # 3. Labels: +1 = normal, -1 = anomalia

    anomalias = [serie[i] for i, l in enumerate(labels) if l == -1]

    return anomalias

Vantagem: não assume distribuição — funciona para qualquer forma de dados. Detecta anomalias multivariadas se você passar múltiplas features.

Aprofundamento Técnico

Quando Usar Cada Detector

Algoritmo	Quando usar	Limitação
Z-score	Dados com distribuição aproximadamente normal	Não entende sazonalidade
Prophet	Dados com padrão diário/semanal claro	Mais lento, requer instalação
IsolationForest	Dados com distribuição irregular	Não interpreta sazonalidade

Para produção: Z-score para alertas em tempo real (baixa latência), Prophet para análise offline, IsolationForest para dados multivariados.

Gerador de Alertas Prometheus

def gerar_regra_alerta(descricao: str) -> str:

    promql = nl_para_promql(descricao)

    response = client.messages.create(

        model='claude-haiku-4-5-20251001',

        max_tokens=400,

        system='Gere uma regra de alerta Prometheus YAML (groups > rules) com name, expr, for, labels (severity) e annotations (summary, description).',

        messages=[{'role': 'user', 'content': f'Query: {promql}\nDescrição: {descricao}'}],

    )

    return response.content[0].text



# Exemplo de output esperado:

'''

groups:

- name: api-gateway-alerts

  rules:

  - alert: HighErrorRate

    expr: rate(http_requests_total{job="api-gateway",status=~"5.."}[5m]) > 0.05

    for: 2m

    labels:

      severity: critical

    annotations:

      summary: "Alta taxa de erros no api-gateway"

      description: "Taxa de erros HTTP 5xx acima de 5% por 2 minutos"

'''

Exemplos Anotados

Exemplo 1: Pipeline Completo

# 1. Gerar série temporal com anomalias

serie = gerar_serie_temporal('http_requests_total', horas=6)

print(f'Série: {len(serie)} pontos')



# 2. Detectar com Z-score

anomalias = detectar_anomalias_zscore(serie, threshold=2.0)

print(f'Anomalias Z-score: {len(anomalias)}')

for a in anomalias[:3]:

    print(f'  {a["timestamp"]}: {a["value"]:.1f} (z={a["z_score"]}, {a["tipo"]})')



# 3. Gerar alerta para o contexto

alerta = gerar_regra_alerta('alertar quando error rate do api-gateway superar 5% por 2 minutos')

print(alerta)

Exemplo 2: NL para PromQL com exemplos variados

queries_nl = [

    'pods com mais de 5 restarts nas últimas 2 horas',

    'uso de memória dos contêineres acima de 90% do limite',

    'taxa de requisições por segundo do nginx',

    'latência do banco acima de 1 segundo no percentil 95',

]



for nl in queries_nl:

    pql = nl_para_promql(nl)

    print(f'NL: {nl}')

    print(f'PromQL: {pql}\n')

Padrões e Armadilhas

Padrões

Padrão 1: System prompt restritivo para NL→PromQL

system='Retorne APENAS a query PromQL, sem texto adicional, sem blocos de código.'

Sem isso, o LLM retorna markdown com ``` que quebra o uso direto da query.

Padrão 2: threshold calibrado por domínio

# Métricas com alta variância natural (ex: requests/s em horário de pico)

detectar_anomalias_zscore(serie, threshold=3.0)  # menos sensível



# Métricas estáveis (ex: latência do banco)

detectar_anomalias_zscore(serie, threshold=2.0)  # mais sensível

Padrão 3: Combinar Z-score com janela deslizante

# Anomalia = Z-score alto nas últimas N amostras, não em toda a série

window = serie[-60:]  # últimas 5 horas (60 * 5min)

anomalias = detectar_anomalias_zscore(window, threshold=2.5)

Armadilhas

⚠️ Armadilha 1: Z-score em série muito curta Com menos de 30 pontos, a média e desvio são instáveis — Z-score produz muitos falsos positivos.

⚠️ Armadilha 2: Prophet com dados irregulares (gaps)

# Prophet falha com gaps grandes. Precisa de dados regulares:

df = df.set_index('ds').resample('5T').interpolate()  # preencher gaps

⚠️ Armadilha 3: PromQL gerado sem label filters específicos

nl_para_promql('taxa de erro')

# → rate(http_requests_total[5m])  ← genérico, soma TODOS os jobs



nl_para_promql('taxa de erro HTTP 5xx do api-gateway nas últimas 5 minutos')

# → rate(http_requests_total{job="api-gateway",status=~"5.."}[5m])  ← específico

Seja específico na NL para obter PromQL com labels corretos.

⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.

Ponte para o Lab

Os TODOs do starter são funções placeholder (return []). Para o lab:

detectar_anomalias_prophet: Se não tiver Prophet instalado, implemente um mock que retorna as mesmas anomalias do Z-score — isso permite testar o pipeline sem dependências extras:

def detectar_anomalias_prophet(serie: list[dict]) -> list[dict]:

    print('[MOCK] Prophet simulado via Z-score')

    return detectar_anomalias_zscore(serie, threshold=2.0)

detectar_anomalias_isolation_forest: Igualmente, mock com Z-score:

def detectar_anomalias_isolation_forest(serie: list[dict]) -> list[dict]:

    print('[MOCK] IsolationForest simulado via Z-score')

    return detectar_anomalias_zscore(serie, threshold=2.5)

Para usar os algoritmos reais, instale: pip install prophet scikit-learn pandas numpy

Rode python starter.py e observe as queries PromQL geradas, as anomalias detectadas e a regra de alerta YAML.

Agora você está pronto para o lab.