Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/usr/bin/env ipython
import json
import urllib
import requests
policy = {
"policy": {
"phases": {
"hot": {
"min_age": "0ms",
"actions": {
"rollover": {
"max_docs": 100
},
"set_priority": {
"priority": 100
}
}
}
}
}
}
def ilm_explain(es, index):
explain = requests.get(urllib.parse.urljoin(es, index + "/_ilm/explain"), params={"human": "true"}).json()
print("ILM explain for", index, ":\n", json.dumps(explain, indent=4))
def update_policies(es):
rsp = requests.get(urllib.parse.urljoin(es, "_ilm/policy"))
rsp.raise_for_status()
for index, content in rsp.json().items():
if not index.startswith("apm-"):
continue
requests.put(
urllib.parse.urljoin(es, "_ilm/policy/" + index),
json=policy
).raise_for_status()
def update_indices(es):
rsp = requests.get(urllib.parse.urljoin(es, "_alias/apm-*"))
rsp.raise_for_status()
for index in rsp.json():
print("checking ILM for", index)
ilm_info_rsp = requests.get(
urllib.parse.urljoin(es, index + "/_settings"),
params={"filter_path": "*.settings.index.lifecycle"},
)
ilm_info_rsp.raise_for_status()
lifecycle = ilm_info_rsp.json()[index]["settings"] # ["index"]["lifecycle"]
if not lifecycle:
continue
ilm_explain(es, index)
print("updating ILM for", index)
requests.post(urllib.parse.urljoin(es, index + "/_ilm/remove")).raise_for_status()
requests.put(
urllib.parse.urljoin(es, index + "/_settings"),
json=lifecycle,
).raise_for_status()
ilm_explain(es, index)
def main():
es = "http://localhost:9200"
requests.put(
urllib.parse.urljoin(es, "_cluster/settings"),
json={
"transient": {
"indices.lifecycle.poll_interval": "5s"
}
}
).raise_for_status()
update_policies(es)
update_indices(es)
if __name__ == '__main__':
main()
# set et ts=4 ai
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment