Monitoring Geometry Validity Rates as an SLO

from __future__ import annotations

from dataclasses import dataclass

from prometheus_client import Counter, Gauge

# Emitted once per feature at the boundary. Three outcomes, bounded labels.
VALIDATION = Counter(
    "feature_validation_total",
    "Validation outcomes at the ingest boundary",
    ("layer", "source_group", "outcome"),      # accepted | repaired | quarantined
)

VALIDITY_SLI = Gauge("geometry_validity_sli", "Share accepted without repair",
                     ("layer",))
VALIDITY_TARGET = Gauge("geometry_validity_target", "The objective", ("layer",))


@dataclass(frozen=True)
class ValidityObjective:
    layer: str
    target: float             # share accepted without repair
    repair_ceiling: float     # share that may be repaired before it is a problem
    window_days: int = 30


OBJECTIVES = [
    # A cadastral register is authoritative: repairs there are a real concern.
    ValidityObjective("parcels", target=0.999, repair_ceiling=0.0009),
    # Hand-digitised habitat polygons are messier by nature, and that is expected.
    ValidityObjective("habitats", target=0.98, repair_ceiling=0.019),
    ValidityObjective("buildings", target=0.995, repair_ceiling=0.0045),
]


def record_outcome(layer: str, source_group: str, outcome: str) -> None:
    VALIDATION.labels(layer=layer, source_group=source_group, outcome=outcome).inc()

groups:
  - name: validity-slo
    rules:
      - record: slo:validity_sli
        expr: |
          sum by (layer) (rate(feature_validation_total{outcome="accepted"}[1h]))
          / sum by (layer) (rate(feature_validation_total[1h]))

      - record: slo:validity_error_ratio
        expr: 1 - slo:validity_sli

      # Repairs are not failures, but a rising repair share is an early warning
      # that a publisher's export has changed. Worth a ticket well before the
      # quarantine share moves.
      - record: slo:validity_repair_ratio
        expr: |
          sum by (layer) (rate(feature_validation_total{outcome="repaired"}[1h]))
          / sum by (layer) (rate(feature_validation_total[1h]))

      - alert: ValidityBudgetBurningFast
        expr: slo:validity_error_ratio > 14.4 * (1 - geometry_validity_target)
        for: 20m
        labels: { severity: page, team: data-stewardship }
        annotations:
          summary: "{{ $labels.layer }} validity budget burning at 14x"

      - alert: RepairShareClimbing
        expr: slo:validity_repair_ratio > on (layer) geometry_validity_repair_ceiling
        for: 2h
        labels: { severity: ticket, team: data-stewardship }
        annotations:
          summary: >
            {{ $labels.layer }} is repairing {{ $value | humanizePercentage }} of features —
            check whether the publisher's export changed

def test_outcomes_are_exhaustive(sample_delivery) -> None:
    """Every feature must land in exactly one bucket, or the SLI is wrong."""
    counts = validate_delivery(sample_delivery)
    assert counts["accepted"] + counts["repaired"] + counts["quarantined"] == \
        sample_delivery.feature_count


def test_sli_excludes_repairs_from_the_numerator() -> None:
    """A repaired feature loaded successfully and still counts against validity."""
    sli = compute_sli(accepted=980, repaired=15, quarantined=5)
    assert sli == pytest.approx(0.98)          # not 0.995


def test_repair_alert_fires_before_the_quarantine_alert(promtool) -> None:
    series = ramp_repair_share(start=0.0005, end=0.004, days=21)
    fired = promtool.evaluate(series)
    repair_at = fired["RepairShareClimbing"]
    validity_at = fired.get("ValidityBudgetBurningFast")
    assert repair_at is not None
    assert validity_at is None or repair_at < validity_at

SELECT source_group, defect_class,
       count(*) FILTER (WHERE outcome = 'repaired')    AS repaired,
       count(*) FILTER (WHERE outcome = 'quarantined') AS quarantined,
       min(checked_at) AS first_seen
FROM   validation_log
WHERE  checked_at > now() - interval '14 days' AND outcome <> 'accepted'
GROUP  BY source_group, defect_class
ORDER  BY repaired + quarantined DESC
LIMIT  15;