-
Notifications
You must be signed in to change notification settings - Fork 0
/
healthcheck.py
51 lines (38 loc) · 1.37 KB
/
healthcheck.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
import asyncio
import logging
import os
import requests
import telegram
from retrying import retry
telegram_token = os.environ["TELEGRAM_TOKEN"]
telegram_chat_id = os.environ["TELEGRAM_DESTINATION_CHAT_ID"]
endpoint_url = os.environ["URL_TO_CHECK"]
def get_logger(log_level=logging.INFO):
log = logging.getLogger()
log.setLevel(log_level)
formatter = logging.Formatter(
fmt="[%(asctime)s][%(lineno)d][%(pathname)s][%(levelname)s]: %(message)s",
datefmt="%m/%d/%Y,%H:%M:%S",
)
handler = logging.StreamHandler()
handler.setLevel(log_level)
handler.setFormatter(formatter)
log.addHandler(handler)
return log
logger = get_logger()
async def send_telegram_message(message):
bot = telegram.Bot(token=telegram_token)
await bot.send_message(chat_id=telegram_chat_id, text=message)
@retry(stop_max_attempt_number=4, wait_fixed=5 * 1000)
def connect():
response = requests.get(endpoint_url, timeout=30)
response.raise_for_status()
async def check_remote_server():
try:
connect()
logger.info(f"Connection with {endpoint_url} successfully established")
except requests.exceptions.RequestException:
await send_telegram_message(f"Unable to connect to {endpoint_url}")
logger.warning("Server is not available. Telegram message sent.")
if __name__ == "__main__":
asyncio.run(check_remote_server())