dir/mikrotik-monitor
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: читаем IP из address-list MikroTik вместо IP_LIST

Browse Source
This commit is contained in:
Sergey
2026-05-16 14:55:16 +07:00
parent 6e3c7afab8
commit 994b95f3e9
4 changed files with 79 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+2 -2
View File
@@ -13,8 +13,8 @@ MIKROTIK_PASSWORD=your_password
TELEGRAM_BOT_TOKEN=123456789:AABBccDDeeFFggHH
TELEGRAM_CHAT_ID=-1001234567890
# ─── Список адресов (через запятую) ──────────────────────────────────────────
IP_LIST=192.168.1.1,192.168.1.10,192.168.1.20
# Имя address-list из /ip firewall address-list на MikroTik
ADDRESS_LIST_NAME=monitor
# ─── Параметры мониторинга ────────────────────────────────────────────────────
CHECK_INTERVAL=60
Dockerfile
+3 -157
View File
@@ -8,166 +8,12 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
tzdata \
&& rm -rf /var/lib/apt/lists/*
# Зависимости
RUN pip install --no-cache-dir \
routeros-api==0.17.0 \
httpx==0.27.0 \
python-dotenv==1.0.1
httpx==0.27.0
# Скрипт встроен прямо в Dockerfile — не нужны дополнительные файлы
RUN cat > /app/monitor.py << 'PYEOF'
#!/usr/bin/env python3
"""MikroTik IP Monitor — пингует адреса через MikroTik API и шлёт уведомления в Telegram."""
import os
import time
import logging
import asyncio
from datetime import datetime
from typing import Optional
import routeros_api
import httpx
# ── Настройки ──────────────────────────────────────────────────────────────
MIKROTIK_HOST = os.environ["MIKROTIK_HOST"]
MIKROTIK_PORT = int(os.getenv("MIKROTIK_PORT", "8728"))
MIKROTIK_USER = os.environ["MIKROTIK_USER"]
MIKROTIK_PASSWORD = os.environ["MIKROTIK_PASSWORD"]
TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
TELEGRAM_CHAT_ID = os.environ["TELEGRAM_CHAT_ID"]
IP_LIST = [ip.strip() for ip in os.environ["IP_LIST"].split(",") if ip.strip()]
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60"))
PING_COUNT = int(os.getenv("PING_COUNT", "3"))
PING_TIMEOUT_MS = int(os.getenv("PING_TIMEOUT_MS", "1000"))
FAILURE_THRESHOLD = int(os.getenv("FAILURE_THRESHOLD", "2"))
# ── Логирование ────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger(__name__)
# ── Telegram ───────────────────────────────────────────────────────────────
async def send_telegram(message: str) -> None:
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message, "parse_mode": "HTML"}
async with httpx.AsyncClient(timeout=10) as client:
try:
resp = await client.post(url, json=payload)
resp.raise_for_status()
log.info("Telegram: сообщение отправлено")
except Exception as exc:
log.error("Telegram: ошибка отправки — %s", exc)
def notify(message: str) -> None:
asyncio.run(send_telegram(message))
# ── MikroTik ping ──────────────────────────────────────────────────────────
def mikrotik_ping(connection, address: str) -> Optional[float]:
try:
api = connection.get_api()
ping_resource = api.get_resource("/ping")
results = ping_resource.call(
"",
{"address": address, "count": str(PING_COUNT), "timeout": str(PING_TIMEOUT_MS)},
)
for row in results:
if "avg-rtt" in row and row.get("avg-rtt"):
try:
return float(row["avg-rtt"].replace("ms", "").strip())
except ValueError:
pass
if row.get("packet-loss") == "100":
return None
return None
except Exception as exc:
log.warning("Ошибка пинга %s: %s", address, exc)
return None
# ── Основной цикл ─────────────────────────────────────────────────────────
def main() -> None:
log.info("=== MikroTik Monitor запущен ===")
log.info("Хосты: %s", IP_LIST)
log.info("Интервал: %d сек", CHECK_INTERVAL)
failure_count: dict[str, int] = {ip: 0 for ip in IP_LIST}
host_down: dict[str, bool] = {ip: False for ip in IP_LIST}
notify(
f"🟢 <b>MikroTik Monitor запущен</b>\n"
f"Хосты: <code>{', '.join(IP_LIST)}</code>\n"
f"Интервал: {CHECK_INTERVAL} сек"
)
while True:
connection = None
try:
log.info("Подключение к MikroTik %s:%d …", MIKROTIK_HOST, MIKROTIK_PORT)
connection = routeros_api.RouterOsApiPool(
MIKROTIK_HOST,
username=MIKROTIK_USER,
password=MIKROTIK_PASSWORD,
port=MIKROTIK_PORT,
plaintext_login=True,
)
connection.get_api()
log.info("Подключено к MikroTik")
for ip in IP_LIST:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
rtt = mikrotik_ping(connection, ip)
if rtt is not None:
log.info("✅ %s avg-rtt=%.1f ms", ip, rtt)
if host_down[ip]:
host_down[ip] = False
failure_count[ip] = 0
notify(
f"✅ <b>Хост восстановился</b>\n"
f"IP: <code>{ip}</code>\n"
f"RTT: {rtt:.1f} ms\n"
f"Время: {now}"
)
else:
failure_count[ip] = 0
else:
failure_count[ip] += 1
log.warning("❌ %s не отвечает (%d/%d)", ip, failure_count[ip], FAILURE_THRESHOLD)
if failure_count[ip] >= FAILURE_THRESHOLD and not host_down[ip]:
host_down[ip] = True
notify(
f"🔴 <b>Хост недоступен!</b>\n"
f"IP: <code>{ip}</code>\n"
f"Не отвечает {failure_count[ip]} проверки подряд\n"
f"Время: {now}"
)
except Exception as exc:
log.error("Ошибка подключения к MikroTik: %s", exc)
notify(f"⚠️ <b>Ошибка подключения к MikroTik</b>\n<code>{exc}</code>")
finally:
if connection:
try:
connection.disconnect()
except Exception:
pass
log.info("Следующая проверка через %d сек …\n", CHECK_INTERVAL)
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
main()
PYEOF
# Копируем скрипт из репозитория
COPY monitor.py .
ENV TZ=Europe/Moscow
CMD ["python", "-u", "/app/monitor.py"]
monitor.py
+72 -46
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
"""
MikroTik IP Monitor — пингует адреса через MikroTik API и шлёт уведомления в Telegram.
MikroTik IP Monitor — читает адреса из /ip firewall address-list,
пингует их через MikroTik API и шлёт уведомления в Telegram.
"""
import os
@@ -12,9 +13,6 @@ from typing import Optional
import routeros_api
import httpx
from dotenv import load_dotenv
load_dotenv()
# ──────────────────────────────────────────────
# Настройки из переменных окружения
@@ -27,14 +25,12 @@ MIKROTIK_PASSWORD = os.environ["MIKROTIK_PASSWORD"]
TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
TELEGRAM_CHAT_ID = os.environ["TELEGRAM_CHAT_ID"]
# Список IP-адресов для мониторинга (через запятую в .env или задать напрямую)
IP_LIST = [ip.strip() for ip in os.environ["IP_LIST"].split(",") if ip.strip()]
# Имя address-list в MikroTik (/ip firewall address-list)
ADDRESS_LIST_NAME = os.environ["ADDRESS_LIST_NAME"]
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60")) # секунды между проверками
PING_COUNT = int(os.getenv("PING_COUNT", "3")) # количество пингов
PING_TIMEOUT_MS = int(os.getenv("PING_TIMEOUT_MS", "1000")) # таймаут пинга (мс)
# Сколько раз подряд хост должен не ответить, чтобы считаться DOWN
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60"))
PING_COUNT = int(os.getenv("PING_COUNT", "3"))
PING_TIMEOUT_MS = int(os.getenv("PING_TIMEOUT_MS", "1000"))
FAILURE_THRESHOLD = int(os.getenv("FAILURE_THRESHOLD", "2"))
# ──────────────────────────────────────────────
@@ -53,11 +49,7 @@ log = logging.getLogger(__name__)
# ──────────────────────────────────────────────
async def send_telegram(message: str) -> None:
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "HTML",
}
payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message, "parse_mode": "HTML"}
async with httpx.AsyncClient(timeout=10) as client:
try:
resp = await client.post(url, json=payload)
@@ -71,15 +63,41 @@ def notify(message: str) -> None:
asyncio.run(send_telegram(message))
# ──────────────────────────────────────────────
# MikroTik: получить список IP из address-list
# ──────────────────────────────────────────────
def fetch_address_list(api) -> list[str]:
"""
Читает /ip firewall address-list и возвращает список IP
из записей с именем ADDRESS_LIST_NAME (пропускает отключённые).
"""
resource = api.get_resource("/ip/firewall/address-list")
entries = resource.get(list=ADDRESS_LIST_NAME)
ip_list = []
for entry in entries:
# Пропускаем отключённые записи
if entry.get("disabled", "false") == "true":
log.debug("Пропущена (disabled): %s", entry.get("address"))
continue
address = entry.get("address", "").strip()
# Убираем префикс /32 если есть — пингуем чистый IP
if "/" in address:
address = address.split("/")[0]
if address:
ip_list.append(address)
return ip_list
# ──────────────────────────────────────────────
# MikroTik: пинг через API
# ──────────────────────────────────────────────
def mikrotik_ping(connection, address: str) -> Optional[float]:
def mikrotik_ping(api, address: str) -> Optional[float]:
"""
Запускает ping на MikroTik. Возвращает средний avg-rtt (мс) или None если недоступен.
Возвращает avg-rtt (мс) или None если хост недоступен.
"""
try:
api = connection.get_api()
ping_resource = api.get_resource("/ping")
results = ping_resource.call(
"",
@@ -89,18 +107,14 @@ def mikrotik_ping(connection, address: str) -> Optional[float]:
"timeout": str(PING_TIMEOUT_MS),
},
)
# Ищем строку с итогами (packet-loss и avg-rtt)
for row in results:
if "avg-rtt" in row and row.get("avg-rtt"):
avg_rtt_str = row["avg-rtt"].replace("ms", "").strip()
try:
return float(avg_rtt_str)
return float(row["avg-rtt"].replace("ms", "").strip())
except ValueError:
pass
if row.get("packet-loss") == "100":
return None
return None
except Exception as exc:
log.warning("Ошибка пинга %s: %s", address, exc)
@@ -112,41 +126,56 @@ def mikrotik_ping(connection, address: str) -> Optional[float]:
# ──────────────────────────────────────────────
def main() -> None:
log.info("=== MikroTik Monitor запущен ===")
log.info("Хосты для проверки: %s", IP_LIST)
log.info("Интервал проверки: %d сек", CHECK_INTERVAL)
log.info("Address-list: '%s'", ADDRESS_LIST_NAME)
log.info("Интервал: %d сек", CHECK_INTERVAL)
# Состояние: счётчик последовательных ошибок и флаг «уже DOWN»
failure_count: dict[str, int] = {ip: 0 for ip in IP_LIST}
host_down: dict[str, bool] = {ip: False for ip in IP_LIST}
# Состояние хостов: счётчик ошибок и флаг DOWN
# Обновляется динамически при каждом чтении address-list
failure_count: dict[str, int] = {}
host_down: dict[str, bool] = {}
notify(
f"🟢 <b>MikroTik Monitor запущен</b>\n"
f"Хосты: <code>{', '.join(IP_LIST)}</code>\n"
f"Address-list: <code>{ADDRESS_LIST_NAME}</code>\n"
f"Интервал: {CHECK_INTERVAL} сек"
)
while True:
connection = None
pool = None
try:
log.info("Подключение к MikroTik %s:%d …", MIKROTIK_HOST, MIKROTIK_PORT)
connection = routeros_api.RouterOsApiPool(
pool = routeros_api.RouterOsApiPool(
MIKROTIK_HOST,
username=MIKROTIK_USER,
password=MIKROTIK_PASSWORD,
port=MIKROTIK_PORT,
plaintext_login=True,
)
connection.get_api() # проверяем соединение
api = pool.get_api()
log.info("Подключено к MikroTik")
for ip in IP_LIST:
# Читаем актуальный список адресов с MikroTik
ip_list = fetch_address_list(api)
if not ip_list:
log.warning("Address-list '%s' пуст или не найден!", ADDRESS_LIST_NAME)
else:
log.info("Адресов для проверки: %d → %s", len(ip_list), ip_list)
# Добавляем новые хосты в state (без сброса существующих)
for ip in ip_list:
if ip not in failure_count:
failure_count[ip] = 0
host_down[ip] = False
# Пингуем каждый адрес
for ip in ip_list:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
rtt = mikrotik_ping(connection, ip)
rtt = mikrotik_ping(api, ip)
if rtt is not None:
log.info("✅ %s avg-rtt=%.1f ms", ip, rtt)
log.info("✅ %-18s avg-rtt=%.1f ms", ip, rtt)
if host_down[ip]:
# Хост восстановился
host_down[ip] = False
failure_count[ip] = 0
notify(
@@ -160,29 +189,26 @@ def main() -> None:
else:
failure_count[ip] += 1
log.warning(
"❌ %s не отвечает (попытка %d/%d)",
"❌ %-18s не отвечает (%d/%d)",
ip, failure_count[ip], FAILURE_THRESHOLD,
)
if failure_count[ip] >= FAILURE_THRESHOLD and not host_down[ip]:
host_down[ip] = True
notify(
f"🔴 <b>Хост недоступен!</b>\n"
f"IP: <code>{ip}</code>\n"
f"Список: <code>{ADDRESS_LIST_NAME}</code>\n"
f"Не отвечает {failure_count[ip]} проверки подряд\n"
f"Время: {now}"
)
except Exception as exc:
log.error("Ошибка подключения к MikroTik: %s", exc)
notify(
f"⚠️ <b>Ошибка подключения к MikroTik</b>\n"
f"<code>{exc}</code>"
)
log.error("Ошибка: %s", exc)
notify(f"⚠️ <b>Ошибка подключения к MikroTik</b>\n<code>{exc}</code>")
finally:
if connection:
if pool:
try:
connection.disconnect()
pool.disconnect()
except Exception:
pass
stack.yml
+2 -2
View File
@@ -33,8 +33,8 @@ services:
- TELEGRAM_BOT_TOKEN=123456:AABBcc # ← токен от @BotFather
- TELEGRAM_CHAT_ID=-1001234567890 # ← ID чата/группы
# Список IP через запятую
- IP_LIST=192.168.1.1,192.168.1.10,192.168.1.20
# Имя address-list из /ip firewall address-list на MikroTik
- ADDRESS_LIST_NAME=monitor
# Параметры мониторинга
- CHECK_INTERVAL=60 # интервал между проверками (сек)