ILM & Reindexing Automation Runbooks

Monitoring Reindex Task Status with Kibana Dev Tools: Production-Grade Diagnostics & Recovery

In high-throughput log analytics and search clusters, reindex operations serve as the execution backbone for Automated Reindexing Pipelines & Workflows. When a reindex task stalls, degrades, or triggers cascading allocation failures, passive Kibana UI dashboards introduce unacceptable diagnostic latency and obscure root-cause telemetry. Monitoring Reindex Task Status with Kibana Dev Tools mandates direct, low-overhead interaction with the _tasks API, enforcing precise state inspection, thread pool correlation, and immediate intervention. This guide establishes exact diagnostic queries, failure-pattern recognition, and idempotent recovery patterns for production environments. Adherence to these protocols ensures rapid restoration and strict compliance with operational SLAs.

flowchart TD
  A["Poll task status"] --> B{"failures or conflicts?"}
  B -->|"no"| C{"completed?"}
  C -->|"yes"| D["Done"]
  C -->|"no"| A
  B -->|"yes"| E{"cluster health red?"}
  E -->|"yes"| F["Halt and escalate"]
  E -->|"no"| G["Rethrottle or safe reroute"]
  G --> A

1. Atomic Task Inspection & Baseline Diagnostics

The foundation of reliable monitoring is querying the task lifecycle directly. Do not rely on periodic _cat/tasks snapshots. Use the task-specific endpoint to retrieve atomic progress metrics:

GET _tasks/<task_id>

The response payload contains non-negotiable state indicators. Monitor status.total against status.created and status.updated to calculate real-time throughput. A divergence where status.total remains static while status.created plateaus indicates a source index scan bottleneck, typically caused by heavy segment merging, max_result_window constraints, or aggressive refresh_interval settings on the source. If status.failures increments, immediately inspect the error array. Common culprits include version_conflict_engine_exception or mapper_parsing_exception. For granular batch analysis, append ?detailed=true to expose the underlying bulk request structure and exact document routing failures. Establish baseline throughput thresholds as documented in Tracking Reindex Progress & Performance to trigger automated alerts before degradation impacts downstream consumers.

2. Thread Pool Exhaustion & Dynamic Throttling

A reindex task reporting running: true with zero progress for >300 seconds typically signals thread pool exhaustion or a circuit breaker trip. Verify the search thread pool status immediately:

GET _nodes/stats/thread_pool/search

If queue and rejected counters spike, the reindex operation is self-throttling. Adjust the requests_per_second parameter dynamically via the rethrottle endpoint (the value is a query parameter, not a request body):

POST _reindex/<task_id>/_rethrottle?requests_per_second=500

Simultaneously, inspect the circuit breaker state:

GET _nodes/stats/breaker

If fielddata or request limits are breached, the node will reject bulk operations. Reduce requests_per_second to 100–200 and monitor indices.breaker.total.limit. Do not bypass circuit breakers; they exist to prevent OOM crashes.

3. Cluster State Verification & Safe Manual Reroutes

Mid-reindex shard allocation failures require immediate cluster state inspection. Execute the following diagnostics in sequence:

GET _cluster/health

Expected Output (Degraded State):

{
  "cluster_name": "prod-logs-cluster",
  "status": "yellow",
  "timed_out": false,
  "number_of_nodes": 5,
  "active_shards": 142,
  "active_primary_shards": 71,
  "unassigned_shards": 3,
  "delayed_unassigned_shards": 0,
  "number_of_pending_tasks": 0,
  "number_of_in_flight_fetch": 0,
  "task_max_waiting_in_queue_millis": 0,
  "active_shards_percent_as_number": 97.9
}

If unassigned_shards > 0, isolate the cause:

GET _cluster/allocation/explain

Look for ALLOCATION_FAILED or NODE_LEFT triggers. For ILM-managed target indices, verify policy execution state:

GET <target_index>/_ilm/explain

Expected Output (ILM Stuck):

{
  "indices": {
    "logs-prod-2024.01": {
      "index": "logs-prod-2024.01",
      "managed": true,
      "policy": "logs-policy",
      "lifecycle_date_millis": 1706745600000,
      "phase": "hot",
      "action": "rollover",
      "step": "check-rollover-ready",
      "step_info": {
        "message": "index has exceeded [max_primary_shard_size=50gb] - will rollover"
      }
    }
  }
}

If replicas remain unassigned due to insufficient disk space or node constraints, execute a safe manual reroute. Do not use accept_data_loss: true unless explicitly authorized by platform engineering.

POST _cluster/reroute
{
  "commands": [
    {
      "allocate_replica": {
        "index": "<target_index>",
        "shard": 0,
        "node": "data-node-03"
      }
    }
  ]
}

4. Conflict Handling & Emergency Bypass Protocols

The _reindex API defaults to abort on error. Emergency schema migrations or bulk data migrations may require "conflicts": "proceed" in the initial request body. This bypasses version_conflict_engine_exception but requires strict post-migration reconciliation. Document versioning conflicts require explicit script overrides or version_type: "external" to maintain ordering guarantees. Never deploy conflicts: proceed to production without a validated reconciliation script and change-management approval.

5. Automated Python v8+ Recovery Orchestration

Manual Dev Tools intervention is insufficient for sustained outages. Deploy the following Python v8+ recovery script to automate task polling, dynamic throttling, cluster health validation, and safe reroute execution. This script utilizes the official elasticsearch client and enforces exponential backoff on transient failures.

import logging
import time
from elasticsearch import Elasticsearch
from elasticsearch.exceptions import ConnectionError, RequestError

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("reindex_recovery")

class ReindexRecoveryOrchestrator:
    def __init__(self, es_host: str, api_key: str, task_id: str):
        self.es = Elasticsearch(hosts=[es_host], api_key=api_key, verify_certs=True)
        self.task_id = task_id
        self.max_retries = 5
        self.backoff_factor = 2

    def poll_task_state(self) -> dict:
        """Retrieve atomic task status with failure inspection."""
        return self.es.tasks.get(task_id=self.task_id, detailed=True)

    def apply_dynamic_throttle(self, rps: int) -> bool:
        """Adjust requests_per_second dynamically on thread pool rejection.
        Rethrottling a running reindex uses reindex_rethrottle — not update_by_query."""
        try:
            self.es.reindex_rethrottle(task_id=self.task_id, requests_per_second=rps)
            logger.info(f"Throttle adjusted to {rps} rps for task {self.task_id}")
            return True
        except RequestError as e:
            logger.error(f"Throttle adjustment failed: {e}")
            return False

    def verify_cluster_health(self) -> bool:
        """Validate cluster allocation status before proceeding."""
        health = self.es.cluster.health(wait_for_status="yellow", timeout="30s")
        if health["status"] == "red":
            logger.critical("Cluster status RED. Halting recovery.")
            return False
        return health["unassigned_shards"] == 0

    def execute_safe_reroute(self, index: str, shard: int, node: str) -> bool:
        """Attempt replica allocation without data loss."""
        try:
            self.es.cluster.reroute(
                body={
                    "commands": [{
                        "allocate_replica": {
                            "index": index,
                            "shard": shard,
                            "node": node
                        }
                    }]
                }
            )
            logger.info(f"Safe reroute executed for {index} shard {shard} on {node}")
            return True
        except RequestError as e:
            logger.error(f"Reroute failed: {e}")
            return False

    def run_recovery_cycle(self):
        """Main orchestration loop with exponential backoff."""
        for attempt in range(self.max_retries):
            try:
                task = self.poll_task_state()
                status = task["task"]["status"]
                # Mid-flight conflicts surface as status.version_conflicts; document-level
                # failures appear under response.failures once the task completes.
                conflicts = status.get("version_conflicts", 0)
                failures = task.get("response", {}).get("failures", [])
                
                if conflicts > 0 or failures:
                    logger.warning(f"Conflicts: {conflicts}, document failures: {len(failures)}")
                    if not self.verify_cluster_health():
                        time.sleep(self.backoff_factor ** attempt)
                        continue
                    
                    # Dynamic throttle on rejection
                    if status.get("throttled_millis", 0) > 5000:
                        self.apply_dynamic_throttle(250)
                        
                elif status.get("total", 0) > 0 and status.get("created", 0) == 0:
                    logger.critical("Source scan bottleneck detected. Escalating to manual triage.")
                    break
                    
                if task["task"]["completed"]:
                    logger.info("Reindex task completed successfully.")
                    return True
                    
                time.sleep(10)
                
            except ConnectionError:
                logger.error("Cluster connection lost. Retrying...")
                time.sleep(self.backoff_factor ** attempt)
                
        logger.critical("Recovery cycle exhausted. Manual intervention required.")
        return False

if __name__ == "__main__":
    orchestrator = ReindexRecoveryOrchestrator(
        es_host="https://prod-es-cluster:9200",
        api_key="YOUR_API_KEY",
        task_id="your_task_id_here"
    )
    orchestrator.run_recovery_cycle()

6. Escalation Matrix & Compliance Enforcement

Automated recovery must operate within strict boundaries. When the orchestrator exhausts retries or detects RED cluster status, execute the following escalation path:

  1. Immediate Containment: Cancel the stalled task via POST _tasks/<task_id>/_cancel to free thread pool resources.
  2. Snapshot Verification: Confirm the snapshot repository is reachable via POST _snapshot/<repo_name>/_verify. Do not proceed with reindex if source segments are corrupted.
  3. Watermark Audit: Inspect disk thresholds with GET _cluster/settings?include_defaults=true&filter_path=*.cluster.routing.allocation.disk.watermark.*. If flood_stage is triggered, manually clear read-only blocks only after disk utilization drops below 85%.
  4. Platform Engineering Handoff: Provide full _cluster/allocation/explain output, thread pool rejection metrics, and ILM policy state. Do not attempt accept_data_loss reroutes without explicit written authorization.
  5. Post-Incident Reconciliation: Execute document count validation (GET _cat/count/<source_index> vs GET _cat/count/<target_index>) and verify field mappings against the Elasticsearch Task Management API specifications. All recovery actions must be logged for audit compliance.