1.通过Alertmanager的api mute全部告警

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Usage: python mute_alerts.py

import json
import requests
import time

root_url = "http://alerts-engine:9093"

comment = "mute_alerts-" + time.strftime("%Y%m%d%H%M%S", time.localtime())

silence = {
    "matchers": [
        {
            "name": "alertname",
            "value": ".+",
            "isRegex": True
        }
    ],
    "endsAt": "2100-01-01T00:00:00Z",
    "createdBy": "test",
    "comment": comment
}

retries = 3
timeout = 3
delay = 3

for i in range(retries):
    request = requests.post(root_url + "/api/v1/silences",
                            data=json.dumps(silence), timeout=timeout).json()
    if request["status"] != "success":
        if i < retries - 1:
            time.sleep(delay)
            continue
        print(json.dumps(request))
        exit(1)
    break

print("OK")

2. 通过Alertmanager的API unmute所有告警

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Usage: python unmute_alerts.py

import json
import requests
import time
import threading

root_url = "http://alerts-engine:9093"

unmute_success = 0
retries = 3
timeout = 3
delay = 3

def unmute_alerts():
    silences = requests.get(root_url + "/api/v1/silences", timeout=3).json()

    for silence in [x for x in silences["data"] if
                    x["status"]["state"] == "active"]:
        for matcher in silence["matchers"]:
            if matcher["name"] == "alertname" and matcher["value"] == ".+" and \
                    matcher["isRegex"] is True:
                for i in range(retries):
                    request = requests.delete(
                        root_url + "/api/v1/silence/" + silence["id"],
                        timeout=timeout).json()
                    if request["status"] != "success":
                        if i < retries - 1:
                            time.sleep(delay)
                            continue
                        print(json.dumps(request))
                        exit(1)
                    break
                continue

    print("OK")
    global unmute_success
    unmute_success = 1

if __name__ == "__main__":
    timer = threading.Timer(300, unmute_alerts)
    timer.start()

    while True:
        if unmute_success == 1:
            exit(0)
        alert_list=requests.get(root_url + "/api/v1/alerts", timeout=3).json()
        if not alert_list['data']:
            unmute_alerts()
            timer.cancel()
            exit(0)
        else:
            time.sleep(10)

 

Logo

CSDN联合极客时间,共同打造面向开发者的精品内容学习社区,助力成长!

更多推荐