Resolve a Nomad Deployment Stuck Waiting on Manual Canary Promotion

id: nomad-deployment-stuck-canary-promotion-resolver
namespace: company.infrastructure

description: |
  Detects a Nomad job deployment stuck in status "running" because its canary
  allocations are waiting on manual promotion (or have gone unhealthy) and
  nobody has acted, tracks how long it has actually been stuck via Kestra KV
  across runs, then, after a timeout and operator approval, calls Nomad's real
  promote or fail deployment API to resolve it and verifies the deployment
  left the running state.

inputs:
  - id: nomad_address
    type: STRING
    displayName: "Nomad Address"
    defaults: "http://localhost:4646"
    description: Base URL of the Nomad HTTP API.
  - id: nomad_namespace
    type: STRING
    displayName: "Nomad Namespace"
    defaults: "default"
    description: Nomad namespace the job runs in.
  - id: job_id
    type: STRING
    displayName: "Nomad Job ID"
    description: The job whose latest deployment is audited.
  - id: stuck_threshold_minutes
    type: INT
    displayName: "Stuck Threshold (minutes)"
    defaults: 15
    description: How long a deployment must remain in status "running" with the same
      deployment ID before this flow treats it as stuck and offers to act,
      instead of acting on the first observation.

triggers:
  - id: periodic_deployment_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the job's latest deployment every 5 minutes. Shipped
      disabled; enable after a manual run confirms nomad_address and job_id are
      correct.
    cron: "*/5 * * * *"
    disabled: true
    inputs:
      job_id: "my-service"

tasks:
  - id: fetch_latest_deployment
    type: io.kestra.plugin.core.http.Request
    description: Requests the job's latest deployment - confirmed from Nomad's own
      source (command/agent/job_endpoint.go, jobLatestDeployment) that this
      handler returns the flat Deployment object directly, not wrapped under a
      "Deployment" key, unlike the raw SingleDeploymentResponse RPC type it
      reads from internally.
    uri: "{{ inputs.nomad_address }}/v1/job/{{ inputs.job_id
      }}/deployment?namespace={{ inputs.nomad_namespace }}"
    method: GET
    headers:
      X-Nomad-Token: "{{ secret('NOMAD_TOKEN') }}"

  - id: evaluate_deployment_state
    type: io.kestra.plugin.scripts.python.Script
    description: Classifies the deployment's real
      Status/StatusDescription/TaskGroups fields (verified against
      nomad/structs/deployment.go) and compares the deployment ID against the
      previous run's Kestra KV snapshot to compute actual elapsed stuck time,
      not just a single-run snapshot.
    env:
      DEPLOYMENT_JSON: "{{ outputs.fetch_latest_deployment.body }}"
      STUCK_THRESHOLD_MINUTES: "{{ inputs.stuck_threshold_minutes }}"
      PREVIOUS_STATE_JSON: "{{ kv('nomad_deployment_tracker_' ~ (inputs.job_id |
        slugify), errorOnMissing=false) ?? '{}' }}"
    script: |
      import json
      import os
      from datetime import datetime, timezone

      now = datetime.now(timezone.utc)
      now_iso = now.isoformat()

      deployment_raw = os.environ.get("DEPLOYMENT_JSON", "null")
      stuck_threshold_minutes = int(os.environ.get("STUCK_THRESHOLD_MINUTES", "15"))

      prev_raw = os.environ.get("PREVIOUS_STATE_JSON", "{}")
      try:
          prev_state = json.loads(prev_raw) if prev_raw.strip() else {}
      except Exception:
          prev_state = {}

      try:
          deployment = json.loads(deployment_raw) if deployment_raw.strip() else None
      except Exception:
          deployment = None

      if not deployment:
          # No deployment exists for this job at all (e.g. job never deployed, or was purged).
          new_state = {}
          needs_action = False
          status = "NONE"
          status_description = ""
          deployment_id = None
          minutes_stuck = 0
          recommended_action = "NONE"
          reasoning = f"No deployment found for this job."
      else:
          status = deployment.get("Status", "unknown")
          status_description = deployment.get("StatusDescription", "")
          deployment_id = deployment.get("ID")
          task_groups = deployment.get("TaskGroups") or {}

          if status != "running":
              # Deployment already resolved on its own (successful, failed, cancelled, paused) - clear tracking.
              new_state = {}
              needs_action = False
              minutes_stuck = 0
              recommended_action = "NONE"
              reasoning = f"Deployment status is '{status}', not running. No action needed; tracking cleared."
          else:
              if prev_state.get("deployment_id") == deployment_id and prev_state.get("first_seen_stuck_at"):
                  first_seen = datetime.fromisoformat(prev_state["first_seen_stuck_at"])
                  minutes_stuck = (now - first_seen).total_seconds() / 60.0
                  first_seen_stuck_at = prev_state["first_seen_stuck_at"]
              else:
                  minutes_stuck = 0.0
                  first_seen_stuck_at = now_iso

              new_state = {
                  "deployment_id": deployment_id,
                  "first_seen_stuck_at": first_seen_stuck_at,
                  "last_checked_at": now_iso,
              }

              all_canaries_healthy = True
              any_unhealthy = False
              for group_name, group_state in task_groups.items():
                  desired_canaries = group_state.get("DesiredCanaries", 0)
                  healthy = group_state.get("HealthyAllocs", 0)
                  unhealthy = group_state.get("UnhealthyAllocs", 0)
                  if unhealthy > 0:
                      any_unhealthy = True
                  if desired_canaries > 0 and healthy < desired_canaries:
                      all_canaries_healthy = False

              if minutes_stuck >= stuck_threshold_minutes:
                  if any_unhealthy:
                      needs_action = True
                      recommended_action = "FAIL"
                      reasoning = (
                          f"Deployment {deployment_id} has been running for {minutes_stuck:.1f} minute(s) "
                          f"(threshold {stuck_threshold_minutes}) with at least one unhealthy canary allocation. "
                          f"Recommending FAIL to trigger auto-revert where the job has it configured."
                      )
                  elif all_canaries_healthy:
                      needs_action = True
                      recommended_action = "PROMOTE"
                      reasoning = (
                          f"Deployment {deployment_id} has been running for {minutes_stuck:.1f} minute(s) "
                          f"(threshold {stuck_threshold_minutes}) with all canaries healthy and status "
                          f"description '{status_description}'. Recommending PROMOTE to advance the rollout."
                      )
                  else:
                      needs_action = False
                      recommended_action = "NONE"
                      reasoning = (
                          f"Deployment {deployment_id} has been running for {minutes_stuck:.1f} minute(s) "
                          f"but canary health is still ambiguous (not all healthy, none explicitly unhealthy yet). Waiting."
                      )
              else:
                  needs_action = False
                  recommended_action = "NONE"
                  reasoning = (
                      f"Deployment {deployment_id} has been running for {minutes_stuck:.1f} minute(s), "
                      f"under the {stuck_threshold_minutes} minute threshold. Tracking, not acting yet."
                  )

      print(reasoning)

      print("::" + json.dumps({
          "outputs": {
              "needs_action": needs_action,
              "recommended_action": recommended_action,
              "deployment_id": deployment_id,
              "status": status,
              "status_description": status_description,
              "minutes_stuck": round(minutes_stuck, 1),
              "reasoning": reasoning,
              "new_state": new_state,
          }
      }) + "::")

  - id: save_tracking_state
    type: io.kestra.plugin.core.kv.Set
    description: Persist the deployment-tracking snapshot (or clear it to empty if
      the deployment already resolved) so the next run can compute real elapsed
      stuck time instead of starting over.
    key: "nomad_deployment_tracker_{{ inputs.job_id | slugify }}"
    value: "{{ outputs.evaluate_deployment_state.vars.new_state | toJson }}"
    kvType: JSON

  - id: evaluate_action_needed
    type: io.kestra.plugin.core.flow.If
    description: Gate the whole remediation path behind the computed needs_action verdict.
    condition: "{{ outputs.evaluate_deployment_state.vars.needs_action == true }}"
    then:
      - id: pause_for_operator_approval
        type: io.kestra.plugin.core.flow.Pause
        description: Wait indefinitely for an operator to review and resume before
          promoting or failing the deployment. No pauseDuration is set, so an
          unattended execution waits rather than silently proceeding.

      - id: execute_remediation
        type: io.kestra.plugin.core.flow.If
        description: Call Nomad's real promote or fail deployment endpoint per the
          computed recommendation.
        condition: "{{ outputs.evaluate_deployment_state.vars.recommended_action ==
          'PROMOTE' }}"
        then:
          - id: promote_deployment
            type: io.kestra.plugin.core.http.Request
            description: Promotes all task groups in the deployment, confirmed from Nomad's
              api/deployments.go DeploymentPromoteRequest shape (DeploymentID,
              All).
            uri: "{{ inputs.nomad_address }}/v1/deployment/promote/{{
              outputs.evaluate_deployment_state.vars.deployment_id
              }}?namespace={{ inputs.nomad_namespace }}"
            method: PUT
            headers:
              X-Nomad-Token: "{{ secret('NOMAD_TOKEN') }}"
            body: |
              {
                "DeploymentID": "{{ outputs.evaluate_deployment_state.vars.deployment_id }}",
                "All": true
              }
        else:
          - id: fail_deployment
            type: io.kestra.plugin.core.http.Request
            description: Force-fails the deployment, confirmed from Nomad's
              api/deployments.go DeploymentFailRequest shape (DeploymentID
              only). Triggers the job's configured auto_revert if set.
            uri: "{{ inputs.nomad_address }}/v1/deployment/fail/{{
              outputs.evaluate_deployment_state.vars.deployment_id
              }}?namespace={{ inputs.nomad_namespace }}"
            method: PUT
            headers:
              X-Nomad-Token: "{{ secret('NOMAD_TOKEN') }}"
            body: |
              {
                "DeploymentID": "{{ outputs.evaluate_deployment_state.vars.deployment_id }}"
              }

      - id: verify_resolution
        type: io.kestra.plugin.core.http.Request
        description: Re-fetches the job's latest deployment after the mutation to
          confirm it actually left the running state, rather than assuming the
          promote/fail call succeeded.
        uri: "{{ inputs.nomad_address }}/v1/job/{{ inputs.job_id
          }}/deployment?namespace={{ inputs.nomad_namespace }}"
        method: GET
        headers:
          X-Nomad-Token: "{{ secret('NOMAD_TOKEN') }}"

      - id: evaluate_verification
        type: io.kestra.plugin.core.flow.If
        description: Branch on whether the post-action status is still "running"
          (verification failed) or has moved on (verification succeeded).
        condition: "{{ (outputs.verify_resolution.body | jq('.Status') | first) ==
          'running' }}"
        then:
          - id: alert_remediation_did_not_resolve
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: The promote or fail call was made but the deployment is still
              running; this needs human intervention, not a silent success
              claim.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Nomad deployment resolver: called {{ outputs.evaluate_deployment_state.vars.recommended_action }} on deployment {{ outputs.evaluate_deployment_state.vars.deployment_id }} for job {{ inputs.job_id }}, but it is STILL in status running after the call. Manual investigation needed. Execution {{ execution.id }}."
              }
        else:
          - id: notify_resolution_succeeded
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Confirms the real, verified outcome rather than just that the API
              call returned 2xx.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Nomad deployment resolver: {{ outputs.evaluate_deployment_state.vars.recommended_action }} applied to deployment {{ outputs.evaluate_deployment_state.vars.deployment_id }} for job {{ inputs.job_id }}, which had been stuck running for {{ outputs.evaluate_deployment_state.vars.minutes_stuck }} minute(s). Verified it left the running state. Execution {{ execution.id }}."
              }
    else:
      - id: log_no_action_needed
        type: io.kestra.plugin.core.log.Log
        message: "{{ outputs.evaluate_deployment_state.vars.reasoning }}"

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts Slack separately if the flow itself fails, so a broken check
      never silently reads as "deployment healthy".
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Nomad deployment resolver FAILED in execution {{ execution.id }} for job {{ inputs.job_id }}. The check did not complete; do not trust the last known status until this is fixed."
      }

outputs:
  - id: needs_action
    type: BOOL
    value: "{{ outputs.evaluate_deployment_state.vars.needs_action }}"
  - id: recommended_action
    type: STRING
    value: "{{ outputs.evaluate_deployment_state.vars.recommended_action }}"