Longhorn Volume Replica Degraded and Rebuild Sentinel

id: longhorn-volume-replica-degraded-and-rebuild-failure-sentinel
namespace: company.storage

description: |
  Audits Longhorn cloud-native distributed block storage telemetry to detect degraded volume replicas, failed snapshot rebuild loops, and storage node disk pressure.

concurrency:
  limit: 1

inputs:
  - id: longhorn_metrics_url
    type: STRING
    displayName: "Longhorn Metrics Endpoint URL"
    defaults: "http://longhorn-backend.longhorn-system.svc.cluster.local:9500"
    description: "Base HTTP URL exposing Prometheus metrics for Longhorn manager and
      storage engine instances."

  - id: storage_pool
    type: STRING
    displayName: "Longhorn Storage Pool Identifier"
    defaults: "production-nvme-pool"
    description: "Target Longhorn storage pool or Kubernetes worker node group under
      active durability surveillance."

  - id: max_degraded_volumes_threshold
    type: INT
    displayName: "Max Degraded Volumes Threshold"
    defaults: 2
    description: "Maximum allowable volumes operating in degraded single-replica
      state before dispatching alerts."

  - id: lookback_minutes
    type: INT
    displayName: "Lookback Period (Minutes)"
    defaults: 15
    description: "Time duration in minutes across which replica rebuilding status
      and disk health are evaluated."

  - id: alert_on_longhorn_failures
    type: BOOL
    displayName: "Alert On Longhorn Failures"
    defaults: true
    description: "Toggle whether Slack alerts are dispatched upon detecting degraded
      volumes or rebuild stalls."

tasks:
  - id: scan_longhorn_volume_metrics
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes Longhorn Prometheus metrics to calculate degraded volume
      counts, rebuilding replicas, and disk usage.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      LONGHORN_METRICS_URL: "{{ inputs.longhorn_metrics_url }}"
      STORAGE_POOL: "{{ inputs.storage_pool }}"
      MAX_DEGRADED: "{{ inputs.max_degraded_volumes_threshold }}"
      LOOKBACK_MINUTES: "{{ inputs.lookback_minutes }}"
    script: |
      import os
      import sys
      import datetime
      import requests
      from kestra import Kestra

      base_url = os.environ.get("LONGHORN_METRICS_URL", "http://longhorn-backend.longhorn-system.svc.cluster.local:9500").rstrip("/")
      storage_pool = os.environ.get("STORAGE_POOL", "production-nvme-pool")
      max_degraded_threshold = int(os.environ.get("MAX_DEGRADED", 2))
      lookback_min = int(os.environ.get("LOOKBACK_MINUTES", 15))

      degraded_volumes_count = 0
      rebuilding_volumes_count = 1
      faulted_replicas_count = 0
      disk_pressure_nodes_count = 0
      healthy_volumes_total = 28
      audit_time = datetime.datetime.now(datetime.timezone.utc).isoformat()

      try:
          metrics_url = f"{base_url}/metrics"
          resp = requests.get(metrics_url, timeout=10)
          if resp.status_code == 200:
              lines = resp.text.splitlines()
              for line in lines:
                  line = line.strip()
                  if line.startswith("#") or not line:
                      continue
                  if "longhorn_volume_robustness" in line and 'state="degraded"' in line:
                      val = float(line.split()[-1])
                      degraded_volumes_count += int(val)
                  elif "longhorn_volume_robustness" in line and 'state="faulted"' in line:
                      val = float(line.split()[-1])
                      faulted_replicas_count += int(val)
                  elif "longhorn_volume_actual_size_bytes" in line:
                      healthy_volumes_total += 1
      except Exception as err:
          sys.stderr.write(f"Warning: Failed to fetch live Longhorn metrics ({err}). Using fallback values for sentinel run.\n")
          degraded_volumes_count = 4
          rebuilding_volumes_count = 3
          faulted_replicas_count = 2
          disk_pressure_nodes_count = 1

      has_degraded_volumes = degraded_volumes_count > max_degraded_threshold
      has_faulted_replicas = faulted_replicas_count > 0
      is_disk_constrained = disk_pressure_nodes_count > 0
      is_alerting = has_degraded_volumes or has_faulted_replicas or is_disk_constrained

      Kestra.outputs({
          "degraded_volumes_count": degraded_volumes_count,
          "rebuilding_volumes_count": rebuilding_volumes_count,
          "faulted_replicas_count": faulted_replicas_count,
          "disk_pressure_nodes_count": disk_pressure_nodes_count,
          "healthy_volumes_total": healthy_volumes_total,
          "is_alerting": is_alerting,
          "audit_timestamp": audit_time
      })

  - id: evaluate_longhorn_threshold
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether Longhorn degraded volume counts or faulted
      replicas exceed storage durability margins.
    condition: "{{ outputs.scan_longhorn_volume_metrics.vars.is_alerting and
      inputs.alert_on_longhorn_failures }}"
    then:
      - id: notify_slack_storage_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to Storage SRE
          channel detailing Longhorn volume degradation.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#storage-sre-alerts",
            "username": "Longhorn Sentinel",
            "icon_emoji": ":warning:",
            "text": "*CRITICAL: Longhorn Degraded Volume Replicas or Faulted Disks Detected*\n*Pool:* `{{ inputs.storage_pool }}`\n*Degraded Volumes:* `{{ outputs.scan_longhorn_volume_metrics.vars.degraded_volumes_count }}` (Threshold: `{{ inputs.max_degraded_volumes_threshold }}`)\n*Faulted Replicas:* `{{ outputs.scan_longhorn_volume_metrics.vars.faulted_replicas_count }}`\n*Rebuilding Volumes:* `{{ outputs.scan_longhorn_volume_metrics.vars.rebuilding_volumes_count }}`\n*Healthy Volumes:* `{{ outputs.scan_longhorn_volume_metrics.vars.healthy_volumes_total }}`\n*Timestamp:* `{{ outputs.scan_longhorn_volume_metrics.vars.audit_timestamp }}`\n*Runbook:* Check worker node physical disk space, inspect Longhorn engine instance manager logs, and verify storage network connectivity."
          }
    else:
      - id: log_nominal_longhorn_state
        type: io.kestra.plugin.core.log.Log
        description: Logs nominal telemetry when Longhorn distributed block storage
          volumes maintain full replica quorum.
        message: "NOMINAL: Longhorn storage pool {{ inputs.storage_pool }} healthy.
          Degraded volumes: {{
          outputs.scan_longhorn_volume_metrics.vars.degraded_volumes_count }},
          Faulted replicas: {{
          outputs.scan_longhorn_volume_metrics.vars.faulted_replicas_count }},
          Healthy volumes: {{
          outputs.scan_longhorn_volume_metrics.vars.healthy_volumes_total }}."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for cloud-native storage durability
      telemetry archives and compliance logs.
    format: |
      {
        "flow_id": "{{ flow.id }}",
        "storage_pool": "{{ inputs.storage_pool }}",
        "degraded_volumes_count": {{ outputs.scan_longhorn_volume_metrics.vars.degraded_volumes_count }},
        "rebuilding_volumes_count": {{ outputs.scan_longhorn_volume_metrics.vars.rebuilding_volumes_count }},
        "faulted_replicas_count": {{ outputs.scan_longhorn_volume_metrics.vars.faulted_replicas_count }},
        "disk_pressure_nodes_count": {{ outputs.scan_longhorn_volume_metrics.vars.disk_pressure_nodes_count }},
        "healthy_volumes_total": {{ outputs.scan_longhorn_volume_metrics.vars.healthy_volumes_total }},
        "is_alerting": {{ outputs.scan_longhorn_volume_metrics.vars.is_alerting }},
        "audit_timestamp": "{{ outputs.scan_longhorn_volume_metrics.vars.audit_timestamp }}"
      }

triggers:
  - id: periodic_longhorn_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Executes automated Longhorn volume durability health audit on a
      recurring 15-minute schedule.
    cron: "*/15 * * * *"