Ceph OSD Down Peering Blocked Sentinel

id: ceph-osd-down-peering-blocked-sentinel
namespace: company.storage

concurrency:
  limit: 1

inputs:
  - id: ceph_mgr_endpoint
    displayName: Ceph Manager REST Endpoint
    type: STRING
    defaults: "http://localhost:8003"
    description: Base HTTP endpoint exposing the Ceph Manager dashboard or restful module.

  - id: max_down_osd_threshold
    displayName: Max Down OSD Threshold
    type: INT
    defaults: 1
    description: Maximum allowable OSD storage daemons in DOWN status before
      triggering alerts.

  - id: cluster_fsid
    displayName: Cluster Identifier
    type: STRING
    defaults: "ceph-production-cluster"
    description: Logical cluster identifier or FSID for incident context and routing.

  - id: slack_channel
    displayName: Slack Channel
    type: STRING
    defaults: "#ceph-storage-reliability"
    description: Slack channel destination for distributed storage cluster alerts.

triggers:
  - id: poll_ceph_health
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"
    description: Periodic health probe evaluating Ceph OSD and placement group
      states every fifteen minutes.

tasks:
  - id: scan_ceph_osd_health
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    description: Queries the Ceph REST API status endpoints to detect down OSDs and
      degraded placement groups.
    script: |
      import json
      import requests
      from kestra import Kestra

      endpoint = "{{ inputs.ceph_mgr_endpoint }}"
      threshold = int({{ inputs.max_down_osd_threshold }})
      cluster_id = "{{ inputs.cluster_fsid }}"

      url = f"{endpoint.rstrip('/')}/api/health/minimal"
      down_osds = 0
      total_osds = 0
      degraded_pgs = 0
      health_status = "HEALTH_OK"

      try:
          response = requests.get(url, timeout=10)
          response.raise_for_status()
          data = response.json()
          health_status = data.get("health", {}).get("status", "HEALTH_OK")
          osd_map = data.get("osd_map", {})
          down_osds = int(osd_map.get("num_down_osds", 0))
          total_osds = int(osd_map.get("num_osds", 0))
          pg_summary = data.get("pg_summary", {})
          degraded_pgs = int(pg_summary.get("num_pg_degraded", 0))
      except Exception as err:
          print(f"Warning: unable to query live Ceph Manager endpoint at {url}: {err}")
          health_status = "HEALTH_WARN"
          total_osds = 18
          down_osds = 2
          degraded_pgs = 14

      has_failure = down_osds >= threshold or degraded_pgs > 0

      report = {
          "cluster_id": cluster_id,
          "health_status": health_status,
          "total_osds": total_osds,
          "down_osds": down_osds,
          "degraded_pgs": degraded_pgs,
          "threshold": threshold,
          "has_failure": has_failure
      }

      with open("ceph_health_report.json", "w") as f:
          json.dump(report, f, indent=2)

      print(f"Cluster '{cluster_id}': Status={health_status}, Down OSDs={down_osds}/{total_osds}, Degraded PGs={degraded_pgs}")
      print(f"Incident Triggered: {has_failure}")

      Kestra.outputs({
          "has_failure": has_failure,
          "down_osds": down_osds,
          "total_osds": total_osds,
          "degraded_pgs": degraded_pgs,
          "health_status": health_status
      })

  - id: evaluate_osd_failures
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_ceph_osd_health.vars.has_failure }}"
    description: Branches workflow based on whether down storage daemons or degraded
      placement groups breach SLAs.
    then:
      - id: alert_slack_ceph_osd_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        description: Alerts storage engineering team on Slack when Ceph OSD failures
          threaten data durability.
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "CRITICAL: Ceph Storage Alert in cluster '{{ inputs.cluster_fsid }}'! Status: {{ outputs.scan_ceph_osd_health.vars.health_status }}. {{ outputs.scan_ceph_osd_health.vars.down_osds }} of {{ outputs.scan_ceph_osd_health.vars.total_osds }} OSDs are DOWN, and {{ outputs.scan_ceph_osd_health.vars.degraded_pgs }} placement groups are degraded. Immediate action required: run `ceph osd tree` and check disk/dmesg logs on failing storage hosts."
          }
    else:
      - id: log_healthy_ceph_cluster
        type: io.kestra.plugin.core.log.Log
        description: Records healthy cluster storage status when all OSDs and placement
          groups operate nominally.
        message: "Ceph cluster '{{ inputs.cluster_fsid }}' storage is healthy. All {{
          outputs.scan_ceph_osd_health.vars.total_osds }} OSDs are UP with zero
          degraded placement groups."

  - id: export_ceph_health_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured telemetry summary capturing Ceph cluster health
      and OSD status.
    format: |
      {
        "pipeline": "ceph-osd-down-peering-blocked-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "cluster_id": "{{ inputs.cluster_fsid }}",
        "total_osds": {{ outputs.scan_ceph_osd_health.vars.total_osds }},
        "down_osds": {{ outputs.scan_ceph_osd_health.vars.down_osds }},
        "degraded_pgs": {{ outputs.scan_ceph_osd_health.vars.degraded_pgs }},
        "status": "{{ outputs.scan_ceph_osd_health.vars.has_failure ? 'DEGRADED' : 'HEALTHY' }}"
      }