es_delete_lifecycle_indices.py

es_delete_lifecycle_indices.py

import requests
import datetime

ES_HOST = 'http://localhost:9200'

lifecycle_rules = [
    {
        'days': 0,
        'format': '%Y.%m.%d',
        'indices': [
            'ignore-index', 
        ],
    },
    {
        'days': 3,
        'format': '%Y.%m.%d',
        'indices': [
            'three-days-index'
        ],
    },
    {
        'days': 7,
        'format': '%Y.%m.%d',
        'indices': [
            'seven-days-index', 
        ],
    },
    {
        'days': 15,
        'format': '%Y.%m.%d',
        'indices': [
            'fifteen-days-index',
        ]
    },
    {
        'days': 30,
        'format': '%Y.%m.%d',
        'indices': [
            'thirty-days-index',
        ]
    }
]

def get_indices():
    url = f"{ES_HOST}/_cat/indices?h=index"
    resp = requests.get(url)
    if resp.status_code != 200:
        print(f"Failed to get index list: {resp.status_code} {resp.text}")
        return []
    return [line.strip() for line in resp.text.splitlines()]

def get_index_stats(index_name):
    url = f"{ES_HOST}/{index_name}/_stats"
    resp = requests.get(url)
    if resp.status_code != 200:
        return None
    data = resp.json()
    stats = data.get('indices', {}).get(index_name, {}).get('total', {})
    docs_count = stats.get('docs', {}).get('count', 0)
    store_size = stats.get('store', {}).get('size_in_bytes', 0)
    return docs_count, store_size

def delete_index(index_name):
    url = f"{ES_HOST}/{index_name}"
    resp = requests.delete(url)
    return resp

def main():
    now = datetime.datetime.now()
    print(f"====[{now.strftime('%Y-%m-%d %H:%M:%S')}] Elasticsearch lifecycle index deletion started====")

    all_indices = get_indices()
    for rule in lifecycle_rules:
        days = rule["days"]
        date_fmt = rule["format"]
        prefix_list = rule["indices"]

        for prefix in prefix_list:
            for idx in all_indices:
                if not idx.startswith(prefix + "-"):
                    continue

                date_str = idx[len(prefix)+1:]
                try:
                    idx_date = datetime.datetime.strptime(date_str, date_fmt)
                except Exception:
                    print(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Index {idx} date format invalid, skip.")
                    continue

                if (now - idx_date).days > days:
                    stats = get_index_stats(idx)
                    if stats is None:
                        print(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Index not found: {idx}")
                        continue

                    docs_count, store_size = stats
                    print(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Deleting index: {idx} | Docs: {docs_count} | Size: {store_size/1024/1024:.2f} MB")

                    resp = delete_index(idx)
                    if resp.status_code == 200:
                        print(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Deleted index: {idx}")
                    elif resp.status_code == 404:
                        print(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Index not found during delete: {idx}")
                    else:
                        print(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Failed to delete index {idx}: {resp.status_code} {resp.text}")

    print(f"====[{now.strftime('%Y-%m-%d %H:%M:%S')}] Elasticsearch lifecycle index deletion finished====")

if __name__ == '__main__':
    main()