OpenSearch Cluster Health and Unassigned Shard Sentinel

id: opensearch-cluster-health-and-unassigned-shard-sentinel
namespace: company.team

description: |
  Continuously audit OpenSearch cluster health, detect degraded yellow or red states,
  and identify unassigned shard accumulation before search latency spikes or data
  unavailability occurs. Evaluates cluster health diagnostics using Pebble, routes
  through a Switch flow control task, dispatches tiered Slack incident alerts, and
  records an immutable operational audit log.

inputs:
  - id: opensearch_url
    type: STRING
    displayName: OpenSearch URL
    description: Base URL of the OpenSearch cluster (including scheme, host, and port).
    defaults: "http://localhost:9200"

  - id: unassigned_shard_threshold
    type: INT
    displayName: Unassigned Shard Threshold
    description: Maximum allowable unassigned shards before a degraded warning is triggered.
    defaults: 0

triggers:
  - id: every_15_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled. Enable in production to continuously evaluate
      cluster health.
    cron: "*/15 * * * *"
    disabled: true

tasks:
  - id: probe_cluster_health
    type: io.kestra.plugin.opensearch.Request
    description: Query cluster health endpoint. allowFailure ensures network
      timeouts or authentication issues are captured for classification.
    allowFailure: true
    connection:
      hosts:
        - "{{ inputs.opensearch_url }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USERNAME') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    method: GET
    endpoint: "_cluster/health"

  - id: probe_cluster_allocation
    type: io.kestra.plugin.opensearch.Request
    description: Query shard allocation explanation to capture diagnostic root cause
      when unassigned shards exist.
    allowFailure: true
    connection:
      hosts:
        - "{{ inputs.opensearch_url }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USERNAME') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    method: GET
    endpoint: "_cluster/allocation/explain"

  - id: classify_health_status
    type: io.kestra.plugin.core.debug.Return
    description: Classify health state into PROBE_FAILED, CLUSTER_RED,
      CLUSTER_YELLOW, or HEALTHY based on cluster status and shard counts.
    format: "{{ outputs.probe_cluster_health is not defined or
      outputs.probe_cluster_health.response is not defined or
      outputs.probe_cluster_health.response.status is not defined ?
      'PROBE_FAILED' : (outputs.probe_cluster_health.response.status == 'red' ?
      'CLUSTER_RED' : (outputs.probe_cluster_health.response.status == 'yellow'
      or outputs.probe_cluster_health.response.unassigned_shards >
      inputs.unassigned_shard_threshold ? 'CLUSTER_YELLOW' : 'HEALTHY')) }}"

  - id: route_by_health_status
    type: io.kestra.plugin.core.flow.Switch
    description: Route execution based on classified OpenSearch cluster health status.
    value: "{{ outputs.classify_health_status.value }}"
    cases:
      CLUSTER_RED:
        - id: alert_cluster_red
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Dispatch critical alert when OpenSearch reports red status
            indicating missing primary shards.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "🚨 *OpenSearch Cluster RED Alert*\n• Cluster: `" ~ inputs.opensearch_url ~ "`\n• Status: RED\n• Active Shards: " ~ outputs.probe_cluster_health.response.active_shards ~ "\n• Unassigned Shards: " ~ outputs.probe_cluster_health.response.unassigned_shards ~ "\n• Impact: One or more primary shards are unassigned. Immediate investigation required to prevent data loss."} | toJson }}

      CLUSTER_YELLOW:
        - id: alert_cluster_yellow
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Dispatch warning alert when OpenSearch reports yellow status
            indicating unassigned replica shards.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "⚠️ *OpenSearch Cluster YELLOW Warning*\n• Cluster: `" ~ inputs.opensearch_url ~ "`\n• Status: YELLOW\n• Active Shards: " ~ outputs.probe_cluster_health.response.active_shards ~ "\n• Unassigned Shards: " ~ outputs.probe_cluster_health.response.unassigned_shards ~ " (Threshold: " ~ inputs.unassigned_shard_threshold ~ ")\n• Impact: Replica shards are unallocated. Reduced cluster redundancy and fault tolerance."} | toJson }}

      PROBE_FAILED:
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Dispatch connectivity alert when OpenSearch endpoints cannot be
            reached.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "🚨 *OpenSearch Connectivity Failure*\n• Target URL: `" ~ inputs.opensearch_url ~ "`\n• Status: Probe task failed to execute or connect.\n• Action: Verify OpenSearch container/service status and network reachability."} | toJson }}

      HEALTHY:
        - id: log_healthy_pass
          type: io.kestra.plugin.core.log.Log
          description: Record operational log confirming green cluster health.
          message: "OpenSearch cluster health verified: status is GREEN with {{
            outputs.probe_cluster_health.response.active_shards }} active shards
            across {{ outputs.probe_cluster_health.response.number_of_nodes }}
            nodes."

    defaults:
      - id: log_unexpected_state
        type: io.kestra.plugin.core.log.Log
        message: "Unexpected classify_health_status value: {{
          outputs.classify_health_status.value }}"

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final immutable audit log confirming sentinel evaluation details
      across all branches.
    message: "OpenSearch sentinel completed. execution={{ execution.id }} status={{
      outputs.classify_health_status.value }} active_shards={{
      outputs.probe_cluster_health.response.active_shards ?? 'N/A' }}
      unassigned_shards={{
      outputs.probe_cluster_health.response.unassigned_shards ?? 'N/A' }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Catch-all notification if workflow encounters fatal execution
      errors outside handled probe tasks.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "❌ *OpenSearch Sentinel Flow Error*\nFlow `" ~ flow.id ~ "` failed during execution `" ~ execution.id ~ "`. Check Kestra execution logs."} | toJson }}