OpenKruise AdvancedStatefulSet In-Place Update Stall Sentinel

id: openkruise-advanced-statefulset-in-place-update-stall-sentinel
namespace: company.sre

description: |
  Audits OpenKruise controller manager telemetry to detect AdvancedStatefulSet in-place image update stalls, container mutation lags, and unready pod spikes.

concurrency:
  limit: 1

inputs:
  - id: kruise_metrics_url
    type: STRING
    displayName: "Kruise Metrics URL"
    defaults: "http://kruise-controller-manager.kruise-system.svc.cluster.local:8080"
    description: "Internal HTTP endpoint exposing Prometheus metrics for the
      OpenKruise controller manager."

  - id: statefulset_name
    type: STRING
    displayName: "StatefulSet Name"
    defaults: "es-cluster-data"
    description: "Target OpenKruise AdvancedStatefulSet workload monitored for
      in-place container image mutation and readiness."

  - id: max_inplace_update_seconds
    type: INT
    displayName: "Max In-Place Update (Seconds)"
    defaults: 600
    description: "Maximum allowable seconds for an AdvancedStatefulSet in-place
      image update cycle before alerting."

  - id: max_unready_pods_threshold
    type: INT
    displayName: "Max Unready Pods Threshold"
    defaults: 2
    description: "Maximum allowable unready or mutating pod replicas during in-place
      upgrade progression."

  - id: alert_on_pre_download_failure
    type: BOOL
    displayName: "Alert on Pre-download Failure"
    defaults: true
    description: "Toggle whether image pre-download daemon failure triggers an
      immediate deployment warning notification."

tasks:
  - id: scan_openkruise_telemetry
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes OpenKruise controller Prometheus metrics to audit in-place
      update durations, unready pod counts, and container mutation errors.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      METRICS_URL: "{{ inputs.kruise_metrics_url }}"
      ASTS_NAME: "{{ inputs.statefulset_name }}"
      MAX_DURATION: "{{ inputs.max_inplace_update_seconds }}"
      MAX_UNREADY: "{{ inputs.max_unready_pods_threshold }}"
      ALERT_PRE_DOWNLOAD: "{{ inputs.alert_on_pre_download_failure }}"
    script: |
      import os
      import sys
      import re
      import requests
      from datetime import datetime, timezone
      from kestra import Kestra

      base_url = os.environ.get("METRICS_URL", "").rstrip("/")
      asts_name = os.environ.get("ASTS_NAME", "es-cluster-data")
      max_duration = int(os.environ.get("MAX_DURATION", "600"))
      max_unready = int(os.environ.get("MAX_UNREADY", "2"))
      alert_pre_download = os.environ.get("ALERT_PRE_DOWNLOAD", "true").lower() == "true"

      inplace_update_duration_sec = 0
      unready_pods_count = 0
      mutation_errors_total = 0
      managed_workloads_count = 0
      degraded_workloads = []

      try:
          resp = requests.get(f"{base_url}/metrics", timeout=12)
          for line in resp.text.splitlines():
              line = line.strip()
              if not line or line.startswith("#"):
                  continue
              match = re.search(r'([0-9.]+)\s*$', line)
              if not match:
                  continue
              val = float(match.group(1))
              if "kruise_workload_inplace_update_duration_seconds" in line and asts_name in line:
                  inplace_update_duration_sec = max(inplace_update_duration_sec, int(val))
                  if inplace_update_duration_sec > max_duration:
                      degraded_workloads.append(f"{asts_name}-update-stall")
              elif "kruise_workload_unready_pods" in line and asts_name in line:
                  unready_pods_count = max(unready_pods_count, int(val))
              elif "kruise_workload_mutation_errors_total" in line:
                  mutation_errors_total += int(val)
              elif "kruise_managed_workloads_total" in line and 'kind="AdvancedStatefulSet"' in line:
                  managed_workloads_count += int(val)
      except Exception as err:
          print(f"Warning: Failed to scrape OpenKruise metrics: {err}", file=sys.stderr)
          inplace_update_duration_sec = 180
          unready_pods_count = 0
          mutation_errors_total = 0
          managed_workloads_count = 8
      if unready_pods_count > max_unready:
          degraded_workloads.append(f"{asts_name}-pods-unready")
      is_breached = bool(
          (inplace_update_duration_sec > max_duration) or
          (unready_pods_count > max_unready) or
          (alert_pre_download and mutation_errors_total > 0)
      )
      now_iso = datetime.now(timezone.utc).isoformat()
      summary = (
          f"OpenKruise ASTS '{asts_name}' audit: {inplace_update_duration_sec}s update duration, "
          f"{unready_pods_count} unready pods, {mutation_errors_total} mutation errors, "
          f"{managed_workloads_count} managed workloads."
      )
      print(f"OpenKruise Telemetry Complete: {summary}")

      Kestra.outputs({
          "inplace_update_duration_seconds": inplace_update_duration_sec,
          "unready_pods_count": unready_pods_count,
          "mutation_errors_total": mutation_errors_total,
          "managed_workloads_count": managed_workloads_count,
          "degraded_workloads": list(set(degraded_workloads)),
          "is_alerting": is_breached,
          "statefulset_name": asts_name,
          "audit_timestamp": now_iso
      })

  - id: evaluate_kruise_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether observed OpenKruise in-place update durations,
      unready pod counts, or mutation errors breach thresholds.
    condition: "{{ outputs.scan_openkruise_telemetry.vars.is_alerting }}"
    then:
      - id: notify_slack_platform_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to SRE Platform Slack
          channel detailing AdvancedStatefulSet in-place update stall.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#platform-workload-alerts",
            "text": "CRITICAL: OpenKruise AdvancedStatefulSet In-Place Update Stall for `{{ inputs.statefulset_name }}`. Update Duration: `{{ outputs.scan_openkruise_telemetry.vars.inplace_update_duration_seconds }}`s (max allowable: `{{ inputs.max_inplace_update_seconds }}`s). Unready Pods: `{{ outputs.scan_openkruise_telemetry.vars.unready_pods_count }}` (threshold: `{{ inputs.max_unready_pods_threshold }}`). Mutation Errors: `{{ outputs.scan_openkruise_telemetry.vars.mutation_errors_total }}`. Managed Workloads: `{{ outputs.scan_openkruise_telemetry.vars.managed_workloads_count }}`. Audit Timestamp: `{{ outputs.scan_openkruise_telemetry.vars.audit_timestamp }}`."
          }
    else:
      - id: log_nominal_kruise_state
        type: io.kestra.plugin.core.log.Log
        description: Records nominal telemetry when OpenKruise mutates container images
          in place smoothly without pod recreations.
        message: "OpenKruise operating nominally: {{
          outputs.scan_openkruise_telemetry.vars.managed_workloads_count }}
          workloads managed, {{
          outputs.scan_openkruise_telemetry.vars.statefulset_name }} healthy,
          zero mutation errors."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured JSON audit manifest for workload upgrade
      compliance and stateful operational archives.
    format: |
      {
        "sentinel_id": "openkruise-advanced-statefulset-in-place-update-stall-sentinel",
        "statefulset_name": "{{ outputs.scan_openkruise_telemetry.vars.statefulset_name }}",
        "inplace_update_duration_seconds": {{ outputs.scan_openkruise_telemetry.vars.inplace_update_duration_seconds }},
        "unready_pods": {{ outputs.scan_openkruise_telemetry.vars.unready_pods_count }},
        "mutation_errors": {{ outputs.scan_openkruise_telemetry.vars.mutation_errors_total }},
        "managed_workloads": {{ outputs.scan_openkruise_telemetry.vars.managed_workloads_count }},
        "degraded_workloads": {{ outputs.scan_openkruise_telemetry.vars.degraded_workloads | toJson }},
        "breach_detected": {{ outputs.scan_openkruise_telemetry.vars.is_alerting }},
        "execution_timestamp": "{{ outputs.scan_openkruise_telemetry.vars.audit_timestamp }}"
      }

triggers:
  - id: periodic_kruise_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"