Como Implementar Limitação de Taxa e Tarefas em Segundo Plano Usando FastAPI e Redis

A limitação de taxa é um mecanismo essencial para proteger APIs contra abusos, prevenir sobrecarga do servidor e garantir uma distribuição justa de recursos. O uso adequado desse mecanismo pode melhorar a escalabilidade e estabilidade da aplicação, além de evitar ataques de negação de serviço distribuída (DDoS). O FastAPI, uma das frameworks mais rápidas para construção de APIs em Python, facilita a implementação de tais técnicas quando combinada com o Redis.

Limitação de Taxa com Janela Fixa

python
Copy code
import time
from fastapi import HTTPException, Request, status
from app.redis_client import get_redis

async def fixed_window_rate_limit(request: Request, user_id: int = None, limit: int = 100, window_seconds: int = 60):

    redis = await get_redis()
    key = get_rate_limit_key(request, user_id, window="minute")
    current_count = await redis.incr(key)
    if current_count == 1:
        await redis.expire(key, window_seconds)
    if current_count > limit:
        ttl = await redis.ttl(key)
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Too many requests. Please try again later.",
            headers={"Retry-After": str(ttl if ttl > 0 else window_seconds)},
        )

Limitação de Taxa com Janela Deslizante

Uma limitação de taxa mais precisa é oferecida pela "janela deslizante", que rastreia os timestamps de cada requisição individualmente. Em vez de trabalhar com intervalos fixos, esse método permite observar um intervalo de tempo que "desliza" à medida que novas requisições chegam, evitando os efeitos indesejados do limite de taxa no limite da janela.

A implementação de uma janela deslizante no Redis pode ser feita com o uso de conjuntos ordenados (sorted sets):

python
Copy code
async def sliding_window_rate_limit(request: Request, user_id: int = None, limit: int = 100, window_seconds: int = 60):

    redis = await get_redis()
    key = get_rate_limit_key(request, user_id, window="sliding")
    now = int(time.time())
    window_start = now - window_seconds
    await redis.zadd(key, {str(now): now})
    await redis.zremrangebyscore(key, 0, window_start)
    req_count = await redis.zcard(key)
    await redis.expire(key, window_seconds)
    if req_count > limit:
        oldest = await redis.zrange(key, 0, 0, withscores=True)

        if oldest and len(oldest[0]) > 1:

            retry_after = window_seconds - (now - int(float(oldest[0][1])))
        else:
            retry_after = window_seconds
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded. Please try again later.",

            headers={"Retry-After": str(max(1, int(retry_after)))},

        )

Essa técnica permite uma janela de tempo "rolante", sem os problemas de "efeito de borda" que a janela fixa pode causar, especialmente quando muitas requisições acontecem perto do limite da janela.

Aplicando a Limitação de Taxa nas Rotas do FastAPI

As funções de limitação de taxa podem ser facilmente integradas às rotas do FastAPI, seja como dependências ou como middleware. No exemplo abaixo, a limitação de taxa é aplicada a uma rota pública, com um limite de 20 requisições por minuto por usuário:

python
Copy code
from fastapi import APIRouter, Request, Depends
router = APIRouter()
@router.get("/public-data")

async def public_data(request: Request):

    await fixed_window_rate_limit(request, limit=20, window_seconds=60)
    return {"data": "This endpoint is rate-limited to 20 requests per minute per client."}

A limitação de taxa pode ser personalizada para diferentes tipos de endpoints, dependendo das necessidades de segurança e desempenho da aplicação. Por exemplo, endpoints mais sensíveis ou pagos podem utilizar o algoritmo de janela deslizante, enquanto endpoints públicos podem ser limitados com uma janela fixa.

Cabeçalhos Padrão "Retry-After" para Limitação de Taxa

Quando o limite de requisições é excedido, o servidor retorna um código HTTP 429, juntamente com um cabeçalho "Retry-After", que informa ao cliente quanto tempo ele deve aguardar antes de tentar novamente. Esse cabeçalho é respeitado por muitos navegadores e bibliotecas HTTP, permitindo que os clientes se ajustem automaticamente e reduzam a carga no servidor. Além disso, cabeçalhos como "X-RateLimit-Limit" e "X-RateLimit-Remaining" podem ser adicionados para maior transparência:

python
Copy code
from fastapi.responses import JSONResponse

@router.get("/user-resource")
async def user_resource(request: Request, user_id: int = 123):
    await sliding_window_rate_limit(request, user_id=user_id, limit=10, window_seconds=60)
    redis = await get_redis()
    key = get_rate_limit_key(request, user_id, "sliding")
    count = await redis.zcard(key)
    remaining = max(0, 10 - count)
    headers = {
        "X-RateLimit-Limit": "10",
        "X-RateLimit-Remaining": str(remaining)
    }
    return JSONResponse({"msg": "User resource accessed."}, headers=headers)

Adaptando a Limitação de Taxa para Diferentes Necessidades

Embora a limitação de taxa já seja uma técnica poderosa, há muitos ajustes que podem ser feitos para torná-la ainda mais eficiente:

• Testes com ferramentas automatizadas: Isso permite garantir que os limites estão sendo aplicados corretamente em diferentes cenários de uso.

• Ajustes finos nos limites e janelas: Dependendo da utilização real dos endpoints, pode ser necessário adaptar os limites e janelas de tempo para equilibrar desempenho e usabilidade.

• Combinação de diferentes algoritmos: Pode ser vantajoso utilizar a janela fixa para APIs públicas e a janela deslizante para rotas sensíveis ou pagas.

• Auditoria e alertas: Registros de uso excessivo podem ser úteis para monitoramento e alertas, ajudando na identificação de abusos ou comportamentos inesperados.

Tarefas em Segundo Plano com Celery e Redis

O Celery, combinado com o Redis como broker e backend de resultados, facilita a execução de tarefas longas e complexas, como a geração de relatórios, exportação de dados ou limpeza de banco de dados, sem impactar diretamente o desempenho da API. A integração do Celery permite que esses trabalhos sejam executados de maneira assíncrona, sem bloquear o processo principal da aplicação.

python
Copy code
from celery import Celery

import os
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
celery_app = Celery(
    "myapp_tasks",
    broker=REDIS_URL,
    backend=REDIS_URL
)

Essa abordagem torna a aplicação mais responsiva e escalável, uma vez que as tarefas pesadas são executadas de forma independente.

Como Configurar um Stack de Log com ELK e Kubernetes: Monitoramento e Orquestração Eficientes

Ao implementar uma solução de monitoramento robusta, a integração do Elasticsearch, Logstash e Kibana (ELK) oferece uma plataforma poderosa para visualização e análise de logs em tempo real. Com o suporte do Filebeat para coleta contínua de logs, você é capaz de obter uma visibilidade total dos dados de sua aplicação, facilitando a detecção de erros, métricas de desempenho e muito mais. O uso do Kubernetes para orquestrar esses contêineres transforma a gestão e a escalabilidade de ambientes complexos, tornando-os mais eficientes e seguros.

Uma vez que a pilha ELK está em funcionamento, o Kibana se torna o ponto principal de visualização. Por meio de dashboards interativos, é possível acompanhar a saúde do sistema, identificar falhas e analisar o uso da aplicação. A primeira etapa após configurar o stack é criar padrões de índice no Kibana, que permitem o mapeamento de dados de log. Ao logar pela primeira vez no Kibana (http://localhost:5601), navegue até "Stack Management" → "Index Patterns", onde você cria um padrão que corresponda aos logs, como por exemplo myapp-logs-*. Selecione o campo de timestamp como o campo de tempo para garantir que os logs sejam ordenados corretamente.

Uma vez que os logs começam a fluir para o Elasticsearch, a visualização dos erros em tempo real é essencial. Para isso, é possível criar gráficos de barras ou linhas no Kibana, filtrando pelos níveis de erro (levelname: ERROR) ou pelos códigos de status de resposta (status_code >= 500). Esses gráficos podem ser adicionados a um novo dashboard chamado "Application Health", permitindo uma visão contínua sobre a estabilidade da aplicação.

Além disso, o monitoramento de métricas de requisições, como a contagem de request_id ou a análise de códigos de status, pode fornecer insights valiosos sobre o desempenho do sistema. Criando tabelas ou gráficos de pizza, é possível agrupar dados por endpoint, tipo de erro ou até mesmo por usuário, facilitando a identificação de anomalias e picos de tráfego. Após configurar todos os widgets necessários, salve o dashboard e configure o auto-refresh para garantir uma atualização em tempo real das métricas.

A pilha ELK oferece visibilidade instantânea para depuração, auditorias pós-incidente e conformidade. A capacidade de realizar buscas completas de texto nos logs permite traçar requisições, correlacionar logs por request_id ou identificar consultas lentas. O filtro detalhado por serviço, host ou campos personalizados da aplicação facilita ainda mais a análise. Além disso, a retenção de longo prazo é essencial para a análise de tendências, auditorias e ajustes de desempenho.

Caso os logs não estejam aparecendo, verifique primeiro os logs do contêiner Filebeat para identificar problemas com caminhos ou permissões. Em seguida, avalie a saúde do Elasticsearch e o uso de disco, e por fim, verifique os volumes do Docker para garantir que os caminhos estejam corretos.

Kubernetes: Orquestração de Contêineres e Automação da Infraestrutura

À medida que o número de contêineres cresce, a simples gestão manual se torna impossível. O Kubernetes, como orquestrador de contêineres padrão da indústria, surge como a solução para automatizar o deployment, escalabilidade, rede e auto-cura de serviços em clusters de nós. No centro de suas operações, estão os manifests YAML, que descrevem exatamente o que o Kubernetes deve executar e como ele deve comportar-se. Estes manifests são a nossa "fonte da verdade", garantindo que cada mudança na infraestrutura seja realizada de maneira segura, repetível e auditável.

Por exemplo, um manifest de Deployment no Kubernetes define como os Pods (contêineres) devem ser executados e gerenciados. Abaixo está um exemplo de manifest YAML para um Deployment simples de uma aplicação:

yaml
Copy code
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
  labels:
    app: myapp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
  template:
    metadata:
      labels:
        app: myapp
    spec:
      containers:
        - name: myapp
          image: myregistry/myapp:1.0.0
          ports:
            - containerPort: 8000
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: myapp-secret
                  key: database-url
          resources:
            requests:
              cpu: "250m"
              memory: "512Mi"
            limits:
              cpu: "500m"
              memory: "1Gi"

Esse manifest especifica a criação de três réplicas da aplicação, cada uma expondo a porta 8000. Ele também define um ambiente variável, DATABASE_URL, que será recuperado de um Secret, e estabelece limites de recursos para a aplicação, como CPU e memória.

Outro componente fundamental do Kubernetes é o Service, que expõe os Pods a outras partes do cluster, garantindo rede estável, mesmo que os Pods sejam efêmeros. Abaixo, um exemplo simples de manifest para um Service:

yaml
Copy code
# service.yaml

apiVersion: v1
kind: Service
metadata:
  name: myapp-service
spec:
  selector:
    app: myapp
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000

Aqui, o tráfego destinado à porta 80 do Service é redirecionado para a porta 8000 de cada Pod. Caso precise de acesso externo, você pode optar por usar type: LoadBalancer ou configurar um Ingress para roteamento de tráfego HTTP.

O Ingress é responsável pelo roteamento avançado de tráfego HTTP(S), incluindo a terminação TLS, roteamento de domínios e gestão de caminhos de URL. Um exemplo de manifest para um Ingress seria o seguinte:

yaml
Copy code
# ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: myapp-ingress
  annotations:
    nginx.ingress.kubernetes.io/rewrite-target: /
spec:
  rules:
    - host: myapp.local
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: myapp-service
                port:
                  number: 80

Ao configurar esse Ingress, todas as requisições para myapp.local/ são encaminhadas para o Service na porta 80.

yaml
Copy code
# configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: myapp-config
data:
  LOG_LEVEL: "info"
  CACHE_HOST: "redis.default.svc.cluster.local"

Esses valores podem ser referenciados diretamente no Deployment, garantindo que a configuração seja separada da lógica da aplicação e permitindo que mudanças sejam feitas de forma centralizada e sem afetar a aplicação diretamente.