Python Orchestration Frameworks for OpenSearch ISM & CCR Operations
Python Orchestration Frameworks serve as the deterministic control plane for automating OpenSearch Index State Management (ISM) and Cross-Cluster Replication (CCR) at production scale. Declarative JSON policies alone cannot handle dynamic ingestion spikes, replication topology shifts, or state drift across multi-tenant clusters. Orchestrators bridge the gap between static OpenSearch configurations and runtime operational requirements by enforcing idempotent API execution, explicit state reconciliation, and structured failure recovery. This guide details how to architect, deploy, and debug Python-driven automation that directly interfaces with ISM and CCR plugin endpoints, ensuring precise phase transitions, controlled rollover execution, and resilient replication workflows. For foundational policy structures and baseline configuration, reference the ISM Policy Implementation & Python Automation baseline before extending into programmatic control.
Execution Architecture & Async State Reconciliation
Production-grade orchestration requires deterministic scheduling, connection pool management, and explicit state tracking. Traditional cron-based scripts fail under partial network partitions and lack visibility into OpenSearch background thread execution. Modern Python Orchestration Frameworks leverage async execution patterns to poll _plugins/_ism/explain and _plugins/_replication endpoints without blocking worker threads or exhausting HTTP keep-alive connections.
The execution model must separate policy definition from policy enforcement. Implement a directed acyclic graph (DAG) or finite state machine where each node represents an ISM phase (hot → warm → cold → delete) or a CCR topology action (follower creation, pause, resume, failover). Configure the orchestrator to run state reconciliation loops at 15–30 second intervals. Enforce strict timeout boundaries on _cluster/settings mutations and implement circuit breakers that halt downstream workflows when OpenSearch thread pool queue depths exceed 80% capacity. Understanding how the cluster evaluates index age, document count, and shard size is critical when mapping these states programmatically; consult the Phase Transition Logic documentation to align your state machine thresholds with OpenSearch’s internal evaluation engine.
flowchart LR
A["Reconciliation loop (15-30s)"] --> B["GET _plugins/_ism/explain"]
B --> C{"State drift or failed action?"}
C -- "failed" --> D["retry / change_policy"]
C -- "on track" --> E["GET _plugins/_replication status"]
D --> E
E --> F{"Queue depth > 80%?"}
F -- "yes" --> G["Circuit breaker: pause"]
F -- "no" --> A
G --> A
Production-Grade Python Implementation
The opensearch-py client provides the necessary primitives for programmatic ISM and CCR control. Since native wrapper methods for plugin endpoints are not yet standardized, use transport.perform_request for direct API routing. The following implementation demonstrates an async orchestrator that attaches policies, monitors phase progression, manages CCR follower topology, and enforces exponential backoff on transient failures.
import asyncio
import logging
from typing import Dict, List, Optional
from opensearchpy import AsyncOpenSearch
from opensearchpy.exceptions import ConnectionTimeout, TransportError, NotFoundError
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
logger = logging.getLogger("opensearch_ism_ccr")
class ISMCCROrchestrator:
def __init__(self, hosts: List[str], http_auth: tuple, max_retries: int = 3):
self._max_retries = max_retries
self.client = AsyncOpenSearch(
hosts=hosts,
http_auth=http_auth,
timeout=30,
max_retries=max_retries,
retry_on_timeout=True
)
self._circuit_open = False
self._reconciliation_interval = 15.0
async def _request_with_retry(self, method: str, path: str, body: Optional[Dict] = None):
"""Idempotent request wrapper with exponential backoff and circuit breaker."""
if self._circuit_open:
raise RuntimeError("Circuit breaker open: halting downstream requests.")
for attempt in range(self._max_retries):
try:
return await self.client.transport.perform_request(
method=method, url=path, body=body
)
except ConnectionTimeout:
delay = 2 ** attempt
logger.warning(f"Timeout on {path}, retrying in {delay}s (attempt {attempt+1})")
await asyncio.sleep(delay)
except TransportError as e:
if e.status_code == 429:
await asyncio.sleep(5)
continue
raise
raise RuntimeError("Max retries exceeded for request.")
async def attach_ism_policy(self, index_pattern: str, policy_id: str):
"""Attach ISM policy to matching indices."""
path = f"/_plugins/_ism/add/{index_pattern}"
body = {"policy_id": policy_id}
return await self._request_with_retry("POST", path, body)
async def get_ism_state(self, index: str) -> Dict:
"""Retrieve current ISM phase and action status."""
path = f"/_plugins/_ism/explain/{index}"
try:
return await self._request_with_retry("GET", path)
except NotFoundError:
return {}
async def manage_ccr_follower(self, follower_index: str, leader_index: str, leader_cluster: str):
"""Initialize or verify CCR replication topology."""
path = f"/_plugins/_replication/follow/{follower_index}"
body = {
"leader_alias": leader_cluster,
"leader_index": leader_index,
"use_roles_map": True
}
return await self._request_with_retry("PUT", path, body)
async def evaluate_rollover_triggers(self, index: str, size_gb: float, age_hours: float) -> bool:
"""
Evaluate dynamic rollover conditions before delegating to OpenSearch.
Aligns with standard [Rollover Trigger Configuration](/ism-policy-implementation-python-automation/rollover-trigger-configuration/)
patterns for size and age thresholds.
"""
stats = await self.client.indices.stats(index=index, metric="store")
primary_bytes = stats["indices"][index]["total"]["store"]["size_in_bytes"]
current_gb = primary_bytes / (1024 ** 3)
if current_gb >= size_gb:
logger.info(f"Size threshold breached for {index}: {current_gb:.2f}GB")
return True
return False
async def run_reconciliation_loop(self):
"""Main async loop for state polling and policy enforcement."""
logger.info("Starting ISM/CCR reconciliation loop.")
while True:
try:
# Example: Poll active indices and reconcile state
# Replace with your index discovery logic
indices_to_monitor = ["logs-app-*", "metrics-*"]
for idx in indices_to_monitor:
state = await self.get_ism_state(idx)
if state.get(idx, {}).get("state", {}).get("name") == "hot":
await self.evaluate_rollover_triggers(idx, size_gb=50.0, age_hours=24.0)
await asyncio.sleep(self._reconciliation_interval)
except Exception as e:
logger.error(f"Reconciliation loop failed: {e}")
await asyncio.sleep(30)
async def main():
orchestrator = ISMCCROrchestrator(
hosts=["https://opensearch-cluster.local:9200"],
http_auth=("admin", "secure_password")
)
await orchestrator.run_reconciliation_loop()
if __name__ == "__main__":
asyncio.run(main())
Operational Safeguards & Runtime Policy Mutation
Deterministic automation requires strict boundaries around state mutations. When orchestrating ISM policies across hundreds of indices, race conditions and partial failures can leave clusters in inconsistent states. Implement idempotent policy attachment by verifying existing policy bindings before issuing POST requests. Use OpenSearch’s _cluster/settings API to temporarily disable automated actions during maintenance windows, and always capture policy snapshots before executing bulk mutations.
For environments requiring dynamic threshold adjustments based on real-time ingestion velocity, programmatic policy updates become necessary. The orchestrator can fetch active policies via GET _plugins/_ism/policies/{id}, modify conditions or actions in memory, and push the updated JSON back to the cluster. This approach eliminates manual policy drift and ensures alignment with capacity planning forecasts. Detailed implementation patterns for runtime policy mutation are covered in Python automation for dynamic ISM policy updates.
Deploy the orchestrator as a systemd service or containerized workload with health checks mapped to the reconciliation loop’s heartbeat. Expose Prometheus metrics for queue depth, retry counts, and phase transition latency. Reference the official OpenSearch Python Client documentation for connection pooling best practices, and align your async event loop configuration with Python’s asyncio standards to prevent thread starvation during high-concurrency polling cycles.