How to Automatically Delete Old UAT Elasticsearch Indices with Python

First, import the required modules and configure logging:

import logging
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch

logging.basicConfig(
    filename='delete_uat_indices.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

Define the connection parameters for the Elasticsearch cluster and the index pattern to match UAT indices:

ELASTICSEARCH_HOST = "http://192.178.18.209:9200"
USERNAME = "elastic"
PASSWORD = "7F9L%mYWjWtb"
INDEX_PATTERN = "*uat*"

Implement the delete_old_indices function, which connects to the cluster, checks the connection, calculates the cutoff date, retrieves matching indices, filters them by name and date, and deletes the qualifying ones:

def delete_old_indices():
    # Connect to Elasticsearch
    es = Elasticsearch([ELASTICSEARCH_HOST], basic_auth=(USERNAME, PASSWORD))
    if not es.ping():
        logging.error("Unable to connect to Elasticsearch cluster")
        return
    logging.info("Successfully connected to Elasticsearch cluster")

    # Compute the date 5 days ago (the code uses 2 days)
    five_days_ago = datetime.now() - timedelta(days=2)
    date_str = five_days_ago.strftime('%Y.%m.%d')

    # Retrieve all indices matching the pattern
    try:
        indices = es.indices.get(index="*uat*")
    except Exception as e:
        logging.error(f"Failed to get index list: {e}")
        return

    # Filter indices that are not hidden and contain 'uat'
    indices_to_delete = []
    for index_name in indices.keys():
        if not index_name.startswith('.') and "uat" in index_name:
            try:
                index_date_str = index_name.split("-")[-1]  # assume date is the last part
                index_date = datetime.strptime(index_date_str, '%Y.%m.%d')
                if index_date < five_days_ago:
                    indices_to_delete.append(index_name)
            except Exception as e:
                logging.warning(f"Failed to parse date from index {index_name}: {e}")

    if not indices_to_delete:
        logging.info(f"No indices found matching 'uat' older than {date_str}")
        return

    # Delete the selected indices
    for index in indices_to_delete:
        try:
            es.indices.delete(index=index)
            logging.info(f"Successfully deleted index: {index}")
        except Exception as e:
            logging.error(f"Failed to delete index {index}: {e}")

Finally, run the function when the script is executed directly:

if __name__ == "__main__":
    delete_old_indices()