Keep GitLab runner pools balanced so pipelines stop queuing

id: gitlab-runner-balancer
namespace: company.team

description: |
  Reports how a GitLab CI runner pool is balanced across tags and suggests
  scaling so pipelines stop queuing. A webhook fires when a pipeline has been
  pending past its threshold, the project's online runners and its pending and
  running jobs are read from the GitLab API, and one Python task merges them
  into a per-tag busy versus idle picture with a queue depth and a verdict. An
  If gate escalates when the busy ratio crosses the imbalance ratio or the
  queue backs up, fanning out a reassignment suggestion per idle runner,
  opening a GitHub scaling issue and pinging Slack, while a healthy pool is
  simply logged. The verdict, online runner count, busy ratio and queue depth
  are exposed as typed outputs.

triggers:
  - id: pipeline_pending
    type: io.kestra.plugin.core.trigger.Webhook
    description: Receives the pipeline event posted by the relay once a job has been
      pending longer than the alerting threshold and starts the balance check.
    key: "{{ secret('GITLAB_WEBHOOK_KEY') }}"
  - id: hourly_rebalance_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Disabled hourly net that would sweep the same runner pool until
      someone switches it on.
    cron: "0 * * * *"
    disabled: true

inputs:
  - id: project_path
    type: STRING
    defaults: acme-inc%2Fpayments-api
    description: URL-encoded group and project path, or numeric project id, used in
      both GitLab REST calls, with %2F standing for the slash.
  - id: imbalance_ratio
    type: FLOAT
    defaults: 0.8
    description: Busy ratio above which a tag counts as imbalanced and every idle
      runner gets a reassignment suggestion.

tasks:
  # --- 1. online runner pool ---
  - id: fetch_online_runners
    type: io.kestra.plugin.core.http.Request
    description: Reads every runner that is currently online for the project so busy
      and idle runners can be counted per tag.
    method: GET
    uri: "https://gitlab.com/api/v4/projects/{{ inputs.project_path
      }}/runners?status=online"
    headers:
      PRIVATE-TOKEN: "{{ secret('GITLAB_TOKEN') }}"
      Accept: "application/json"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT15S
      maxAttempts: 3

  # --- 2. queued and executing jobs ---
  - id: fetch_queued_jobs
    type: io.kestra.plugin.core.http.Request
    description: Reads the pending and running jobs of the project, which give the
      queue depth and the runner ids that are busy right now.
    method: GET
    uri: "https://gitlab.com/api/v4/projects/{{ inputs.project_path
      }}/jobs?scope[]=pending&scope[]=running"
    headers:
      PRIVATE-TOKEN: "{{ secret('GITLAB_TOKEN') }}"
      Accept: "application/json"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT15S
      maxAttempts: 3

  # --- 3. merge both payloads into a balance verdict ---
  - id: compute_balance
    type: io.kestra.plugin.scripts.python.Script
    description: Splits online runners into busy and idle per tag, measures the
      queue and emits the verdict the gate branches on.
    containerImage: python:3.11-slim
    env:
      IMBALANCE_RATIO: "{{ inputs.imbalance_ratio }}"
    inputFiles:
      runners.json: "{{ outputs.fetch_online_runners.body | toJson }}"
      jobs.json: "{{ outputs.fetch_queued_jobs.body | toJson }}"
    script: |
      import json
      import os

      RUNNERS_FILE = "runners.json"
      JOBS_FILE = "jobs.json"
      QUEUE_LIMIT = 5


      def load(path):
          try:
              with open(path, "r", encoding="utf-8") as handle:
                  raw = handle.read().strip()
          except OSError:
              return None
          if not raw:
              return None
          try:
              data = json.loads(raw)
          except ValueError:
              return None
          while isinstance(data, str):
              try:
                  data = json.loads(data)
              except ValueError:
                  return None
          return data


      def rows_of(payload, key):
          if isinstance(payload, list):
              return payload, True
          if isinstance(payload, dict):
              if isinstance(payload.get(key), list):
                  return payload[key], True
              return [], False
          return [], False


      def safe_ratio(busy, total):
          if total <= 0:
              return 0.0
          return round(float(busy) / float(total), 4)


      def tag_names(runner):
          names = []
          for tag in runner.get("tags") or []:
              name = str(tag).strip().lower()
              if name and name not in names:
                  names.append(name)
          return names


      def imbalance_limit():
          raw = os.environ.get("IMBALANCE_RATIO", "0.8")
          try:
              value = float(raw)
          except ValueError:
              return 0.8
          if value < 0:
              return 0.0
          if value > 1:
              return 1.0
          return value


      def main():
          limit = imbalance_limit()

          runner_rows, runners_ok = rows_of(load(RUNNERS_FILE), "runners")
          job_rows, jobs_ok = rows_of(load(JOBS_FILE), "jobs")

          online = []
          for row in runner_rows:
              if not isinstance(row, dict):
                  continue
              if row.get("status") == "offline" or row.get("online") is False:
                  continue
              online.append(row)

          jobs = [row for row in job_rows if isinstance(row, dict)]
          pending = [j for j in jobs if str(j.get("status") or "").lower() == "pending"]
          running = [j for j in jobs if str(j.get("status") or "").lower() == "running"]
          queue_depth = len(pending)

          busy_ids = set()
          for job in running:
              runner = job.get("runner")
              if isinstance(runner, dict) and runner.get("id") is not None:
                  busy_ids.add(runner.get("id"))
              elif isinstance(runner, (int, float)):
                  busy_ids.add(int(runner))

          busy_runners = [r for r in online if r.get("id") in busy_ids]
          idle_runners = [r for r in online if r.get("id") not in busy_ids]

          tag_totals = {}
          tag_busy = {}
          for runner in online:
              for name in tag_names(runner):
                  tag_totals[name] = tag_totals.get(name, 0) + 1
          for runner in busy_runners:
              for name in tag_names(runner):
                  tag_busy[name] = tag_busy.get(name, 0) + 1

          busy_ratio = safe_ratio(len(busy_runners), len(online))
          if tag_totals:
              worst = 0.0
              for name in sorted(tag_totals):
                  value = safe_ratio(tag_busy.get(name, 0), tag_totals[name])
                  if value > worst:
                      worst = value
              busy_ratio = worst

          if not runners_ok or not jobs_ok:
              verdict = "NO_DATA"
          elif busy_ratio > limit or queue_depth > QUEUE_LIMIT:
              verdict = "SCALE_UP"
          elif not online:
              verdict = "NO_RUNNERS"
          else:
              verdict = "BALANCED"

          result = {
              "verdict": verdict,
              "online_runners": len(online),
              "busy_ratio": busy_ratio,
              "idle_runners": [
                  {
                      "id": runner.get("id"),
                      "description": str(
                          runner.get("description") or runner.get("id") or "unknown"
                      ),
                      "tags": tag_names(runner),
                  }
                  for runner in idle_runners
              ],
              "queue_depth": queue_depth,
          }
          print("::" + json.dumps({"outputs": result}, ensure_ascii=True) + "::")


      main()

  # --- 4. the balance gate ---
  - id: balance_gate
    type: io.kestra.plugin.core.flow.If
    description: Opens when the busiest tag crosses the imbalance ratio or more than
      five jobs are waiting, otherwise records a healthy pool.
    condition: "{{ outputs.compute_balance.vars.busy_ratio > inputs.imbalance_ratio
      or outputs.compute_balance.vars.queue_depth > 5 }}"
    then:
      - id: suggest_reassignment
        type: io.kestra.plugin.core.flow.Loop
        description: Walks every idle runner and prints which queued job could be moved
          onto it.
        values: "{{ outputs.compute_balance.vars.idle_runners | toJson }}"
        concurrencyLimit: 2
        tasks:
          - id: log_reassignment
            type: io.kestra.plugin.core.log.Log
            description: Prints one reassignment suggestion naming the idle runner, its tags
              and the queue it would relieve.
            message: "Reassign one pending job to idle runner {{ item.value.description }}
              (id {{ item.value.id }}, tags {{ item.value.tags | join(', ') }})
              because {{ outputs.compute_balance.vars.queue_depth }} jobs are
              queued and the busiest tag sits at {{
              outputs.compute_balance.vars.busy_ratio }}."
      - id: open_scaling_issue
        type: io.kestra.plugin.github.issues.Create
        description: Files the scaling request in the ops backlog with the verdict, the
          counts and the idle runners to use first.
        oauthToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "acme-inc/platform-ops"
        title: "Scale the GitLab runner pool for {{ inputs.project_path }}"
        body: |
          ## Runner pool balance

          - Verdict: {{ outputs.compute_balance.vars.verdict }}
          - Online runners: {{ outputs.compute_balance.vars.online_runners }}
          - Busiest tag busy ratio: {{ outputs.compute_balance.vars.busy_ratio }} against a threshold of {{ inputs.imbalance_ratio }}
          - Queued jobs: {{ outputs.compute_balance.vars.queue_depth }}
          - Idle runners: {{ outputs.compute_balance.vars.idle_runners | toJson }}

          Add runners for the busiest tag, or point the queued jobs at the idle
          runners listed above, so pipelines stop waiting in the queue.
        labels:
          - gitlab-ci
          - runners
      - id: notify_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Pings the build channel so an on-call engineer rebalances the pool
          before the queue grows further.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"channel": "#ci-ops", "text": "GitLab runner pool for {{ inputs.project_path }} is skewed: verdict {{ outputs.compute_balance.vars.verdict }}, busiest tag at {{ outputs.compute_balance.vars.busy_ratio }}, {{ outputs.compute_balance.vars.queue_depth }} jobs queued, {{ outputs.compute_balance.vars.idle_runners | length }} idle runners ready to take work."}
    else:
      - id: log_balanced
        type: io.kestra.plugin.core.log.Log
        description: Records a healthy pool where the gate stayed closed and no scaling
          is needed this run.
        message: "Runner pool for {{ inputs.project_path }} is in balance: verdict {{
          outputs.compute_balance.vars.verdict }}, {{
          outputs.compute_balance.vars.online_runners }} online runners, busy
          ratio {{ outputs.compute_balance.vars.busy_ratio }}, queue depth {{
          outputs.compute_balance.vars.queue_depth }}."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalates 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
    value: "{{ outputs.compute_balance.vars.verdict }}"
  - id: online_runners
    type: INT
    value: "{{ outputs.compute_balance.vars.online_runners }}"
  - id: busy_ratio
    type: FLOAT
    value: "{{ outputs.compute_balance.vars.busy_ratio }}"
  - id: queue_depth
    type: INT
    value: "{{ outputs.compute_balance.vars.queue_depth }}"