Right-Size Kubernetes HPA Replica Bounds from Live CPU Utilization

id: hpa-utilization-tuner
namespace: company.team
description: |
  Right-size Kubernetes Horizontal Pod Autoscalers by comparing observed CPU
  utilization against a target, typically 60 percent, from a daily schedule
  that ships disabled or from an on-demand webhook. A kubectl container writes
  a deterministic HPA usage fixture and scrapes cluster metrics best-effort,
  then a Python pass recommends minReplicas and maxReplicas changes for idle
  floors and for autoscalers pinned at their ceiling. When a change is
  recommended the flow opens a GitHub issue, alerts Slack, and applies every
  kubectl patch best-effort behind a documented dry-run preview; when the
  fleet is already optimal it records the all-clear instead. The verdict, the
  recommendation list, the HPA count, and the target become flow outputs.

triggers:
  - id: daily_hpa_review
    type: io.kestra.plugin.core.trigger.Schedule
    description: Primary trigger, a daily HPA review at 06:00 UTC; shipped disabled
      until an operator switches it on.
    cron: "0 6 * * *"
    disabled: true
  - id: on_demand_review
    type: io.kestra.plugin.core.trigger.Webhook
    description: POST here to review a namespace on demand, for example after a
      deployment or a traffic spike.
    key: "{{ secret('HPA_TUNER_WEBHOOK_KEY') }}"

inputs:
  - id: namespace
    type: STRING
    defaults: production
    description: Kubernetes namespace whose autoscalers are inspected and, when a
      change is recommended, patched.
  - id: cpu_target
    type: INT
    defaults: 60
    description: Average CPU utilization percentage that every replica bound
      recommendation is measured against.
  - id: github_repository
    type: STRING
    defaults: your-org/platform-ops
    description: Repository in owner/repo form that receives the tuning issue when a
      change is recommended.

tasks:
  # --- 1. collect utilization ---
  - id: usage_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the scrape, the analyzer, and the patcher
      all read the same HPA usage file.
    tasks:
      - id: collect_hpa_usage
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write the hpa-usage.json fixture with a heredoc, then scrape
          cluster metrics best-effort so the review stays deterministic with or
          without a reachable cluster.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: bitnami/kubectl:latest
        env:
          HPA_NAMESPACE: "{{ inputs.namespace }}"
          CPU_TARGET: "{{ inputs.cpu_target }}"
        outputFiles:
          - hpa-usage.json
        commands:
          - |
            set -u
            cat > hpa-usage.json <<'JSON'
            {
              "targetCPU": 60,
              "namespace": "production",
              "hpas": [
                {"name": "web-frontend", "minReplicas": 2, "maxReplicas": 10, "desiredReplicas": 5, "cpuUtilization": 40},
                {"name": "checkout-api", "minReplicas": 2, "maxReplicas": 6, "desiredReplicas": 6, "cpuUtilization": 95},
                {"name": "report-generator", "minReplicas": 3, "maxReplicas": 8, "desiredReplicas": 3, "cpuUtilization": 2}
              ]
            }
            JSON

            echo "reviewing HPA utilization in ${HPA_NAMESPACE} against a ${CPU_TARGET}% target"
            kubectl get --raw /apis/metrics.k8s.io/v1beta1/nodes || true
            echo "usage fixture bytes $(wc -c < hpa-usage.json)"

      - id: analyze_utilization
        type: io.kestra.plugin.scripts.python.Script
        description: Read the HPA usage file, recommend minReplicas and maxReplicas
          changes against the CPU target, and emit verdict, recommendations, and
          the HPA count through the outputs protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          HPA_NAMESPACE: "{{ inputs.namespace }}"
          CPU_TARGET: "{{ inputs.cpu_target }}"
        script: |
          import json
          import math
          import os

          USAGE_FILE = "hpa-usage.json"
          WARM_FLOOR = 2
          IDLE_CEILING = 10.0
          PINNED_FLOOR = 90.0

          namespace = os.environ.get("HPA_NAMESPACE", "default")

          usage = {}
          if os.path.exists(USAGE_FILE) and os.path.getsize(USAGE_FILE) > 0:
              try:
                  with open(USAGE_FILE, "r") as handle:
                      usage = json.load(handle)
              except ValueError:
                  usage = {}
          if not isinstance(usage, dict):
              usage = {}

          try:
              target = int(os.environ.get("CPU_TARGET", ""))
          except ValueError:
              target = 0
          if target <= 0:
              try:
                  target = int(usage.get("targetCPU"))
              except (TypeError, ValueError):
                  target = 60
          if target <= 0:
              target = 60

          hpas = usage.get("hpas")
          if not isinstance(hpas, list):
              hpas = []

          recommendations = []
          for entry in hpas:
              if not isinstance(entry, dict):
                  continue
              name = str(entry.get("name") or "").strip()
              if not name:
                  continue
              try:
                  current_min = int(entry.get("minReplicas"))
                  current_max = int(entry.get("maxReplicas"))
                  cpu = float(entry.get("cpuUtilization"))
              except (TypeError, ValueError):
                  continue
              if current_min < 0 or current_max < 1 or current_min > current_max:
                  continue
              try:
                  desired = int(entry.get("desiredReplicas"))
              except (TypeError, ValueError):
                  desired = current_min

              suggested_min = current_min
              suggested_max = current_max
              reason = ""

              if cpu <= IDLE_CEILING and current_min != WARM_FLOOR:
                  suggested_min = WARM_FLOOR
                  if current_min > WARM_FLOOR:
                      reason = ("idle at {cpu:g}% against a {target}% target: lower the minReplicas "
                                "floor from {cur} to {floor} replicas").format(
                                  cpu=cpu, target=target, cur=current_min, floor=WARM_FLOOR)
                  else:
                      reason = ("idle at {cpu:g}% against a {target}% target: raise minReplicas to "
                                "{floor} replicas so a warm pair stays ready").format(
                                  cpu=cpu, target=target, floor=WARM_FLOOR)
              elif cpu >= PINNED_FLOOR and desired >= current_max:
                  headroom = max(2, int(math.ceil(current_max * 0.5)))
                  suggested_max = current_max + headroom
                  reason = ("pinned at {cpu:g}% with {desired} desired replicas against maxReplicas "
                            "{cur}: raise maxReplicas to {sug}").format(
                              cpu=cpu, desired=desired, cur=current_max, sug=suggested_max)

              if reason:
                  recommendations.append({
                      "hpa": name,
                      "current_min": current_min,
                      "current_max": current_max,
                      "suggested_min": suggested_min,
                      "suggested_max": suggested_max,
                      "reason": reason,
                  })

          verdict = "TUNE" if recommendations else "OPTIMAL"
          print("namespace=%s target=%d hpas=%d recommendations=%d verdict=%s" %
                (namespace, target, len(hpas), len(recommendations), verdict))
          print("::" + json.dumps({"outputs": {"verdict": verdict, "recommendations": recommendations, "hpas": len(hpas)}}) + "::")

  # --- 2. verdict gate ---
  - id: tuning_gate
    type: io.kestra.plugin.core.flow.If
    description: Escalate and patch only when at least one autoscaler carries a
      recommendation, a fleet already sitting at target is recorded and left
      alone.
    condition: "{{ outputs.analyze_utilization.vars.recommendations | length > 0 }}"
    then:
      - id: open_tuning_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a labeled GitHub issue listing every suggested min and max
          change so tuning is tracked where the team already works.
        jwtToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ inputs.github_repository }}"
        title: "HPA tuning recommendations for {{ inputs.namespace }}"
        body: |
          ## HPA utilization tuning

          Namespace **{{ inputs.namespace }}** measured **{{ outputs.analyze_utilization.vars.hpas }}** autoscalers against a **{{ inputs.cpu_target }}%** CPU target.

          ### Recommended replica bound changes

          {% for rec in outputs.analyze_utilization.vars.recommendations %}
          - **{{ rec.hpa }}**: min {{ rec.current_min }} -> {{ rec.suggested_min }}, max {{ rec.current_max }} -> {{ rec.suggested_max }}. {{ rec.reason }}
          {% endfor %}

          This issue was opened by execution {{ execution.id }}. Review each change, run the printed `kubectl patch` command with `--dry-run=server` first, then close the issue once the fleet settles.
        labels:
          - kubernetes
          - hpa-tuning
      - id: alert_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the platform channel how many autoscalers need a bound change
          and that a GitHub issue is waiting.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "HPA tuning recommended in {{ inputs.namespace }}: {{ outputs.analyze_utilization.vars.recommendations | length }} of {{ outputs.analyze_utilization.vars.hpas }} autoscalers need a replica bound change against a {{ inputs.cpu_target }}% CPU target. Issue opened in {{ inputs.github_repository }}. Execution {{ execution.id }}."
          }
      - id: apply_recommendations
        type: io.kestra.plugin.core.flow.Loop
        description: Apply each suggested replica bound one at a time so a single
          unreachable autoscaler cannot abort the whole sweep.
        values: "{{ outputs.analyze_utilization.vars.recommendations }}"
        concurrencyLimit: 1
        tasks:
          - id: patch_hpa
            type: io.kestra.plugin.scripts.shell.Commands
            description: Merge the suggested bounds into the autoscaler with kubectl patch,
              echoing the dry-run preview first and absorbing kubectl failures
              with a trailing guard.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              entryPoint: []
            containerImage: bitnami/kubectl:latest
            env:
              HPA_NAMESPACE: "{{ inputs.namespace }}"
              HPA_NAME: "{{ item.hpa }}"
              SUGGESTED_MIN: "{{ item.suggested_min }}"
              SUGGESTED_MAX: "{{ item.suggested_max }}"
            commands:
              - |
                set -u
                echo "dry run preview: kubectl patch horizontalpodautoscaler ${HPA_NAME} -n ${HPA_NAMESPACE} --dry-run=server --type merge -p {\"spec\":{\"minReplicas\":${SUGGESTED_MIN},\"maxReplicas\":${SUGGESTED_MAX}}}"
                kubectl patch horizontalpodautoscaler "${HPA_NAME}" -n "${HPA_NAMESPACE}" --type merge -p "{\"spec\":{\"minReplicas\":${SUGGESTED_MIN},\"maxReplicas\":${SUGGESTED_MAX}}}" || true
                echo "applied min ${SUGGESTED_MIN} max ${SUGGESTED_MAX} to ${HPA_NAME} in ${HPA_NAMESPACE}"
    else:
      - id: log_optimal
        type: io.kestra.plugin.core.log.Log
        description: Record the all-clear when every autoscaler already sits sensibly
          around the target.
        message: "HPA fleet in {{ inputs.namespace }} is optimal against a {{
          inputs.cpu_target }}% CPU target, no replica bound changes
          recommended."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: TUNE when at least one replica bound change is recommended,
      otherwise OPTIMAL.
    value: "{{ outputs.analyze_utilization.vars.verdict }}"
  - id: recommendations
    type: ARRAY
    description: 'Suggested bound changes, for example [{"hpa": "checkout-api",
      "suggested_max": 9}].'
    value: "{{ outputs.analyze_utilization.vars.recommendations | toJson }}"
  - id: hpas
    type: INT
    description: Number of Horizontal Pod Autoscalers inspected during this run, for
      example 3.
    value: "{{ outputs.analyze_utilization.vars.hpas }}"
  - id: cpu_target
    type: INT
    description: Average CPU utilization percentage every recommendation was
      measured against, for example 60.
    value: "{{ inputs.cpu_target }}"