I have such python code, to notice by TG when do my docker containers fall:
import subprocess
import requests
import time
TELEGRAM_BOT_TOKEN = '##############'
CHAT_ID = '##################'
def get_containers():
cmd = "docker ps -a --format '{{.Names}}'"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
return result.stdout
def get_containers_status(container_name):
cmd = "docker inspect --format '{{.State.Health.Status}}' "+container_name+" 2>/dev/null"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
return result.stdout
def send_telegram_message(message):
url = f'https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage'
data = {'chat_id': CHAT_ID, 'text': message}
requests.post(url, data=data)
def log_to_journald(message):
cmd = f"echo '{message}' | systemd-cat -t docker-monitor"
subprocess.run(cmd, shell=True)
def check_arrays_for_ident(arr1,arr2):
if len(arr1) == len(arr2):
for value in arr1:
if value not in arr2:
return False
return True
else:
return False
if __name__ == "__main__":
print("Start")
old_broke_containers = []
container_dict = {}
while True:
containers_status = get_containers()
container_list = containers_status.split()
broke_containers = []
for container in container_list:
status = get_containers_status(container)
container_dict[container] = status
if status=="unhealthy\n":
broke_containers.append(container)
message = "empty"
if not check_arrays_for_ident(broke_containers, old_broke_containers):
if 'unhealthy\n' in container_dict.values():
message = "———————————————————\n"
message+="⚠️⚠️⚠️Недоступные контейнеры:\n\n"
message+= "".join(map(lambda item: f'❌ {item[0]}\n', filter(lambda item: item[1] == "unhealthy\n", container_dict.items())))
message+="———————————————————\n\n"
message+="Доступные контейнеры:\n\n"
message+= "".join(map(lambda item: f'✅ {item[0]}\n', filter(lambda item: item[1] == "healthy\n", container_dict.items())))
print(message)
send_telegram_message(message)
container_dict.clear()
old_broke_containers = broke_containers[:]
time.sleep(3)
if time.time() % 60 == 0:
log_to_journald("Status update")
It needs to run all the time and load after reboot of server, so I have decided to make a system service out of it. Here is my .service file in /etc/systemd/system
[Unit]
Description=tg-bot
After=multi-user.target
[Service]
User=root
Group=root
Type=simple
Restart=always
ExecStart=/usr/bin/python3 /custom_scripts/docker_checker.py
[Install]
WantedBy=multi-user.target
Then I did sudo systemctl daemon-reload, sudo systemctl enable bot.service and sudo systemctl start bot.service. In sudo systemctl status bot.service it was written, that service works well, but it did nothing, I didn't get notifications. If I run script in console, everything works well