feat(discovery): network auto-discovery with rule-based provisioning

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)

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)

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)

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

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

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()