-
Notifications
You must be signed in to change notification settings - Fork 0
/
find_missing_write_alias.py
48 lines (39 loc) · 1.5 KB
/
find_missing_write_alias.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
"""
Simple script that will scroll through all aliases/indices patterns and try to locate any that do not have a set write index for the entire set.
"""
from getpass import getpass, getuser
import requests
import argparse
def main(args):
auth = (args.username, args.password or getpass())
api = args.api.rstrip("/")
ignore = args.ignore
requests.get(api, auth=auth).raise_for_status()
# Get all aliases, check if each set has at
# least one index with a write index set to true.
alias_write_status = {}
current_aliases = requests.get(f"{api}/_alias", auth=auth).json()
for index, index_config in current_aliases.items():
for alias, alias_config in index_config["aliases"].items():
if alias in ignore:
continue
alias_write_status.setdefault(alias, set())
alias_write_status[alias].add(alias_config.get("is_write_index", False))
for alias, is_or_not_set in alias_write_status.items():
if True not in is_or_not_set:
print(f"{alias} is missing is_write_index=True on one of its indices")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("-u", "--username", default=getuser())
parser.add_argument("-p", "--password", default=None)
parser.add_argument(
"-i",
"--ignore",
nargs="*",
default=[
".kibana",
],
)
parser.add_argument("api")
args = parser.parse_args()
main(args)