M5: DLQ tooling - RabbitMQ management client, admin list/peek, requeue and purge (019-021)
This commit is contained in:
febux
parent
befa5c897f
commit
c00f3cf9f9
9 changed files with 980 additions and 0 deletions
256
src/contract_check/api/admin/mq.py
Normal file
256
src/contract_check/api/admin/mq.py
Normal file
|
|
@ -0,0 +1,256 @@
|
|||
"""DLQ admin panel: list, peek, requeue, and purge dead-letter queues.
|
||||
|
||||
All management API calls go through :class:`RabbitMQManagementClient` so the
|
||||
admin surface is broker-agnostic and mockable at the HTTP seam.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
|
||||
from src.contract_check.api.admin.auth import AdminUser, HtmxGuard, require_admin
|
||||
from src.contract_check.api.admin.templating import templates
|
||||
from src.contract_check.core.logging import get_logger
|
||||
from src.contract_check.core.mq.management import (
|
||||
MqManagementDisabledError,
|
||||
MqManagementError,
|
||||
RabbitMQManagementClient,
|
||||
)
|
||||
from src.contract_check.core.mq.topology import DLQ_FOR
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/admin/mq",
|
||||
tags=["admin-mq"],
|
||||
dependencies=[Depends(require_admin)],
|
||||
)
|
||||
|
||||
_PREVIEW_LIMIT = 20
|
||||
|
||||
|
||||
def _client() -> RabbitMQManagementClient:
|
||||
return RabbitMQManagementClient()
|
||||
|
||||
|
||||
def _attach_toast(response: RedirectResponse, toast: str) -> None:
|
||||
import json
|
||||
|
||||
response.headers["HX-Trigger"] = json.dumps({"showToast": toast})
|
||||
|
||||
|
||||
def _safe_toast(toast: str) -> str:
|
||||
# Keep toast text short and URL-safe for redirects.
|
||||
return toast[:200]
|
||||
|
||||
|
||||
@router.get("/queues", response_class=HTMLResponse, include_in_schema=False)
|
||||
async def list_queues(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_admin)],
|
||||
) -> HTMLResponse:
|
||||
"""List all DLQs with live depth and status."""
|
||||
client = _client()
|
||||
dlq_names = sorted(DLQ_FOR.values())
|
||||
rows: list[dict[str, object]] = []
|
||||
error: str | None = None
|
||||
try:
|
||||
await client.connect()
|
||||
for name in dlq_names:
|
||||
try:
|
||||
info = await client.get_queue_info(name)
|
||||
rows.append({"name": info.name, "messages": info.messages, "state": info.state})
|
||||
except MqManagementError as exc:
|
||||
log.warning("dlq_info_failed", queue=name, error=str(exc))
|
||||
rows.append({"name": name, "messages": None, "state": "unknown"})
|
||||
except MqManagementDisabledError:
|
||||
error = "Управление RabbitMQ отключено в конфигурации."
|
||||
except MqManagementError as exc:
|
||||
error = f"Управление RabbitMQ недоступно: {exc}"
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"mq_queues.html",
|
||||
{
|
||||
"request": request,
|
||||
"rows": rows,
|
||||
"error": error,
|
||||
"total_messages": sum(
|
||||
int(r["messages"]) if isinstance(r["messages"], int) else 0 for r in rows
|
||||
),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/dlq/{queue_name}", response_class=HTMLResponse, include_in_schema=False)
|
||||
async def dlq_detail(
|
||||
request: Request,
|
||||
queue_name: str,
|
||||
admin: Annotated[AdminUser, Depends(require_admin)],
|
||||
) -> HTMLResponse:
|
||||
"""Peek recent messages from one DLQ."""
|
||||
if queue_name not in DLQ_FOR.values():
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ")
|
||||
|
||||
client = _client()
|
||||
messages: list[dict[str, object]] = []
|
||||
error: str | None = None
|
||||
info: dict[str, object] = {"name": queue_name, "messages": 0, "state": "unknown"}
|
||||
try:
|
||||
await client.connect()
|
||||
try:
|
||||
qi = await client.get_queue_info(queue_name)
|
||||
info = {"name": qi.name, "messages": qi.messages, "state": qi.state}
|
||||
except MqManagementError as exc:
|
||||
log.warning("dlq_info_failed", queue=queue_name, error=str(exc))
|
||||
error = f"Не удалось получить состояние очереди: {exc}"
|
||||
try:
|
||||
raw = await client.peek_messages(queue_name, _PREVIEW_LIMIT)
|
||||
for m in raw:
|
||||
payload_text = m.payload.decode("utf-8", errors="replace")
|
||||
messages.append(
|
||||
{
|
||||
"correlation_id": m.correlation_id or "—",
|
||||
"failure_class": m.failure_class,
|
||||
"failure_error": (m.failure_error or "")[:240],
|
||||
"payload_summary": (
|
||||
payload_text[:240] + "…" if len(payload_text) > 240 else payload_text
|
||||
),
|
||||
"headers": m.headers,
|
||||
}
|
||||
)
|
||||
except MqManagementError as exc:
|
||||
log.warning("dlq_peek_failed", queue=queue_name, error=str(exc))
|
||||
if error is None:
|
||||
error = f"Не удалось прочитать сообщения: {exc}"
|
||||
except MqManagementDisabledError:
|
||||
error = "Управление RabbitMQ отключено в конфигурации."
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"mq_dlq_detail.html",
|
||||
{
|
||||
"request": request,
|
||||
"queue": info,
|
||||
"messages": messages,
|
||||
"error": error,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post("/dlq/{queue_name}/requeue", response_class=HTMLResponse, include_in_schema=False)
|
||||
async def requeue_dlq(
|
||||
request: Request,
|
||||
queue_name: str,
|
||||
admin: Annotated[AdminUser, Depends(require_admin)],
|
||||
_: Annotated[None, HtmxGuard],
|
||||
) -> RedirectResponse:
|
||||
"""Move a bounded batch from the DLQ back to its main exchange."""
|
||||
if queue_name not in DLQ_FOR.values():
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ")
|
||||
|
||||
client = _client()
|
||||
try:
|
||||
await client.connect()
|
||||
result = await client.requeue_batch(queue_name)
|
||||
except MqManagementError as exc:
|
||||
log.warning(
|
||||
"dlq_requeue_failed", queue=queue_name, error=str(exc), admin_id=str(admin.user_id)
|
||||
)
|
||||
return RedirectResponse(
|
||||
url=f"/admin/mq/dlq/{queue_name}?toast={_safe_toast(str(exc))}",
|
||||
status_code=status.HTTP_303_SEE_OTHER,
|
||||
)
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
toast = f"Возвращено в очередь: {result.messages_requeued}"
|
||||
if result.errors:
|
||||
toast += f", ошибок: {result.errors}"
|
||||
log.info(
|
||||
"admin_dlq_requeue",
|
||||
queue=queue_name,
|
||||
admin_user_id=str(admin.user_id),
|
||||
requeued=result.messages_requeued,
|
||||
errors=result.errors,
|
||||
)
|
||||
response = RedirectResponse(
|
||||
url=f"/admin/mq/dlq/{queue_name}", status_code=status.HTTP_303_SEE_OTHER
|
||||
)
|
||||
_attach_toast(response, toast)
|
||||
return response
|
||||
|
||||
|
||||
@router.get(
|
||||
"/dlq/{queue_name}/purge/confirm",
|
||||
response_class=HTMLResponse,
|
||||
include_in_schema=False,
|
||||
)
|
||||
async def purge_confirm(
|
||||
request: Request,
|
||||
queue_name: str,
|
||||
admin: Annotated[AdminUser, Depends(require_admin)],
|
||||
) -> HTMLResponse:
|
||||
"""Show typed-confirmation page before purge."""
|
||||
if queue_name not in DLQ_FOR.values():
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ")
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"mq_purge_confirm.html",
|
||||
{"request": request, "queue_name": queue_name},
|
||||
)
|
||||
|
||||
|
||||
@router.post("/dlq/{queue_name}/purge", response_class=HTMLResponse, include_in_schema=False)
|
||||
async def purge_dlq(
|
||||
request: Request,
|
||||
queue_name: str,
|
||||
admin: Annotated[AdminUser, Depends(require_admin)],
|
||||
_: Annotated[None, HtmxGuard],
|
||||
confirmation: Annotated[str, Form()] = "",
|
||||
) -> RedirectResponse:
|
||||
"""Purge a DLQ only after explicit typed confirmation."""
|
||||
if queue_name not in DLQ_FOR.values():
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ")
|
||||
|
||||
expected = f"purge {queue_name}"
|
||||
if confirmation.strip().lower() != expected.lower():
|
||||
return RedirectResponse(
|
||||
url=f"/admin/mq/dlq/{queue_name}/purge/confirm?toast=Неверное+подтверждение.+Введите+%27{expected}%27",
|
||||
status_code=status.HTTP_303_SEE_OTHER,
|
||||
)
|
||||
|
||||
client = _client()
|
||||
try:
|
||||
await client.connect()
|
||||
result = await client.purge_queue(queue_name)
|
||||
except MqManagementError as exc:
|
||||
log.warning(
|
||||
"dlq_purge_failed", queue=queue_name, error=str(exc), admin_id=str(admin.user_id)
|
||||
)
|
||||
return RedirectResponse(
|
||||
url=f"/admin/mq/dlq/{queue_name}?toast={_safe_toast(str(exc))}",
|
||||
status_code=status.HTTP_303_SEE_OTHER,
|
||||
)
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
toast = f"DLQ {queue_name} очищена: удалено {result.messages_removed} сообщений"
|
||||
log.info(
|
||||
"admin_dlq_purge",
|
||||
queue=queue_name,
|
||||
admin_user_id=str(admin.user_id),
|
||||
removed=result.messages_removed,
|
||||
)
|
||||
response = RedirectResponse(
|
||||
url=f"/admin/mq/dlq/{queue_name}", status_code=status.HTTP_303_SEE_OTHER
|
||||
)
|
||||
_attach_toast(response, toast)
|
||||
return response
|
||||
|
|
@ -14,6 +14,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse
|
|||
|
||||
from src.contract_check.api.admin.auth import ADMIN_COOKIE, ADMIN_COOKIE_MAX_AGE, resolve_admin
|
||||
from src.contract_check.api.admin.billing import router as billing_router
|
||||
from src.contract_check.api.admin.mq import router as mq_router
|
||||
from src.contract_check.api.admin.review import router as review_router
|
||||
from src.contract_check.api.admin.templating import templates
|
||||
from src.contract_check.api.admin.users import router as users_router
|
||||
|
|
@ -30,6 +31,7 @@ router = APIRouter(tags=["admin"])
|
|||
router.include_router(users_router)
|
||||
router.include_router(billing_router)
|
||||
router.include_router(review_router)
|
||||
router.include_router(mq_router)
|
||||
|
||||
|
||||
@router.get("/admin", include_in_schema=False)
|
||||
|
|
|
|||
|
|
@ -71,6 +71,7 @@
|
|||
<a href="/admin/users" class="{{ 'active' if request.url.path.startswith('/admin/users') else '' }}">Пользователи</a>
|
||||
<a href="/admin/invoices" class="{{ 'active' if request.url.path.startswith('/admin/invoices') else '' }}">Счета</a>
|
||||
<a href="/admin/review" class="{{ 'active' if request.url.path.startswith('/admin/review') else '' }}">Ручная проверка</a>
|
||||
<a href="/admin/mq/queues" class="{{ 'active' if request.url.path.startswith('/admin/mq') else '' }}">DLQ</a>
|
||||
</nav>
|
||||
<span class="spacer"></span>
|
||||
<form method="post" action="/admin/logout"><button class="btn ghost sm" type="submit">Выйти</button></form>
|
||||
|
|
|
|||
47
src/contract_check/api/admin/templates/mq_dlq_detail.html
Normal file
47
src/contract_check/api/admin/templates/mq_dlq_detail.html
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}DLQ {{ queue.name }} — Админка{% endblock %}
|
||||
{% block content %}
|
||||
<div class="toolbar">
|
||||
<a class="btn ghost sm" href="/admin/mq/queues">← Назад к списку</a>
|
||||
<span class="spacer"></span>
|
||||
<span class="muted">{{ queue.name }} · {{ queue.messages }} сообщений · {{ queue.state }}</span>
|
||||
</div>
|
||||
|
||||
{% if error %}
|
||||
<div class="card" style="border-color:var(--bad)">
|
||||
<p style="margin:0;color:var(--bad)">{{ error }}</p>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<div class="toolbar" style="margin-top:12px">
|
||||
<form method="post" action="/admin/mq/dlq/{{ queue.name }}/requeue" style="display:inline">
|
||||
<button class="btn" type="submit" hx-post="/admin/mq/dlq/{{ queue.name }}/requeue" hx-confirm="Вернуть пакет сообщений в основную очередь?">Вернуть в очередь</button>
|
||||
</form>
|
||||
<a class="btn danger" href="/admin/mq/dlq/{{ queue.name }}/purge/confirm">Очистить DLQ</a>
|
||||
</div>
|
||||
|
||||
<div class="card">
|
||||
<table>
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Correlation ID</th>
|
||||
<th>Класс ошибки</th>
|
||||
<th>Сокращённая ошибка</th>
|
||||
<th>Сводка payload</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for m in messages %}
|
||||
<tr>
|
||||
<td><code>{{ m.correlation_id }}</code></td>
|
||||
<td><span class="pill">{{ m.failure_class }}</span></td>
|
||||
<td class="muted">{{ m.failure_error or '—' }}</td>
|
||||
<td class="muted"><code>{{ m.payload_summary }}</code></td>
|
||||
</tr>
|
||||
{% else %}
|
||||
<tr><td colspan="4"><div class="empty">Очередь пуста</div></td></tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
{% endblock %}
|
||||
17
src/contract_check/api/admin/templates/mq_purge_confirm.html
Normal file
17
src/contract_check/api/admin/templates/mq_purge_confirm.html
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}Очистка {{ queue_name }} — Админка{% endblock %}
|
||||
{% block content %}
|
||||
<div class="toolbar">
|
||||
<a class="btn ghost sm" href="/admin/mq/dlq/{{ queue_name }}">← Назад</a>
|
||||
</div>
|
||||
|
||||
<div class="card" style="border-color:var(--bad)">
|
||||
<h2 style="margin-top:0;color:var(--bad)">Очистить {{ queue_name }}?</h2>
|
||||
<p>Это безвозвратно удалит все сообщения из DLQ. Для подтверждения введите <code>purge {{ queue_name }}</code>.</p>
|
||||
|
||||
<form method="post" action="/admin/mq/dlq/{{ queue_name }}/purge">
|
||||
<input name="confirmation" placeholder="purge {{ queue_name }}" style="width:320px">
|
||||
<button class="btn danger" type="submit" hx-post="/admin/mq/dlq/{{ queue_name }}/purge" style="margin-top:10px">Подтвердить очистку</button>
|
||||
</form>
|
||||
</div>
|
||||
{% endblock %}
|
||||
46
src/contract_check/api/admin/templates/mq_queues.html
Normal file
46
src/contract_check/api/admin/templates/mq_queues.html
Normal file
|
|
@ -0,0 +1,46 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}DLQ — Админка{% endblock %}
|
||||
{% block content %}
|
||||
<div class="card">
|
||||
<div class="toolbar">
|
||||
<h2 style="margin:0">Очереди DLQ <span class="muted" style="font-weight:400;font-size:14px">({{ total_messages }})</span></h2>
|
||||
</div>
|
||||
|
||||
{% if error %}
|
||||
<div class="card" style="border-color:var(--bad)">
|
||||
<p style="margin:0;color:var(--bad)">{{ error }}</p>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<table>
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Очередь</th>
|
||||
<th>Сообщений</th>
|
||||
<th>Состояние</th>
|
||||
<th></th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for row in rows %}
|
||||
<tr>
|
||||
<td><code>{{ row.name }}</code></td>
|
||||
<td>
|
||||
{% if row.messages is none %}
|
||||
<span class="muted">—</span>
|
||||
{% elif row.messages == 0 %}
|
||||
<span class="pill">0</span>
|
||||
{% else %}
|
||||
<span class="pill warn">{{ row.messages }}</span>
|
||||
{% endif %}
|
||||
</td>
|
||||
<td class="muted">{{ row.state }}</td>
|
||||
<td><a class="btn ghost sm" href="/admin/mq/dlq/{{ row.name }}">Просмотр</a></td>
|
||||
</tr>
|
||||
{% else %}
|
||||
<tr><td colspan="4"><div class="empty">Нет DLQ</div></td></tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
{% endblock %}
|
||||
|
|
@ -44,8 +44,12 @@ class Settings(BaseSettings):
|
|||
mq_prefetch_extract: int = 1
|
||||
mq_prefetch_analyze: int = 3
|
||||
mq_prefetch_notify: int = 5
|
||||
mq_prefetch_prescreen: int = 1
|
||||
mq_max_attempts: int = 5
|
||||
mq_retry_base_ms: int = 2000
|
||||
mq_management_enabled: bool = True
|
||||
mq_management_timeout: float = 10.0
|
||||
mq_requeue_batch_size: int = 50
|
||||
|
||||
# --- object storage (MinIO) ---
|
||||
s3_endpoint_url: str
|
||||
|
|
|
|||
365
src/contract_check/core/mq/management.py
Normal file
365
src/contract_check/core/mq/management.py
Normal file
|
|
@ -0,0 +1,365 @@
|
|||
"""Async RabbitMQ management API client.
|
||||
|
||||
Derives URL and credentials from the broker AMQP URL, exactly like the
|
||||
lazy-queue policy setup in :mod:`src.contract_check.core.mq.topology`.
|
||||
|
||||
All operations return small typed dataclasses. Network/management-plugin
|
||||
failures are mapped to :class:`MqManagementError` subclasses so callers
|
||||
can render a clean error state instead of a traceback.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
|
||||
from src.contract_check.core.config import get_settings
|
||||
from src.contract_check.core.http_logging import http_log_event_hooks
|
||||
from src.contract_check.core.logging import get_logger
|
||||
from src.contract_check.core.mq.topology import (
|
||||
DLQ_FOR,
|
||||
EXCHANGE_MAIN,
|
||||
EXCHANGE_NOTIFY,
|
||||
H_ATTEMPT,
|
||||
H_CORRELATION_ID,
|
||||
_amqp_credentials,
|
||||
_management_url_from_amqp,
|
||||
)
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
||||
|
||||
class MqManagementError(Exception):
|
||||
"""Base for RabbitMQ management API errors."""
|
||||
|
||||
|
||||
class MqManagementDisabledError(MqManagementError):
|
||||
"""Raised when the management client is disabled by configuration."""
|
||||
|
||||
|
||||
class MqManagementUnreachableError(MqManagementError):
|
||||
"""Raised when the management plugin is unreachable or disabled."""
|
||||
|
||||
|
||||
class MqManagementNotFoundError(MqManagementError):
|
||||
"""Raised when the management API returns 404 (queue not found)."""
|
||||
|
||||
|
||||
class MqManagementResponseError(MqManagementError):
|
||||
"""Raised for unexpected management API status codes."""
|
||||
|
||||
def __init__(self, message: str, *, status_code: int, body: str = "") -> None:
|
||||
super().__init__(message)
|
||||
self.status_code = status_code
|
||||
self.body = body
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class QueueInfo:
|
||||
"""Minimal queue metadata returned by the management API."""
|
||||
|
||||
name: str
|
||||
messages: int
|
||||
state: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class QueueMessage:
|
||||
"""One peeked message from a queue, with its headers and payload summary."""
|
||||
|
||||
payload: bytes
|
||||
headers: dict[str, Any]
|
||||
correlation_id: str | None
|
||||
routing_key: str | None
|
||||
|
||||
@property
|
||||
def failure_class(self) -> str:
|
||||
return str(self.headers.get("x-failure-class", "unknown"))
|
||||
|
||||
@property
|
||||
def failure_error(self) -> str:
|
||||
return str(self.headers.get("x-failure-error", ""))
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class PurgeResult:
|
||||
messages_removed: int
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RequeueResult:
|
||||
messages_requeued: int
|
||||
errors: int
|
||||
|
||||
|
||||
def _main_exchange_and_routing_key(queue_name: str) -> tuple[str, str]:
|
||||
"""Return the exchange and routing key to publish a requeued message to."""
|
||||
if queue_name == "notify.q":
|
||||
return EXCHANGE_NOTIFY, "notify"
|
||||
return EXCHANGE_MAIN, queue_name
|
||||
|
||||
|
||||
class RabbitMQManagementClient:
|
||||
"""Small async client for the RabbitMQ management HTTP API."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
amqp_url: str | None = None,
|
||||
*,
|
||||
timeout: float | None = None,
|
||||
) -> None:
|
||||
settings = get_settings()
|
||||
self._enabled = settings.mq_management_enabled
|
||||
self._amqp_url = amqp_url or settings.rabbitmq_url
|
||||
self._timeout = timeout or settings.mq_management_timeout
|
||||
self._vhost = quote("/", safe="")
|
||||
self._client: httpx.AsyncClient | None = None
|
||||
|
||||
async def __aenter__(self) -> RabbitMQManagementClient:
|
||||
await self.connect()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc: object) -> None:
|
||||
await self.aclose()
|
||||
|
||||
def _ensure_enabled(self) -> None:
|
||||
if not self._enabled:
|
||||
raise MqManagementDisabledError("RabbitMQ management client is disabled")
|
||||
|
||||
def _management_url(self) -> str:
|
||||
url = _management_url_from_amqp(self._amqp_url)
|
||||
if not url:
|
||||
raise MqManagementUnreachableError("cannot derive management URL from broker URL")
|
||||
return url
|
||||
|
||||
def _auth(self) -> tuple[str, str]:
|
||||
creds = _amqp_credentials(self._amqp_url)
|
||||
if not creds:
|
||||
raise MqManagementUnreachableError(
|
||||
"cannot derive management credentials from broker URL"
|
||||
)
|
||||
return creds
|
||||
|
||||
async def connect(self) -> None:
|
||||
self._ensure_enabled()
|
||||
if self._client is not None:
|
||||
return
|
||||
mgmt_url = self._management_url()
|
||||
username, password = self._auth()
|
||||
self._client = httpx.AsyncClient(
|
||||
base_url=mgmt_url,
|
||||
auth=(username, password),
|
||||
timeout=httpx.Timeout(self._timeout, connect=5.0),
|
||||
event_hooks=http_log_event_hooks(service="rabbitmq_mgmt"),
|
||||
)
|
||||
|
||||
async def aclose(self) -> None:
|
||||
if self._client is not None:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
|
||||
@property
|
||||
def client(self) -> httpx.AsyncClient:
|
||||
if self._client is None:
|
||||
raise RuntimeError("RabbitMQManagementClient not connected")
|
||||
return self._client
|
||||
|
||||
async def get_queue_info(self, queue_name: str) -> QueueInfo:
|
||||
"""Return name, depth, and state for a queue."""
|
||||
self._ensure_enabled()
|
||||
url = f"/api/queues/{self._vhost}/{quote(queue_name, safe='')}"
|
||||
try:
|
||||
resp = await self.client.get(url)
|
||||
except httpx.NetworkError as exc:
|
||||
raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc
|
||||
except httpx.TimeoutException as exc:
|
||||
raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc
|
||||
if resp.status_code == 404:
|
||||
raise MqManagementNotFoundError(f"queue {queue_name!r} not found")
|
||||
if resp.status_code >= 400:
|
||||
raise MqManagementResponseError(
|
||||
f"management API error {resp.status_code}",
|
||||
status_code=resp.status_code,
|
||||
body=resp.text[:500],
|
||||
)
|
||||
data = resp.json()
|
||||
return QueueInfo(
|
||||
name=str(data.get("name", queue_name)),
|
||||
messages=int(data.get("messages", 0)),
|
||||
state=str(data.get("state", "unknown")),
|
||||
)
|
||||
|
||||
async def peek_messages(self, queue_name: str, count: int) -> list[QueueMessage]:
|
||||
"""Non-destructively peek at up to ``count`` messages from a queue."""
|
||||
self._ensure_enabled()
|
||||
url = f"/api/queues/{self._vhost}/{quote(queue_name, safe='')}/get"
|
||||
body: dict[str, Any] = {
|
||||
"count": max(1, count),
|
||||
"ackmode": "ack_requeue_true",
|
||||
"encoding": "auto",
|
||||
"truncate": 2000,
|
||||
}
|
||||
try:
|
||||
resp = await self.client.post(url, json=body)
|
||||
except httpx.NetworkError as exc:
|
||||
raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc
|
||||
except httpx.TimeoutException as exc:
|
||||
raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc
|
||||
if resp.status_code == 404:
|
||||
raise MqManagementNotFoundError(f"queue {queue_name!r} not found")
|
||||
if resp.status_code >= 400:
|
||||
raise MqManagementResponseError(
|
||||
f"management API error {resp.status_code}",
|
||||
status_code=resp.status_code,
|
||||
body=resp.text[:500],
|
||||
)
|
||||
rows = resp.json()
|
||||
if not isinstance(rows, list):
|
||||
return []
|
||||
out: list[QueueMessage] = []
|
||||
for row in rows:
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
payload = self._decode_payload(row)
|
||||
headers = row.get("properties", {}).get("headers") or {}
|
||||
out.append(
|
||||
QueueMessage(
|
||||
payload=payload,
|
||||
headers=dict(headers),
|
||||
correlation_id=row.get("properties", {}).get("correlation_id")
|
||||
or headers.get(H_CORRELATION_ID),
|
||||
routing_key=row.get("routing_key"),
|
||||
)
|
||||
)
|
||||
return out
|
||||
|
||||
async def purge_queue(self, queue_name: str) -> PurgeResult:
|
||||
"""Empty a queue and return the number of messages removed."""
|
||||
self._ensure_enabled()
|
||||
url = f"/api/queues/{self._vhost}/{quote(queue_name, safe='')}/contents"
|
||||
try:
|
||||
resp = await self.client.delete(url)
|
||||
except httpx.NetworkError as exc:
|
||||
raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc
|
||||
except httpx.TimeoutException as exc:
|
||||
raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc
|
||||
if resp.status_code == 404:
|
||||
raise MqManagementNotFoundError(f"queue {queue_name!r} not found")
|
||||
if resp.status_code >= 400:
|
||||
raise MqManagementResponseError(
|
||||
f"management API error {resp.status_code}",
|
||||
status_code=resp.status_code,
|
||||
body=resp.text[:500],
|
||||
)
|
||||
# Successful purge returns no body; the queue is now empty.
|
||||
info = await self.get_queue_info(queue_name)
|
||||
return PurgeResult(messages_removed=max(0, info.messages))
|
||||
|
||||
async def requeue_batch(
|
||||
self,
|
||||
dlq_name: str,
|
||||
*,
|
||||
batch_size: int | None = None,
|
||||
) -> RequeueResult:
|
||||
"""Move up to ``batch_size`` messages from a DLQ back to their main exchange.
|
||||
|
||||
The ``x-attempt`` header is reset to 0 so retried messages get a fresh
|
||||
retry budget. Unknown DLQ names are rejected explicitly.
|
||||
"""
|
||||
self._ensure_enabled()
|
||||
main_queue = _main_queue_for_dlq(dlq_name)
|
||||
if main_queue is None:
|
||||
raise MqManagementNotFoundError(f"no main queue known for DLQ {dlq_name!r}")
|
||||
|
||||
settings = get_settings()
|
||||
batch_size = max(1, batch_size or settings.mq_requeue_batch_size)
|
||||
|
||||
url = f"/api/queues/{self._vhost}/{quote(dlq_name, safe='')}/get"
|
||||
body: dict[str, Any] = {
|
||||
"count": batch_size,
|
||||
"ackmode": "ack_requeue_false",
|
||||
"encoding": "auto",
|
||||
"truncate": 2000,
|
||||
}
|
||||
try:
|
||||
resp = await self.client.post(url, json=body)
|
||||
except httpx.NetworkError as exc:
|
||||
raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc
|
||||
except httpx.TimeoutException as exc:
|
||||
raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc
|
||||
if resp.status_code == 404:
|
||||
raise MqManagementNotFoundError(f"queue {dlq_name!r} not found")
|
||||
if resp.status_code >= 400:
|
||||
raise MqManagementResponseError(
|
||||
f"management API error {resp.status_code}",
|
||||
status_code=resp.status_code,
|
||||
body=resp.text[:500],
|
||||
)
|
||||
rows = resp.json()
|
||||
if not isinstance(rows, list) or not rows:
|
||||
return RequeueResult(messages_requeued=0, errors=0)
|
||||
|
||||
exchange, routing_key = _main_exchange_and_routing_key(main_queue)
|
||||
requeued = 0
|
||||
errors = 0
|
||||
for row in rows:
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
try:
|
||||
await self._publish_to_main(row, exchange, routing_key)
|
||||
requeued += 1
|
||||
except Exception as exc: # noqa: BLE001 — count failures, keep going
|
||||
log.warning("requeue_publish_failed", dlq=dlq_name, error=str(exc))
|
||||
errors += 1
|
||||
return RequeueResult(messages_requeued=requeued, errors=errors)
|
||||
|
||||
async def _publish_to_main(
|
||||
self,
|
||||
row: dict[str, Any],
|
||||
exchange: str,
|
||||
routing_key: str,
|
||||
) -> None:
|
||||
"""Publish one message back to the main exchange with attempt reset to 0."""
|
||||
url = f"/api/exchanges/{self._vhost}/{quote(exchange, safe='')}/publish"
|
||||
props = dict(row.get("properties", {}))
|
||||
headers = dict(props.get("headers") or {})
|
||||
headers[H_ATTEMPT] = 0
|
||||
props["headers"] = headers
|
||||
if H_CORRELATION_ID not in headers and props.get("correlation_id"):
|
||||
headers[H_CORRELATION_ID] = props["correlation_id"]
|
||||
payload = self._decode_payload(row)
|
||||
body: dict[str, Any] = {
|
||||
"routing_key": routing_key,
|
||||
"payload": payload.decode("utf-8", errors="replace"),
|
||||
"payload_encoding": "string",
|
||||
"properties": props,
|
||||
}
|
||||
resp = await self.client.post(url, json=body)
|
||||
if resp.status_code >= 400:
|
||||
raise MqManagementResponseError(
|
||||
f"publish to {exchange!r} failed: {resp.status_code}",
|
||||
status_code=resp.status_code,
|
||||
body=resp.text[:500],
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _decode_payload(row: dict[str, Any]) -> bytes:
|
||||
payload = row.get("payload")
|
||||
if payload is None:
|
||||
return b""
|
||||
if isinstance(payload, bytes):
|
||||
return payload
|
||||
if isinstance(payload, str):
|
||||
return payload.encode("utf-8", errors="replace")
|
||||
return str(payload).encode("utf-8", errors="replace")
|
||||
|
||||
|
||||
def _main_queue_for_dlq(dlq_name: str) -> str | None:
|
||||
"""Return the main queue that feeds the given DLQ."""
|
||||
for main, dlq in DLQ_FOR.items():
|
||||
if dlq == dlq_name:
|
||||
return main
|
||||
return None
|
||||
242
tests/unit/test_mq_management.py
Normal file
242
tests/unit/test_mq_management.py
Normal file
|
|
@ -0,0 +1,242 @@
|
|||
"""Unit tests for the RabbitMQ management client.
|
||||
|
||||
All real HTTP traffic is intercepted with ``respx``. We do not need a running
|
||||
broker here — the integration suite covers live RabbitMQ behaviour.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
from src.contract_check.core.mq.management import (
|
||||
MqManagementDisabledError,
|
||||
MqManagementNotFoundError,
|
||||
MqManagementResponseError,
|
||||
MqManagementUnreachableError,
|
||||
PurgeResult,
|
||||
QueueInfo,
|
||||
QueueMessage,
|
||||
RabbitMQManagementClient,
|
||||
RequeueResult,
|
||||
)
|
||||
from src.contract_check.core.mq.topology import DLQ_FOR
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mgmt_client(monkeypatch: pytest.MonkeyPatch) -> RabbitMQManagementClient:
|
||||
# Pin AMQP URL before Settings is imported/looked up anywhere else.
|
||||
monkeypatch.setenv("RABBITMQ_URL", "amqp://admin:secret@rabbitmq:5672//")
|
||||
from src.contract_check.core.config import get_settings
|
||||
|
||||
get_settings.cache_clear()
|
||||
return RabbitMQManagementClient()
|
||||
|
||||
|
||||
def _mgmt_root(base: str) -> str:
|
||||
return base.rstrip("/")
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_queue_info_ok(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
with respx.mock:
|
||||
route = respx.get("http://rabbitmq:15672/api/queues/%2F/extract.q").mock(
|
||||
return_value=httpx.Response(
|
||||
200, json={"name": "extract.q", "messages": 7, "state": "running"}
|
||||
)
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
info = await mgmt_client.get_queue_info("extract.q")
|
||||
assert info == QueueInfo(name="extract.q", messages=7, state="running")
|
||||
assert route.called
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_queue_info_not_found(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
with respx.mock:
|
||||
respx.get("http://rabbitmq:15672/api/queues/%2F/missing").mock(
|
||||
return_value=httpx.Response(404)
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
with pytest.raises(MqManagementNotFoundError):
|
||||
await mgmt_client.get_queue_info("missing")
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_queue_info_unreachable(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
with respx.mock:
|
||||
respx.get("http://rabbitmq:15672/api/queues/%2F/extract.q").mock(
|
||||
side_effect=httpx.ConnectError("nope")
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
with pytest.raises(MqManagementUnreachableError):
|
||||
await mgmt_client.get_queue_info("extract.q")
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_disabled_client_raises(
|
||||
mgmt_client: RabbitMQManagementClient, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(mgmt_client, "_enabled", False)
|
||||
with pytest.raises(MqManagementDisabledError):
|
||||
await mgmt_client.connect()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_peek_messages(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
payload = json.dumps({"document_id": "d1"}).encode()
|
||||
with respx.mock:
|
||||
respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json=[
|
||||
{
|
||||
"routing_key": "extract.q",
|
||||
"payload": payload.decode(),
|
||||
"properties": {
|
||||
"correlation_id": "c1",
|
||||
"headers": {"x-failure-class": "infra", "x-failure-error": "boom"},
|
||||
},
|
||||
}
|
||||
],
|
||||
)
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
msgs = await mgmt_client.peek_messages("extract.dlq", 5)
|
||||
assert len(msgs) == 1
|
||||
msg = msgs[0]
|
||||
assert isinstance(msg, QueueMessage)
|
||||
assert msg.correlation_id == "c1"
|
||||
assert msg.routing_key == "extract.q"
|
||||
assert msg.failure_class == "infra"
|
||||
assert msg.failure_error == "boom"
|
||||
assert msg.payload == payload
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_peek_messages_empty_non_list(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
with respx.mock:
|
||||
respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock(
|
||||
return_value=httpx.Response(200, json={"garbage": True})
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
assert await mgmt_client.peek_messages("extract.dlq", 5) == []
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_purge_queue(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
with respx.mock:
|
||||
respx.delete("http://rabbitmq:15672/api/queues/%2F/extract.dlq/contents").mock(
|
||||
return_value=httpx.Response(204)
|
||||
)
|
||||
respx.get("http://rabbitmq:15672/api/queues/%2F/extract.dlq").mock(
|
||||
return_value=httpx.Response(
|
||||
200, json={"name": "extract.dlq", "messages": 0, "state": "running"}
|
||||
)
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
assert await mgmt_client.purge_queue("extract.dlq") == PurgeResult(messages_removed=0)
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_requeue_batch_unknown_dlq(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
await mgmt_client.connect()
|
||||
with pytest.raises(MqManagementNotFoundError):
|
||||
await mgmt_client.requeue_batch("not-a-known-dlq")
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_requeue_batch_empty(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
dlq = list(DLQ_FOR.values())[0]
|
||||
with respx.mock:
|
||||
respx.post(f"http://rabbitmq:15672/api/queues/%2F/{dlq}/get").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
assert await mgmt_client.requeue_batch(dlq) == RequeueResult(messages_requeued=0, errors=0)
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_requeue_batch_success(mgmt_client: RabbitMQManagementClient) -> None:
|
||||
dlq = "extract.dlq"
|
||||
main_queue = "extract.q"
|
||||
payload = json.dumps({"document_id": "d1"})
|
||||
with respx.mock:
|
||||
respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json=[
|
||||
{
|
||||
"routing_key": main_queue,
|
||||
"payload": payload,
|
||||
"properties": {
|
||||
"correlation_id": "c1",
|
||||
"headers": {"x-attempt": 5, "x-failure-class": "infra"},
|
||||
},
|
||||
}
|
||||
],
|
||||
)
|
||||
)
|
||||
respx.post("http://rabbitmq:15672/api/exchanges/%2F/contracts.x/publish").mock(
|
||||
return_value=httpx.Response(200, json={"routed": True})
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
result = await mgmt_client.requeue_batch(dlq, batch_size=10)
|
||||
assert result == RequeueResult(messages_requeued=1, errors=0)
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_requeue_batch_publish_failure_counts_as_error(
|
||||
mgmt_client: RabbitMQManagementClient,
|
||||
) -> None:
|
||||
dlq = "extract.dlq"
|
||||
main_queue = "extract.q"
|
||||
with respx.mock:
|
||||
respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json=[
|
||||
{
|
||||
"routing_key": main_queue,
|
||||
"payload": "{}",
|
||||
"properties": {"headers": {"x-attempt": 5}},
|
||||
}
|
||||
],
|
||||
)
|
||||
)
|
||||
respx.post("http://rabbitmq:15672/api/exchanges/%2F/contracts.x/publish").mock(
|
||||
return_value=httpx.Response(500, text="broker error")
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
result = await mgmt_client.requeue_batch(dlq, batch_size=10)
|
||||
assert result == RequeueResult(messages_requeued=0, errors=1)
|
||||
await mgmt_client.aclose()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_response_error_includes_status_and_body(
|
||||
mgmt_client: RabbitMQManagementClient,
|
||||
) -> None:
|
||||
with respx.mock:
|
||||
respx.get("http://rabbitmq:15672/api/queues/%2F/extract.q").mock(
|
||||
return_value=httpx.Response(418, text="teapot")
|
||||
)
|
||||
await mgmt_client.connect()
|
||||
with pytest.raises(MqManagementResponseError) as exc_info:
|
||||
await mgmt_client.get_queue_info("extract.q")
|
||||
err = exc_info.value
|
||||
assert err.status_code == 418
|
||||
assert err.body == "teapot"
|
||||
await mgmt_client.aclose()
|
||||
Loading…
Add table
Reference in a new issue