OpenSearch ISM & Cross-Cluster Replication

Automating ISM drift remediation with scheduled jobs

This walkthrough wraps the drift detector in a Kubernetes CronJob that runs on a fixed cadence, emits an alert-grade metric the moment the live ISM policy diverges from its git source of truth, and — only when a dry-run guard is explicitly lifted — auto-reconciles the policy and re-applies it to stuck indices with change_policy. It operationalizes the single-run detector from Detecting and reconciling ISM policy drift into an unattended control loop, following the model laid out in Policy Drift Detection.

Prerequisites

Confirm each of these before scheduling the job — an unset dry-run guard or a credential baked into the image is the usual reason a first deployment either does nothing or does far too much.

What the scheduled job does each cycle

Every scheduled run performs the same sequence: check each policy for drift, push a gauge that alerting watches, and then consult the dry-run guard. With the guard on (the default) it stops at alerting; with the guard lifted it reconciles the document and re-applies the current version to any index still pinned to an old policy_seq_no.

Scheduled drift remediation cycle with a dry-run guard Flowchart. A CronJob triggers the runner on a fixed cadence. The runner iterates each policy and diffs it against the git desired document, then pushes a drift gauge to the metrics sink which feeds an alert rule. Decision: is the dry-run guard on? Yes ends the cycle after alerting only. No performs a version-guarded reconcile PUT, then runs change_policy on indices pinned to an old policy_seq_no, then ends the cycle. guard on guard off next policy CronJob */10m Diff each policy vs git desired Push drift gauge metrics sink → alert rule dry-run guard? Alert only no writes Reconcile PUT (guarded) if_seq_no / if_primary_term change_policy stuck idx move to current version

Step-by-step procedure

Build the runner that iterates policies and emits metrics, then package it as a CronJob with the dry-run guard exposed as an environment variable.

1. Iterate policies and emit a per-policy drift gauge

Wrap the detector in a loop over every policy id you manage, and record a 0/1 drift value per id. The gauge — not a log line — is what alerting keys on, so a silent job that stops running still fires a staleness alert.

Python
import json, logging, os
from opensearchpy import OpenSearch, exceptions

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

SERVER_FIELDS = {"policy_id", "last_updated_time", "last_updated_user", "schema_version"}
APPLY = os.environ.get("DRIFT_APPLY", "false").lower() == "true"

def canonical(body: dict) -> str:
    cleaned = {k: v for k, v in body.items() if k not in SERVER_FIELDS}
    return json.dumps(cleaned, sort_keys=True, separators=(",", ":"))

def check_policy(client, policy_id, desired_path) -> int:
    """Return 1 if drifted, 0 if in sync; reconcile when APPLY is set."""
    with open(desired_path) as fh:
        desired = json.load(fh)
    try:
        live = client.transport.perform_request(
            "GET", f"/_plugins/_ism/policies/{policy_id}")
    except exceptions.NotFoundError:
        logger.error("policy_missing id=%s", policy_id)
        return 1
    if canonical(live["policy"]) == canonical(desired["policy"]):
        logger.info("in_sync id=%s", policy_id)
        return 0
    logger.warning("drift id=%s apply=%s", policy_id, APPLY)
    if APPLY:
        reconcile(client, policy_id, desired, live["_seq_no"], live["_primary_term"])
    return 1
Text
2026-07-18 02:10:03 [INFO]  in_sync id=logs-hot-warm-delete
2026-07-18 02:10:03 [WARNING] drift id=metrics-rollup apply=false

Gotcha: return the drift value even after reconciling — the gauge should reflect that drift was seen this cycle, so a dashboard shows the event and the subsequent return to 0. Suppressing the gauge on the reconcile path hides the very events you deployed the job to catch.

2. Reconcile and re-apply to stuck indices

When APPLY is set, the reconcile step does two things the alert-only path never does: a version-guarded PUT to fix the document, then a change_policy for each index whose policy_seq_no trails the current version.

Python
def reconcile(client, policy_id, desired, seq_no, primary_term) -> None:
    try:
        client.transport.perform_request(
            "PUT", f"/_plugins/_ism/policies/{policy_id}",
            params={"if_seq_no": seq_no, "if_primary_term": primary_term},
            body=desired)
        logger.info("reconciled id=%s from_seq_no=%s", policy_id, seq_no)
    except exceptions.ConflictError:
        logger.error("conflict id=%s; retry next cycle", policy_id)
        return
    move_stuck_indices(client, policy_id)

def move_stuck_indices(client, policy_id) -> None:
    explain = client.transport.perform_request(
        "GET", f"/_plugins/_ism/explain/{policy_id}-*")
    current = client.transport.perform_request(
        "GET", f"/_plugins/_ism/policies/{policy_id}")["_seq_no"]
    for index, meta in explain.items():
        if not isinstance(meta, dict) or "policy_seq_no" not in meta:
            continue
        if meta["policy_seq_no"] < current:
            state = meta.get("state", {}).get("name")
            client.transport.perform_request(
                "POST", f"/_plugins/_ism/change_policy/{index}",
                body={"policy_id": policy_id, "state": state})
            logger.info("moved index=%s to seq_no=%s at state=%s", index, current, state)

Gotcha: pass the index’s current state into change_policy so a reconcile does not rewind an index that is mid-warm back to hot. Re-applying without a state can re-run actions the index already completed. The state semantics are the ones described in Policy Drift Detection.

3. Publish the gauge to the metrics sink

Push the per-policy drift value with the policy id as a label. A max by (policy_id) alert over a short window then pages on any policy that is drifted for more than one cycle.

Python
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway

def publish(results: dict[str, int], gateway: str) -> None:
    registry = CollectorRegistry()
    g = Gauge("ism_policy_drift", "1 if live policy differs from git",
              ["policy_id"], registry=registry)
    for policy_id, drifted in results.items():
        g.labels(policy_id=policy_id).set(drifted)
    push_to_gateway(gateway, job="ism-drift-remediation", registry=registry)
Text
ism_policy_drift{policy_id="logs-hot-warm-delete"} 0.0
ism_policy_drift{policy_id="metrics-rollup"} 1.0

Gotcha: use push_to_gateway (replace), not pushadd_to_gateway, so a policy that returns to sync overwrites its stale 1 with a 0. With pushadd an old drift value lingers in the gateway and the alert never clears.

4. Package as a CronJob with the dry-run guard

Run the job on a cadence aligned with the ISM sweep, inject credentials from a Secret, and expose the guard as DRIFT_APPLY. Keeping it "false" in the committed spec means auto-reconcile can only be turned on by an explicit, reviewable edit.

YAML
apiVersion: batch/v1
kind: CronJob
metadata:
  name: ism-drift-remediation
spec:
  schedule: "*/10 * * * *"        # align with the 5m ISM sweep, one cycle of slack
  concurrencyPolicy: Forbid        # never overlap remediation cycles
  jobTemplate:
    spec:
      backoffLimit: 1
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: remediate
              image: registry.internal/ism-drift:2.1.0
              envFrom:
                - secretRef: { name: opensearch-drift-creds }
              env:
                - { name: OPENSEARCH_HOST, value: "https://opensearch.data.svc:9200" }
                - { name: PUSHGATEWAY, value: "http://pushgateway.monitoring.svc:9091" }
                - { name: DRIFT_APPLY, value: "false" }   # detect + alert only; flip to enable repair
              volumeMounts:
                - { name: desired, mountPath: /policies, readOnly: true }
          volumes:
            - name: desired
              configMap: { name: ism-desired-policies }

Gotcha: concurrencyPolicy: Forbid matters more here than for a read-only job — two overlapping reconcile cycles can each read seq_no: 42, and the second PUT will 409 against the first. Serializing cycles keeps the optimistic guard from tripping on your own job. For building and shipping this image, see CI/CD Pipeline Integration.

Verification

After the first scheduled runs, confirm the gauge reflects reality and that a flip of the guard actually repairs drift.

Shell
# 1. The gauge should show 0 for in-sync policies, 1 for drifted ones.
curl -s "http://pushgateway.monitoring.svc:9091/metrics" | grep ism_policy_drift
# ism_policy_drift{policy_id="logs-hot-warm-delete"} 0
Shell
# 2. With DRIFT_APPLY=true, a drifted policy returns to sync and stuck indices move up.
GET _plugins/_ism/explain/metrics-rollup-*
# "metrics-rollup-000004": { "policy_seq_no": 51, "state": { "name": "warm" } }
# policy_seq_no now equals the current policy _seq_no -> no longer pinned

A healthy result shows every managed policy’s gauge at 0 within one cycle of a deploy, and — after enabling the guard on a genuinely drifted policy — a policy_seq_no on each index that matches the policy’s current _seq_no. If the gauge flips between 0 and 1 every cycle, a writer outside git is fighting the reconciler; freeze DRIFT_APPLY and find the other actor first.

Common failures

Symptom Root cause Fix command
Alert never clears after a policy is fixed Gauge pushed with pushadd, stale 1 retained Switch to push_to_gateway (replace) so a 0 overwrites the old value
Job 409s against itself every cycle Overlapping runs both read the same seq_no Set concurrencyPolicy: Forbid; ensure the prior job finished
Drifted policy reconciled but indices still stale change_policy skipped or missing state Re-run with the per-index state from _plugins/_ism/explain
Gauge flaps 0/1 every run A non-git actor keeps editing the policy Freeze DRIFT_APPLY=false; audit last_updated_user in GET _plugins/_ism/policies/<id>
CronJob runs but pushes nothing Pushgateway unreachable or wrong label set curl the gateway from the pod; confirm PUSHGATEWAY env and network policy

Frequently asked questions

Should the scheduled job auto-reconcile by default?

No. Ship it with DRIFT_APPLY=false so every deployment starts in detect-and-alert mode. Auto-reconcile writes to a live cluster and, on a bad desired document, can rewrite every policy in one cycle. Turn the guard off only for policies whose git source is itself gated behind review, and treat flipping it as a change worth its own pull request.

How often should the job run relative to the ISM sweep?

Run it at roughly the ISM job_interval cadence or a little slower — every 10 minutes against a 5-minute sweep is a good default. Running much faster just re-observes indices that ISM has not had a chance to advance yet, producing noisy transient “stuck” readings. One sweep of slack lets a legitimate transition settle before the detector judges it.

Why alert on a gauge instead of parsing the job's logs?

A gauge tells you two things logs cannot cheaply express: which policies are currently drifted, and whether the job ran at all. A max by (policy_id) (ism_policy_drift) alert fires on real drift, and an absent()-style alert on the same metric fires when the CronJob silently stops scheduling. Log scraping catches neither the “job died” case nor the clean per-policy dimensionality.

Up: Policy Drift Detection