titan-iac/scripts/tests/test_monitoring_query_capacity.py

"""Protect the monitoring backend from known query-starvation regressions."""

from pathlib import Path

import yaml


REPO_ROOT = Path(__file__).resolve().parents[2]


def _documents(path: Path) -> list[dict]:
    """Load every non-empty YAML document from a repository manifest."""
    return [document for document in yaml.safe_load_all(path.read_text()) if document]


def test_victoria_metrics_has_dashboard_burst_headroom() -> None:
    """Keep search capacity and pod resources above the proven failure floor."""
    manifests = _documents(REPO_ROOT / "services/monitoring/helmrelease.yaml")
    release = next(
        manifest
        for manifest in manifests
        if manifest.get("kind") == "HelmRelease"
        and manifest.get("metadata", {}).get("name") == "victoria-metrics-single"
    )
    server = release["spec"]["values"]["server"]

    assert int(server["extraArgs"]["search.maxConcurrentRequests"]) >= 4
    assert server["extraArgs"]["search.maxQueryDuration"] == "1m"
    assert server["extraArgs"]["search.maxQueueDuration"] == "30s"
    assert server["resources"]["requests"]["memory"] == "2Gi"
    assert server["resources"]["limits"]["cpu"] == "2"
    assert server["resources"]["limits"]["memory"] == "4Gi"


def test_yearly_availability_reuses_the_hourly_rollup() -> None:
    """Prevent the yearly rule from rescanning raw cluster metrics for 365 days."""
    manifest = _documents(
        REPO_ROOT / "services/monitoring/vmalert-atlas-availability.yaml"
    )[0]
    rules = yaml.safe_load(manifest["data"]["atlas-availability.yaml"])["groups"][0]
    yearly = next(
        rule
        for rule in rules["rules"]
        if rule["record"] == "atlas:availability:ratio_365d"
    )

    assert "atlas:availability:ratio_1h" in yearly["expr"]
    assert "kube_node_status_condition" not in yearly["expr"]
    assert "[365d:1h]" not in yearly["expr"]


def test_quality_rollups_do_not_run_every_minute() -> None:
    """Keep high-cardinality quality rollups below the backend saturation cadence."""
    manifest = _documents(
        REPO_ROOT / "services/monitoring/vmalert-atlas-availability.yaml"
    )[0]
    quality = yaml.safe_load(manifest["data"]["platform-quality.yaml"])["groups"][0]

    assert quality["interval"] == "5m"


def test_vmalert_reloads_updated_rule_files() -> None:
    """Make Flux ConfigMap updates take effect without manual pod revision bumps."""
    manifests = _documents(
        REPO_ROOT / "services/monitoring/vmalert-atlas-availability.yaml"
    )
    deployment = next(
        manifest for manifest in manifests if manifest.get("kind") == "Deployment"
    )
    args = deployment["spec"]["template"]["spec"]["containers"][0]["args"]

    assert "-configCheckInterval=30s" in args