digiops/call-center
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enrich Telegram error alerts with request and upstream context

Browse Source
This commit is contained in:
Magzhan Zhumabayev
2026-05-11 13:52:13 +05:00
parent bf75a005c2
commit 53f9845195
2 changed files with 284 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/app.py
+204 -10
View File
@@ -6,6 +6,7 @@ import os
from pathlib import Path
from datetime import datetime, timezone
from typing import Any
from urllib.parse import parse_qsl
import httpx
from fastapi import FastAPI, HTTPException, Request
@@ -17,6 +18,9 @@ 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] = {}
_ERROR_ALERT_MAX_MESSAGE_LENGTH = 3800
_ERROR_ALERT_HEADER_PREVIEW_LIMIT = 120
_ERROR_ALERT_BODY_PREVIEW_LIMIT = 1200
def _env_flag(name: str, default: bool = False) -> bool:
@@ -102,6 +106,168 @@ def _error_alert_key(service: str, status: int, path: str, method: str) -> str:
return f"{service}:{status}:{method}:{path}"
def _error_alert_mask_value(value: str) -> str:
text = value.strip()
if len(text) <= 6:
return "<hidden>"
return f"{text[:3]}...{text[-2:]}"
def _error_alert_headers_preview(headers: Any) -> list[str]:
sensitive = {
"authorization",
"x-telegram-bot-api-secret-token",
"x-whatsapp-webhook-secret",
"x-hub-signature-256",
"cookie",
"set-cookie",
"proxy-authorization",
"x-api-key",
"api-key",
"x-auth-token",
"x-service-token",
}
result: list[str] = []
for key, value in headers.items():
key_lower = key.lower()
if key_lower in sensitive:
value = _error_alert_mask_value(value)
result.append(f"{key}: {value}")
return result[:_ERROR_ALERT_HEADER_PREVIEW_LIMIT]
def _error_alert_mask_query(query: str) -> str:
if not query:
return "-"
sensitive = {
"token",
"access_token",
"refresh_token",
"authorization",
"password",
"secret",
"api_key",
"apikey",
"x_api_key",
"x-telegram-bot-api-secret-token",
"x-whatsapp-webhook-secret",
"signature",
}
parsed = parse_qsl(query, keep_blank_values=True)
if not parsed:
return query
values = []
for key, value in parsed:
key_lower = key.lower()
if key_lower in sensitive:
values.append(f"{key}=<hidden>")
else:
values.append(f"{key}={value}")
return "&".join(values)
def _error_alert_request_context(request: Request) -> dict[str, Any]:
query = request.url.query
client = request.client
client_addr = f"{client.host}:{client.port}" if client else "unknown"
return {
"client": client_addr,
"http_version": request.scope.get("http_version", "-"),
"query": _error_alert_mask_query(query),
"user_agent": request.headers.get("user-agent", "-"),
"referer": request.headers.get("referer", "-"),
"host": request.headers.get("host", "-"),
"request_id": request.headers.get("x-request-id")
or request.headers.get("x-correlation-id")
or request.headers.get("x-amzn-trace-id")
or "-",
"x_user": request.headers.get("x-user", "-"),
"x_role": request.headers.get("x-role", "-"),
"x_tenant_id": request.headers.get("x-tenant-id", "-"),
"x_forwarded_for": request.headers.get("x-forwarded-for", "-"),
"headers_count": len(request.headers),
"headers": _error_alert_headers_preview(request.headers),
}
def _error_alert_message_payload(
*,
service: str,
method: str,
path: str,
status: int,
detail: str,
request: Request,
extra: dict[str, Any] | None = None,
) -> str:
timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
request_context = _error_alert_request_context(request)
lines = [
f"🚨 [{service}] {status}",
f"time: {timestamp}",
f"method: {method}",
f"path: {path}",
f"service: {service}",
f"client: {request_context['client']}",
f"http_version: {request_context['http_version']}",
f"query: {request_context['query']}",
f"request_id: {request_context['request_id']}",
f"user_agent: {request_context['user_agent']}",
f"referer: {request_context['referer']}",
f"host: {request_context['host']}",
f"x_user: {request_context['x_user']}",
f"x_role: {request_context['x_role']}",
f"x_tenant_id: {request_context['x_tenant_id']}",
f"x_forwarded_for: {request_context['x_forwarded_for']}",
"",
"headers:",
]
lines.extend(f" {item}" for item in request_context["headers"])
if extra:
lines.extend(("", "extra:"))
for key, value in extra.items():
lines.append(f" {key}: {value}")
lines.append("")
lines.append(f"detail: {detail}")
full_message = "\n".join(lines)
if len(full_message) <= _ERROR_ALERT_MAX_MESSAGE_LENGTH:
return full_message
cutoff = _ERROR_ALERT_MAX_MESSAGE_LENGTH - 80
if cutoff < 100:
return full_message[:_ERROR_ALERT_MAX_MESSAGE_LENGTH]
return f"{full_message[:cutoff]}\n... (truncated)"
def _error_alert_body_preview(payload: bytes | None) -> str:
if payload is None:
return "-"
if len(payload) > _ERROR_ALERT_BODY_PREVIEW_LIMIT:
payload = payload[:_ERROR_ALERT_BODY_PREVIEW_LIMIT]
truncated = True
else:
truncated = False
try:
text = payload.decode("utf-8")
except UnicodeDecodeError:
text = payload.decode("utf-8", errors="replace")
if truncated:
return f"{text}\n... (truncated body preview)"
return text
def _error_alert_is_throttled(key: str) -> bool:
now = datetime.now(tz=timezone.utc).timestamp()
cooldown = _error_alert_cooldown_seconds()
@@ -136,7 +302,15 @@ async def _send_telegram_alert(message: str) -> None:
)
def _queue_telegram_error_alert(service: str, method: str, path: str, status: int, detail: str) -> None:
def _queue_telegram_error_alert(
service: str,
method: str,
path: str,
status: int,
detail: str,
request: Request | None = None,
extra: dict[str, Any] | None = None,
) -> None:
if not _error_alert_enabled():
return
if not _error_alert_should_alert_path(path):
@@ -147,13 +321,17 @@ def _queue_telegram_error_alert(service: str, method: str, path: str, status: in
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}"
if request is None:
return
message = _error_alert_message_payload(
service=service,
method=method,
path=path,
status=status,
detail=detail,
request=request,
extra=extra,
)
try:
@@ -344,6 +522,7 @@ async def error_notification_middleware(request: Request, call_next):
path=str(request.url.path),
status=status_code,
detail=f"{type(exc).__name__}: {exc}",
request=request,
)
raise
@@ -354,6 +533,11 @@ async def error_notification_middleware(request: Request, call_next):
path=str(request.url.path),
status=response.status_code,
detail=f"HTTP {response.status_code} from {request.method} {request.url.path}",
request=request,
extra={
"response_content_type": response.headers.get("content-type", "-"),
"response_content_length": response.headers.get("content-length", "-"),
},
)
return response
@@ -574,6 +758,7 @@ async def _forward(method: str, service: str, path: str, request: Request) -> Re
if payload == b"":
payload = None
body_preview = "-"
try:
async with httpx.AsyncClient(timeout=20) as client:
resp = await client.request(
@@ -583,13 +768,22 @@ async def _forward(method: str, service: str, path: str, request: Request) -> Re
content=payload,
headers=headers,
)
except httpx.RequestError:
except httpx.RequestError as exc:
body_preview = _error_alert_body_preview(payload)
_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}",
detail=f"{type(exc).__name__}: {exc}",
request=request,
extra={
"upstream_service": service,
"upstream_base_url": base,
"upstream_target_path": target_path,
"upstream_url": url,
"request_body_preview": body_preview,
},
)
logger.warning(
"Upstream service unavailable: service=%s base_url=%s method=%s path=%s",
tests/test_gateway_proxy.py
+80 -12
View File
@@ -217,13 +217,21 @@ def test_gateway_queues_error_alert_for_proxy_unreachable(monkeypatch):
alerts: list[dict] = []
def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001
def queue_alert(
service: str,
method: str,
path: str,
status: int,
detail: str,
**kwargs: object,
) -> None: # noqa: ARG001
alerts.append({
'service': service,
'method': method,
'path': path,
'status': status,
'detail': detail,
'kwargs': kwargs,
})
monkeypatch.setattr(DummyAsyncClient, 'request', request)
@@ -237,15 +245,11 @@ def test_gateway_queues_error_alert_for_proxy_unreachable(monkeypatch):
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',
},
]
assert alerts[0]['service'] == 'api-gateway'
assert alerts[0]['method'] == 'GET'
assert alerts[0]['path'] == '/proxy/sales/pipelines'
assert alerts[0]['status'] == 502
assert 'All connection attempts failed' in alerts[0]['detail']
def test_gateway_queues_error_alert_for_channel_alias(monkeypatch):
@@ -254,13 +258,21 @@ def test_gateway_queues_error_alert_for_channel_alias(monkeypatch):
alerts: list[dict] = []
def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001
def queue_alert(
service: str,
method: str,
path: str,
status: int,
detail: str,
**kwargs: object,
) -> None: # noqa: ARG001
alerts.append({
'service': service,
'method': method,
'path': path,
'status': status,
'detail': detail,
'kwargs': kwargs,
})
monkeypatch.setattr(DummyAsyncClient, 'request', request)
@@ -283,13 +295,21 @@ def test_gateway_queues_error_alert_for_internal_exception(monkeypatch):
alerts: list[dict] = []
def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001
def queue_alert(
service: str,
method: str,
path: str,
status: int,
detail: str,
**kwargs: object,
) -> None: # noqa: ARG001
alerts.append({
'service': service,
'method': method,
'path': path,
'status': status,
'detail': detail,
'kwargs': kwargs,
})
monkeypatch.setattr(gateway_module, '_forward', broken_forward)
@@ -309,6 +329,54 @@ def test_gateway_queues_error_alert_for_internal_exception(monkeypatch):
assert alerts[0]['path'] == '/proxy/sales/pipelines'
def test_gateway_queues_error_alert_with_body_and_extra_context(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,
request=None,
extra=None,
**kwargs: object,
) -> None: # noqa: ARG001
alerts.append({
'service': service,
'method': method,
'path': path,
'status': status,
'detail': detail,
'request': request,
'extra': extra,
})
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.post(
'/proxy/sales/pipelines?name=demo&token=super-secret',
headers={'Authorization': 'Bearer secret-token', 'X-Request-Id': 'req-123'},
json={'foo': 'bar'},
)
assert response.status_code == 502
assert alerts[0]['status'] == 502
assert alerts[0]['extra']['upstream_service'] == 'sales'
assert alerts[0]['extra']['upstream_target_path'] == 'api/v1/pipelines'
assert alerts[0]['extra']['request_body_preview'] == '{"foo":"bar"}'
assert alerts[0]['request'] is not None
def test_gateway_normalizes_channel_alias_for_t_me(monkeypatch):
monkeypatch.setenv('ERROR_TELEGRAM_ALERT_CHANNEL_ID', 't.me/konturaitelecom')