🏠 Início / 📚 Documentação / Operations / Integracao Slos Business Metrics

Integracao Slos Business Metrics

🔗 Guia de Integração - SLOs e Business Metrics

Data: 2025-12-01
Status: ✅ Guia Criado

🎯 Objetivo

Este guia mostra como integrar os módulos de SLOs e Business Metrics nos apps do ecossistema C-Suite.

📦 Módulos Disponíveis

1. SLOs (Service Level Objectives)

2. Business Metrics

🚀 Integração de SLOs

Passo 1: Importar e Inicializar

import sys
from pathlib import Path

# Adiciona common ao path
COMMON_PATH = str(Path(__file__).resolve().parents[N])  # Ajuste N
sys.path.insert(0, COMMON_PATH)

try:
    from common.common_slos import SLOTracker, SLO, SLIType
    SLO_ENABLED = True
except ImportError:
    SLO_ENABLED = False

# Criar instância do tracker
if SLO_ENABLED:
    slo_tracker = SLOTracker()

Passo 2: Definir SLOs no Startup

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    if SLO_ENABLED:
        # SLO de Disponibilidade (99.9% uptime)
        slo_tracker.register_slo(SLO(
            name="api_availability",
            description="API availability target",
            sli_type=SLIType.AVAILABILITY,
            target=0.999,  # 99.9%
            window_minutes=60
        ))

        # SLO de Latência (P95 < 200ms)
        slo_tracker.register_slo(SLO(
            name="api_latency_p95",
            description="API P95 latency target",
            sli_type=SLIType.LATENCY,
            target=0.200,  # 200ms
            window_minutes=60
        ))

        # SLO de Taxa de Erro (< 0.1%)
        slo_tracker.register_slo(SLO(
            name="api_error_rate",
            description="API error rate target",
            sli_type=SLIType.ERROR_RATE,
            target=0.001,  # 0.1%
            window_minutes=60
        ))

Passo 3: Registrar Métricas em Endpoints

from fastapi import Request
import time

@app.middleware("http")
async def slo_middleware(request: Request, call_next):
    start_time = time.time()
    error_occurred = False

    try:
        response = await call_next(request)

        # Registrar disponibilidade (1 = sucesso, 0 = erro)
        if SLO_ENABLED:
            slo_tracker.record_metric("api_availability", 1.0)

            # Registrar latência
            latency = time.time() - start_time
            slo_tracker.record_metric("api_latency_p95", latency)

            # Registrar taxa de erro
            if response.status_code >= 500:
                error_occurred = True
                slo_tracker.record_metric("api_error_rate", 1.0)
            else:
                slo_tracker.record_metric("api_error_rate", 0.0)

        return response
    except Exception as e:
        error_occurred = True
        if SLO_ENABLED:
            slo_tracker.record_metric("api_availability", 0.0)
            slo_tracker.record_metric("api_error_rate", 1.0)
        raise

Passo 4: Endpoint para Verificar SLOs

@app.get("/slo/status")
async def get_slo_status():
    """Retorna status de todos os SLOs"""
    if not SLO_ENABLED:
        return {"error": "SLOs not enabled"}

    return {
        "slos": slo_tracker.get_all_slo_status()
    }

@app.get("/slo/status/{slo_name}")
async def get_slo_status_by_name(slo_name: str):
    """Retorna status de um SLO específico"""
    if not SLO_ENABLED:
        return {"error": "SLOs not enabled"}

    return slo_tracker.check_slo(slo_name)

📊 Integração de Business Metrics

Passo 1: Importar e Inicializar

try:
    from common.common_business_metrics import BusinessMetricsTracker
    BUSINESS_METRICS_ENABLED = True
except ImportError:
    BUSINESS_METRICS_ENABLED = False

# Criar instância do tracker
if BUSINESS_METRICS_ENABLED:
    business_tracker = BusinessMetricsTracker()

Passo 2: Registrar Métricas de Negócio

@app.post("/decide")
async def decide(request: DecisionRequest):
    """Endpoint de decisão"""

    # ... lógica de decisão ...

    decision_result = await make_decision(request)

    # Registrar métricas de negócio
    if BUSINESS_METRICS_ENABLED:
        # Incrementar contador de decisões
        business_tracker.record_decision(
            org_id=str(request.org_id),
            decision_type=decision_result.get("type", "unknown"),
            channel=decision_result.get("channel", "unknown")
        )

        # Se decisão foi aceita
        if decision_result.get("accepted", False):
            business_tracker.record_decision_accepted(
                org_id=str(request.org_id),
                decision_type=decision_result.get("type", "unknown")
            )
        else:
            business_tracker.record_decision_rejected(
                org_id=str(request.org_id),
                decision_type=decision_result.get("type", "unknown")
            )

    return decision_result

Passo 3: Métricas Customizadas

# Registrar métrica customizada
if BUSINESS_METRICS_ENABLED:
    business_tracker.record_custom_metric(
        metric_name="revenue",
        value=revenue_amount,
        labels={"org_id": str(org_id), "period": "daily"}
    )

Passo 4: Endpoint para Métricas de Negócio

@app.get("/business-metrics")
async def get_business_metrics():
    """Retorna métricas de negócio"""
    if not BUSINESS_METRICS_ENABLED:
        return {"error": "Business metrics not enabled"}

    return {
        "metrics": business_tracker.get_all_metrics()
    }

🔄 Integração Completa - Exemplo

Exemplo: 4C Decision API

from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware
import time
import sys
from pathlib import Path

COMMON_PATH = str(Path(__file__).resolve().parents[N])
sys.path.insert(0, COMMON_PATH)

# Imports com fallback
try:
    from common.common_slos import SLOTracker, SLO, SLIType
    from common.common_business_metrics import BusinessMetricsTracker
    SLO_ENABLED = True
    BUSINESS_METRICS_ENABLED = True
except ImportError:
    SLO_ENABLED = False
    BUSINESS_METRICS_ENABLED = False

app = FastAPI()

# Inicializar trackers
if SLO_ENABLED:
    slo_tracker = SLOTracker()
if BUSINESS_METRICS_ENABLED:
    business_tracker = BusinessMetricsTracker()

@app.on_event("startup")
async def startup():
    # Registrar SLOs
    if SLO_ENABLED:
        slo_tracker.register_slo(SLO(
            name="decision_api_availability",
            description="Decision API availability",
            sli_type=SLIType.AVAILABILITY,
            target=0.999,
            window_minutes=60
        ))
        slo_tracker.register_slo(SLO(
            name="decision_api_latency",
            description="Decision API latency P95",
            sli_type=SLIType.LATENCY,
            target=0.500,  # 500ms
            window_minutes=60
        ))

# Middleware para SLOs
class SLOMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()

        try:
            response = await call_next(request)

            if SLO_ENABLED:
                # Disponibilidade
                slo_tracker.record_metric("decision_api_availability", 1.0)

                # Latência
                latency = time.time() - start_time
                slo_tracker.record_metric("decision_api_latency", latency)

                # Taxa de erro
                if response.status_code >= 500:
                    slo_tracker.record_metric("decision_api_error_rate", 1.0)
                else:
                    slo_tracker.record_metric("decision_api_error_rate", 0.0)

            return response
        except Exception:
            if SLO_ENABLED:
                slo_tracker.record_metric("decision_api_availability", 0.0)
                slo_tracker.record_metric("decision_api_error_rate", 1.0)
            raise

app.middleware("http")(SLOMiddleware())

@app.post("/decide")
async def decide(request: DecisionRequest):
    # ... lógica de decisão ...

    result = await make_decision(request)

    # Business metrics
    if BUSINESS_METRICS_ENABLED:
        business_tracker.record_decision(
            org_id=str(request.org_id),
            decision_type=result.get("type", "offer"),
            channel=result.get("channel", "whatsapp")
        )

    return result

@app.get("/slo/status")
async def slo_status():
    if SLO_ENABLED:
        return {"slos": slo_tracker.get_all_slo_status()}
    return {"error": "SLOs not enabled"}

📈 Dashboards Grafana

Métricas Prometheus Expostas

Os módulos expõem automaticamente métricas Prometheus:

SLOs:
- slo_violation{slo_name="..."} - Status de violação (1=violado, 0=atendido)
- sli_value{slo_name="...", sli_type="..."} - Valor atual do SLI

Business Metrics:
- business_decisions_total{org_id="...", decision_type="...", channel="..."} - Total de decisões
- business_decisions_accepted_total{org_id="...", decision_type="..."} - Decisões aceitas
- business_decisions_rejected_total{org_id="...", decision_type="..."} - Decisões rejeitadas
- business_conversion_rate{org_id="...", period="..."} - Taxa de conversão

Queries Prometheus Exemplo

# Taxa de violação de SLO
slo_violation{slo_name="api_availability"}

# Valor atual do SLI
sli_value{slo_name="api_latency_p95"}

# Total de decisões por organização
sum(business_decisions_total) by (org_id)

# Taxa de conversão
business_conversion_rate{org_id="1", period="daily"}

✅ Checklist de Integração

SLOs

Business Metrics

🎯 Próximos Passos

  1. ⏳ Integrar SLOs em todos os apps principais
  2. ⏳ Integrar Business Metrics em apps que fazem decisões
  3. ⏳ Criar dashboards Grafana unificados
  4. ⏳ Configurar alertas baseados em SLOs
  5. ⏳ Documentar SLOs específicos por serviço

Última atualização: 2025-12-01

🔊 Text-to-Speech

1.0x
1.0
Pronto para reproduzir