diff --git a/.env.example b/.env.example index dde7e83..9ab35dc 100644 --- a/.env.example +++ b/.env.example @@ -3,4 +3,10 @@ ENVIRONMENT=local DATABASE_URL=sqlite:///./dataops_observability.db API_PREFIX=/api/v1 # Optional comma-separated keys. When set, write/ingestion endpoints require X-DataOps-API-Key. -INGESTION_API_KEYS= \ No newline at end of file +INGESTION_API_KEYS= +PUBLIC_BASE_URL=http://127.0.0.1:8000 +# Optional comma-separated alert receivers. Failed/canceled runs and warning/failed checks are posted here. +ALERT_WEBHOOK_URLS= +# Optional shared secret sent as X-DataOps-Webhook-Secret to alert receivers. +ALERT_WEBHOOK_SECRET= +ALERT_WEBHOOK_TIMEOUT_SECONDS=5 \ No newline at end of file diff --git a/README.md b/README.md index 333768a..4ea2a6a 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ Most pipeline monitoring tools stop at raw events, logs, or generic charts. This - a small dashboard that makes the API useful immediately after seeding sample data - integration-friendly endpoints for Airflow, dbt, cron jobs, GitHub Actions, or custom ETL scripts - optional API-key protection for ingestion writes when the service is exposed outside a local machine +- webhook alerts for failed pipeline runs and quality checks that need operator attention - a codebase small enough for teams to adapt instead of adopting a heavy platform See [docs/integrations.md](docs/integrations.md) for copy-paste examples that report pipeline events from Python jobs, GitHub Actions, Airflow, and dbt. @@ -73,6 +74,8 @@ Full setup notes live in [docs/integrations.md](docs/integrations.md). Set `INGESTION_API_KEYS` to require external reporters to send `X-DataOps-API-Key` on write requests. Read-only dashboard and metrics endpoints stay open so operators can inspect service health without sharing ingestion credentials. +Set `ALERT_WEBHOOK_URLS` to send operational alerts to Slack-compatible bridges, incident tooling, or custom automation when runs fail/cancel or quality checks warn/fail. + ## Run Quality Gates ```powershell @@ -168,6 +171,10 @@ Settings are loaded from environment variables. | `DATABASE_URL` | `sqlite:///./dataops_observability.db` | SQLAlchemy database URL | | `API_PREFIX` | `/api/v1` | Versioned API prefix | | `INGESTION_API_KEYS` | empty | Comma-separated keys accepted by write/ingestion endpoints. When empty, local writes are open. | +| `PUBLIC_BASE_URL` | `http://127.0.0.1:8000` | Base URL used in generated dashboard and API links. | +| `ALERT_WEBHOOK_URLS` | empty | Comma-separated webhook URLs that receive operational alerts. | +| `ALERT_WEBHOOK_SECRET` | empty | Optional shared secret sent as `X-DataOps-Webhook-Secret` on alert deliveries. | +| `ALERT_WEBHOOK_TIMEOUT_SECONDS` | `5` | Timeout for each outbound webhook delivery. | When `INGESTION_API_KEYS` is set, these write endpoints require `X-DataOps-API-Key`: @@ -177,6 +184,27 @@ When `INGESTION_API_KEYS` is set, these write endpoints require `X-DataOps-API-K Use different keys during rotation by setting a comma-separated value such as `INGESTION_API_KEYS=old-key,new-key`. Store real keys in your deployment secret manager, CI secrets, or local `.env` file rather than committing them. +## Webhook Alerts + +When `ALERT_WEBHOOK_URLS` is configured, the API sends background webhook alerts for events that normally need human attention: + +- pipeline run status changes to `failed` +- pipeline run status changes to `canceled` +- quality check status is `failed` +- quality check status is `warning` + +Successful runs and passed quality checks do not send alerts. This keeps notification volume focused on work that needs follow-up. + +Example configuration: + +```powershell +$env:PUBLIC_BASE_URL = "https://dataops.example.com" +$env:ALERT_WEBHOOK_URLS = "https://hooks.example.com/dataops,https://backup.example.com/dataops" +$env:ALERT_WEBHOOK_SECRET = "shared-secret" +``` + +Alert payloads include `event_type`, `severity`, `message`, the affected pipeline run, optional quality-check details, and links back to the dashboard, run detail endpoint, and timeline endpoint. + ## Notes SQLite is the default for local development. The SQLAlchemy and Alembic setup is structured so the service can be moved to Postgres by changing `DATABASE_URL` and adding the relevant deployment configuration. \ No newline at end of file diff --git a/app/api/pipeline_runs.py b/app/api/pipeline_runs.py index 38d4b7e..cb4aaa9 100644 --- a/app/api/pipeline_runs.py +++ b/app/api/pipeline_runs.py @@ -1,8 +1,9 @@ from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, Query, status +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status from sqlalchemy.orm import Session +from app.core.config import Settings, get_settings from app.core.security import require_ingestion_api_key from app.db.session import get_db from app.models.pipeline import PipelineRun @@ -35,6 +36,7 @@ list_quality_checks, update_pipeline_run, ) +from app.services.webhook_alerts import queue_pipeline_run_alert, queue_quality_check_alert router = APIRouter(tags=["pipeline runs"]) @@ -54,9 +56,13 @@ def _get_run_or_404(db: Session, run_id: int) -> PipelineRun: ) def create_run( payload: PipelineRunCreate, + background_tasks: BackgroundTasks, db: Annotated[Session, Depends(get_db)], + settings: Annotated[Settings, Depends(get_settings)], ) -> PipelineRun: - return create_pipeline_run(db, payload) + run = create_pipeline_run(db, payload) + queue_pipeline_run_alert(background_tasks, settings, run) + return run @router.get("/pipeline-runs", response_model=list[PipelineRunRead]) @@ -101,10 +107,15 @@ def read_run_timeline( def patch_run( run_id: int, payload: PipelineRunUpdate, + background_tasks: BackgroundTasks, db: Annotated[Session, Depends(get_db)], + settings: Annotated[Settings, Depends(get_settings)], ) -> PipelineRun: run = _get_run_or_404(db, run_id) - return update_pipeline_run(db, run, payload) + updated_run = update_pipeline_run(db, run, payload) + if payload.status in {PipelineRunStatus.failed, PipelineRunStatus.canceled}: + queue_pipeline_run_alert(background_tasks, settings, updated_run) + return updated_run @router.post( @@ -116,10 +127,14 @@ def patch_run( def create_check( run_id: int, payload: QualityCheckCreate, + background_tasks: BackgroundTasks, db: Annotated[Session, Depends(get_db)], + settings: Annotated[Settings, Depends(get_settings)], ) -> QualityCheckRead: run = _get_run_or_404(db, run_id) - return QualityCheckRead.model_validate(create_quality_check(db, run, payload)) + check = create_quality_check(db, run, payload) + queue_quality_check_alert(background_tasks, settings, run, check) + return QualityCheckRead.model_validate(check) @router.get("/pipeline-runs/{run_id}/quality-checks", response_model=list[QualityCheckRead]) diff --git a/app/core/config.py b/app/core/config.py index 9171b80..5c21825 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -13,6 +13,23 @@ class Settings(BaseSettings): default="", description="Comma-separated API keys accepted for write/ingestion endpoints.", ) + public_base_url: str = Field( + default="http://127.0.0.1:8000", + description="Base URL used in generated dashboard and API links.", + ) + alert_webhook_urls: str = Field( + default="", + description="Comma-separated webhook URLs that receive operational alerts.", + ) + alert_webhook_secret: str = Field( + default="", + description="Optional shared secret sent with webhook alert deliveries.", + ) + alert_webhook_timeout_seconds: float = Field( + default=5.0, + gt=0, + description="Timeout in seconds for outbound webhook alert deliveries.", + ) model_config = SettingsConfigDict(env_file=".env", extra="ignore") diff --git a/app/services/webhook_alerts.py b/app/services/webhook_alerts.py new file mode 100644 index 0000000..1812ac9 --- /dev/null +++ b/app/services/webhook_alerts.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import json +import logging +from datetime import UTC, datetime +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +from fastapi import BackgroundTasks + +from app.core.config import Settings +from app.models.pipeline import PipelineRun, QualityCheck +from app.schemas.pipeline import PipelineRunStatus, QualityCheckStatus + +WEBHOOK_SECRET_HEADER = "X-DataOps-Webhook-Secret" +WEBHOOK_USER_AGENT = "dataops-observability-api/0.1" +ACTIONABLE_RUN_STATUSES = { + PipelineRunStatus.failed.value, + PipelineRunStatus.canceled.value, +} +ACTIONABLE_CHECK_STATUSES = { + QualityCheckStatus.failed.value, + QualityCheckStatus.warning.value, +} + +AlertPayload = dict[str, object] + +logger = logging.getLogger(__name__) + + +def get_configured_alert_webhook_urls(settings: Settings) -> tuple[str, ...]: + return tuple( + webhook_url.strip() + for webhook_url in settings.alert_webhook_urls.split(",") + if webhook_url.strip() + ) + + +def queue_pipeline_run_alert( + background_tasks: BackgroundTasks, + settings: Settings, + run: PipelineRun, +) -> None: + if run.status not in ACTIONABLE_RUN_STATUSES: + return + + queue_webhook_alert( + background_tasks, + settings, + build_pipeline_run_alert_payload(run, settings), + ) + + +def queue_quality_check_alert( + background_tasks: BackgroundTasks, + settings: Settings, + run: PipelineRun, + check: QualityCheck, +) -> None: + if check.status not in ACTIONABLE_CHECK_STATUSES: + return + + queue_webhook_alert( + background_tasks, + settings, + build_quality_check_alert_payload(run, check, settings), + ) + + +def queue_webhook_alert( + background_tasks: BackgroundTasks, + settings: Settings, + payload: AlertPayload, +) -> None: + webhook_urls = get_configured_alert_webhook_urls(settings) + if not webhook_urls: + return + + background_tasks.add_task( + send_webhook_alerts, + webhook_urls, + payload, + settings.alert_webhook_secret, + settings.alert_webhook_timeout_seconds, + ) + + +def build_pipeline_run_alert_payload(run: PipelineRun, settings: Settings) -> AlertPayload: + event_type = f"pipeline_run_{run.status}" + severity = "critical" if run.status == PipelineRunStatus.failed.value else "warning" + return { + "event_type": event_type, + "severity": severity, + "message": f"Pipeline run {run.name} is {run.status}.", + "occurred_at": datetime.now(UTC).isoformat(), + "pipeline_run": serialize_pipeline_run(run), + "links": build_alert_links(run, settings), + } + + +def build_quality_check_alert_payload( + run: PipelineRun, + check: QualityCheck, + settings: Settings, +) -> AlertPayload: + event_type = f"quality_check_{check.status}" + severity = check.severity + return { + "event_type": event_type, + "severity": severity, + "message": f"Quality check {check.check_name} is {check.status} for {run.name}.", + "occurred_at": datetime.now(UTC).isoformat(), + "pipeline_run": serialize_pipeline_run(run), + "quality_check": serialize_quality_check(check), + "links": build_alert_links(run, settings), + } + + +def serialize_pipeline_run(run: PipelineRun) -> dict[str, object]: + return { + "id": run.id, + "name": run.name, + "source_system": run.source_system, + "status": run.status, + "records_processed": run.records_processed, + "started_at": format_datetime(run.started_at), + "finished_at": format_datetime(run.finished_at), + "error_message": run.error_message, + "created_at": run.created_at.isoformat(), + "updated_at": run.updated_at.isoformat(), + } + + +def serialize_quality_check(check: QualityCheck) -> dict[str, object]: + return { + "id": check.id, + "pipeline_run_id": check.pipeline_run_id, + "check_name": check.check_name, + "status": check.status, + "severity": check.severity, + "expected_value": check.expected_value, + "observed_value": check.observed_value, + "details": check.details, + "created_at": check.created_at.isoformat(), + } + + +def build_alert_links(run: PipelineRun, settings: Settings) -> dict[str, str]: + base_url = settings.public_base_url.rstrip("/") + return { + "dashboard": f"{base_url}/dashboard", + "pipeline_run": f"{base_url}{settings.api_prefix}/pipeline-runs/{run.id}", + "timeline": f"{base_url}{settings.api_prefix}/pipeline-runs/{run.id}/timeline", + } + + +def format_datetime(value: datetime | None) -> str | None: + if value is None: + return None + return value.isoformat() + + +def send_webhook_alerts( + webhook_urls: tuple[str, ...], + payload: AlertPayload, + shared_secret: str, + timeout_seconds: float, +) -> None: + for webhook_url in webhook_urls: + send_webhook_alert(webhook_url, payload, shared_secret, timeout_seconds) + + +def send_webhook_alert( + webhook_url: str, + payload: AlertPayload, + shared_secret: str, + timeout_seconds: float, +) -> None: + data = json.dumps(payload).encode("utf-8") + headers = { + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": WEBHOOK_USER_AGENT, + } + if shared_secret: + headers[WEBHOOK_SECRET_HEADER] = shared_secret + + request = Request( + url=webhook_url, + data=data, + headers=headers, + method="POST", + ) + try: + with urlopen(request, timeout=timeout_seconds): + return + except (HTTPError, URLError, TimeoutError, OSError) as error: + logger.warning("Webhook alert delivery failed for %s: %s", webhook_url, error) \ No newline at end of file diff --git a/docs/integrations.md b/docs/integrations.md index 024b5d0..95e0c4b 100644 --- a/docs/integrations.md +++ b/docs/integrations.md @@ -25,6 +25,64 @@ $env:DATAOPS_API_KEY = "dev-ingest-key" Read-only endpoints such as `/dashboard`, `/health`, and `/api/v1/metrics/*` remain open. Only ingestion writes require the header when keys are configured. +## Webhook Alerts + +Webhook alerts turn ingested events into operator notifications. Configure one or more receivers with `ALERT_WEBHOOK_URLS`: + +```powershell +$env:PUBLIC_BASE_URL = "https://dataops.example.com" +$env:ALERT_WEBHOOK_URLS = "https://hooks.example.com/dataops" +$env:ALERT_WEBHOOK_SECRET = "shared-secret" +``` + +The service sends alerts in the background after the API write succeeds. Alerts fire for: + +- pipeline runs patched to `failed` or `canceled` +- quality checks created as `failed` or `warning` + +Successful runs and passed checks are recorded without sending notifications. + +Every configured receiver gets a JSON payload like this: + +```json +{ + "event_type": "quality_check_failed", + "severity": "critical", + "message": "Quality check freshness_sla is failed for orders_daily_load.", + "occurred_at": "2026-06-07T21:00:00+00:00", + "pipeline_run": { + "id": 42, + "name": "orders_daily_load", + "source_system": "airflow", + "status": "running", + "records_processed": 1284, + "started_at": "2026-06-07T20:55:00+00:00", + "finished_at": null, + "error_message": null, + "created_at": "2026-06-07T20:55:00+00:00", + "updated_at": "2026-06-07T20:55:00+00:00" + }, + "quality_check": { + "id": 9, + "pipeline_run_id": 42, + "check_name": "freshness_sla", + "status": "failed", + "severity": "critical", + "expected_value": "less than 2 hours", + "observed_value": "4 hours", + "details": "Warehouse table is stale.", + "created_at": "2026-06-07T21:00:00+00:00" + }, + "links": { + "dashboard": "https://dataops.example.com/dashboard", + "pipeline_run": "https://dataops.example.com/api/v1/pipeline-runs/42", + "timeline": "https://dataops.example.com/api/v1/pipeline-runs/42/timeline" + } +} +``` + +If `ALERT_WEBHOOK_SECRET` is set, deliveries include `X-DataOps-Webhook-Secret`. Receivers should verify that header before acting on an alert. + ## Common Event Shapes Create a run: diff --git a/tests/test_webhook_alerts.py b/tests/test_webhook_alerts.py new file mode 100644 index 0000000..e034eff --- /dev/null +++ b/tests/test_webhook_alerts.py @@ -0,0 +1,179 @@ +from typing import Any + +from fastapi.testclient import TestClient + +import app.services.webhook_alerts as webhook_alerts +from app.core.config import Settings, get_settings +from app.main import app + + +def override_alert_settings(**values: Any) -> None: + app.dependency_overrides[get_settings] = lambda: Settings(**values) + + +def create_running_run(client: TestClient) -> dict[str, Any]: + response = client.post( + "/api/v1/pipeline-runs", + json={ + "name": "daily_orders_load", + "source_system": "warehouse", + "status": "running", + "records_processed": 0, + }, + ) + assert response.status_code == 201 + return response.json() + + +def test_failed_pipeline_run_sends_webhook_alert( + client: TestClient, + monkeypatch: Any, +) -> None: + deliveries: list[tuple[tuple[str, ...], dict[str, object], str, float]] = [] + + def fake_send_webhook_alerts( + webhook_urls: tuple[str, ...], + payload: dict[str, object], + shared_secret: str, + timeout_seconds: float, + ) -> None: + deliveries.append((webhook_urls, payload, shared_secret, timeout_seconds)) + + monkeypatch.setattr(webhook_alerts, "send_webhook_alerts", fake_send_webhook_alerts) + override_alert_settings( + alert_webhook_urls="https://alerts.example/dataops, https://backup.example/dataops", + alert_webhook_secret="local-secret", + alert_webhook_timeout_seconds=2.5, + public_base_url="https://dataops.example", + ) + created = create_running_run(client) + + response = client.patch( + f"/api/v1/pipeline-runs/{created['id']}", + json={"status": "failed", "error_message": "warehouse load failed"}, + ) + + assert response.status_code == 200 + assert len(deliveries) == 1 + webhook_urls, payload, shared_secret, timeout_seconds = deliveries[0] + assert webhook_urls == ("https://alerts.example/dataops", "https://backup.example/dataops") + assert shared_secret == "local-secret" + assert timeout_seconds == 2.5 + assert payload["event_type"] == "pipeline_run_failed" + assert payload["severity"] == "critical" + assert payload["message"] == "Pipeline run daily_orders_load is failed." + assert payload["pipeline_run"] == { + **payload["pipeline_run"], + "id": created["id"], + "name": "daily_orders_load", + "status": "failed", + "error_message": "warehouse load failed", + } + assert payload["links"] == { + "dashboard": "https://dataops.example/dashboard", + "pipeline_run": f"https://dataops.example/api/v1/pipeline-runs/{created['id']}", + "timeline": f"https://dataops.example/api/v1/pipeline-runs/{created['id']}/timeline", + } + + +def test_warning_quality_check_sends_webhook_alert( + client: TestClient, + monkeypatch: Any, +) -> None: + deliveries: list[dict[str, object]] = [] + + def fake_send_webhook_alerts( + webhook_urls: tuple[str, ...], + payload: dict[str, object], + shared_secret: str, + timeout_seconds: float, + ) -> None: + deliveries.append(payload) + + monkeypatch.setattr(webhook_alerts, "send_webhook_alerts", fake_send_webhook_alerts) + override_alert_settings(alert_webhook_urls="https://alerts.example/dataops") + created = create_running_run(client) + + response = client.post( + f"/api/v1/pipeline-runs/{created['id']}/quality-checks", + json={ + "check_name": "freshness_sla", + "status": "warning", + "severity": "medium", + "expected_value": "less than 2 hours", + "observed_value": "95 minutes", + }, + ) + + assert response.status_code == 201 + assert len(deliveries) == 1 + payload = deliveries[0] + assert payload["event_type"] == "quality_check_warning" + assert payload["severity"] == "medium" + assert payload["message"] == "Quality check freshness_sla is warning for daily_orders_load." + assert payload["quality_check"] == { + **payload["quality_check"], + "check_name": "freshness_sla", + "status": "warning", + "severity": "medium", + "observed_value": "95 minutes", + } + + +def test_non_actionable_events_do_not_send_webhook_alert( + client: TestClient, + monkeypatch: Any, +) -> None: + deliveries: list[dict[str, object]] = [] + + def fake_send_webhook_alerts( + webhook_urls: tuple[str, ...], + payload: dict[str, object], + shared_secret: str, + timeout_seconds: float, + ) -> None: + deliveries.append(payload) + + monkeypatch.setattr(webhook_alerts, "send_webhook_alerts", fake_send_webhook_alerts) + override_alert_settings(alert_webhook_urls="https://alerts.example/dataops") + created = create_running_run(client) + + patch_response = client.patch( + f"/api/v1/pipeline-runs/{created['id']}", + json={"status": "succeeded", "records_processed": 1284}, + ) + check_response = client.post( + f"/api/v1/pipeline-runs/{created['id']}/quality-checks", + json={ + "check_name": "row_count_minimum", + "status": "passed", + "severity": "high", + }, + ) + + assert patch_response.status_code == 200 + assert check_response.status_code == 201 + assert deliveries == [] + + +def test_failed_pipeline_run_without_configured_webhooks_does_not_deliver( + client: TestClient, + monkeypatch: Any, +) -> None: + def fail_if_called( + webhook_urls: tuple[str, ...], + payload: dict[str, object], + shared_secret: str, + timeout_seconds: float, + ) -> None: + raise AssertionError("No delivery should be queued without webhook URLs") + + monkeypatch.setattr(webhook_alerts, "send_webhook_alerts", fail_if_called) + created = create_running_run(client) + + response = client.patch( + f"/api/v1/pipeline-runs/{created['id']}", + json={"status": "failed", "error_message": "warehouse load failed"}, + ) + + assert response.status_code == 200 \ No newline at end of file