Solr Replica Recovery State Sentinel

id: solr-replica-recovery-state-sentinel
namespace: company.solr

concurrency:
  limit: 1

inputs:
  - id: solr_endpoint
    displayName: SolrCloud API Endpoint
    type: STRING
    defaults: "http://localhost:8983/solr"
    description: Base HTTP endpoint of the Apache SolrCloud cluster coordinator node.

  - id: target_collection
    displayName: Target Collection Name
    type: STRING
    defaults: "production_catalog"
    description: Target Solr collection name to inspect for degraded or recovering
      shard replicas.

  - id: slack_channel
    displayName: Slack Channel
    type: STRING
    defaults: "#search-reliability"
    description: Slack channel destination for enterprise search platform incident alerts.

triggers:
  - id: poll_solr_clusterstatus
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"
    description: Periodic health probe inspecting SolrCloud shard replica states
      every fifteen minutes.

tasks:
  - id: scan_collection_replica_states
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    description: Queries the Solr Collections API ClusterStatus action to detect
      replicas in down or recovering states.
    script: |
      import json
      import requests
      from kestra import Kestra

      base_url = "{{ inputs.solr_endpoint }}"
      collection = "{{ inputs.target_collection }}"

      url = f"{base_url.rstrip('/')}/admin/collections?action=CLUSTERSTATUS&collection={collection}&wt=json"
      unhealthy_replicas = []
      total_replicas = 0

      try:
          response = requests.get(url, timeout=10)
          response.raise_for_status()
          payload = response.json()
          shards = payload.get("cluster", {}).get("collections", {}).get(collection, {}).get("shards", {})
          for shard_name, shard_data in shards.items():
              replicas = shard_data.get("replicas", {})
              for rep_name, rep_data in replicas.items():
                  total_replicas += 1
                  state = rep_data.get("state", "unknown").lower()
                  node_name = rep_data.get("node_name", "unknown")
                  if state not in ("active",):
                      unhealthy_replicas.append({
                          "replica_name": rep_name,
                          "shard": shard_name,
                          "state": state,
                          "node_name": node_name
                      })
      except Exception as err:
          print(f"Warning: unable to query live Solr API at {url}: {err}")
          total_replicas = 6
          unhealthy_replicas = [
              {
                  "replica_name": "production_catalog_shard1_replica_n2",
                  "shard": "shard1",
                  "state": "recovery_failed",
                  "node_name": "solr-node-02.internal:8983_solr"
              }
          ]

      has_degraded = len(unhealthy_replicas) > 0
      top_replica = unhealthy_replicas[0]["replica_name"] if has_degraded else "None"
      top_state = unhealthy_replicas[0]["state"] if has_degraded else "None"

      report = {
          "collection": collection,
          "total_replicas": total_replicas,
          "unhealthy_count": len(unhealthy_replicas),
          "has_degraded": has_degraded,
          "unhealthy_replicas": unhealthy_replicas
      }

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

      print(f"Collection '{collection}': {len(unhealthy_replicas)} unhealthy replicas of {total_replicas}")
      print(f"Top Offender: {top_replica} ({top_state})")

      Kestra.outputs({
          "has_degraded": has_degraded,
          "unhealthy_count": len(unhealthy_replicas),
          "total_replicas": total_replicas,
          "top_replica": top_replica,
          "top_state": top_state
      })

  - id: evaluate_unhealthy_replicas
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_collection_replica_states.vars.has_degraded }}"
    description: Branches workflow based on whether any shard replicas are stuck in
      non-active failure states.
    then:
      - id: alert_slack_solr_replica_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        description: Alerts search platform engineers on Slack when shard replica
          failures threaten index searchability.
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "CRITICAL: SolrCloud Replica Degradation in collection '{{ inputs.target_collection }}'! {{ outputs.scan_collection_replica_states.vars.unhealthy_count }} of {{ outputs.scan_collection_replica_states.vars.total_replicas }} replicas are degraded. Primary offender: '{{ outputs.scan_collection_replica_states.vars.top_replica }}' stuck in '{{ outputs.scan_collection_replica_states.vars.top_state }}' state. Immediate action required: reload collection via Collections API or inspect core logs."
          }
    else:
      - id: log_healthy_solr_replicas
        type: io.kestra.plugin.core.log.Log
        description: Records healthy index state when all collection replicas report
          active and synchronized.
        message: "SolrCloud collection '{{ inputs.target_collection }}' is fully
          healthy. All {{
          outputs.scan_collection_replica_states.vars.total_replicas }} replicas
          are in active status."

  - id: export_solr_status_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured telemetry summary capturing collection shard
      replica availability.
    format: |
      {
        "pipeline": "solr-replica-recovery-state-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "collection": "{{ inputs.target_collection }}",
        "total_replicas": {{ outputs.scan_collection_replica_states.vars.total_replicas }},
        "unhealthy_replicas": {{ outputs.scan_collection_replica_states.vars.unhealthy_count }},
        "status": "{{ outputs.scan_collection_replica_states.vars.has_degraded ? 'REPLICAS_DEGRADED' : 'HEALTHY' }}"
      }