grafana_alert_group_cleanup

alert_cleanup

#!/usr/local/bin/python


import os
import subprocess
import yaml
import re
import datetime

alert_to_be_deleted = {}
all_migrated_alerts_excel_file = pd.ExcelFile('/Users/m.badruzzaman/Downloads/Datadog_grafana_migrated_alerts.xlsx')

os.environ["MIMIR_ADDRESS"] = "https://mimir_endpoint"
os.environ["MIMIR_TENANT_ID"] = "tenant"
os.environ["MIMIR_API_KEY"] = os.environ.get("MIMIR_API_KEY")


def get_pandroa_alerts_to_be_deleted():
    with open("/Users/m.badruzzaman/Downloads/SRE_imported_Bat_Included_alert_ids_to_delete.txt", 'r') as file:
    # with open("/Users/m.badruzzaman/Downloads/short_pandora_alert_list.txt", 'r') as file:
        for line in file:
            line = line.strip()
            if len(line) > 4 and line.isdigit():
                alert_id = int(line)
                alert_to_be_deleted[alert_id] = {}

                
def get_platform_alerts_to_be_deleted():
    platform_alerts = pd.read_excel(all_migrated_alerts_excel_file, sheet_name='Platform', usecols=[
        "Datadog monitor Id/ Grafana alert group",
        "Action",
        "Link in Grafana UI",
        "Alert name"
    ])
    platform_alerts_to_be_deleted = platform_alerts[platform_alerts.isin(['Will delete']).any(axis=1)]
    for i, row in platform_alerts_to_be_deleted.iterrows():
        try:
            alert_id = int(row['Datadog monitor Id/ Grafana alert group'])
            alert_to_be_deleted[alert_id] = {
                'grafana_link': row['Link in Grafana UI'],
                'alert_name': row['Alert name']
            }
        except ValueError:
            print(f"Skipping row with non-integer id: {row['Datadog monitor Id/ Grafana alert group']}")
            

def get_pedidosya_alerts_to_be_deleted():
    platform_alerts = pd.read_excel(all_migrated_alerts_excel_file, sheet_name='Pedidosya', usecols=[
        "Datadog monitor Id/ Grafana alert group",
        "Action",
        "Link in Grafana UI",
        "Alert name"
    ])
    platform_alerts_to_be_deleted = platform_alerts[platform_alerts.isin(['Will delete']).any(axis=1)]
    for i, row in platform_alerts_to_be_deleted.iterrows():
        try:
            alert_id = int(row['Datadog monitor Id/ Grafana alert group'])
            alert_to_be_deleted[alert_id] = {
                'grafana_link': row['Link in Grafana UI'],
                'alert_name': row['Alert name']
            }
        except ValueError:
            print(f"Skipping row with non-integer id: {row['Datadog monitor Id/ Grafana alert group']}")
            

def update_alert_to_be_deleted_map():
    get_pandroa_alerts_to_be_deleted()
    # get_platform_alerts_to_be_deleted()
    # get_pedidosya_alerts_to_be_deleted()


def is_rule_group_deletable(namespace, rule_group_name):
    command = "mimirtool rules get " + namespace + " " + rule_group_name
    result = ""
    try:
        result = subprocess.run(command, capture_output=True, text=True, shell=True, env=os.environ)
        if result.stderr:
            print("Failed to get rule group '" + rule_group_name + "' due to error: \n", result.stderr)

    except subprocess.CalledProcessError as e:
        print(f"Error executing command '{str(command)}': {e}")
    except Exception as e:
        print(f"An unexpected error occurred: '{str(e)}'")

    if not result.stdout.splitlines()[1:] or result == "":
        return False


    rule_group_yaml_content = '\n'.join(result.stdout.splitlines()[1:])
    description_modified_yaml = re.sub(r'description: (.*)', r'description: "\1"', rule_group_yaml_content,
                                       flags=re.DOTALL)
    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
    loadable_yaml_content = ansi_escape.sub('', description_modified_yaml)
    try:
        rule_group = yaml.safe_load(loadable_yaml_content)
        if len(rule_group['rules']) == 1:
            print("Rule group '" + rule_group_name + "' is deletable")
            return True
        else:
            print("Rule group '" + rule_group_name + "' is not deletable. Total rules in group are " + str(len(
                rule_group['rules'])))
            return False
        # print(data['rules'])
        # print("Finished printing rule")
    except yaml.YAMLError as exc:
        print(f"Error parsing YAML: '{str(exc)}'")


def delete_rule_group(namespace, rule_group_name):
    command = "mimirtool rules delete " + namespace + " " + rule_group_name
    result = subprocess.run(command, capture_output=True, text=True, shell=True, env=os.environ)
    if result.stderr:
        print("Failed to delete rule group '" + str(rule_group_name) + "' due to error: \n", result.stderr)
    else:
        print("Successfully deleted rule group " + str(rule_group_name) + "' in namespace " + " '" + namespace + "'")




def main():
    current_entity = "pa..ra"
    namespace_name = "d**e"
    update_alert_to_be_deleted_map()
    deletable_rule_group = 0
    not_deletable_rule_group = []
    for rule_group in alert_to_be_deleted:
        # Using hardcode namesapce name (example: '***') here as there are only '**' and '**' namespaces
        if is_rule_group_deletable(namespace_name, "group-" + str(rule_group)):
            print("Deleting rule group " + "'group-" + str(rule_group) + "'")
            # delete_rule_group(namespace_name, "group-"+str(rule_group))
            deletable_rule_group += 1
        else:
            print("Rule group " + "'group-" + str(rule_group) + "' not deletable")
            not_deletable_rule_group.append(rule_group)
    print("Total deletable rule group : ", deletable_rule_group)
    print("Total not deletable rule group : ", len(not_deletable_rule_group))

    now = datetime.datetime.now()
    timestamp = now.strftime("%Y-%m-%d_%H-%M-%S")
    filename = f"{current_entity}_not_deletable_{timestamp}.txt"
    with open(filename, 'w') as file:
        for rule_group_id in not_deletable_rule_group:
            file.write(str(rule_group_id) + '\n')


if __name__ == "__main__":
    main()