Загрузить файлы в «alert-receiver/app»

alert-receiver/app/__init__.py

@@ -0,0 +1 @@
+# empty

alert-receiver/app/config.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+from dotenv import load_dotenv
+
+
+load_dotenv()
+
+
+def _parse_bool(value: str | None, default: bool = False) -> bool:
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+@dataclass(frozen=True)
+class Settings:
+    app_name: str = os.getenv("APP_NAME", "alert-receiver")
+    app_host: str = os.getenv("APP_HOST", "0.0.0.0")
+    app_port: int = int(os.getenv("APP_PORT", "8080"))
+
+    webhook_token: str = os.getenv("WEBHOOK_TOKEN", "").strip()
+    require_webhook_token: bool = _parse_bool(
+        os.getenv("REQUIRE_WEBHOOK_TOKEN", "true"),
+        default=True,
+    )
+
+    forward_to_processor: bool = _parse_bool(
+        os.getenv("FORWARD_TO_PROCESSOR", "false"),
+        default=False,
+    )
+    alert_processor_url: str = os.getenv(
+        "ALERT_PROCESSOR_URL",
+        "http://alert-processor:8081/internal/events",
+    )
+    alert_processor_token: str = os.getenv(
+        "ALERT_PROCESSOR_TOKEN",
+        "",
+    ).strip()
+    forward_timeout_seconds: float = float(
+        os.getenv("FORWARD_TIMEOUT_SECONDS", "5")
+    )
+
+alert_processor_token: str = os.getenv("ALERT_PROCESSOR_TOKEN", "").strip()
+
+settings = Settings()

alert-receiver/app/forwarder.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import logging
+
+import httpx
+
+from app.config import settings
+from app.models import ProcessorForwardEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+async def forward_to_processor(envelope: ProcessorForwardEnvelope) -> bool:
+    if not settings.forward_to_processor:
+        logger.info(
+            "Forwarding disabled, event accepted locally only: correlation_id=%s",
+            envelope.event.correlation_id,
+        )
+        return False
+
+    headers = {}
+    if settings.alert_processor_token:
+        headers["X-Internal-Token"] = settings.alert_processor_token
+
+    try:
+        async with httpx.AsyncClient(timeout=settings.forward_timeout_seconds) as client:
+            response = await client.post(
+                settings.alert_processor_url,
+                json=envelope.model_dump(mode="json"),
+                headers=headers,
+            )
+            response.raise_for_status()
+
+        logger.info(
+            "Forwarded event to alert-processor: correlation_id=%s status=%s",
+            envelope.event.correlation_id,
+            response.status_code,
+        )
+        return True
+    except Exception as exc:
+        logger.exception(
+            "Failed to forward event to alert-processor: correlation_id=%s error=%s",
+            envelope.event.correlation_id,
+            exc,
+        )
+        return False

alert-receiver/app/main.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import logging
+import uuid
+
+from fastapi import FastAPI, Header, HTTPException, Request, status
+
+from app.config import settings
+from app.forwarder import forward_to_processor
+from app.models import NormalizedEvent, ProcessorForwardEnvelope, ReceiverAck
+from app.normalizer import normalize_zabbix_payload
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+
+logger = logging.getLogger(__name__)
+
+app = FastAPI(
+    title="alert-receiver",
+    version="0.1.0",
+    description="Receives Zabbix webhooks and normalizes them for alert-processor.",
+)
+
+
+@app.on_event("startup")
+async def startup_check() -> None:
+    if settings.require_webhook_token and not settings.webhook_token:
+        raise RuntimeError(
+            "WEBHOOK_TOKEN is required, but not set. "
+            "Set it in environment variables or in .env file."
+        )
+
+    logger.info(
+        "Startup configuration loaded: require_webhook_token=%s forward_to_processor=%s",
+        settings.require_webhook_token,
+        settings.forward_to_processor,
+    )
+
+def _extract_token(
+    x_webhook_token: str | None,
+    authorization: str | None,
+) -> str | None:
+    if x_webhook_token:
+        return x_webhook_token.strip()
+
+    if authorization:
+        auth = authorization.strip()
+        if auth.lower().startswith("bearer "):
+            return auth[7:].strip()
+        return auth
+
+    return None
+
+
+def _extract_token(
+    x_webhook_token: str | None,
+    authorization: str | None,
+) -> str | None:
+    if x_webhook_token:
+        return x_webhook_token.strip()
+
+    if authorization:
+        auth = authorization.strip()
+        if auth.lower().startswith("bearer "):
+            return auth[7:].strip()
+        return auth
+
+    return None
+
+def _validate_token(
+    x_webhook_token: str | None,
+    authorization: str | None,
+) -> None:
+    if not settings.require_webhook_token:
+        return
+
+    provided = _extract_token(x_webhook_token, authorization)
+    if not provided or provided != settings.webhook_token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or missing webhook token",
+        )
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {
+        "status": "ok",
+        "service": settings.app_name,
+    }
+
+@app.post(
+    "/webhook/zabbix",
+    response_model=ReceiverAck,
+    status_code=status.HTTP_202_ACCEPTED,
+)
+async def receive_zabbix_webhook(
+    request: Request,
+    x_webhook_token: str | None = Header(default=None),
+    authorization: str | None = Header(default=None),
+    x_correlation_id: str | None = Header(default=None),
+) -> ReceiverAck:
+    _validate_token(x_webhook_token, authorization)
+
+    try:
+        payload = await request.json()
+    except Exception as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid JSON payload: {exc}",
+        ) from exc
+
+    if not isinstance(payload, dict):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Payload must be a JSON object",
+        )
+
+    correlation_id = x_correlation_id or str(uuid.uuid4())
+    remote_addr = request.client.host if request.client else None
+
+    normalized_dict = normalize_zabbix_payload(
+        payload=payload,
+        correlation_id=correlation_id,
+        remote_addr=remote_addr,
+    )
+
+    event = NormalizedEvent.model_validate(normalized_dict)
+    envelope = ProcessorForwardEnvelope(event=event)
+
+    logger.info(
+        "Accepted Zabbix event: correlation_id=%s event_id=%s severity=%s host=%s trigger=%s",
+        event.correlation_id,
+        event.event_id,
+        event.severity,
+        event.host,
+        event.trigger_name,
+    )
+
+    forwarded = await forward_to_processor(envelope)
+
+    return ReceiverAck(
+        accepted=True,
+        correlation_id=event.correlation_id,
+        forwarded_to_processor=forwarded,
+        message="Webhook accepted",
+    )

alert-receiver/app/models.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class NormalizedEvent(BaseModel):
+    source: str = "zabbix"
+    correlation_id: str
+    received_at: datetime
+    remote_addr: str | None = None
+
+    event_id: str | None = None
+    problem_id: str | None = None
+    event_type: str = "problem"
+    timestamp: datetime | None = None
+
+    severity: str | None = None
+    severity_code: int | None = None
+
+    host: str | None = None
+    host_id: str | None = None
+    service: str | None = None
+
+    trigger_name: str | None = None
+    trigger_id: str | None = None
+    item_id: str | None = None
+
+    value: str | None = None
+    opdata: str | None = None
+    zabbix_url: str | None = None
+
+    tags: dict[str, str] = Field(default_factory=dict)
+    raw_payload: dict[str, Any]
+
+
+class ReceiverAck(BaseModel):
+    accepted: bool
+    correlation_id: str
+    forwarded_to_processor: bool
+    message: str
+
+
+class ProcessorForwardEnvelope(BaseModel):
+    event: NormalizedEvent