Notifications and Alerts - C-Suite
Este documento descreve o sistema de notificações e alertas centralizado para o ecossistema C-Suite.
Visão Geral
O módulo common/common_notifications.py fornece um sistema centralizado de notificações suportando múltiplos canais (email, Slack, webhooks) com diferentes níveis de prioridade.
Características
- ✅ Múltiplos canais (Email, Slack, Webhook, Console)
- ✅ Níveis de prioridade (Low, Medium, High, Critical)
- ✅ Suporte a metadata customizada
- ✅ Integração com sistema de logging
- ✅ Fallback graceful quando serviços não disponíveis
Configuração
Variáveis de Ambiente
# Habilitar notificações
NOTIFICATIONS_ENABLED=true
# Email (SMTP)
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@gmail.com
SMTP_PASSWORD=your-password
SMTP_FROM=noreply@csuite.local
SMTP_USE_TLS=true
# Slack
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL
SLACK_CHANNEL=#alerts
SLACK_DEFAULT_CHANNEL=#alerts
# Webhook
WEBHOOK_URL=https://your-webhook-endpoint.com/notify
WEBHOOK_SECRET=your-secret-key
# Default recipient
NOTIFICATION_DEFAULT_EMAIL=admin@csuite.local
Uso
Enviar Notificação Básica
from common.common_notifications import send_notifications import send_notification, NotificationChannel, NotificationPriority
# Email
send_notification(
title="Backup Completed",
message="Database backup completed successfully",
priority=NotificationPriority.MEDIUM,
channel=NotificationChannel.EMAIL,
recipient="admin@csuite.local"
)
# Slack
send_notification(
title="High CPU Usage",
message="CPU usage is above 80%",
priority=NotificationPriority.HIGH,
channel=NotificationChannel.SLACK,
recipient="#alerts"
)
Enviar Alertas
from common.common_notifications import send_alert, send_critical_alert
# Alerta de alta prioridade (Slack)
send_alert(
title="Service Down",
message="csuite-api service is not responding",
recipient="#ops-team",
metadata={"service": "csuite-api", "status_code": 503}
)
# Alerta crítico (múltiplos canais)
send_critical_alert(
title="Database Connection Lost",
message="Cannot connect to primary database",
metadata={"database": "csuite", "error": "Connection timeout"}
)
Usando Notification Object
from common.common_notifications import Notification, NotificationService
notification = Notification(
title="Policy Updated",
message="Policy 'min_margin' was updated",
priority=NotificationPriority.MEDIUM,
channel=NotificationChannel.EMAIL,
recipient="admin@csuite.local",
metadata={
"policy_id": "min_margin",
"updated_by": "user123",
"old_value": 0.10,
"new_value": 0.15
}
)
service = NotificationService()
service.send(notification)
Canais Suportados
Envia notificações via SMTP.
send_notification(
title="Test Email",
message="This is a test",
channel=NotificationChannel.EMAIL,
recipient="user@example.com"
)
Slack
Envia notificações para canais do Slack via webhook.
send_notification(
title="Test Slack",
message="This is a test",
channel=NotificationChannel.SLACK,
recipient="#alerts"
)
Webhook
Envia notificações para um endpoint HTTP customizado.
send_notification(
title="Test Webhook",
message="This is a test",
channel=NotificationChannel.WEBHOOK,
metadata={"custom_field": "value"}
)
Console
Para desenvolvimento/testing, imprime no console.
send_notification(
title="Test Console",
message="This is a test",
channel=NotificationChannel.CONSOLE
)
Níveis de Prioridade
- LOW: Informações gerais
- MEDIUM: Avisos moderados
- HIGH: Problemas que requerem atenção
- CRITICAL: Problemas críticos que requerem ação imediata
Integração com Outros Módulos
Com Audit Log
from common.common_audit import audit_log
from common.common_notifications import send_alert, NotificationPriority
def update_policy(policy_id: str, new_value: float, user_id: str):
# Atualizar política
# ...
# Log de auditoria
audit_log(
action="policy_updated",
user_id=user_id,
resource_type="policy",
resource_id=policy_id
)
# Notificar administradores
send_alert(
title="Policy Updated",
message=f"Policy {policy_id} was updated to {new_value}",
metadata={"policy_id": policy_id, "new_value": new_value}
)
Com Health Checks
from common.common_health import HealthChecker
from common.common_notifications import send_critical_alert
def check_service_health():
checker = HealthChecker()
status = checker.check_dependency("database:csuite")
if status.status == "unhealthy":
send_critical_alert(
title="Database Unhealthy",
message=f"Database health check failed: {status.message}",
metadata={"dependency": "database:csuite", "error": status.message}
)
Exemplos de Uso
Backup Failed
from common.common_notifications import send_critical_alert
try:
backup_database()
except Exception as e:
send_critical_alert(
title="Database Backup Failed",
message=f"Backup failed: {str(e)}",
metadata={"error": str(e), "timestamp": datetime.utcnow().isoformat()}
)
Rate Limit Exceeded
from common.common_notifications import send_alert
if rate_limit_exceeded:
send_alert(
title="Rate Limit Exceeded",
message=f"User {user_id} exceeded rate limit",
metadata={"user_id": user_id, "limit": 100, "requests": 150}
)
Policy Violation
from common.common_notifications import send_notification, NotificationPriority
if policy_violated:
send_notification(
title="Policy Violation",
message=f"Decision violated policy: {policy_name}",
priority=NotificationPriority.HIGH,
channel=NotificationChannel.SLACK,
metadata={"policy": policy_name, "decision_id": decision_id}
)
Troubleshooting
Email não está sendo enviado
- Verifique configurações SMTP
- Verifique credenciais
- Verifique firewall/portas
- Teste com
SMTP_USE_TLS=falseprimeiro
Slack não está funcionando
- Verifique webhook URL
- Verifique formato do payload
- Verifique permissões do webhook
Notificações desabilitadas
Se NOTIFICATIONS_ENABLED=false, todas as notificações serão ignoradas (apenas logadas).
Dependências
Opcionais (instaladas automaticamente se disponíveis):
smtplib(built-in Python)httpx(para Slack/webhooks)