Scope boundary

Owns health, completeness, and integrity findings for AI-security sensors, collectors, schemas, clocks, sampling policies, queues, exporters, heartbeats, and end-to-end dead-man events. Telemetry loss proves impaired visibility, not service compromise or outage.

AID-D-005.002: AID-D-005.010 monitors shared telemetry-path and sensor health; AID-D-005.002 monitors the lifecycle, execution, and delivery health of individual detection rules. Sensor loss can blind many healthy rules, while one broken rule can fail on an otherwise complete pipeline.

AID-D-017: AID-D-005.010 determines whether telemetry is arriving completely and correctly; AID-D-017 measures end-to-end detection, alert-delivery, acknowledgement, triage, and coverage SLOs. Pipeline visibility is an input to service-level quality, not the complete detection outcome.

Implementation guidance

• AID-D-005.010-G001: Monitor sensor heartbeats, collector throughput and loss, schema rejection, clock offset, configured versus effective sampling, and an end-to-end synthetic dead-man event; fail closed to an explicit visibility-impaired state.

AID-D-005.010-G001

Concept and topology assumptions

A healthy AI service can still be security-blind if a sensor, collector, exporter, or backend path silently stops. This example assumes OpenTelemetry Collector v0.156.0, one Prometheus scrape target per Collector instance, a unique service.instance.id, one health series per sensor_id, exporter sending queues pinned to 1000 batches, and an expected-inventory exporter that publishes aidefend_expected_sensor_info and aidefend_expected_deadman_info. Collector receiver and exporter counters have different component labels and can diverge legitimately under routing, fan-out, retry, filtering, and sampling, so the rules do not subtract or ratio incompatible ingress and egress vectors. End-to-end loss is proved by the dead-man probe and expected-sensor reconciliation.

Step 1: Export current Collector metric names deterministically

Current Collector internal telemetry uses a pull reader. Explicitly disable Prometheus type and unit suffixes so the exported names remain the documented OTLP-style names used by the rules below.

# File: otelcol/config.yaml
service:
  telemetry:
    resource:
      service.name: otelcol
      # Inject a unique immutable value per pod/VM from deployment configuration.
      service.instance.id: otelcol-prod-a
      deployment.environment.name: production
    metrics:
      level: basic
      readers:
        - pull:
            exporter:
              prometheus:
                host: 0.0.0.0
                port: 8888
                without_type_suffix: true
                without_units: true
                with_resource_constant_labels:
                  included:
                    - service.instance.id
                    - deployment.environment.name

With this explicit configuration, counters are exported without Prometheus _total and unit suffixes: for example otelcol_receiver_refused_spans, otelcol_exporter_send_failed_spans, and otelcol_process_uptime. Pin the Collector image and fail deployment if a scrape-contract test cannot find these names plus the normalized service_instance_id label.

Step 2: Use label-safe Prometheus alerts

# File: monitoring/telemetry-pipeline-alerts.yaml
groups:
  - name: ai-telemetry-pipeline-health
    interval: 30s
    rules:
      # Detect an individual expected sensor whose heartbeat series disappeared.
      - alert: ExpectedAiSecuritySensorMissing
        expr: (aidefend_expected_sensor_info == 1) unless on(sensor_id) aidefend_sensor_last_heartbeat_seconds
        for: 2m
        labels: {severity: critical}

      - alert: AiSecuritySensorHeartbeatStale
        expr: (time() - aidefend_sensor_last_heartbeat_seconds > 120) and on(sensor_id) (aidefend_expected_sensor_info == 1)
        for: 2m
        labels: {severity: critical}

      # 'up == 0' catches a known target; absent() catches loss of the whole job.
      - alert: OTelCollectorTargetDown
        expr: up{job="otelcol-internal"} == 0
        for: 2m
        labels: {severity: critical}

      - alert: OTelCollectorJobMissing
        expr: absent(up{job="otelcol-internal"})
        for: 2m
        labels: {severity: critical}

      - alert: OTelReceiverRefusingSpans
        expr: sum by(service_instance_id, receiver, transport) (increase(otelcol_receiver_refused_spans[5m])) > 0
        for: 1m
        labels: {severity: high}

      - alert: OTelExporterSendFailures
        expr: sum by(service_instance_id, exporter) (increase(otelcol_exporter_send_failed_spans[5m])) > 0
        for: 1m
        labels: {severity: high}

      - alert: OTelExporterQueueSaturation
        # This deployment pins every exporter sending queue to 1000 batches.
        # Generate this threshold from deployed config if capacities differ.
        expr: max by(service_instance_id, exporter) (otelcol_exporter_queue_size) > 800
        for: 5m
        labels: {severity: high}

      - alert: TelemetrySchemaDrift
        expr: sum by(sensor_id) (increase(aidefend_sensor_schema_rejections_total[5m])) > 0
        for: 1m
        labels: {severity: high}

      - alert: TelemetryClockSkew
        expr: abs(aidefend_sensor_clock_offset_seconds) > 2
        for: 5m
        labels: {severity: high}

      - alert: TelemetrySamplingPolicyChanged
        expr: abs(aidefend_sensor_effective_sample_ratio - on(sensor_id) aidefend_sensor_configured_sample_ratio) > 0.01
        for: 5m
        labels: {severity: high}

      - alert: ExpectedDeadManProbeMissing
        expr: (aidefend_expected_deadman_info == 1) unless on(probe_id) aidefend_deadman_last_seen_seconds
        for: 2m
        labels: {severity: critical}

      - alert: TelemetryDeadManStale
        expr: (time() - aidefend_deadman_last_seen_seconds > 180) and on(probe_id) (aidefend_expected_deadman_info == 1)
        for: 1m
        labels: {severity: critical}

Step 3: Reconcile every expected sensor and sign the evidence

from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
PROMETHEUS_REQUEST_TIMEOUT_SECONDS = _aidefend_positive_float("D005010_PROMETHEUS_REQUEST_TIMEOUT_SECONDS")
SIGNATURE_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D005010_SIGNATURE_COMMAND_TIMEOUT_SECONDS")
PROMETHEUS_MAX_RESPONSE_BYTES = _aidefend_positive_int("D005010_PROMETHEUS_MAX_RESPONSE_BYTES")
PROMETHEUS_RESPONSE_CHUNK_BYTES = _aidefend_positive_int("D005010_PROMETHEUS_RESPONSE_CHUNK_BYTES")
if PROMETHEUS_MAX_RESPONSE_BYTES < 2 or PROMETHEUS_RESPONSE_CHUNK_BYTES > PROMETHEUS_MAX_RESPONSE_BYTES:
    raise RuntimeError("signed Prometheus response bounds are invalid")

# File: monitoring/reconcile_telemetry_health.py

import argparse
import base64
import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Any

import requests
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_write_private(path: Path, payload: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def _aidefend_strict_json(raw: bytes) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )


def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                ["cosign", "verify-blob", "--key", key_argument,
                 "--bundle", str(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw


def canonical_json(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")


def reject_duplicate_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate Prometheus JSON key: {key}")
        value[key] = item
    return value


def read_prometheus_json(response):
    try:
        chunks, observed = [], 0
        for chunk in response.iter_content(chunk_size=PROMETHEUS_RESPONSE_CHUNK_BYTES):
            observed += len(chunk)
            if observed > PROMETHEUS_MAX_RESPONSE_BYTES:
                raise RuntimeError("Prometheus response exceeds signed bound")
            chunks.append(chunk)
        return json.loads(b"".join(chunks).decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)))
    finally:
        response.close()


def query_vector(prometheus_url: str, query: str) -> list[dict]:
    response = requests.get(
        prometheus_url.rstrip("/") + "/api/v1/query",
        params={"query": query},
        timeout=PROMETHEUS_REQUEST_TIMEOUT_SECONDS,
        allow_redirects=False,
        stream=True,
        headers={"Accept-Encoding": "identity"},
    )
    response.raise_for_status()
    payload = read_prometheus_json(response)
    if payload.get("status") != "success":
        raise RuntimeError(f"Prometheus query failed: {query}")
    data = payload.get("data", {})
    if data.get("resultType") != "vector":
        raise RuntimeError(f"Expected instant vector for: {query}")
    return data.get("result", [])


def index_unique(rows: list[dict], label: str, metric_name: str) -> dict[str, dict]:
    indexed: dict[str, dict] = {}
    for row in rows:
        identity = str(row.get("metric", {}).get(label, ""))
        if not identity:
            raise RuntimeError(f"{metric_name} row lacks {label}")
        if identity in indexed:
            raise RuntimeError(f"{metric_name} has duplicate series for {label}={identity}")
        indexed[identity] = row
    return indexed


def sample_value(row: dict, metric_name: str) -> float:
    try:
        value = float(row["value"][1])
    except (KeyError, IndexError, TypeError, ValueError) as exc:
        raise RuntimeError(f"Malformed {metric_name} sample") from exc
    if not math.isfinite(value):
        raise RuntimeError(f"Non-finite {metric_name} sample")
    return value


def add_finding(findings: list[dict], *, kind: str, identity: str, detail: str) -> None:
    findings.append({"kind": kind, "identity": identity, "detail": detail})


def reconcile(config: dict, prometheus_url: str) -> dict:
    sensors = config.get("sensors")
    probes = config.get("deadman_probes")
    if not isinstance(sensors, list) or not sensors:
        raise RuntimeError("expected-sensor registry is empty")
    if not isinstance(probes, list) or not probes:
        raise RuntimeError("expected dead-man registry is empty")

    sensor_ids = [str(item.get("sensor_id", "")) for item in sensors]
    probe_ids = [str(item.get("probe_id", "")) for item in probes]
    if any(not value for value in sensor_ids + probe_ids):
        raise RuntimeError("registry contains an empty identity")
    if len(sensor_ids) != len(set(sensor_ids)) or len(probe_ids) != len(set(probe_ids)):
        raise RuntimeError("registry contains duplicate identities")

    heartbeats = index_unique(
        query_vector(prometheus_url, "aidefend_sensor_last_heartbeat_seconds"),
        "sensor_id",
        "sensor heartbeat",
    )
    schemas = index_unique(
        query_vector(prometheus_url, "aidefend_sensor_schema_info == 1"),
        "sensor_id",
        "sensor schema",
    )
    clock_offsets = index_unique(
        query_vector(prometheus_url, "aidefend_sensor_clock_offset_seconds"),
        "sensor_id",
        "sensor clock offset",
    )
    effective_sampling = index_unique(
        query_vector(prometheus_url, "aidefend_sensor_effective_sample_ratio"),
        "sensor_id",
        "effective sampling",
    )
    deadman = index_unique(
        query_vector(prometheus_url, "aidefend_deadman_last_seen_seconds"),
        "probe_id",
        "dead-man last seen",
    )

    now = time.time()
    findings: list[dict] = []
    for expected in sensors:
        sensor_id = str(expected["sensor_id"])
        required = {
            "heartbeat": heartbeats.get(sensor_id),
            "schema": schemas.get(sensor_id),
            "clock": clock_offsets.get(sensor_id),
            "sampling": effective_sampling.get(sensor_id),
        }
        for metric_name, row in required.items():
            if row is None:
                add_finding(
                    findings,
                    kind="missing_sensor_metric",
                    identity=sensor_id,
                    detail=metric_name,
                )
        if any(row is None for row in required.values()):
            continue

        heartbeat = sample_value(required["heartbeat"], "sensor heartbeat")
        max_age = int(expected.get("max_heartbeat_age_seconds", 120))
        if heartbeat > now + 30 or now - heartbeat > max_age:
            add_finding(findings, kind="stale_or_future_heartbeat", identity=sensor_id, detail=str(heartbeat))

        actual_schema = str(required["schema"]["metric"].get("schema_version", ""))
        if actual_schema != str(expected["schema_version"]):
            add_finding(findings, kind="schema_drift", identity=sensor_id, detail=actual_schema)

        offset = sample_value(required["clock"], "sensor clock offset")
        if abs(offset) > float(expected.get("max_clock_offset_seconds", 2)):
            add_finding(findings, kind="clock_skew", identity=sensor_id, detail=str(offset))

        actual_ratio = sample_value(required["sampling"], "effective sampling")
        configured_ratio = float(expected["configured_sample_ratio"])
        if not 0.0 <= configured_ratio <= 1.0:
            raise RuntimeError(f"invalid configured sampling ratio for {sensor_id}")
        if abs(actual_ratio - configured_ratio) > 0.01:
            add_finding(findings, kind="sampling_drift", identity=sensor_id, detail=str(actual_ratio))

    for expected in probes:
        probe_id = str(expected["probe_id"])
        row = deadman.get(probe_id)
        if row is None:
            add_finding(findings, kind="missing_deadman_probe", identity=probe_id, detail="no series")
            continue
        last_seen = sample_value(row, "dead-man last seen")
        max_age = int(expected.get("max_age_seconds", 180))
        if last_seen > now + 30 or now - last_seen > max_age:
            add_finding(findings, kind="stale_or_future_deadman", identity=probe_id, detail=str(last_seen))

    return {
        "schema_version": "aidefend.telemetry_health_evidence.v1",
        "generated_at_unix": int(now),
        "registry_sha256": hashlib.sha256(canonical_json(config)).hexdigest(),
        "expected_sensor_count": len(sensors),
        "expected_deadman_count": len(probes),
        "finding_count": len(findings),
        "control_status": "PASS",
        "finding_status": "FINDING" if findings else "NO_FINDING",
        "visibility_state": (
            "visibility_impaired" if findings else "visibility_verified"
        ),
        "findings": findings,
    }


def sign_report(report: dict, private_key_path: str, key_id: str) -> dict:
    loaded = serialization.load_pem_private_key(
        Path(private_key_path).read_bytes(),
        password=None,
    )
    if not isinstance(loaded, Ed25519PrivateKey):
        raise RuntimeError("evidence signing key must be Ed25519")
    signature = loaded.sign(canonical_json(report))
    return {
        "report": report,
        "signature": {
            "algorithm": "Ed25519",
            "key_id": key_id,
            "value_base64": base64.b64encode(signature).decode("ascii"),
        },
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--registry", required=True)
    parser.add_argument("--registry-signature", required=True)
    parser.add_argument("--registry-verify-key", required=True)
    parser.add_argument("--prometheus-url", required=True)
    parser.add_argument("--signing-key", required=True)
    parser.add_argument("--key-id", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()

    registry_bytes = None
    try:
        registry_bytes = _aidefend_capture_signed_bytes(
            Path(args.registry), Path(args.registry_signature),
            args.registry_verify_key, SIGNATURE_COMMAND_TIMEOUT_SECONDS,
        )
        registry = _aidefend_strict_json(registry_bytes)
        report = reconcile(registry, args.prometheus_url)
    except Exception as exc:
        report = {
            "schema_version": "aidefend.telemetry_health_evidence.v1",
            "generated_at_unix": int(time.time()),
            "registry_sha256": (
                hashlib.sha256(registry_bytes).hexdigest()
                if registry_bytes is not None else None
            ),
            "expected_sensor_count": None,
            "expected_deadman_count": None,
            "finding_count": 1,
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "visibility_state": "visibility_impaired",
            "findings": [{
                "kind": "reconciliation_error",
                "identity": "telemetry_pipeline",
                "detail": (exc.__class__.__name__ + ": " + str(exc))[:500],
            }],
        }
    envelope = sign_report(report, args.signing_key, args.key_id)
    Path(args.output).write_text(
        json.dumps(envelope, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    if (
        report["control_status"] != "PASS"
        or report["finding_status"] == "FINDING"
    ):
        raise SystemExit("telemetry visibility is impaired")


if __name__ == "__main__":
    main()

Verify the detector and retain evidence

Run the reconciler every five minutes with a registry that names every required sensor and dead-man probe. Store its signed envelope, registry, Collector config digest, raw scrape, and alert evaluations in the append-only archive from AID-D-005.001. An independent auditor verifies the Ed25519 signature, replays every Prometheus query, confirms no expected identity was omitted, and traces each synthetic event through the final backend.

Action: Pin the Collector version and scrape contract, deploy the label-safe alerts, and fail to visibility_impaired on any missing individual sensor or dead-man probe. Use separate controls for protected-service availability.

Open-source tools

• OpenTelemetry Collector
• Prometheus
• Prometheus Alertmanager
• Grafana
• Vector
• Healthchecks

Commercial / hosted tools

• Datadog
• Splunk Observability Cloud
• New Relic
• AWS CloudWatch
• Google Cloud Monitoring
• Azure Monitor

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• Compromised Observability Tools (L5)
• Poisoning Observability Data (L5)
• Evasion of Detection (L5)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AITech-9.2 Detection Evasion (sensor loss, deadman failures and reduced observation coverage)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Agents - Core 13.8: Repudiation & Untraceability (dead-man monitoring detects missing action telemetry)