diff --git a/app/api/routes/__init__.py b/app/api/routes/__init__.py index 65ad861..7f90875 100644 --- a/app/api/routes/__init__.py +++ b/app/api/routes/__init__.py @@ -2,6 +2,7 @@ from app.api.routes.auth import router as auth_router from app.api.routes.channels import router as channels_router +from app.api.routes.discovery import router as discovery_router from app.api.routes.escalations import router as escalations_router from app.api.routes.health import router as health_router from app.api.routes.hosts import router as hosts_router @@ -21,6 +22,7 @@ api_router.include_router(maintenances_router) api_router.include_router(problems_router) api_router.include_router(escalations_router) +api_router.include_router(discovery_router) api_router.include_router(hosts_router) api_router.include_router(ingest_router) api_router.include_router(item_triggers_router) diff --git a/app/api/routes/discovery.py b/app/api/routes/discovery.py new file mode 100644 index 0000000..715820d --- /dev/null +++ b/app/api/routes/discovery.py @@ -0,0 +1,81 @@ +import uuid +from datetime import UTC, datetime + +from fastapi import APIRouter, HTTPException, Response, status + +from app.api.deps import CurrentUser, DBSession +from app.core.schemas.discovery import DiscoveryRuleCreate, DiscoveryRuleRead +from app.core.services.discovery_service import DiscoveryService +from app.core.services.template_service import TemplateService + +router = APIRouter(prefix="/discovery-rules", tags=["discovery"]) + + +async def _validate_template( + session: DBSession, owner_id: uuid.UUID, payload: DiscoveryRuleCreate +) -> None: + if payload.template_id is not None: + if await TemplateService(session).get(payload.template_id, owner_id) is None: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="unknown or unowned template", + ) + + +@router.get("", response_model=list[DiscoveryRuleRead], summary="List discovery rules") +async def list_rules(session: DBSession, current_user: CurrentUser) -> list[DiscoveryRuleRead]: + rules = await DiscoveryService(session).list_for_owner(current_user.id) + return [DiscoveryRuleRead.model_validate(r) for r in rules] + + +@router.post( + "", + response_model=DiscoveryRuleRead, + status_code=status.HTTP_201_CREATED, + summary="Create a discovery rule", +) +async def create_rule( + payload: DiscoveryRuleCreate, session: DBSession, current_user: CurrentUser +) -> DiscoveryRuleRead: + await _validate_template(session, current_user.id, payload) + rule = await DiscoveryService(session).create(current_user.id, payload) + return DiscoveryRuleRead.model_validate(rule) + + +@router.get("/{rule_id}", response_model=DiscoveryRuleRead, summary="Retrieve a discovery rule") +async def get_rule( + rule_id: uuid.UUID, session: DBSession, current_user: CurrentUser +) -> DiscoveryRuleRead: + rule = await DiscoveryService(session).get(rule_id, current_user.id) + if rule is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Rule not found") + return DiscoveryRuleRead.model_validate(rule) + + +@router.post( + "/{rule_id}/scan", + summary="Scan a discovery rule now and provision newly-reachable hosts", +) +async def scan_rule_now( + rule_id: uuid.UUID, session: DBSession, current_user: CurrentUser +) -> dict[str, int]: + service = DiscoveryService(session) + rule = await service.get(rule_id, current_user.id) + if rule is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Rule not found") + created = await service.scan_rule(rule, datetime.now(UTC)) + return {"discovered": created} + + +@router.delete( + "/{rule_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a discovery rule" +) +async def delete_rule( + rule_id: uuid.UUID, session: DBSession, current_user: CurrentUser +) -> Response: + service = DiscoveryService(session) + rule = await service.get(rule_id, current_user.id) + if rule is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Rule not found") + await service.delete(rule) + return Response(status_code=status.HTTP_204_NO_CONTENT) diff --git a/app/core/models/discovery.py b/app/core/models/discovery.py new file mode 100644 index 0000000..8c0bba6 --- /dev/null +++ b/app/core/models/discovery.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import enum +import uuid +from datetime import datetime + +from sqlalchemy import DateTime, Enum, ForeignKey, Integer, String +from sqlalchemy.dialects.postgresql import UUID +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.db.session import Base +from app.core.models.mixins import Timestamped, UUIDPrimaryKey + + +class DiscoveryMethod(enum.StrEnum): + PING = "ping" + TCP = "tcp" + + +class DiscoveryRule(UUIDPrimaryKey, Timestamped, Base): + """A network range scanned on a schedule: responsive addresses that are not yet + a host are provisioned as hosts, optionally with a template's items applied.""" + + __tablename__ = "discovery_rules" + + owner_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + name: Mapped[str] = mapped_column(String(255), nullable=False) + cidr: Mapped[str] = mapped_column(String(64), nullable=False) + method: Mapped[DiscoveryMethod] = mapped_column( + Enum(DiscoveryMethod, name="discovery_method"), nullable=False + ) + # TCP port to probe when method is TCP (ignored for PING). + port: Mapped[int | None] = mapped_column(Integer, nullable=True) + # Template whose items are applied to each newly discovered host (optional). + template_id: Mapped[uuid.UUID | None] = mapped_column( + UUID(as_uuid=True), + ForeignKey("templates.id", ondelete="SET NULL"), + nullable=True, + ) + interval_seconds: Mapped[int] = mapped_column(Integer, nullable=False) + is_enabled: Mapped[bool] = mapped_column(nullable=False, default=True, server_default="true") + last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) diff --git a/app/core/schemas/discovery.py b/app/core/schemas/discovery.py new file mode 100644 index 0000000..2968c30 --- /dev/null +++ b/app/core/schemas/discovery.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import ipaddress +import uuid +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from app.core.models.discovery import DiscoveryMethod + +# Cap the scan size so a typo (e.g. a /8) can't launch millions of probes. +MAX_SCAN_HOSTS = 1024 + + +class DiscoveryRuleBase(BaseModel): + name: str = Field(min_length=1, max_length=255) + cidr: str = Field(min_length=1, max_length=64) + method: DiscoveryMethod = DiscoveryMethod.PING + port: int | None = Field(default=None, ge=1, le=65535) + template_id: uuid.UUID | None = None + interval_seconds: int = Field(default=3600, ge=60, le=604800) + is_enabled: bool = True + + @model_validator(mode="after") + def _validate(self) -> DiscoveryRuleBase: + try: + network = ipaddress.ip_network(self.cidr, strict=False) + except ValueError as exc: + raise ValueError(f"invalid CIDR: {exc}") from exc + hosts = ( + network.num_addresses + if network.prefixlen >= network.max_prefixlen - 1 + else sum(1 for _ in network.hosts()) + ) + if hosts > MAX_SCAN_HOSTS: + raise ValueError(f"CIDR covers {hosts} hosts; max is {MAX_SCAN_HOSTS}") + if self.method is DiscoveryMethod.TCP and self.port is None: + raise ValueError("port is required for TCP discovery") + return self + + +class DiscoveryRuleCreate(DiscoveryRuleBase): + pass + + +class DiscoveryRuleRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: uuid.UUID + name: str + cidr: str + method: DiscoveryMethod + port: int | None + template_id: uuid.UUID | None + interval_seconds: int + is_enabled: bool + last_run_at: datetime | None + created_at: datetime + updated_at: datetime diff --git a/app/core/services/discovery_service.py b/app/core/services/discovery_service.py new file mode 100644 index 0000000..c982699 --- /dev/null +++ b/app/core/services/discovery_service.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import asyncio +import ipaddress +import logging +import uuid +from collections.abc import Sequence +from datetime import datetime + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.models.discovery import DiscoveryMethod, DiscoveryRule +from app.core.models.host import Host +from app.core.models.monitor_result import ProbeStatus +from app.core.schemas.discovery import MAX_SCAN_HOSTS, DiscoveryRuleCreate +from app.core.services.template_service import TemplateService +from app.tasks.probes import _probe_ping, _probe_tcp + +logger = logging.getLogger(__name__) + +_SCAN_CONCURRENCY = 32 + + +async def check_reachable(address: str, method: DiscoveryMethod, port: int | None) -> bool: + """Probe a single address. Module-level so tests can stub it.""" + if method is DiscoveryMethod.TCP and port is not None: + outcome = await _probe_tcp(f"{address}:{port}") + else: + outcome = await _probe_ping(address) + return outcome.status is ProbeStatus.UP + + +class DiscoveryService: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def list_for_owner(self, owner_id: uuid.UUID) -> Sequence[DiscoveryRule]: + stmt = ( + select(DiscoveryRule) + .where(DiscoveryRule.owner_id == owner_id) + .order_by(DiscoveryRule.created_at.desc()) + ) + return (await self._session.execute(stmt)).scalars().all() + + async def get(self, rule_id: uuid.UUID, owner_id: uuid.UUID) -> DiscoveryRule | None: + stmt = select(DiscoveryRule).where( + DiscoveryRule.id == rule_id, DiscoveryRule.owner_id == owner_id + ) + return (await self._session.execute(stmt)).scalar_one_or_none() + + async def create(self, owner_id: uuid.UUID, data: DiscoveryRuleCreate) -> DiscoveryRule: + rule = DiscoveryRule( + owner_id=owner_id, + name=data.name, + cidr=data.cidr, + method=data.method, + port=data.port, + template_id=data.template_id, + interval_seconds=data.interval_seconds, + is_enabled=data.is_enabled, + ) + self._session.add(rule) + await self._session.commit() + await self._session.refresh(rule) + return rule + + async def delete(self, rule: DiscoveryRule) -> None: + await self._session.delete(rule) + await self._session.commit() + + async def due_rules(self, now: datetime) -> list[DiscoveryRule]: + stmt = select(DiscoveryRule).where(DiscoveryRule.is_enabled.is_(True)) + rules = (await self._session.execute(stmt)).scalars().all() + return [ + r + for r in rules + if r.last_run_at is None or (now - r.last_run_at).total_seconds() >= r.interval_seconds + ] + + async def scan_rule(self, rule: DiscoveryRule, now: datetime) -> int: + """Scan a rule's range, provision hosts for newly-reachable addresses (not + already a host), apply its template, and stamp last_run_at. Returns the count + of hosts created.""" + addresses = [str(a) for a in ipaddress.ip_network(rule.cidr, strict=False).hosts()][ + :MAX_SCAN_HOSTS + ] + + existing_rows = await self._session.execute( + select(Host.address).where(Host.owner_id == rule.owner_id, Host.address.in_(addresses)) + ) + existing = {addr for (addr,) in existing_rows} + candidates = [a for a in addresses if a not in existing] + + semaphore = asyncio.Semaphore(_SCAN_CONCURRENCY) + + async def _probe(address: str) -> tuple[str, bool]: + async with semaphore: + try: + return address, await check_reachable(address, rule.method, rule.port) + except (OSError, TimeoutError): + return address, False + + results = await asyncio.gather(*(_probe(a) for a in candidates)) + reachable = [address for address, ok in results if ok] + + templates = TemplateService(self._session) + created = 0 + for address in reachable: + host = Host( + name=address, + address=address, + owner_id=rule.owner_id, + description=f"Discovered by “{rule.name}”.", + ) + self._session.add(host) + await self._session.flush() + if rule.template_id is not None: + await templates.apply_to_host(rule.template_id, host) + created += 1 + + rule.last_run_at = now + await self._session.commit() + if created: + logger.info("discovery rule %s provisioned %d host(s)", rule.id, created) + return created diff --git a/app/tasks/scheduler.py b/app/tasks/scheduler.py index 27c87d9..705674c 100644 --- a/app/tasks/scheduler.py +++ b/app/tasks/scheduler.py @@ -15,6 +15,7 @@ from app.core.models.monitor import Monitor, MonitorStatus from app.core.models.monitor_result import MonitorResult, ProbeStatus from app.core.models.trigger import TriggerMetric +from app.core.services.discovery_service import DiscoveryService from app.core.services.escalation_service import EscalationService from app.core.services.maintenance_service import MaintenanceService from app.core.services.monitor_host_bridge import ensure_backing_items @@ -32,15 +33,18 @@ RETENTION_INTERVAL_SECONDS = 3600 POLL_ITEMS_INTERVAL_SECONDS = 15 ESCALATION_INTERVAL_SECONDS = 60 +DISCOVERY_INTERVAL_SECONDS = 60 _RECONCILE_JOB_ID = "__reconcile__" _PRUNE_JOB_ID = "__prune__" _POLL_ITEMS_JOB_ID = "__poll_items__" _ESCALATION_JOB_ID = "__escalation__" +_DISCOVERY_JOB_ID = "__discovery__" _RESERVED_JOB_IDS = { _RECONCILE_JOB_ID, _PRUNE_JOB_ID, _POLL_ITEMS_JOB_ID, _ESCALATION_JOB_ID, + _DISCOVERY_JOB_ID, } @@ -78,6 +82,15 @@ async def start(self) -> None: max_instances=1, coalesce=True, ) + self._scheduler.add_job( + _discovery_job, + trigger=IntervalTrigger(seconds=DISCOVERY_INTERVAL_SECONDS), + id=_DISCOVERY_JOB_ID, + replace_existing=True, + next_run_time=datetime.now(UTC) + timedelta(seconds=45), + max_instances=1, + coalesce=True, + ) self._scheduler.add_job( poll_due_items, trigger=IntervalTrigger(seconds=POLL_ITEMS_INTERVAL_SECONDS), @@ -320,5 +333,18 @@ async def _escalation_job() -> None: logger.info("dispatched %d escalation step(s)", len(deliveries)) +async def _discovery_job() -> None: + """Scan due discovery rules and provision hosts for newly-reachable addresses.""" + now = datetime.now(UTC) + async with SessionLocal() as session: + service = DiscoveryService(session) + rules = await service.due_rules(now) + for rule in rules: + try: + await service.scan_rule(rule, now) + except Exception: + logger.exception("discovery scan failed for rule %s", rule.id) + + def build_scheduler() -> ProbeScheduler: return ProbeScheduler() diff --git a/docs/roadmap.md b/docs/roadmap.md index 0fb2a58..319d2f4 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -120,11 +120,17 @@ GhostMonitor's reason to exist over a plain Zabbix clone is privacy (ghost-suite (webhook-only, validated on create and at fire time). - *(next)* long-range charts backed by trends. -## Phase 5 — Discovery & scale - -- Network and host auto-discovery with rule-based item provisioning. -- Distributed collection (proxy/agent fan-in) and history storage scaling - (evaluate TimescaleDB or a dedicated TSDB — measure first). +## Phase 5 — Discovery & scale *(in progress)* + +- ✅ **Network auto-discovery**: a `DiscoveryRule` (CIDR + ping/TCP method + optional + template + interval) is scanned on a schedule; responsive addresses that are not yet + a host are provisioned as hosts, with the rule's template items applied. Guardrails: + CIDR capped (≤ 1024 hosts), bounded scan concurrency, dedupe by address, owner-scoped. + REST CRUD + scan-now under `/api/discovery-rules`. *(next)* a web UI. +- *(deferred — measure first)* Distributed collection (proxy/agent fan-in) and history + storage scaling (TimescaleDB / a dedicated TSDB). Deliberately not built yet: it is + premature without real load data. Add ingestion-volume/throughput metrics first so + the decision is evidence-driven. --- diff --git a/migrations/env.py b/migrations/env.py index a7becf9..1da92dc 100644 --- a/migrations/env.py +++ b/migrations/env.py @@ -7,6 +7,7 @@ from app.core.config import get_settings from app.core.db.session import Base +from app.core.models import discovery as _discovery # noqa: F401 from app.core.models import escalation as _escalation # noqa: F401 from app.core.models import host as _host # noqa: F401 from app.core.models import ingestion_token as _ingestion_token # noqa: F401 diff --git a/migrations/versions/3c013fb9a5b6_discovery_rules.py b/migrations/versions/3c013fb9a5b6_discovery_rules.py new file mode 100644 index 0000000..cb2c62c --- /dev/null +++ b/migrations/versions/3c013fb9a5b6_discovery_rules.py @@ -0,0 +1,60 @@ +"""discovery rules + +Revision ID: 3c013fb9a5b6 +Revises: e2a6636ea1da +Create Date: 2026-06-30 21:56:10.199412 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "3c013fb9a5b6" +down_revision: str | Sequence[str] | None = "e2a6636ea1da" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "discovery_rules", + sa.Column("owner_id", sa.UUID(), nullable=False), + sa.Column("name", sa.String(length=255), nullable=False), + sa.Column("cidr", sa.String(length=64), nullable=False), + sa.Column("method", sa.Enum("PING", "TCP", name="discovery_method"), nullable=False), + sa.Column("port", sa.Integer(), nullable=True), + sa.Column("template_id", sa.UUID(), nullable=True), + sa.Column("interval_seconds", sa.Integer(), nullable=False), + sa.Column("is_enabled", sa.Boolean(), server_default="true", nullable=False), + sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("id", sa.UUID(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["template_id"], ["templates.id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_discovery_rules_owner_id"), "discovery_rules", ["owner_id"], unique=False + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + op.drop_index(op.f("ix_discovery_rules_owner_id"), table_name="discovery_rules") + op.drop_table("discovery_rules") + # Drop the enum type the table created (Alembic doesn't do this automatically). + sa.Enum(name="discovery_method").drop(op.get_bind(), checkfirst=True) diff --git a/tests/test_discovery.py b/tests/test_discovery.py new file mode 100644 index 0000000..eed52b3 --- /dev/null +++ b/tests/test_discovery.py @@ -0,0 +1,106 @@ +"""Network discovery: the scan engine (provision/dedupe), due gating, caps and API.""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any + +import httpx +import pytest + +from app.core.models.discovery import DiscoveryMethod, DiscoveryRule +from app.core.schemas.discovery import DiscoveryRuleCreate +from app.core.services import discovery_service as ds +from app.core.services.discovery_service import DiscoveryService +from app.core.services.host_service import HostService + +NOW = datetime(2026, 6, 30, 12, 0, tzinfo=UTC) + + +def _rule(owner_id: Any, **kw: Any) -> DiscoveryRule: + defaults: dict[str, Any] = { + "name": "lan", + "cidr": "10.9.9.0/30", + "method": DiscoveryMethod.PING, + "interval_seconds": 3600, + } + defaults.update(kw) + return DiscoveryRule(owner_id=owner_id, **defaults) + + +async def test_scan_provisions_reachable_and_dedupes( + session: Any, user: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + rule = _rule(user.id) + session.add(rule) + await session.commit() + + async def fake_reachable(address: str, method: Any, port: Any) -> bool: + return address == "10.9.9.1" # only .1 answers (of .1 and .2) + + monkeypatch.setattr(ds, "check_reachable", fake_reachable) + + created = await DiscoveryService(session).scan_rule(rule, NOW) + assert created == 1 + hosts = list(await HostService(session).list_for_owner(user.id)) + assert {h.address for h in hosts} == {"10.9.9.1"} + assert rule.last_run_at == NOW + + # A second scan finds it already provisioned → nothing new. + assert await DiscoveryService(session).scan_rule(rule, NOW) == 0 + + +async def test_due_rules_gating(session: Any, user: Any) -> None: + never = _rule(user.id, name="never", last_run_at=None) + fresh = _rule(user.id, name="fresh", last_run_at=NOW - timedelta(minutes=30)) + stale = _rule(user.id, name="stale", last_run_at=NOW - timedelta(hours=2)) + off = _rule(user.id, name="off", last_run_at=None, is_enabled=False) + session.add_all([never, fresh, stale, off]) + await session.commit() + + due = await DiscoveryService(session).due_rules(NOW) + assert {r.name for r in due} == {"never", "stale"} + + +def test_cidr_is_capped_and_validated() -> None: + with pytest.raises(ValueError, match="max is"): + DiscoveryRuleCreate(name="huge", cidr="10.0.0.0/8") + with pytest.raises(ValueError, match="invalid CIDR"): + DiscoveryRuleCreate(name="bad", cidr="not-a-cidr") + with pytest.raises(ValueError, match="port is required"): + DiscoveryRuleCreate(name="tcp", cidr="10.0.0.0/30", method=DiscoveryMethod.TCP) + + +async def test_discovery_api_crud_and_scan( + client: httpx.AsyncClient, + auth_headers: dict[str, str], + session: Any, + user: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + created = await client.post( + "/api/discovery-rules", + headers=auth_headers, + json={"name": "lan", "cidr": "10.9.9.0/30", "method": "ping", "interval_seconds": 3600}, + ) + assert created.status_code == 201, created.text + rule_id = created.json()["id"] + + bad_template = await client.post( + "/api/discovery-rules", + headers=auth_headers, + json={"name": "x", "cidr": "10.9.9.0/30", "template_id": str(uuid.uuid4())}, + ) + assert bad_template.status_code == 422 + + async def fake_reachable(address: str, method: Any, port: Any) -> bool: + return True + + monkeypatch.setattr(ds, "check_reachable", fake_reachable) + scanned = await client.post(f"/api/discovery-rules/{rule_id}/scan", headers=auth_headers) + assert scanned.status_code == 200 + assert scanned.json()["discovered"] == 2 # .1 and .2 + + deleted = await client.delete(f"/api/discovery-rules/{rule_id}", headers=auth_headers) + assert deleted.status_code == 204