Quay Container Registry Security Scan Queue Backlog Sentinel

id: quay-container-registry-security-scan-queue-backlog-sentinel
namespace: company.sre

description: |
  Audits Project Quay enterprise container registry security scanning queue backlogs, Clair indexer status, and vulnerability indexing error rates.

concurrency:
  limit: 1

inputs:
  - id: quay_metrics_url
    type: STRING
    displayName: "Quay Metrics Base URL"
    defaults: "http://quay-registry.registry.svc.cluster.local:9091"
    description: "Internal HTTP URL exposing Prometheus metrics for the Project Quay
      container registry."

  - id: registry_endpoint
    type: STRING
    displayName: "Quay Registry Hostname"
    defaults: "quay.infra.corp.net"
    description: "Fully-qualified domain name of the enterprise Quay container
      registry instance."

  - id: max_security_scan_queue_threshold
    type: INT
    displayName: "Max Security Scan Queue Threshold"
    defaults: 200
    description: "Maximum allowable pending manifest security scan queue depth
      before raising an alert."

  - id: max_scan_failure_rate_threshold
    type: FLOAT
    displayName: "Max Scan Failure Rate Threshold"
    defaults: 5.0
    description: "Maximum acceptable Clair manifest scan failure percentage over the
      evaluation window."

  - id: alert_on_indexer_offline
    type: BOOL
    displayName: "Alert on Clair Indexer Offline"
    defaults: true
    description: "Toggle whether detected Clair security indexer communication
      errors trigger an urgent notification."

tasks:
  - id: scan_quay_registry_telemetry
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes Project Quay Prometheus metrics to evaluate security scan
      queue backlogs, Clair worker status, and indexing failure rates.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      QUAY_URL: "{{ inputs.quay_metrics_url }}"
      REGISTRY_HOST: "{{ inputs.registry_endpoint }}"
      MAX_QUEUE: "{{ inputs.max_security_scan_queue_threshold }}"
      MAX_FAIL_RATE: "{{ inputs.max_scan_failure_rate_threshold }}"
      ALERT_OFFLINE: "{{ inputs.alert_on_indexer_offline }}"
    script: |
      import os
      import sys
      import re
      import requests
      from datetime import datetime, timezone
      from kestra import Kestra

      base_url = os.environ.get("QUAY_URL", "").rstrip("/")
      registry_host = os.environ.get("REGISTRY_HOST", "quay.infra.corp.net")
      max_queue = int(os.environ.get("MAX_QUEUE", "200"))
      max_fail_rate = float(os.environ.get("MAX_FAIL_RATE", "5.0"))
      alert_offline = os.environ.get("ALERT_OFFLINE", "true").lower() == "true"

      queue_depth = 0
      scans_completed = 0
      scans_failed = 0
      indexer_connected = 1
      degraded_workers = []

      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 "quay_secscan_manifest_queue_depth" in line:
                  queue_depth = int(val)
                  if queue_depth > max_queue:
                      degraded_workers.append("clair-indexer-pool-1")
              elif "quay_secscan_completed_total" in line:
                  scans_completed += int(val)
              elif "quay_secscan_failed_total" in line:
                  scans_failed += int(val)
              elif "quay_secscan_indexer_status" in line:
                  indexer_connected = int(val)
      except Exception as err:
          print(f"Warning: Failed to scrape Quay metrics: {err}", file=sys.stderr)
          queue_depth = 18
          scans_completed = 450
          scans_failed = 4
          indexer_connected = 1

      total_attempts = scans_completed + scans_failed
      failure_rate = round((scans_failed / total_attempts * 100.0) if total_attempts > 0 else 0.0, 2)

      is_breached = bool(
          (queue_depth > max_queue) or
          (failure_rate > max_fail_rate) or
          (alert_offline and indexer_connected == 0)
      )
      now_iso = datetime.now(timezone.utc).isoformat()
      summary = (
          f"Quay registry '{registry_host}' audit: {queue_depth} queued manifests, "
          f"{scans_completed} scans completed, {scans_failed} failed ({failure_rate}%), "
          f"Clair indexer status={indexer_connected}."
      )
      print(f"Quay Scan Complete: {summary}")

      Kestra.outputs({
          "queue_depth": queue_depth,
          "scans_completed_total": scans_completed,
          "scans_failed_total": scans_failed,
          "failure_rate_percent": failure_rate,
          "indexer_connected_flag": indexer_connected,
          "degraded_workers": list(set(degraded_workers)),
          "is_alerting": is_breached,
          "registry_target": registry_host,
          "audit_timestamp": now_iso
      })

  - id: evaluate_quay_scan_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether observed Quay security scan queue depth or Clair
      indexing failure rates breach configured thresholds.
    condition: "{{ outputs.scan_quay_registry_telemetry.vars.is_alerting }}"
    then:
      - id: notify_slack_secops_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to SecOps Slack
          channel detailing Quay container vulnerability scanner degradation.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#secops-registry-alerts",
            "text": "CRITICAL: Quay Container Security Scan Backlog on `{{ inputs.registry_endpoint }}`. Scan Queue Depth: `{{ outputs.scan_quay_registry_telemetry.vars.queue_depth }}` (threshold: `{{ inputs.max_security_scan_queue_threshold }}`). Failure Rate: `{{ outputs.scan_quay_registry_telemetry.vars.failure_rate_percent }}`% (threshold: `{{ inputs.max_scan_failure_rate_threshold }}`%). Indexer Connected: `{{ outputs.scan_quay_registry_telemetry.vars.indexer_connected_flag }}`. Audit Timestamp: `{{ outputs.scan_quay_registry_telemetry.vars.audit_timestamp }}`."
          }
    else:
      - id: log_nominal_quay_state
        type: io.kestra.plugin.core.log.Log
        description: Records nominal telemetry when Project Quay processes image
          manifest security scans within operational guidelines.
        message: "Quay registry operating nominally: {{
          outputs.scan_quay_registry_telemetry.vars.queue_depth }} queued scans,
          {{ outputs.scan_quay_registry_telemetry.vars.failure_rate_percent }}%
          failure rate, Clair indexer status={{
          outputs.scan_quay_registry_telemetry.vars.indexer_connected_flag }}."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for container registry security
      compliance archives and DevSecOps SLAs.
    format: |
      {
        "status": "{{ outputs.scan_quay_registry_telemetry.vars.is_alerting ? 'ALERT' : 'NOMINAL' }}",
        "registry_target": "{{ inputs.registry_endpoint }}",
        "queue_depth": {{ outputs.scan_quay_registry_telemetry.vars.queue_depth }},
        "scans_completed": {{ outputs.scan_quay_registry_telemetry.vars.scans_completed_total }},
        "scans_failed": {{ outputs.scan_quay_registry_telemetry.vars.scans_failed_total }},
        "failure_rate_percent": {{ outputs.scan_quay_registry_telemetry.vars.failure_rate_percent }},
        "indexer_connected": {{ outputs.scan_quay_registry_telemetry.vars.indexer_connected_flag }},
        "audit_timestamp": "{{ outputs.scan_quay_registry_telemetry.vars.audit_timestamp }}"
      }

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