diff --git a/.env.example b/.env.example index d971788..dda2d19 100644 --- a/.env.example +++ b/.env.example @@ -133,6 +133,15 @@ AI_VOICE_MAX_CONTEXT_SEGMENTS=12 AI_VOICE_HANDOFF_TIMEOUT_SECONDS=8 AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS=svc:ai-voice-runtime +# Gateway error alerts (Telegram) +ERROR_TELEGRAM_ALERTS=0 +ERROR_TELEGRAM_ALERT_BOT_TOKEN= +ERROR_TELEGRAM_ALERT_CHANNEL_ID= +# Alternative: ERROR_TELEGRAM_ALERT_CHAT_IDS=, +ERROR_TELEGRAM_ALERT_MIN_STATUS=500 +ERROR_TELEGRAM_ALERT_COOLDOWN_SECONDS=60 +ERROR_TELEGRAM_ALERT_IGNORE_PATHS=/health + # Event bus (Track 8) EVENT_BUS_ENABLED=0 EVENT_BUS_URL=amqp://guest:guest@localhost:5672/ diff --git a/.env.production.template b/.env.production.template index 8247326..f2492ee 100644 --- a/.env.production.template +++ b/.env.production.template @@ -115,6 +115,15 @@ AI_VOICE_MAX_CONTEXT_SEGMENTS=12 AI_VOICE_HANDOFF_TIMEOUT_SECONDS=8 AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS=svc:ai-voice-runtime +# Gateway error alerts (Telegram) +ERROR_TELEGRAM_ALERTS=0 +ERROR_TELEGRAM_ALERT_BOT_TOKEN= +ERROR_TELEGRAM_ALERT_CHANNEL_ID= +# Alternative: ERROR_TELEGRAM_ALERT_CHAT_IDS=, +ERROR_TELEGRAM_ALERT_MIN_STATUS=500 +ERROR_TELEGRAM_ALERT_COOLDOWN_SECONDS=60 +ERROR_TELEGRAM_ALERT_IGNORE_PATHS=/health + # Event bus compatibility (Track 8 baseline) EVENT_BUS_ENABLED=1 EVENT_BUS_URL=amqp://guest:guest@:5672/ diff --git a/gateway/app.py b/gateway/app.py index e019bac..d6abe27 100644 --- a/gateway/app.py +++ b/gateway/app.py @@ -1,8 +1,10 @@ from __future__ import annotations +import asyncio import logging import os from pathlib import Path +from datetime import datetime, timezone from typing import Any import httpx @@ -14,6 +16,135 @@ from services.shared.models import HealthResponse app = FastAPI(title="api-gateway", version="1.0.0") logger = logging.getLogger("api-gateway.proxy") +_ERROR_ALERT_SENT_AT: dict[str, float] = {} + + +def _env_flag(name: str, default: bool = False) -> bool: + value = os.getenv(name) + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + +def _env_int(name: str, default: int) -> int: + value = os.getenv(name) + if value is None: + return default + try: + parsed = int(value) + except ValueError: + return default + return max(1, parsed) + + +def _error_alert_enabled() -> bool: + return _env_flag("ERROR_TELEGRAM_ALERTS", False) + + +def _error_alert_bot_token() -> str: + return os.getenv("ERROR_TELEGRAM_ALERT_BOT_TOKEN", "").strip() + + +def _error_alert_chat_ids() -> list[str]: + raw = os.getenv("ERROR_TELEGRAM_ALERT_CHAT_IDS", "").strip() + if not raw: + raw = os.getenv("ERROR_TELEGRAM_ALERT_CHANNEL_ID", "").strip() + if not raw: + raw = os.getenv("ERROR_TELEGRAM_ALERT_CHANNEL", "").strip() + return [item.strip() for item in raw.split(",") if item.strip()] + + +def _error_alert_min_status() -> int: + return max(1, _env_int("ERROR_TELEGRAM_ALERT_MIN_STATUS", 500)) + + +def _error_alert_cooldown_seconds() -> int: + return _env_int("ERROR_TELEGRAM_ALERT_COOLDOWN_SECONDS", 30) + + +def _error_alert_ignored_paths() -> tuple[str, ...]: + raw = os.getenv("ERROR_TELEGRAM_ALERT_IGNORE_PATHS", "/health").strip() + paths = tuple(item.strip() for item in raw.split(",") if item.strip()) + return paths if paths else ("/health",) + + +def _error_alert_service_name() -> str: + return os.getenv("ERROR_TELEGRAM_ALERT_SERVICE", "api-gateway").strip() or "api-gateway" + + +def _error_alert_bot_api_base() -> str: + return os.getenv("ERROR_TELEGRAM_ALERT_BOT_API_BASE", "https://api.telegram.org").rstrip("/") + + +def _error_alert_should_alert_path(path: str) -> bool: + ignore = _error_alert_ignored_paths() + return not any(path == item or path.startswith(f"{item}/") for item in ignore) + + +def _error_alert_key(service: str, status: int, path: str, method: str) -> str: + return f"{service}:{status}:{method}:{path}" + + +def _error_alert_is_throttled(key: str) -> bool: + now = datetime.now(tz=timezone.utc).timestamp() + cooldown = _error_alert_cooldown_seconds() + previous = _ERROR_ALERT_SENT_AT.get(key, 0.0) + if now < previous: + return True + _ERROR_ALERT_SENT_AT[key] = now + cooldown + return False + + +async def _send_telegram_alert(message: str) -> None: + chat_ids = _error_alert_chat_ids() + bot_token = _error_alert_bot_token() + if not bot_token or not chat_ids: + return + + url = f"{_error_alert_bot_api_base()}/bot{bot_token}/sendMessage" + payload = {"text": message} + async with httpx.AsyncClient(timeout=3.0) as client: + for chat_id in chat_ids: + payload["chat_id"] = chat_id + try: + response = await client.post(url, json=payload) + except httpx.RequestError: + logger.exception("Unable to send Telegram error notification") + return + if response.status_code >= 300: + logger.warning( + "Telegram error notification failed: status=%s body=%s", + response.status_code, + response.text, + ) + + +def _queue_telegram_error_alert(service: str, method: str, path: str, status: int, detail: str) -> None: + if not _error_alert_enabled(): + return + if not _error_alert_should_alert_path(path): + return + if not _error_alert_bot_token() or not _error_alert_chat_ids(): + return + + if _error_alert_is_throttled(_error_alert_key(service, status, path, method)): + return + + timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + message = ( + f"🚨 [{service}] {status}\n" + f"method: {method}\n" + f"path: {path}\n" + f"time: {timestamp}\n" + f"detail: {detail}" + ) + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + logger.warning("No running loop to queue telegram error notification") + return + loop.create_task(_send_telegram_alert(message)) _LEGAL_HTML = { "privacy": """ @@ -169,13 +300,6 @@ SERVICE_PATH_PREFIX_EXEMPTIONS = { } -def _env_flag(name: str, default: bool = False) -> bool: - value = os.getenv(name) - if value is None: - return default - return value.strip().lower() in {"1", "true", "yes", "on"} - - def _resolve_service_path(service: str, path: str) -> str: normalized = path.lstrip("/") prefix = SERVICE_PATH_PREFIXES.get(service, "").strip("/") @@ -188,6 +312,35 @@ def _resolve_service_path(service: str, path: str) -> str: return f"{prefix}/{normalized}" if normalized else prefix +@app.middleware("http") +async def error_notification_middleware(request: Request, call_next): + try: + response = await call_next(request) + except Exception as exc: + status_code = 500 + if isinstance(exc, HTTPException): + status_code = exc.status_code + if status_code >= _error_alert_min_status(): + _queue_telegram_error_alert( + service=_error_alert_service_name(), + method=request.method, + path=str(request.url.path), + status=status_code, + detail=f"{type(exc).__name__}: {exc}", + ) + raise + + if response.status_code >= _error_alert_min_status(): + _queue_telegram_error_alert( + service=_error_alert_service_name(), + method=request.method, + path=str(request.url.path), + status=response.status_code, + detail=f"HTTP {response.status_code} from {request.method} {request.url.path}", + ) + return response + + @app.get("/health", response_model=HealthResponse) def health() -> HealthResponse: return HealthResponse(status="ok", service="api-gateway") @@ -414,6 +567,13 @@ async def _forward(method: str, service: str, path: str, request: Request) -> Re headers=headers, ) except httpx.RequestError: + _queue_telegram_error_alert( + service=_error_alert_service_name(), + method=method, + path=f"/proxy/{service}/{path}", + status=502, + detail=f"Upstream service unavailable: service={service} base_url={base}", + ) logger.warning( "Upstream service unavailable: service=%s base_url=%s method=%s path=%s", service, diff --git a/tests/test_gateway_proxy.py b/tests/test_gateway_proxy.py index 775329e..74d9388 100644 --- a/tests/test_gateway_proxy.py +++ b/tests/test_gateway_proxy.py @@ -209,3 +209,101 @@ def test_gateway_returns_bad_gateway_when_upstream_unavailable(monkeypatch): 'detail': 'Upstream service unavailable', 'service': 'sales', } + + +def test_gateway_queues_error_alert_for_proxy_unreachable(monkeypatch): + async def request(self, method, url, params=None, content=None, headers=None): # noqa: ANN001, ANN201 + raise gateway_module.httpx.ConnectError("All connection attempts failed") + + alerts: list[dict] = [] + + def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001 + alerts.append({ + 'service': service, + 'method': method, + 'path': path, + 'status': status, + 'detail': detail, + }) + + monkeypatch.setattr(DummyAsyncClient, 'request', request) + monkeypatch.setattr(gateway_module.httpx, 'AsyncClient', DummyAsyncClient) + monkeypatch.setattr(gateway_module, '_queue_telegram_error_alert', queue_alert) + monkeypatch.setenv('ERROR_TELEGRAM_ALERTS', '1') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_BOT_TOKEN', 'token') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_CHAT_IDS', '999') + client = TestClient(gateway_module.app) + + response = client.get('/proxy/sales/pipelines') + + assert response.status_code == 502 + assert alerts == [ + { + 'service': 'api-gateway', + 'method': 'GET', + 'path': '/proxy/sales/pipelines', + 'status': 502, + 'detail': 'Upstream service unavailable: service=sales base_url=http://localhost:8020', + }, + ] + + +def test_gateway_queues_error_alert_for_channel_alias(monkeypatch): + async def request(self, method, url, params=None, content=None, headers=None): # noqa: ANN001, ANN201 + raise gateway_module.httpx.ConnectError("All connection attempts failed") + + alerts: list[dict] = [] + + def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001 + alerts.append({ + 'service': service, + 'method': method, + 'path': path, + 'status': status, + 'detail': detail, + }) + + monkeypatch.setattr(DummyAsyncClient, 'request', request) + monkeypatch.setattr(gateway_module.httpx, 'AsyncClient', DummyAsyncClient) + monkeypatch.setattr(gateway_module, '_queue_telegram_error_alert', queue_alert) + monkeypatch.setenv('ERROR_TELEGRAM_ALERTS', '1') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_BOT_TOKEN', 'token') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_CHANNEL_ID', '@my_alert_channel') + client = TestClient(gateway_module.app) + + client.get('/proxy/sales/pipelines') + + assert alerts + assert alerts[0]['status'] == 502 + + +def test_gateway_queues_error_alert_for_internal_exception(monkeypatch): + async def broken_forward(method: str, service: str, path: str, request: object) -> None: # noqa: ANN001, ANN201 + raise ValueError('proxy exploded') + + alerts: list[dict] = [] + + def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001 + alerts.append({ + 'service': service, + 'method': method, + 'path': path, + 'status': status, + 'detail': detail, + }) + + monkeypatch.setattr(gateway_module, '_forward', broken_forward) + monkeypatch.setattr(gateway_module, '_queue_telegram_error_alert', queue_alert) + monkeypatch.setenv('ERROR_TELEGRAM_ALERTS', '1') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_BOT_TOKEN', 'token') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_CHAT_IDS', '999') + + client = TestClient(gateway_module.app, raise_server_exceptions=False) + + response = client.get('/proxy/sales/pipelines') + + assert response.status_code == 500 + assert alerts + assert alerts[0]['service'] == 'api-gateway' + assert alerts[0]['status'] == 500 + assert alerts[0]['path'] == '/proxy/sales/pipelines'