From 4b122dfee72de6070d6a55760f9657f1bd154679 Mon Sep 17 00:00:00 2001 From: Alexander Zubarev Date: Thu, 6 Aug 2026 18:31:13 +0300 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?alert-receiver/app=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alert-receiver/app/__init__.py | 1 + alert-receiver/app/config.py | 48 ++++++++++ alert-receiver/app/forwarder.py | 46 ++++++++++ alert-receiver/app/main.py | 149 ++++++++++++++++++++++++++++++++ alert-receiver/app/models.py | 47 ++++++++++ 5 files changed, 291 insertions(+) create mode 100644 alert-receiver/app/__init__.py create mode 100644 alert-receiver/app/config.py create mode 100644 alert-receiver/app/forwarder.py create mode 100644 alert-receiver/app/main.py create mode 100644 alert-receiver/app/models.py diff --git a/alert-receiver/app/__init__.py b/alert-receiver/app/__init__.py new file mode 100644 index 0000000..bc63beb --- /dev/null +++ b/alert-receiver/app/__init__.py @@ -0,0 +1 @@ +# empty \ No newline at end of file diff --git a/alert-receiver/app/config.py b/alert-receiver/app/config.py new file mode 100644 index 0000000..41c5d2b --- /dev/null +++ b/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() \ No newline at end of file diff --git a/alert-receiver/app/forwarder.py b/alert-receiver/app/forwarder.py new file mode 100644 index 0000000..d21edef --- /dev/null +++ b/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 \ No newline at end of file diff --git a/alert-receiver/app/main.py b/alert-receiver/app/main.py new file mode 100644 index 0000000..37d4bbb --- /dev/null +++ b/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", + ) \ No newline at end of file diff --git a/alert-receiver/app/models.py b/alert-receiver/app/models.py new file mode 100644 index 0000000..5951679 --- /dev/null +++ b/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 \ No newline at end of file