Health Monitoring

Build plugins that monitor domain health using HealthMonitorHooks.

Overview

Health monitoring allows plugins to:

• Register domain health checks (HTTP ping, SSL expiry, etc.)
• Execute health checks when triggered by the monitor
• React to health status changes (HEALTHY → DEGRADED → UNHEALTHY)
• Detect statistical anomalies in health metrics

Quick Start

from typing import TYPE_CHECKING
from dvp_cms.plugins.base import Plugin
from dvp_cms.plugins.hookspec import HealthMonitorHooks

if TYPE_CHECKING:
    from dvp_cms.plugins.context import HookContext
    from dvp_cms.plugins.hooks.health_monitor import (
        HealthCheckDefinition,
        HealthCheckContext,
        HealthCheckResult,
        HealthSeverity,
        HealthStatus,
    )


class UptimePlugin(Plugin, HealthMonitorHooks):
    """Plugin that monitors domain uptime."""

    name = "uptime"
    version = "1.0.0"

    def dvp_register_health_checks(
        self,
        ctx: "HookContext",
    ) -> list["HealthCheckDefinition"]:
        """Register health checks this plugin provides."""
        from dvp_cms.plugins.hooks.health_monitor import (
            HealthCheckDefinition,
            HealthSeverity,
        )

        return [
            HealthCheckDefinition(
                check_id="uptime.http-ping",
                name="HTTP Ping Check",
                severity=HealthSeverity.CRITICAL,
                description="Verify domain responds to HTTP requests",
                interval_seconds=60,
                timeout_seconds=30,
            ),
        ]

    async def dvp_execute_health_check(
        self,
        check: "HealthCheckDefinition",
        ctx: "HookContext",
        execution_ctx: "HealthCheckContext",
    ) -> "HealthCheckResult":
        """Execute the health check."""
        from dvp_cms.plugins.hooks.health_monitor import (
            HealthCheckDetail,
            HealthCheckResult,
            HealthStatus,
        )

        if check.check_id == "uptime.http-ping":
            response_time, is_ok = await self._ping_domain(execution_ctx.domain)
            status = HealthStatus.HEALTHY if is_ok else HealthStatus.UNHEALTHY

            return HealthCheckResult(
                domain=execution_ctx.domain,
                overall_status=status,
                details=[
                    HealthCheckDetail(
                        check_id=check.check_id,
                        status=status,
                        severity=check.severity,
                        message=f"HTTP {'200 OK' if is_ok else 'Failed'} in {response_time}ms",
                        response_time_ms=response_time,
                    ),
                ],
            )

        return HealthCheckResult(
            domain=execution_ctx.domain,
            overall_status=HealthStatus.UNKNOWN,
        )

Hooks Reference

dvp_register_health_checks

Register health checks from your plugin. Called at startup and when plugins load.

def dvp_register_health_checks(
    self,
    ctx: "HookContext",
) -> list["HealthCheckDefinition"]:
    """
    Type: Collect hook
    Returns: List of health check definitions
    """
    return [
        HealthCheckDefinition(
            check_id="my-plugin.http-ping",
            name="HTTP Ping",
            severity=HealthSeverity.CRITICAL,
            description="Verify domain responds to HTTP requests",
            domains=[],  # Empty = all domains
            interval_seconds=60,
            timeout_seconds=30,
        ),
    ]

dvp_execute_health_check

Execute the actual health check logic.

async def dvp_execute_health_check(
    self,
    check: "HealthCheckDefinition",
    ctx: "HookContext",
    execution_ctx: "HealthCheckContext",
) -> "HealthCheckResult":
    """
    Type: Execute hook
    Returns: HealthCheckResult
    """
    if check.check_id == "my-plugin.http-ping":
        response_time = await self._ping_domain(execution_ctx.domain)
        return HealthCheckResult(
            domain=execution_ctx.domain,
            overall_status=HealthStatus.HEALTHY,
            details=[
                HealthCheckDetail(
                    check_id=check.check_id,
                    status=HealthStatus.HEALTHY,
                    severity=check.severity,
                    message=f"Response in {response_time}ms",
                    response_time_ms=response_time,
                ),
            ],
        )

dvp_health_status_changed

React to health status changes.

async def dvp_health_status_changed(
    self,
    ctx: "HookContext",
    domain: str,
    old_status: "HealthStatus",
    new_status: "HealthStatus",
    details: list["HealthCheckDetail"],
) -> None:
    """
    Type: Action hook
    Returns: None
    """
    if new_status == HealthStatus.UNHEALTHY:
        await self._send_alert(
            f"Domain {domain} is now UNHEALTHY",
            details=details,
        )

Data Classes

HealthSeverity

class HealthSeverity(str, Enum):
    CRITICAL = "critical"   # Blocks publishing
    WARNING = "warning"     # Warning only
    INFO = "info"           # Informational

HealthStatus

class HealthStatus(str, Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"
    UNKNOWN = "unknown"

See Also

• Hook Reference
• Scheduled Tasks
• Plugin Best Practices