Como Implementar Observabilidade (Logs, Métricas, Traces) na Integração com API de CPF

Introdução

Integrar uma API de CPF em produção é apenas o primeiro passo. Para garantir que tudo funcione de forma confiável, você precisa de observabilidade — a capacidade de entender o estado interno do seu sistema a partir dos dados que ele produz. Sem logs estruturados, métricas e traces distribuídos, você está operando no escuro, descobrindo problemas apenas quando seus usuários reclamam.

Os três pilares da observabilidade

Antes de implementar qualquer coisa, é fundamental entender o que cada pilar oferece e como eles se complementam.

• Logs -- registros detalhados de eventos individuais, essenciais para debugging e auditoria
• Métricas -- números agregáveis que permitem dashboards e alertas em tempo real
• Traces -- rastreamento fim-a-fim de requisições distribuídas, crucial para identificar gargalos

Implementando logs estruturados

Logs não estruturados (texto livre) são difíceis de pesquisar e analisar. Use JSON estruturado para facilitar a indexação e consulta.

import logging
import json
import time
import requests

class CPFApiLogger:
    def __init__(self):
    self.logger = logging.getLogger("cpf_api")
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter('%(message)s'))
    self.logger.addHandler(handler)
    self.logger.setLevel(logging.INFO)

    def consultar_cpf(self, cpf: str, correlation_id: str):
    start = time.time()
    log_context = {
    "service": "cpf-validation",
    "correlation_id": correlation_id,
    "cpf_masked": f"***.***.{cpf[8:11]}-{cpf[12:14]}",
    "action": "cpf_lookup"
    }

    try:
    response = requests.get(
    f"https://api.cpfhub.io/cpf/{cpf}",
    headers={"x-api-key": "sua-chave-aqui"},
    timeout=10
    )
    duration_ms = (time.time() - start) * 1000

    log_context.update({
    "status_code": response.status_code,
    "duration_ms": round(duration_ms, 2),
    "success": response.json().get("success", False)
    })
    self.logger.info(json.dumps(log_context))
    return response.json()

    except requests.exceptions.Timeout:
    log_context.update({"error": "timeout", "level": "error"})
    self.logger.error(json.dumps(log_context))
    raise

• correlation_id -- identificador único que conecta logs de diferentes serviços para a mesma requisição
• cpf_masked -- nunca registre CPFs completos em logs por questões de LGPD
• duration_ms -- tempo de resposta é uma das métricas mais importantes para monitorar

Coletando métricas com Prometheus

Métricas permitem criar dashboards e alertas. As quatro métricas essenciais para monitorar sua integração são conhecidas como os Golden Signals.

from prometheus_client import Counter, Histogram, Gauge

# Contador de requisições por status
cpf_requests_total = Counter(
    'cpf_api_requests_total',
    'Total de requisições à API de CPF',
    ['method', 'status', 'endpoint']
)

# Histograma de latência
cpf_request_duration = Histogram(
    'cpf_api_request_duration_seconds',
    'Duração das requisições à API de CPF',
    buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

# Gauge de requisições em andamento
cpf_requests_in_progress = Gauge(
    'cpf_api_requests_in_progress',
    'Requisições em andamento à API de CPF'
)

@cpf_request_duration.time()
@cpf_requests_in_progress.track_inprogress()
def consultar_cpf(cpf: str):
    response = requests.get(
    f"https://api.cpfhub.io/cpf/{cpf}",
    headers={"x-api-key": "sua-chave-aqui"}
    )
    cpf_requests_total.labels(
    method="GET",
    status=response.status_code,
    endpoint="/cpf"
    ).inc()
    return response.json()

Implementando traces distribuídos com OpenTelemetry

Traces permitem visualizar o caminho completo de uma requisição através de múltiplos serviços. O OpenTelemetry é o padrão da indústria para instrumentação.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://collector:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("cpf-service")

def validar_cpf_completo(cpf: str):
    with tracer.start_as_current_span("validar_cpf") as span:
    span.set_attribute("cpf.masked", f"***{cpf[-5:]}")

    with tracer.start_as_current_span("validar_formato"):
    if not cpf_formato_valido(cpf):
    span.set_attribute("validation.result", "formato_invalido")
    return None

    with tracer.start_as_current_span("consultar_api_cpfhub"):
    response = requests.get(
    f"https://api.cpfhub.io/cpf/{cpf}",
    headers={"x-api-key": "sua-chave-aqui"}
    )
    span.set_attribute("http.status_code", response.status_code)
    return response.json()

• Spans -- cada bloco with tracer.start_as_current_span() cria um span que mede o tempo daquela operação
• Atributos -- metadados adicionados aos spans para facilitar a análise posterior
• Propagação -- o contexto do trace é automaticamente propagado entre serviços via headers HTTP

Perguntas frequentes

O que é necessário para implementar validação de CPF neste contexto?

A validação de CPF exige uma chamada à API com o número do documento e a chave de autenticação. A CPFHub.io retorna o status do CPF, nome do titular e data de nascimento em menos de 200ms, permitindo a verificação em tempo real durante o cadastro ou transação.

A API CPFHub.io funciona para todos os volumes de consulta?

Sim. O plano gratuito oferece 50 consultas por mês sem cartão de crédito — ideal para testes e projetos pequenos. Para volumes maiores, o plano Pro inclui 1.000 consultas mensais por R$149. Se o limite for ultrapassado, a API não bloqueia: cobra R$0,15 por consulta adicional.

Como garantir conformidade com a LGPD ao usar uma API de CPF?

Use o CPF apenas para a finalidade declarada ao titular, armazene apenas o necessário (não guarde o CPF cru se um token bastar), implemente controle de acesso aos logs de consulta e documente a base legal para o tratamento. A ANPD orienta que dados de identificação devem ser tratados com o princípio da necessidade.

Quanto tempo leva para integrar a API CPFHub.io?

A integração básica leva menos de 30 minutos: crie uma conta em cpfhub.io, gere a API key no painel e faça uma chamada GET para https://api.cpfhub.io/cpf/{CPF} com o header x-api-key. A documentação inclui exemplos em Python, Node.js, PHP, Java e outras linguagens.

Conclusão

Observabilidade na integração com API de CPF não é opcional para quem opera em produção. Comece pelos logs estruturados — eles resolvem 80% dos problemas de debug. Adicione métricas para alertas proativos e, quando o sistema crescer, implemente traces para entender o comportamento fim-a-fim. Com os três pilares no lugar, você para de descobrir problemas pelos seus usuários. Comece agora em cpfhub.io.