-
Notifications
You must be signed in to change notification settings - Fork 2
/
telegram_bot.py
189 lines (161 loc) · 7.32 KB
/
telegram_bot.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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
import requests
# Global dictionary to store the mapping of "firing" alerts to their message IDs
alert_message_ids = {}
prometheus_alert_message_ids = {}
team_dict = {"hotel": "", "devops": "", "ticket": "", "shared": ""}
bot_token = ""
chat_id_devops = ""
def send_telegram_message(bot_token, chat_id, message, reply_to_message_id=None):
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
params = {
"chat_id": chat_id,
"text": message,
"reply_to_message_id": reply_to_message_id
}
response = requests.post(url, params=params)
if response.status_code != 200:
print(f"Failed to send Telegram message. Error: {response.text}")
else:
print("Telegram message sent successfully!")
return response
def send_telegram_message_devops(bot_token, chat_id_devops, message, reply_to_message_id=None):
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
params = {
"chat_id": chat_id_devops,
"text": message,
"reply_to_message_id": reply_to_message_id
}
response = requests.post(url, params=params)
if response.status_code != 200:
print(f"Failed to send Telegram message. Error: {response.text}")
else:
print("Telegram message sent successfully!")
return response
def handle_telegram_alert(alert):
# Check if the alert is from Prometheus or Zabbix
if 'EVENT.STATUS' in alert:
# Zabbix alert format
send_zabbix_alert_to_telegram(alert)
elif 'alerts' in alert:
handle_alert(alert)
else:
print("Unknown alert format.")
def create_message(alert):
status = alert.get('status', '')
alertname = alert['labels'].get('alertname', '')
instance = alert['labels'].get('instance', 'N/A')
job = alert['labels'].get('job', 'N/A')
summary = alert['annotations'].get('summary', '')
description = alert['annotations'].get('description', '')
# Customize the alert message as per your requirement
status_emoji = "🛑" if status == 'firing' else "✅"
message = f"Prometheus Alert:\n\nStatus: {status_emoji} {status.capitalize()}\nAlert Name: {alertname}\nHost: {instance}\nJob: {job}\nSummary: {summary}\n\nDescription: {description}"
print("Sending message to Telegram...")
print(f"Message: {message}")
if status == 'firing':
return message
elif status == 'resolved':
resolved_message = f"Resolved Alert:\n\nStatus: {status_emoji} {status.capitalize()}\nAlert Name: {alertname}\nHost: {instance}\nJob: {job}\nSummary: {summary}\n\nDescription: {description}"
return resolved_message
return None
def handle_alert(alert):
global prometheus_alert_message_ids, chat_id_devops
alerts = alert.get('alerts', [])
for alert in alerts:
team = alert['labels'].get('teamname')
chat_id = team_dict[team]
status = alert.get('status', '')
alertname = alert['labels'].get('alertname', '')
message = create_message(alert)
if not message:
print(f"Unknown status '{status}' received for alert: {alertname}")
return
messages = {}
if status == 'resolved':
messages = prometheus_alert_message_ids.get(alertname, None)
if messages:
prometheus_alert_message_ids.pop(alertname)
else:
print(f"Failed to find corresponding firing message IDs for resolved alert: {alertname}")
return
response = send_telegram_message(bot_token, chat_id_devops, message, reply_to_message_id=messages.get("devops", None))
if response.status_code == 200 and status == 'firing':
prometheus_alert_message_ids[alertname] = {"devops": response.json().get('result', {}).get('message_id', '')}
# log response
if chat_id != chat_id_devops:
response = send_telegram_message(bot_token, chat_id, message, reply_to_message_id=messages.get(team, None))
if response.status_code == 200 and status == 'firing':
prometheus_alert_message_ids[alertname][team] = response.json().get('result', {}).get('message_id', '')
def send_zabbix_alert_to_telegram(alert):
# Extract Zabbix alert data from the 'alert' dictionary
status = alert.get('EVENT.STATUS', '')
event_message = alert.get('EVENT.MESSAGE', '')
event_id = alert.get('EVENT.ID', '')
print(event_id)
print(event_message)
# Customize the alert message as per your requirement
status_emoji = "🛑" if status == 'firing' else "✅"
message = f"Zabbix Alert:\n\nStatus: {status_emoji} {status.capitalize()}\n{event_message}"
chat_id = chat_id_devops
if status == 'firing':
response = send_telegram_message(bot_token, chat_id, message)
if response.status_code == 200:
# Store the message ID in the dictionary
alert_message_ids[event_id] = response.json().get('result', {}).get('message_id', '')
print(f"alert message id event_message is : {alert_message_ids[event_id]}")
print(f"response json is: {response.json().get('result', {})}")
else:
print(f"Failed to send Telegram message for firing alert: {event_id}")
elif status == 'resolved':
# Check if the corresponding "firing" alert's message ID exists
firing_message_id = alert_message_ids.get(event_id)
event = (f"event message is {event_message}")
print(f"firing message id is:{firing_message_id}")
print(alert_message_ids)
if firing_message_id:
# Create the "resolved" message and send it as a reply to the "firing" message
resolved_message = f"Zabbix Alert:\n\nStatus: {status_emoji} {status.capitalize()}\n{event_message}"
print("Sending resolved message to Telegram...")
print(f"Message: {resolved_message}")
response = send_telegram_message(bot_token, chat_id, resolved_message, reply_to_message_id=firing_message_id)
if response.status_code == 200:
print("Reply message sent successfully!")
alert_message_ids.pop(event_id)
else:
print(f"Failed to send reply message for resolved alert: {event_message}")
else:
print(f"Failed to find corresponding firing message ID for resolved alert: {event_message}")
else:
print(f"Unknown status '{status}' received for alert: {event_message}")
# Example Prometheus Alert
example_prometheus_alert = {
'alerts': [
{
'status': 'firing',
'labels': {
'teamname': 'hotel',
'alertname': 'HighErrorRate',
'instance': 'server1.example.com',
'job': 'web-server',
},
'annotations': {
'summary': 'High error rate detected!',
'description': 'The error rate is above the threshold.',
},
}
]
}
handle_alert(example_prometheus_alert)
# Example Zabbix Alert
example_zabbix_alert = {
'EVENT.STATUS': 'firing',
'EVENT.DATE': '2023-07-29',
'EVENT.TIME': '12:34:56',
'EVENT.NAME': 'High CPU Usage',
'HOST.NAME': 'Server1',
'EVENT.SEVERITY': 'High',
'EVENT.OPDATA': 'Some operational data here...',
'EVENT.ID': '12345',
'TRIGGER.URL': 'https://your_zabbix_instance/triggers/12345',
}
handle_telegram_alert(example_zabbix_alert)