OpenSearch ISM & Cross-Cluster Replication

Detecting and reconciling ISM policy drift

This walkthrough builds a single self-contained Python script that fetches the live ISM policy from one cluster, canonicalizes it, diffs it line-by-line against the version-controlled JSON in your git repository, prints a human-readable report, and — behind an explicit --apply flag — repairs the drift with a PUT guarded by the live seq_no and primary_term. It is the focused, one-policy implementation behind the control loop described in Policy Drift Detection.

Prerequisites

Confirm each of these before running the script — a missing desired file or an over-broad service-account role is the usual reason the first run reports phantom drift or fails to write.

How the script decides

The script has exactly one decision to make per run: after canonicalizing both sides, are they equal? If so it exits clean. If not, it prints the diff and then branches on the --apply flag — report-only, or a version-guarded write that either succeeds or bounces off a 409 when someone edited the policy first.

Single-policy detect and reconcile decision Flowchart. Read the live policy document and its seq_no and primary_term. Canonicalize both the live body and the desired git body. Decision: are the canonical strings equal? Yes ends in a terminal in-sync state. No prints a unified diff, then a decision on the apply flag: report-only ends in exit-with-drift-code; apply performs a version-guarded PUT, which either succeeds and ends in reconciled, or returns 409 and ends in conflict-retry-next-run. equal differ --apply 409 Read live policy + seq_no GET _plugins/_ism/policies/<id> canonical equal? In sync exit 0 Print unified diff difflib on canonical lines apply flag? report-only Exit with drift code report-only run Version-guarded PUT if_seq_no / if_primary_term Conflict retry next run

Step-by-step procedure

Each step is standard library plus opensearch-py. Build the pieces, then wire them into a small CLI.

1. Load the desired document and read live state

Read the git file and fetch the running policy in one place so the two are captured at the same moment. The live read yields the version markers you will need if you decide to write.

Python
import json
from opensearchpy import OpenSearch, exceptions

def load_desired(path: str) -> dict:
    with open(path) as fh:
        return json.load(fh)

def read_live(client: OpenSearch, policy_id: str) -> dict:
    return client.transport.perform_request(
        "GET", f"/_plugins/_ism/policies/{policy_id}")
Text
live _seq_no=42 _primary_term=6  (captured for the guarded PUT)

Gotcha: capture _seq_no and _primary_term from this read, not from a cached value. If you diff against a stale seq_no, a healthy concurrent edit will look like a conflict on write and a real conflict will slip through.

2. Canonicalize both sides identically

The live and desired bodies must pass through the exact same normalization or the diff is meaningless. Strip server-managed fields, sort keys, and emit one JSON object per line so difflib can align them.

Python
SERVER_FIELDS = {"policy_id", "last_updated_time", "last_updated_user", "schema_version"}

def canonical_lines(body: dict) -> list[str]:
    cleaned = {k: v for k, v in body.items() if k not in SERVER_FIELDS}
    pretty = json.dumps(cleaned, sort_keys=True, indent=2)
    return pretty.splitlines()

Gotcha: use the same canonical_lines for both inputs. A subtle bug is stripping server fields from live but not from desired (or vice versa) — the field then shows up as a one-sided diff line that no reconcile can ever close.

3. Diff and render a readable report

Feed the two line lists to difflib.unified_diff. An empty generator means in-sync; any output is the exact semantic delta, ready to print or attach to an alert.

Python
import difflib

def diff_report(live_body: dict, desired_body: dict) -> str:
    live = canonical_lines(live_body)
    want = canonical_lines(desired_body)
    delta = difflib.unified_diff(live, want, fromfile="live", tofile="desired", lineterm="")
    return "\n".join(delta)
Text
--- live
+++ desired
@@ -7,7 +7,7 @@
       "actions": [
         {
           "force_merge": {
-            "max_num_segments": 3
+            "max_num_segments": 1
           }
         }

Gotcha: the diff shows the live value first and the desired value second (the - line is the live cluster, the + line is git). Read it as “the live cluster has 3, git wants 1” — reconciling makes the live cluster match the + side.

4. Reconcile under optimistic concurrency

When --apply is set and the diff is non-empty, PUT the desired document back with the captured if_seq_no/if_primary_term. A 409 means someone wrote the policy between your read and your write; that is a signal to retry on the next run, not to force the write.

Python
def reconcile(client, policy_id, desired, seq_no, primary_term) -> str:
    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,
        )
        return "reconciled"
    except exceptions.ConflictError:
        return "conflict"  # concurrent write; do not force

Gotcha: never fall back to an unguarded PUT after a 409. Dropping the version guard to “make it work” is exactly how you overwrite a colleague’s emergency hotfix without a trace.

5. Wire the CLI

Assemble the pieces behind an argparse interface whose default is detect-only. The exit code encodes the outcome so a CI job or a scheduler can branch on it.

Python
import argparse, sys

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("policy_id")
    ap.add_argument("desired_path")
    ap.add_argument("--apply", action="store_true", help="write the fix (default: detect only)")
    args = ap.parse_args()

    client = OpenSearch(hosts=[{"host": "localhost", "port": 9200}],
                        http_auth=("admin", "admin"), use_ssl=True, verify_certs=True)
    desired = load_desired(args.desired_path)
    live = read_live(client, args.policy_id)

    report = diff_report(live["policy"], desired["policy"])
    if not report:
        print(f"in_sync {args.policy_id}")
        sys.exit(0)

    print(report)
    if not args.apply:
        sys.exit(2)  # drift found, not repaired

    outcome = reconcile(client, args.policy_id, desired,
                        live["_seq_no"], live["_primary_term"])
    print(f"{outcome} {args.policy_id}")
    sys.exit(0 if outcome == "reconciled" else 3)

if __name__ == "__main__":
    main()

Gotcha: exit code 2 (drift found, report-only) is what a scheduled wrapper watches for to decide whether to page. Reserve 3 for a conflict so alerting can distinguish “drift I could not repair” from “drift I chose not to repair.” That wrapper is built in Automating ISM drift remediation with scheduled jobs.

Verification

After a reconcile run, confirm the document is in sync at the version level and that the write actually landed.

Shell
# 1. The live seq_no should have advanced by exactly one after a reconcile.
GET _plugins/_ism/policies/logs-hot-warm-delete
# "_seq_no": 43, "_primary_term": 6   <- one past the 42 we wrote against
Shell
# 2. Re-run the detector; it must now report in_sync with exit 0.
python drift.py logs-hot-warm-delete policies/logs-hot-warm-delete.json
# in_sync logs-hot-warm-delete

A healthy result shows _seq_no incremented by one and a second detector run exiting 0 with no diff. If the seq_no jumped by more than one, another writer is also touching the policy — investigate before enabling --apply on a schedule.

Common failures

Symptom Root cause Fix command
Diff never empties despite identical intent One side not canonicalized (server fields or key order) Run both inputs through the same canonical_lines; strip policy_id/last_updated_time/schema_version
PUT returns 409 version_conflict_engine_exception Policy edited between the read and the write Re-run; the script re-reads a fresh _seq_no and retries the guarded write
Reconcile succeeds but indices still run old rules PUT updates the document, not managed indices POST _plugins/_ism/change_policy/<idx> with the target state
NotFoundError on the live read Policy id typo or policy was deleted Verify id with GET _plugins/_ism/policies; recreate by PUT-ting the desired file
Exit code 2 in CI but no obvious change Whitespace-only edit on the live cluster that canonicalization collapses to nothing Confirm the diff body is truly empty; if empty, treat as in-sync and adjust the comparison to canonical, not raw

Frequently asked questions

Why print a unified diff instead of just "drift detected"?

Because the diff is what a human needs to decide whether to reconcile. “Drift detected” tells you something changed; the unified diff tells you a force_merge dropped from max_num_segments: 1 to 3, which is the difference between a safe formatting change and a retention-altering mistake. The -/+ lines also become the body of a useful alert.

Is it safe to run the detector (without --apply) continuously?

Yes. Without --apply the script only issues a GET and computes a diff in memory — it never writes to the live cluster. That is why detect-only is the default and why a scheduler can run it every sweep interval without risk. Only the --apply path writes, and even then it is guarded by if_seq_no.

What if the desired file itself is wrong?

Then a report-only run is your safety net: it prints the diff and exits 2 without touching the live cluster, so you review the delta before ever passing --apply. This is the whole argument for keeping detection and reconciliation as separate steps — you always see what would change before it changes.

Up: Policy Drift Detection