Como Usar Arquitetura de Software em Produção

Fundamentos de Arquitetura em Ambiente de Produção

Arquitetura de software em produção vai além de desenhar diagramas bonitos. É sobre criar sistemas que funcionam sob pressão real: tráfego variável, falhas de hardware, deploys frequentes e manutenção contínua. A diferença entre um código que funciona no laptop e um sistema produtivo está na resiliência, observabilidade e capacidade de evolução sem quebrar.

Os pilares fundamentais incluem: separação de responsabilidades (cada componente tem um propósito claro), desacoplamento (mudanças em uma parte não quebram outras), escalabilidade (capacidade de crescer sob demanda) e observabilidade (saber o que está acontecendo em tempo real). Estes conceitos transformam aplicações frágeis em sistemas confiáveis que empresas dependem para operar.

Padrões Arquiteturais Essenciais

Em produção, três padrões dominam: Layered Architecture (separação em camadas), Microservices (serviços independentes) e Event-Driven (comunicação por eventos). A escolha depende do contexto: aplicações menores funcionam bem com camadas, sistemas complexos se beneficiam de microservices, e arquiteturas que precisam reagir a mudanças de estado usam eventos.

# Exemplo de Layered Architecture em Python (Flask)
# Camada de Apresentação (API)
from flask import Flask, jsonify, request
from services.user_service import UserService
from repositories.user_repository import UserRepository

app = Flask(__name__)

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    # Controller delega para Service
    repository = UserRepository()
    service = UserService(repository)
    user = service.get_user_by_id(user_id)
    return jsonify(user.to_dict()) if user else ('', 404)

# Camada de Serviço (Lógica de Negócio)
class UserService:
    def __init__(self, repository):
        self.repository = repository

    def get_user_by_id(self, user_id):
        user = self.repository.find_by_id(user_id)
        if user and user.is_active:
            # Regra de negócio: só retorna usuários ativos
            return user
        return None

# Camada de Repositório (Acesso a Dados)
class UserRepository:
    def __init__(self):
        self.db = get_database_connection()

    def find_by_id(self, user_id):
        result = self.db.query(
            "SELECT * FROM users WHERE id = %s", 
            (user_id,)
        )
        return User.from_db(result) if result else None

Resiliência e Tratamento de Falhas

Sistemas em produção falham. Bancos de dados ficam lentos, APIs externas caem, rede tem intermitências. Arquitetura resiliente antecipa essas falhas usando circuit breakers (quebra o circuito quando um serviço falha), retries com backoff exponencial (tenta novamente com intervalos crescentes), timeouts (não espera infinitamente) e fallbacks (plano B quando algo falha).

O padrão Circuit Breaker é crucial: após N falhas consecutivas, para de chamar o serviço problemático temporariamente, evitando sobrecarga. Após um período, tenta novamente (half-open). Se funcionar, volta ao normal (closed); se falhar, abre novamente (open).

# Circuit Breaker com retry e timeout
import time
import requests
from circuitbreaker import circuit
from tenacity import retry, stop_after_attempt, wait_exponential

class PaymentGateway:
    @circuit(failure_threshold=5, recovery_timeout=60)
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def process_payment(self, amount, card_token):
        try:
            response = requests.post(
                'https://api.payment.com/charge',
                json={'amount': amount, 'token': card_token},
                timeout=5  # Timeout de 5 segundos
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            # Fallback: coloca em fila para processar depois
            self.queue_for_retry(amount, card_token)
            raise
        except requests.exceptions.RequestException as e:
            # Log para análise posterior
            logger.error(f"Payment failed: {e}")
            raise

    def queue_for_retry(self, amount, card_token):
        # Publica em fila (RabbitMQ, SQS, etc)
        queue.publish('payment_retry', {
            'amount': amount,
            'token': card_token,
            'attempt_time': time.time()
        })

Observabilidade e Monitoramento

Não se gerencia o que não se mede. Em produção, você precisa de logs estruturados (JSON facilita parsing), métricas (latência, taxa de erro, throughput) e tracing distribuído (rastrear requisições através de múltiplos serviços). A tríade logs-metrics-traces forma a base da observabilidade moderna.

Implemente health checks em todos os serviços, exponha métricas no formato Prometheus, e use correlation IDs para rastrear requisições. Ferramentas como Grafana visualizam métricas, ELK/Loki agregam logs, e Jaeger/Zipkin rastreiam chamadas distribuídas.

# Observabilidade completa em Python
import logging
import uuid
from prometheus_client import Counter, Histogram, generate_latest
from flask import Flask, request, g
import time

# Configuração de logs estruturados
import structlog
logger = structlog.get_logger()

# Métricas Prometheus
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP Requests', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('http_request_duration_seconds', 'HTTP Request Latency', ['endpoint'])

app = Flask(__name__)

@app.before_request
def before_request():
    # Correlation ID para rastreamento
    g.correlation_id = request.headers.get('X-Correlation-ID', str(uuid.uuid4()))
    g.start_time = time.time()

    logger.info("request_started",
                correlation_id=g.correlation_id,
                method=request.method,
                path=request.path,
                user_agent=request.user_agent.string)

@app.after_request
def after_request(response):
    # Calcula latência e registra métricas
    latency = time.time() - g.start_time

    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.path,
        status=response.status_code
    ).inc()

    REQUEST_LATENCY.labels(endpoint=request.path).observe(latency)

    logger.info("request_completed",
                correlation_id=g.correlation_id,
                status=response.status_code,
                latency=latency)

    # Adiciona correlation ID na resposta
    response.headers['X-Correlation-ID'] = g.correlation_id
    return response

@app.route('/health')
def health_check():
    # Health check para Kubernetes/Load Balancers
    checks = {
        'database': check_database(),
        'cache': check_redis(),
        'external_api': check_payment_gateway()
    }

    all_healthy = all(checks.values())
    status_code = 200 if all_healthy else 503

    return jsonify({
        'status': 'healthy' if all_healthy else 'unhealthy',
        'checks': checks
    }), status_code

@app.route('/metrics')
def metrics():
    # Endpoint para Prometheus coletar métricas
    return generate_latest()

Estratégias de Deploy e Versionamento

Deploy em produção não é "subir código e torcer". Use blue-green deployment (dois ambientes, troca instantânea), canary releases (libera para pequeno percentual primeiro) ou rolling updates (atualiza gradualmente). Sempre tenha rollback rápido - a capacidade de voltar à versão anterior em segundos pode salvar seu negócio.

Versionamento de API é crítico: nunca quebre contratos existentes. Use versionamento na URL (/v1/users, /v2/users) ou headers. Mantenha pelo menos duas versões funcionando simultaneamente durante transições.

# Feature Flags para deploy gradual
from flask import Flask, request
import redis

app = Flask(__name__)
redis_client = redis.Redis(host='localhost', port=6379)

def is_feature_enabled(feature_name, user_id=None):
    """
    Feature flag com rollout percentual
    """
    # Verifica flag global
    global_enabled = redis_client.get(f"feature:{feature_name}:enabled")
    if global_enabled == b'false':
        return False

    # Rollout percentual (ex: 10% dos usuários)
    rollout_pct = int(redis_client.get(f"feature:{feature_name}:rollout") or 100)

    if user_id:
        # Hash consistente: mesmo usuário sempre vê mesma versão
        user_hash = hash(f"{feature_name}:{user_id}") % 100
        return user_hash < rollout_pct

    return True

@app.route('/api/v2/recommendations')
def get_recommendations_v2():
    user_id = request.args.get('user_id')

    if is_feature_enabled('new_recommendation_algorithm', user_id):
        # Nova versão com ML
        return jsonify(ml_based_recommendations(user_id))
    else:
        # Versão antiga estável
        return jsonify(rule_based_recommendations(user_id))

# Configuração via Redis (runtime, sem redeploy)
# redis_client.set('feature:new_recommendation_algorithm:enabled', 'true')
# redis_client.set('feature:new_recommendation_algorithm:rollout', '10')  # 10% dos usuários

Conclusão

• Arquitetura em produção exige resiliência por design: implemente circuit breakers, retries inteligentes e fallbacks. Sistemas falham, planeje para isso.
• Observabilidade não é opcional: logs estruturados, métricas e tracing distribuído são essenciais para diagnosticar problemas rapidamente em produção.
• Deploy é um processo contínuo: use feature flags e estratégias graduais (canary, blue-green) para reduzir riscos. A capacidade de rollback rápido é tão importante quanto o deploy em si.

Referências

• The Twelve-Factor App - Metodologia para aplicações SaaS modernas
• Microsoft Azure Architecture Center - Padrões e práticas de arquitetura
• Martin Fowler - Microservices Guide - Referência definitiva sobre microservices
• Site Reliability Engineering Book - Google - Práticas de SRE do Google
• Release It! - Michael Nygard - Padrões de estabilidade e design para produção