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

feat: initial MikroTik monitor setup

Browse Source
This commit is contained in:
Sergey
2026-05-16 13:27:27 +07:00
commit 6e3c7afab8
7 changed files with 695 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+26
View File
@@ -0,0 +1,26 @@
# Скопируйте этот файл в .env и заполните значения
# cp .env.example .env
#
# ⚠️ Файл .env добавлен в .gitignore — не коммитьте его!
# ─── MikroTik ─────────────────────────────────────────────────────────────────
MIKROTIK_HOST=192.168.88.1
MIKROTIK_PORT=8728
MIKROTIK_USER=monitor
MIKROTIK_PASSWORD=your_password
# ─── Telegram ─────────────────────────────────────────────────────────────────
TELEGRAM_BOT_TOKEN=123456789:AABBccDDeeFFggHH
TELEGRAM_CHAT_ID=-1001234567890
# ─── Список адресов (через запятую) ──────────────────────────────────────────
IP_LIST=192.168.1.1,192.168.1.10,192.168.1.20
# ─── Параметры мониторинга ────────────────────────────────────────────────────
CHECK_INTERVAL=60
PING_COUNT=3
PING_TIMEOUT_MS=1000
FAILURE_THRESHOLD=2
# ─── Временная зона ───────────────────────────────────────────────────────────
TZ=Europe/Moscow
.gitignore
+22
View File
@@ -0,0 +1,22 @@
# Секреты — никогда не коммитить!
.env
# Python
__pycache__/
*.pyc
*.pyo
*.pyd
.Python
*.egg-info/
dist/
build/
# IDE
.idea/
.vscode/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
Dockerfile
+173
View File
@@ -0,0 +1,173 @@
FROM python:3.12-slim
LABEL description="MikroTik IP Monitor with Telegram alerts"
WORKDIR /app
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
# Скрипт встроен прямо в 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
ENV TZ=Europe/Moscow
CMD ["python", "-u", "/app/monitor.py"]
PUSH.md
+77
View File
@@ -0,0 +1,77 @@
# Публикация в Gitea — пошаговая инструкция
## 1. Создайте репозиторий в Gitea
1. Откройте Gitea → **+** → **New Repository**
2. Имя: `mikrotik-monitor`
3. Видимость: Private (рекомендуется)
4. **Не инициализировать** репозиторий (без README, .gitignore)
5. Нажмите **Create Repository**
---
## 2. Инициализируйте и запушьте
Скопируйте папку репозитория на свой компьютер, затем:
```bash
cd mikrotik-monitor-repo
# Инициализация
git init
git checkout -b main
# Первый коммит
git add .
git commit -m "feat: initial MikroTik monitor setup"
# Подключение к Gitea (замените URL на свой)
git remote add origin https://gitea.example.com/yourname/mikrotik-monitor.git
# Пуш
git push -u origin main
```
---
## 3. Настройте SSH (опционально, но удобнее)
```bash
# Сгенерируйте ключ (если ещё нет)
ssh-keygen -t ed25519 -C "your@email.com"
# Скопируйте публичный ключ
cat ~/.ssh/id_ed25519.pub
```
Добавьте ключ в Gitea: **Settings → SSH / GPG Keys → Add Key**
Затем используйте SSH-URL:
```bash
git remote set-url origin git@gitea.example.com:yourname/mikrotik-monitor.git
```
---
## 4. Дальнейшая работа
```bash
# Внести изменения и запушить
git add .
git commit -m "fix: описание изменения"
git push
# Посмотреть статус
git status
git log --oneline
```
---
## ⚠️ Важно: секреты в git
Файл `.env` добавлен в `.gitignore` и **никогда не попадёт в репозиторий**.
Для хранения секретов (пароли, токены) используйте:
- Переменные окружения в Portainer (раздел **Environment variables** при создании стека)
- [Gitea Secrets](https://docs.gitea.com/usage/secrets) — если настроите CI/CD
README.md
+132
View File
@@ -0,0 +1,132 @@
# 🔍 MikroTik Monitor
Мониторинг IP-адресов через MikroTik API с уведомлениями в Telegram.
Деплой через Portainer Stack — без лишних зависимостей.
---
## Как это работает
```
[Docker Container]
│
├─→ Подключается к MikroTik по API (порт 8728)
│ └─→ Выполняет /ping для каждого IP из списка
│
└─→ Если хост не отвечает N раз подряд → Telegram уведомление
```
**Логика уведомлений:**
- 🔴 `DOWN` — хост не ответил `FAILURE_THRESHOLD` раз подряд
- ✅ `UP` — хост восстановился после даунтайма
- ⚠️ `ERROR` — не удалось подключиться к MikroTik
---
## Структура репозитория
```
.
├── Dockerfile # Образ с встроенным скриптом
├── monitor.py # Основной скрипт мониторинга
├── stack.yml # Portainer Stack
├── .env.example # Шаблон переменных окружения
└── README.md
```
---
## Быстрый старт
### 1. Настройте MikroTik
```routeros
/ip service enable api
/user group add name=monitor policy=api,read,test
/user add name=monitor password=StrongPass group=monitor
```
### 2. Создайте Telegram-бота
1. [@BotFather](https://t.me/BotFather) → `/newbot` → скопируйте токен
2. Добавьте бота в чат, напишите любое сообщение
3. Получите `chat_id`:
```
https://api.telegram.org/bot<TOKEN>/getUpdates
```
### 3. Соберите Docker-образ на сервере
```bash
git clone https://gitea.example.com/yourname/mikrotik-monitor.git
cd mikrotik-monitor
docker build -t mikrotik-monitor:latest .
```
### 4. Задеплойте через Portainer
1. **Stacks** → **Add Stack**
2. Имя: `mikrotik-monitor`
3. **Web editor** → вставьте содержимое `stack.yml`
4. Заполните переменные окружения
5. **Deploy the stack**
---
## Переменные окружения
| Переменная | Обязательная | По умолчанию | Описание |
|---|:---:|---|---|
| `MIKROTIK_HOST` | ✅ | — | IP или hostname MikroTik |
| `MIKROTIK_PORT` | | `8728` | API-порт (8729 для SSL) |
| `MIKROTIK_USER` | ✅ | — | Логин |
| `MIKROTIK_PASSWORD` | ✅ | — | Пароль |
| `TELEGRAM_BOT_TOKEN` | ✅ | — | Токен бота от @BotFather |
| `TELEGRAM_CHAT_ID` | ✅ | — | ID чата/группы/канала |
| `IP_LIST` | ✅ | — | Адреса через запятую |
| `CHECK_INTERVAL` | | `60` | Интервал проверок (сек) |
| `PING_COUNT` | | `3` | Кол-во пинг-пакетов |
| `PING_TIMEOUT_MS` | | `1000` | Таймаут пинга (мс) |
| `FAILURE_THRESHOLD` | | `2` | Провалов подряд → DOWN |
| `TZ` | | `Europe/Moscow` | Временная зона |
### Рекомендуемый способ: переменные Portainer
В `stack.yml` используйте синтаксис `${VAR}`:
```yaml
environment:
- MIKROTIK_PASSWORD=${MIKROTIK_PASSWORD}
- TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
```
Значения добавляйте в раздел **Environment variables** в UI Portainer — они хранятся зашифрованно.
---
## Обновление
```bash
# Пересобрать образ
cd mikrotik-monitor
git pull
docker build -t mikrotik-monitor:latest .
```
Затем в Portainer: **Stacks → mikrotik-monitor → Update the stack**.
---
## Просмотр логов
```bash
docker logs -f mikrotik-monitor
```
Пример вывода:
```
2024-01-15 14:30:00 [INFO] === MikroTik Monitor запущен ===
2024-01-15 14:30:00 [INFO] Хосты: ['192.168.1.1', '192.168.1.10']
2024-01-15 14:30:01 [INFO] Подключено к MikroTik
2024-01-15 14:30:02 [INFO] ✅ 192.168.1.1 avg-rtt=0.8 ms
2024-01-15 14:30:03 [WARNING] ❌ 192.168.1.10 не отвечает (1/2)
```
monitor.py
+194
View File
@@ -0,0 +1,194 @@
#!/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
from dotenv import load_dotenv
load_dotenv()
# ──────────────────────────────────────────────
# Настройки из переменных окружения
# ──────────────────────────────────────────────
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-адресов для мониторинга (через запятую в .env или задать напрямую)
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")) # таймаут пинга (мс)
# Сколько раз подряд хост должен не ответить, чтобы считаться DOWN
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: пинг через API
# ──────────────────────────────────────────────
def mikrotik_ping(connection, address: str) -> Optional[float]:
"""
Запускает ping на MikroTik. Возвращает средний avg-rtt (мс) или None если недоступен.
"""
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),
},
)
# Ищем строку с итогами (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)
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)
# Состояние: счётчик последовательных ошибок и флаг «уже 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}
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"
f"<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()
stack.yml
+71
View File
@@ -0,0 +1,71 @@
version: "3.9"
# ═══════════════════════════════════════════════════════════════════
# MikroTik Monitor — Portainer Stack
#
# Перед деплоем:
# 1. Соберите образ на сервере (см. README) и запомните его имя
# 2. Заполните все переменные окружения ниже
# 3. В Portainer: Stacks → Add Stack → вставьте этот файл → Deploy
# ═══════════════════════════════════════════════════════════════════
services:
mikrotik-monitor:
# ── Образ ──────────────────────────────────────────────────────
# Соберите локально: docker build -t mikrotik-monitor:latest .
# Или укажите адрес своего реестра:
# registry.example.com/mikrotik-monitor:latest
image: mikrotik-monitor:latest
container_name: mikrotik-monitor
restart: unless-stopped
hostname: mikrotik-monitor
# ── Переменные окружения ────────────────────────────────────────
environment:
# MikroTik
- MIKROTIK_HOST=192.168.88.1 # ← IP вашего роутера
- MIKROTIK_PORT=8728 # API-порт (8728 plain / 8729 SSL)
- MIKROTIK_USER=monitor # ← логин
- MIKROTIK_PASSWORD=your_password # ← пароль
# Telegram
- 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
# Параметры мониторинга
- CHECK_INTERVAL=60 # интервал между проверками (сек)
- PING_COUNT=3 # кол-во пинг-пакетов
- PING_TIMEOUT_MS=1000 # таймаут пинга (мс)
- FAILURE_THRESHOLD=2 # провалов подряд → уведомление DOWN
# Временная зона
- TZ=Europe/Moscow
# ── Тома ───────────────────────────────────────────────────────
volumes:
- /etc/localtime:/etc/localtime:ro
# ── Логирование ─────────────────────────────────────────────────
logging:
driver: json-file
options:
max-size: "10m"
max-file: "5"
# ── Healthcheck ─────────────────────────────────────────────────
# Контейнер помечается healthy, если процесс monitor.py жив
healthcheck:
test: ["CMD-SHELL", "pgrep -f monitor.py || exit 1"]
interval: 30s
timeout: 5s
retries: 3
start_period: 10s
# ── Метки для Portainer ─────────────────────────────────────────
labels:
- "com.docker.compose.project=mikrotik-monitor"
- "description=MikroTik IP monitoring with Telegram alerts"