Запускаем на сервере скрипт Мониторинга Docker с отправкой уведомлений в Телеграм

Больше скриптов и обсуждение в телеграм https://t.me/cryptuworker

Подписывайтесь!

1. Проверка установленной версии Python

Если где то в процессе будет спрашивать согласия, просто нажимаем 'y' и 'enter'.

Для начала, проверяем, какая версия Python установлена сервере:

python3 --version

Если версия ниже 3.11 (последняя), установливаем Python 3.11:

sudo apt install python3.11

2. Установка pip

Для установки pip выполнияем следующую команду:

apt install python3-pip

3. Установка необходимых библиотек

Установливаем необходимые библиотеки с помощью следующей команды:

python3 -m pip install python-dotenv pyTelegramBotAPI

4. Создание файла с переменными окружения

Создаем файл `.env`, чтобы хранить переменные окружения:

nano .env

В открывшееся окно всставляем bot_token и chat_id:

TELEGRAM_BOT_TOKEN=your_telegram_bot_token
TELEGRAM_CHAT_ID=your_telegram_chat_id

Далее сохранием файл, нажав по очереди: `Ctrl + O`, `Enter`, `Ctrl + X`.

5. Создание директории для скрипта

Создаем директорию для скрипта:

mkdir docker_monitor

6. Скопируйте скрипт в нужную директорию

Скопируем скрипт в директорию, например:

nano docker_monitor/docker_monitor.py

В открывшееся окно вставляем скрипт:

import os
import time
import subprocess
from datetime import datetime, timedelta
import telebot
import re
from dotenv import load_dotenv
load_dotenv()
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')
CHECK_INTERVAL = 4 * 60 * 60
LOG_COMMAND = "docker logs --tail 1 {container_name}"
GET_CONTAINERS_COMMAND = "docker ps --format '{{.Names}}'"
bot = telebot.TeleBot(TELEGRAM_BOT_TOKEN)
def send_telegram_message(message):
    for i in range(0, len(message), 3990):
        bot.send_message(TELEGRAM_CHAT_ID, message[i:i + 3990])
        time.sleep(1)
def remove_ansi_escape_sequences(text):
    ansi_escape = re.compile(r'(?:\x1B[@-_][0-?]*[ -/]*[@-~])')
    return ansi_escape.sub('', text)
def check_node(container_name):
    try:
        result = subprocess.run(LOG_COMMAND.format(container_name=container_name), shell=True, capture_output=True, text=True)
        last_log_line = result.stdout.strip()
        last_log_line = remove_ansi_escape_sequences(last_log_line)
        log_time_str = last_log_line.split(' ')[0].strip('[]')
        log_time = datetime.strptime(log_time_str, "%Y-%m-%dT%H:%M:%S.%fZ")
        current_time = datetime.utcnow()
        if current_time - log_time > timedelta(hours=4):
            return False, last_log_line[:50]
        if "Pinging master node" in last_log_line and "successful" in last_log_line:
            return True, None
        else:
            return False, last_log_line[:50]
    except Exception as e:
        print(f"Ошибка при проверке контейнера {container_name}: {e}")
        return False, str(e)[:50]
def get_container_names():
    try:
        result = subprocess.run(GET_CONTAINERS_COMMAND, shell=True, capture_output=True, text=True)
        container_names = result.stdout.strip().splitlines()
        return container_names
    except Exception as e:
        print(f"Ошибка при получении списка контейнеров: {e}")
        return []
def monitor_nodes():
    while True:
        container_names = get_container_names()
        if not container_names:
            send_telegram_message("Не удалось получить список контейнеров.")
            return
        success_count = 0
        failure_reports = []
        for container in container_names:
            is_successful, error_message = check_node(container)
            if is_successful:
                success_count += 1
            else:
                failure_reports.append(f"Контейнер {container} неуспешен: {error_message}")
        report = f"Успешных нод: {success_count}/{len(container_names)}\n"
        if failure_reports:
            report += "\n".join(failure_reports)
        send_telegram_message(report)
        time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
    monitor_nodes()

Сохраняем файл, нажав по очереди: `Ctrl + O`, `Enter`, `Ctrl + X`.

7. Даем скрипту права на исполнение

Выдаем скрипту права на исполнение:

chmod +x docker_monitor/docker_monitor.py

8. Настройка автозапуска

Создаем файл сервиса для автозапуска:

sudo nano /etc/systemd/system/docker_monitor.service

В открывшееся окно вставьте следующее содержимое:

[Unit]
Description=Monitoring Docker Nodes
After=network.target docker.service
Requires=docker.service

[Service]
WorkingDirectory=/root/docker_monitor
ExecStart=/usr/bin/python3 /root/docker_monitor/docker_monitor.py
Restart=always
User=root

[Install]
WantedBy=multi-user.target

Далее сохраняем файл, нажав по очереди: `Ctrl + O`, `Enter`, `Ctrl + X`.

9. Запускаем и включаем сервис

Запускаем и включаем сервис:

sudo systemctl enable docker_monitor.service
sudo systemctl start docker_monitor.service

Теперь скрипт включен как сервис и будет стартовать при перезагрузке сервера автоматически.

10. Запуск вручную для проверки

Запускаем скрипт вручную для проверки:

python3 docker_monitor/docker_monitor.py

Если все запустилось, то в телеграм придет сообщение со статусом контейнеров.