Monitor Kubernetes Pod CPU Spikes with Prometheus and Send Grouped Slack Alerts

id: prometheus-k8s-cpu-spike-alerts
namespace: company.team

triggers:
  - id: every_5_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    description: Poll Prometheus every 5 minutes

inputs:
  - id: prometheus_url
    type: STRING
    displayName: Prometheus URL
    description: Base URL of your Prometheus instance
    defaults: "http://prometheus.monitoring.svc.cluster.local:9090"
  - id: cpu_threshold_cores
    type: FLOAT
    displayName: CPU threshold (cores)
    description: Alert when a pod's CPU usage exceeds this many cores
    defaults: 0.8
  - id: namespace_filter
    type: STRING
    displayName: Kubernetes namespace
    description: Filter pods by namespace (leave empty for all namespaces)
    defaults: ""

tasks:
  - id: query_cpu_spikes
    type: io.kestra.plugin.core.http.Request
    uri: "{{ inputs.prometheus_url }}/api/v1/query"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('PROMETHEUS_TOKEN') }}"
    body: |
      {
        "query": "sum(rate(container_cpu_usage_seconds_total{container!='',pod!=''}[5m])) by (pod, namespace) > {{ inputs.cpu_threshold_cores }}"
      }

  - id: parse_spikes
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      prometheus_response.json: "{{ outputs.query_cpu_spikes.body }}"
    script: |
      import json
      from kestra import Kestra

      with open("prometheus_response.json") as f:
          data = json.load(f)

      namespace_filter = "{{ inputs.namespace_filter }}"
      threshold = {{ inputs.cpu_threshold_cores }}

      results = data.get("data", {}).get("result", [])
      spikes = []

      for item in results:
          metric = item.get("metric", {})
          pod = metric.get("pod", "unknown")
          namespace = metric.get("namespace", "unknown")
          cpu_cores = float(item.get("value", [0, "0"])[1])

          # Apply namespace filter if specified
          if namespace_filter and namespace != namespace_filter:
              continue

          spikes.append({
              "pod": pod,
              "namespace": namespace,
              "cpu_cores": round(cpu_cores, 3)
          })

      # Group by app (prefix before last hyphen-hex suffix)
      apps = {}
      for spike in spikes:
          app_name = "-".join(spike["pod"].split("-")[:-2]) or spike["pod"]
          if app_name not in apps:
              apps[app_name] = {"pods": [], "namespace": spike["namespace"]}
          apps[app_name]["pods"].append(spike)

      Kestra.outputs({
          "spike_count": len(spikes),
          "spikes": spikes,
          "app_groups": apps,
          "has_spikes": len(spikes) > 0
      })

  - id: send_alert
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.parse_spikes.vars.has_spikes }}"
    then:
      - id: format_and_alert
        type: io.kestra.plugin.scripts.python.Script
        inputFiles:
          spikes.json: "{{ outputs.parse_spikes.vars.spikes | json }}"
          apps.json: "{{ outputs.parse_spikes.vars.app_groups | json }}"
        script: |
          import json
          from kestra import Kestra

          with open("spikes.json") as f:
              spikes = json.load(f)
          with open("apps.json") as f:
              apps = json.load(f)

          lines = [f":rotating_light: *Kubernetes CPU Spike Alert*, {len(spikes)} pod(s) above threshold\n"]
          for app, info in apps.items():
              pods = info["pods"]
              ns = info["namespace"]
              if len(pods) == 1:
                  p = pods[0]
                  lines.append(f"• `{p['pod']}` ({ns}): *{p['cpu_cores']} cores*")
              else:
                  total = sum(p["cpu_cores"] for p in pods)
                  lines.append(f"• `{app}` ({ns}): *{len(pods)} pods, {round(total,3)} cores total*")

          Kestra.outputs({"slack_message": "\n".join(lines)})

      - id: slack_notification
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: "{{ outputs.format_and_alert.vars.slack_message }}"