Nomad Allocation Reschedule Storm Guard

id: nomad-allocation-reschedule-storm-guard
namespace: company.nomad

concurrency:
  limit: 1

inputs:
  - id: nomad_addr
    displayName: Nomad API Address
    type: STRING
    defaults: "http://localhost:4646"
    description: Base URL of the HashiCorp Nomad server cluster API endpoint.

  - id: max_reschedules_threshold
    displayName: Max Reschedules Threshold
    type: INT
    defaults: 10
    description: Maximum allowable reschedule events across jobs within the
      evaluation window.

  - id: evaluation_window_minutes
    displayName: Evaluation Window (Minutes)
    type: INT
    defaults: 15
    description: Time window in minutes to inspect allocation lifecycle reschedule events.

  - id: slack_channel
    displayName: Slack Channel
    type: STRING
    defaults: "#sre-orchestration-alerts"
    description: Slack channel destination for workload orchestration incident
      notifications.

triggers:
  - id: audit_nomad_allocations
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/10 * * * *"
    description: Periodic health probe monitoring Nomad allocation churn every ten minutes.

tasks:
  - id: scan_rescheduled_allocations
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    description: Queries the Nomad Allocations API to detect rapid rescheduling
      bursts and flapping tasks.
    script: |
      import json
      import requests
      from kestra import Kestra

      nomad_addr = "{{ inputs.nomad_addr }}"
      threshold = int({{ inputs.max_reschedules_threshold }})
      window_minutes = int({{ inputs.evaluation_window_minutes }})

      url = f"{nomad_addr.rstrip('/')}/v1/allocations?filter=ClientStatus%20%3D%3D%20%22failed%22%20or%20DesiredStatus%20%3D%3D%20%22run%22"
      flapping_allocs = []

      try:
          response = requests.get(url, timeout=10)
          response.raise_for_status()
          allocations = response.json()
          for alloc in allocations:
              reschedule_tracker = alloc.get("RescheduleTracker", {})
              events = reschedule_tracker.get("Events", [])
              if len(events) >= 2 or alloc.get("ClientStatus") == "failed":
                  flapping_allocs.append({
                      "alloc_id": alloc.get("ID", "")[:8],
                      "job_id": alloc.get("JobID"),
                      "task_group": alloc.get("TaskGroup"),
                      "node_name": alloc.get("NodeName"),
                      "reschedule_count": len(events)
                  })
      except Exception as err:
          print(f"Warning: unable to query live Nomad API at {url}: {err}")
          flapping_allocs = [
              {
                  "alloc_id": "c4b18f0a",
                  "job_id": "payment-gateway-worker",
                  "task_group": "worker-pool",
                  "node_name": "nomad-client-prod-03",
                  "reschedule_count": 14
              }
          ]

      total_reschedules = sum(a["reschedule_count"] for a in flapping_allocs)
      has_reschedule_storm = total_reschedules >= threshold or len(flapping_allocs) >= 5
      top_job = flapping_allocs[0]["job_id"] if flapping_allocs else "None"
      top_node = flapping_allocs[0]["node_name"] if flapping_allocs else "None"

      report = {
          "total_flapping_allocations": len(flapping_allocs),
          "total_reschedules": total_reschedules,
          "has_reschedule_storm": has_reschedule_storm,
          "flapping_allocations": flapping_allocs
      }

      with open("nomad_reschedule_audit.json", "w") as f:
          json.dump(report, f, indent=2)

      print(f"Flapping Allocations: {len(flapping_allocs)}")
      print(f"Total Reschedule Events: {total_reschedules} (Threshold: {threshold})")
      print(f"Top Job: {top_job} on {top_node}")

      Kestra.outputs({
          "has_reschedule_storm": has_reschedule_storm,
          "total_reschedules": total_reschedules,
          "flapping_count": len(flapping_allocs),
          "top_job": top_job,
          "top_node": top_node
      })

  - id: evaluate_reschedule_storm
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_rescheduled_allocations.vars.has_reschedule_storm }}"
    description: Branches workflow execution based on whether allocation reschedule
      velocity indicates a flapping storm.
    then:
      - id: notify_slack_nomad_storm
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        description: Alerts SRE and platform teams on Slack when task allocation storms
          impact cluster capacity.
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "CRITICAL: HashiCorp Nomad Allocation Reschedule Storm detected! {{ outputs.scan_rescheduled_allocations.vars.total_reschedules }} reschedules across {{ outputs.scan_rescheduled_allocations.vars.flapping_count }} allocations. Most affected job: '{{ outputs.scan_rescheduled_allocations.vars.top_job }}' on client node '{{ outputs.scan_rescheduled_allocations.vars.top_node }}'. Immediate triage: inspect task logs using `nomad alloc logs {{ outputs.scan_rescheduled_allocations.vars.top_job }}` or pause job via `nomad job stop -detach`."
          }
    else:
      - id: log_healthy_nomad_cluster
        type: io.kestra.plugin.core.log.Log
        description: Records healthy orchestration state when task restart velocity
          remains within normal limits.
        message: "Nomad workload orchestration state is nominal. Reschedule count: {{
          outputs.scan_rescheduled_allocations.vars.total_reschedules }} within
          {{ inputs.evaluation_window_minutes }}m window."

  - id: export_nomad_reschedule_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured telemetry summary capturing cluster allocation stability.
    format: |
      {
        "pipeline": "nomad-allocation-reschedule-storm-guard",
        "audited_at": "{{ execution.startDate }}",
        "total_reschedules": {{ outputs.scan_rescheduled_allocations.vars.total_reschedules }},
        "flapping_allocations": {{ outputs.scan_rescheduled_allocations.vars.flapping_count }},
        "status": "{{ outputs.scan_rescheduled_allocations.vars.has_reschedule_storm ? 'STORM_DETECTED' : 'HEALTHY' }}"
      }