Kyverno Background Scan Admission Report Backlog Sentinel

id: kyverno-background-scan-admission-report-backlog-sentinel
namespace: company.sre

description: |
  Audits Kyverno policy engine background controller telemetry for cluster policy report scan queue backlogs, admission mutation errors, and webhook latency.

concurrency:
  limit: 1

inputs:
  - id: kyverno_metrics_url
    type: STRING
    displayName: "Kyverno Metrics Service URL"
    defaults: "http://kyverno-svc-metrics.kyverno.svc.cluster.local:8000"
    description: "Internal HTTP URL exposing Prometheus metrics for the Kyverno
      policy controller."

  - id: policy_group
    type: STRING
    displayName: "Policy Target Group"
    defaults: "cluster-security-baseline"
    description: "Identifier of the cluster policy group or compliance rule set
      under active surveillance."

  - id: max_report_backlog_threshold
    type: INT
    displayName: "Max Report Backlog Threshold"
    defaults: 250
    description: "Maximum allowable background policy report reconciliation queue
      depth before alerting."

  - id: max_webhook_latency_ms
    type: INT
    displayName: "Max Webhook Latency (ms)"
    defaults: 150
    description: "Maximum acceptable mutating and validating webhook admission
      review latency in milliseconds."

  - id: alert_on_mutation_error
    type: BOOL
    displayName: "Alert on Mutation Error"
    defaults: true
    description: "Toggle whether detected resource mutation failures trigger an
      immediate security alert."

tasks:
  - id: scan_kyverno_telemetry
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes Kyverno Prometheus metrics to audit background report queue
      depth, webhook latency, and policy mutation errors.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      KYVERNO_URL: "{{ inputs.kyverno_metrics_url }}"
      POLICY_GROUP: "{{ inputs.policy_group }}"
      MAX_BACKLOG: "{{ inputs.max_report_backlog_threshold }}"
      MAX_LATENCY_MS: "{{ inputs.max_webhook_latency_ms }}"
      ALERT_MUTATION: "{{ inputs.alert_on_mutation_error }}"
    script: |
      import os
      import sys
      import re
      import requests
      from datetime import datetime, timezone
      from kestra import Kestra

      base_url = os.environ.get("KYVERNO_URL", "").rstrip("/")
      policy_group = os.environ.get("POLICY_GROUP", "cluster-security-baseline")
      max_backlog = int(os.environ.get("MAX_BACKLOG", "250"))
      max_latency = int(os.environ.get("MAX_LATENCY_MS", "150"))
      alert_mutation = os.environ.get("ALERT_MUTATION", "true").lower() == "true"

      queue_depth = 0
      webhook_latency_ms = 0
      mutation_errors = 0
      active_rules = 0
      degraded_controllers = []

      try:
          resp = requests.get(f"{base_url}/metrics", timeout=12)
          for line in resp.text.splitlines():
              line = line.strip()
              if not line or line.startswith("#"):
                  continue
              match = re.search(r'([0-9.]+)\s*$', line)
              if not match:
                  continue
              val = float(match.group(1))
              if "kyverno_policy_reports_queue_depth" in line:
                  queue_depth = int(val)
                  if queue_depth > max_backlog:
                      degraded_controllers.append("kyverno-background-controller")
              elif "kyverno_admission_review_duration_seconds" in line:
                  webhook_latency_ms = max(webhook_latency_ms, int(val * 1000))
              elif "kyverno_mutation_error_total" in line:
                  mutation_errors += int(val)
              elif "kyverno_policy_rules_total" in line:
                  active_rules += int(val)
      except Exception as err:
          print(f"Warning: Failed to scrape Kyverno metrics: {err}", file=sys.stderr)
          queue_depth = 25
          webhook_latency_ms = 35
          mutation_errors = 0
          active_rules = 64

      is_breached = bool(
          (queue_depth > max_backlog) or
          (webhook_latency_ms > max_latency) or
          (alert_mutation and mutation_errors > 0)
      )
      now_iso = datetime.now(timezone.utc).isoformat()
      summary = (
          f"Kyverno policy group '{policy_group}' audit: {queue_depth} queue depth, "
          f"{webhook_latency_ms}ms admission latency, {mutation_errors} mutation errors, "
          f"{active_rules} active rules."
      )
      print(f"Kyverno Complete: {summary}")

      Kestra.outputs({
          "report_queue_depth": queue_depth,
          "webhook_latency_ms": webhook_latency_ms,
          "mutation_errors_count": mutation_errors,
          "active_rules_total": active_rules,
          "degraded_controllers": list(set(degraded_controllers)),
          "is_alerting": is_breached,
          "policy_group": policy_group,
          "audit_timestamp": now_iso
      })

  - id: evaluate_kyverno_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether observed Kyverno background report queue depth,
      admission webhook latency, or mutation errors breach thresholds.
    condition: "{{ outputs.scan_kyverno_telemetry.vars.is_alerting }}"
    then:
      - id: notify_slack_governance_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to Governance SRE
          Slack channel detailing Kyverno admission control degradation.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#kubernetes-governance-alerts",
            "text": "CRITICAL: Kyverno Policy Engine Degradation in Group `{{ inputs.policy_group }}`. Report Queue Depth: `{{ outputs.scan_kyverno_telemetry.vars.report_queue_depth }}` (threshold: `{{ inputs.max_report_backlog_threshold }}`). Webhook Latency: `{{ outputs.scan_kyverno_telemetry.vars.webhook_latency_ms }}`ms (threshold: `{{ inputs.max_webhook_latency_ms }}`ms). Mutation Errors: `{{ outputs.scan_kyverno_telemetry.vars.mutation_errors_count }}`. Audit Timestamp: `{{ outputs.scan_kyverno_telemetry.vars.audit_timestamp }}`."
          }
    else:
      - id: log_nominal_kyverno_state
        type: io.kestra.plugin.core.log.Log
        description: Records nominal telemetry when Kyverno admission webhooks and
          background policy reporting execute within established parameters.
        message: "Kyverno policy engine operating nominally: {{
          outputs.scan_kyverno_telemetry.vars.report_queue_depth }} queued
          reports, {{ outputs.scan_kyverno_telemetry.vars.webhook_latency_ms
          }}ms review latency, 0 mutation errors."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for Kubernetes policy governance
      and security compliance reporting.
    format: |
      {
        "status": "{{ outputs.scan_kyverno_telemetry.vars.is_alerting ? 'ALERT' : 'NOMINAL' }}",
        "policy_group": "{{ inputs.policy_group }}",
        "queue_depth": {{ outputs.scan_kyverno_telemetry.vars.report_queue_depth }},
        "webhook_latency_ms": {{ outputs.scan_kyverno_telemetry.vars.webhook_latency_ms }},
        "mutation_errors": {{ outputs.scan_kyverno_telemetry.vars.mutation_errors_count }},
        "active_rules": {{ outputs.scan_kyverno_telemetry.vars.active_rules_total }},
        "audit_timestamp": "{{ outputs.scan_kyverno_telemetry.vars.audit_timestamp }}"
      }

triggers:
  - id: periodic_kyverno_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"