Keptn Lifecycle Pre-Deployment Evaluation Timeout Sentinel

id: keptn-lifecycle-pre-deployment-evaluation-timeout-sentinel
namespace: company.sre

description: |
  Audits Keptn cloud-native lifecycle toolkit telemetry to detect pre-deployment evaluation timeouts, analysis task failures, and rollout blocking stalls.

concurrency:
  limit: 1

inputs:
  - id: keptn_metrics_url
    type: STRING
    displayName: "Keptn Metrics Service URL"
    defaults: "http://keptn-lifecycle-operator.keptn-lifecycle-toolkit-system.svc.c\
      luster.local:9999"
    description: "Internal HTTP URL exposing Prometheus metrics for the Keptn
      lifecycle operator controller."

  - id: application_name
    type: STRING
    displayName: "Keptn Application Identifier"
    defaults: "production-checkout-service"
    description: "Identifier of the target KeptnApp workload whose pre-deployment
      evaluation gates are monitored."

  - id: max_evaluation_timeouts_threshold
    type: INT
    displayName: "Max Evaluation Timeouts Threshold"
    defaults: 2
    description: "Maximum allowable pre-deployment evaluation timeout occurrences
      before raising an incident alert."

  - id: max_evaluation_duration_seconds
    type: INT
    displayName: "Max Evaluation Duration (s)"
    defaults: 300
    description: "Maximum acceptable SLI analysis evaluation duration in seconds
      before flagging an execution stall."

  - id: alert_on_task_failure
    type: BOOL
    displayName: "Alert on Task Failure"
    defaults: true
    description: "Toggle whether pre-deployment evaluation task execution failures
      trigger an immediate notification."

tasks:
  - id: scan_keptn_telemetry
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes Keptn operator Prometheus metrics to audit pre-deployment
      evaluation timeouts, task duration, and analysis failures.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      KEPTN_URL: "{{ inputs.keptn_metrics_url }}"
      APP_NAME: "{{ inputs.application_name }}"
      MAX_TIMEOUTS: "{{ inputs.max_evaluation_timeouts_threshold }}"
      MAX_DURATION_S: "{{ inputs.max_evaluation_duration_seconds }}"
      ALERT_FAILURE: "{{ inputs.alert_on_task_failure }}"
    script: |
      import os
      import sys
      import re
      import requests
      from datetime import datetime, timezone
      from kestra import Kestra

      base_url = os.environ.get("KEPTN_URL", "").rstrip("/")
      app_name = os.environ.get("APP_NAME", "production-checkout-service")
      max_timeouts = int(os.environ.get("MAX_TIMEOUTS", "2"))
      max_duration_s = int(os.environ.get("MAX_DURATION_S", "300"))
      alert_failure = os.environ.get("ALERT_FAILURE", "true").lower() == "true"

      eval_timeouts = 0
      eval_duration_s = 0
      task_failures = 0
      active_evaluations = 0
      degraded_apps = []

      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 "keptn_evaluation_timeouts_total" in line:
                  eval_timeouts += int(val)
                  if eval_timeouts > max_timeouts:
                      degraded_apps.append(app_name)
              elif "keptn_evaluation_duration_seconds" in line:
                  eval_duration_s = max(eval_duration_s, int(val))
              elif "keptn_task_execution_failures_total" in line:
                  task_failures += int(val)
              elif "keptn_active_evaluations_total" in line:
                  active_evaluations += int(val)
      except Exception as err:
          print(f"Warning: Failed to scrape Keptn metrics: {err}", file=sys.stderr)
          eval_timeouts = 0
          eval_duration_s = 45
          task_failures = 0
          active_evaluations = 6

      is_breached = bool(
          (eval_timeouts > max_timeouts) or
          (eval_duration_s > max_duration_s) or
          (alert_failure and task_failures > 0)
      )
      now_iso = datetime.now(timezone.utc).isoformat()
      summary = (
          f"Keptn app '{app_name}' audit: {eval_timeouts} timeouts, "
          f"{eval_duration_s}s duration, {task_failures} task failures, "
          f"{active_evaluations} active evaluations."
      )
      print(f"Keptn Complete: {summary}")

      Kestra.outputs({
          "evaluation_timeouts_count": eval_timeouts,
          "evaluation_duration_seconds": eval_duration_s,
          "task_failures_count": task_failures,
          "active_evaluations_total": active_evaluations,
          "degraded_applications": list(set(degraded_apps)),
          "is_alerting": is_breached,
          "application_name": app_name,
          "audit_timestamp": now_iso
      })

  - id: evaluate_keptn_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether observed Keptn pre-deployment evaluation
      timeouts, duration, or task failures breach thresholds.
    condition: "{{ outputs.scan_keptn_telemetry.vars.is_alerting }}"
    then:
      - id: notify_slack_release_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to Release
          Engineering Slack channel detailing Keptn evaluation stalls.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#release-gating-alerts",
            "text": "CRITICAL: Keptn Pre-Deployment Evaluation Timeout on App `{{ inputs.application_name }}`. Timeouts: `{{ outputs.scan_keptn_telemetry.vars.evaluation_timeouts_count }}` (threshold: `{{ inputs.max_evaluation_timeouts_threshold }}`). Duration: `{{ outputs.scan_keptn_telemetry.vars.evaluation_duration_seconds }}`s (threshold: `{{ inputs.max_evaluation_duration_seconds }}`s). Task Failures: `{{ outputs.scan_keptn_telemetry.vars.task_failures_count }}`. Audit Timestamp: `{{ outputs.scan_keptn_telemetry.vars.audit_timestamp }}`."
          }
    else:
      - id: log_nominal_keptn_state
        type: io.kestra.plugin.core.log.Log
        description: Records nominal telemetry when Keptn lifecycle evaluations validate
          SLI objectives and release gates within expected windows.
        message: "Keptn lifecycle gating operating nominally: {{
          outputs.scan_keptn_telemetry.vars.evaluation_timeouts_count }}
          timeouts, {{
          outputs.scan_keptn_telemetry.vars.evaluation_duration_seconds }}s
          duration, {{
          outputs.scan_keptn_telemetry.vars.active_evaluations_total }} active
          evaluations."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for continuous deployment
      governance and SLI release verification records.
    format: |
      {
        "status": "{{ outputs.scan_keptn_telemetry.vars.is_alerting ? 'ALERT' : 'NOMINAL' }}",
        "application_name": "{{ inputs.application_name }}",
        "evaluation_timeouts": {{ outputs.scan_keptn_telemetry.vars.evaluation_timeouts_count }},
        "evaluation_duration_seconds": {{ outputs.scan_keptn_telemetry.vars.evaluation_duration_seconds }},
        "task_failures": {{ outputs.scan_keptn_telemetry.vars.task_failures_count }},
        "active_evaluations": {{ outputs.scan_keptn_telemetry.vars.active_evaluations_total }},
        "audit_timestamp": "{{ outputs.scan_keptn_telemetry.vars.audit_timestamp }}"
      }

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