commit 8dca7eb46b55c191021875639583535de57a493f Author: Maksim Totmin Date: Fri Jun 19 21:04:05 2026 +0700 Initial commit: SES Monitor Watchdog-демон для непрерывного мониторинга SES. - Liveness-проверки API, лицензии, списка роботов - Диалоговые сценарии из YAML с агрегацией в 6 фиксированных Zabbix-метрик - Доставка: zabbix_sender (push) или UserParameter (poll) - Hot-reload сценариев без перезапуска diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..13a85a5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,25 @@ +# --- Python --- +__pycache__/ +*.py[cod] +*.egg-info/ +*.egg +dist/ +build/ + +# --- Virtual environment --- +.venv/ +venv/ + +# --- IDE / editors --- +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# --- OS --- +.DS_Store +Thumbs.db + +# --- Project specific --- +config/config.yaml diff --git a/README.md b/README.md new file mode 100644 index 0000000..629bcfb --- /dev/null +++ b/README.md @@ -0,0 +1,330 @@ +# SES Monitor + +Система непрерывного мониторинга диалоговой сценарной машины **SES** (Service Engine Server). + +## Что делает + +Watchdog-демон, который: +- **Проверяет доступность** SES API каждые 30 секунд +- **Имитирует диалоги пользователей** по сценариям, описанным в YAML-файлах +- **Отправляет метрики в Zabbix** через zabbix_sender или UserParameter + +SES — универсальная сценарная машина, поэтому мониторинг не привязан к конкретному проекту. Один и тот же демон может обслуживать любой сценарий: голосового ассистента, чат-бота, IVR, диалогов поддержки и любых других. + +## Архитектура + +``` +┌─────────────────────────────────────────────────────┐ +│ ses-monitor (watchdog-демон) │ +│ │ +│ ┌──────────────┐ ┌────────────────┐ │ +│ │ LivenessCheck │ │ ScenarioRunner │ │ +│ │ (30 сек) │ │ (по YAML) │ │ +│ │ │ │ │ │ +│ │ GET /hostname│ │ POST /ses/ask │ │ +│ │ GET /license │ │ шаги+expect │ │ +│ └──────┬───────┘ └───────┬─────────┘ │ +│ │ │ │ +│ ┌──────▼──────────────────▼──────┐ │ +│ │ Zabbix (sender / UP) │ │ +│ └──────────────┬─────────────────┘ │ +└─────────────────┼────────────────────────────────────┘ + │ TCP 10051 / JSON file + ┌───────────▼───────────┐ + │ Zabbix Server │ + │ (триггеры, алерты) │ + └───────────────────────┘ +``` + + +## Режимы доставки в Zabbix + +Монитор поддерживает два режима отправки метрик: + +### sender (по умолчанию) — zabbix_sender (активный push) +Монитор запускает утилиту `zabbix_sender` и передаёт метрики через stdin. +Данные отправляются на Zabbix сервер или прокси (порт 10051, trapper). +В Zabbix элементы данных имеют тип "Zabbix траппер". +Требуется: установленный zabbix_sender на сервере с монитором. + +### userparameter — Zabbix агент (пассивный опрос) +Монитор пишет JSON со всеми метриками в файл (`/var/tmp/ses_monitor/metrics.json`). +Zabbix агент по запросу сервера запускает скрипт `scripts/zbx_get.py`, +который возвращает содержимое JSON-файла. +В Zabbix используется master-элемент `ses.metrics` (тип "Zabbix агент", текст) +и зависимые элементы с препроцессингом JSONPath. + +Настройка: +1. Импортировать шаблон `zabbix/zbx_ses_template_5.0_up.xml` в Zabbix +2. Привязать шаблон к хосту (Configuration → Hosts → выберите хост → Templates) +3. В конфиге монитора `config/config.yaml` установить `mode: "userparameter"` +4. В `/etc/zabbix/zabbix_agentd.conf` добавить строку: + ``` + Include=/opt/ses-monitor/config/zbx_agentd_ses.conf + ``` +5. Перезапустить Zabbix агента: `sudo systemctl restart zabbix-agent` + +## Структура проекта + +``` +/opt/ses-monitor/ +├── src/ +│ └── ses_monitor/ # Python-пакет (src-layout) +│ ├── __init__.py # Версия, метаданные +│ ├── __main__.py # Точка входа: python -m ses_monitor +│ ├── main.py # Главный asyncio watchdog-цикл +│ ├── config.py # Загрузка YAML-конфигурации +│ ├── checker.py # Liveness-проверки (API, лицензия, роботы) +│ ├── session_client.py # HTTP-клиент для SES API +│ ├── scenario_runner.py # Движок выполнения сценариев +│ ├── log_client.py # Чтение логов SES (задел на будущее) +│ ├── zabbix.py # Интеграция с Zabbix (sender / UserParameter) +│ └── utils.py # Утилиты (логирование, JSON-пути, проверки) +├── config/ +│ ├── config.yaml.example # Шаблон конфигурации (скопировать в config.yaml) +│ └── scenarios/ +│ ├── example_liveness.yaml # Пример: быстрая проверка +│ └── example_dialog.yaml # Пример: многошаговый диалог +├── zabbix/ +│ ├── zbx_ses_template_5.0.xml # Шаблон для sender (trapper) +│ ├── zbx_ses_template_5.0_up.xml # Шаблон для UserParameter +│ └── zbx_ses_monitor.conf # UserParameter для Zabbix агента +├── scripts/ +│ └── zbx_get.py # Скрипт чтения JSON для UserParameter +├── systemd/ +│ └── ses-monitor.service # Systemd unit +├── tests/ +│ ├── __init__.py +│ └── test_scenario_runner.py # 48 тестов +├── pyproject.toml # Зависимости и метаданные пакета +└── README.md +``` + +## Установка на сервере + +### 1. Требования + +- **Python 3.9+** +- **Git** — для клонирования репозитория +- **Zabbix Agent** с утилитой `zabbix_sender` +- Доступ к SES API (`http://:6190`) +- Доступ к Zabbix Server по порту 10051 +- SSH-ключ, добавленный в Git-репозиторий (`git.totmin.ru`) + +### 2. Установка пакета + +```bash +# Клонируем репозиторий на сервер +git clone git@git.totmin.ru:en2zmax/ses-monitor.git /opt/ses-monitor + +# Переходим в каталог проекта +cd /opt/ses-monitor + +# Создаём виртуальное окружение и устанавливаем пакет +python3 -m venv .venv +source .venv/bin/activate +pip install -e . + +# Создаём каталог для логов +mkdir -p /var/log/ses-monitor +``` + +### 3. Настройка конфигурации + +Скопируйте шаблон и отредактируйте: + +```bash +cp config/config.yaml.example config/config.yaml +nano config/config.yaml +``` + +Основные поля: + +```yaml +ses: + base_url: "http://YOUR_SES_HOST:6190" # URL вашего SES API + channel: "monitoring" # Канал для мониторинговых сессий + request_timeout: 30 + +zabbix: + enabled: true + mode: "userparameter" # или "sender" + host: "my-server-name" # Имя хоста в Zabbix + server: "127.0.0.1" # IP Zabbix сервера (только sender) + port: 10051 + sender_path: "/usr/bin/zabbix_sender" + metrics_file: "/var/tmp/ses_monitor/metrics.json" # (только userparameter) +``` + +### 4. Создание сценария + +Скопируйте пример и адаптируйте под ваш проект: + +```bash +cp config/scenarios/example_liveness.yaml config/scenarios/my_project.yaml +``` + +### 5. Запуск + +```bash +# Проверка вручную: +cd /opt/ses-monitor +.venv/bin/python -m ses_monitor + +# Установка как systemd-сервиса: +cp systemd/ses-monitor.service /etc/systemd/system/ +systemctl daemon-reload +systemctl enable --now ses-monitor + +# Просмотр логов: +journalctl -u ses-monitor -f +``` + +### 6. Импорт шаблона Zabbix + +1. Zabbix Frontend → **Configuration → Templates → Import** +2. Выбрать файл `zabbix/zbx_ses_template_5.0.xml` (для sender) или `zabbix/zbx_ses_template_5.0_up.xml` (для UserParameter) +3. Нажать Import +4. Привязать шаблон "SES Monitor" к нужному хосту + +## Формат сценариев + +Сценарии описываются в YAML-файлах в каталоге `config/scenarios/`. Это гибкий DSL, который не зависит от конкретного проекта SES. + +### Полный формат + +```yaml +# Имя сценария (используется в логах) +name: "example: проверка диалога" + +# Описание для документации +description: "Проверяет что бот корректно отвечает на вопросы" + +# Включён ли сценарий (false — отключён) +enabled: true + +# Интервал запуска в секундах +interval: 300 + +# Таймаут HTTP-запроса в секундах +timeout: 30 + +# ID робота (обязательно). Узнать: GET /ses/robot/listall +robot_id: "uuid-вашего-робота" + +# Канал (если не указан — берётся из config.yaml) +channel: "monitoring" + +# Не переоткрывать закрытые сессии (false — создаём новую сессию каждый раз) +restart_closed: false + +# Шаги сценария +steps: + - send: "Привет" + expect: + # Каждое ожидание проверяет значение в JSON-ответе + - path: "error" + equals: 0 + + - path: "answer[0].messages[0]" + contains: "Здрав" + + - send: "Как дела?" + expect: + - path: "error" + equals: 0 + - path: "answer" + not_empty: true +``` + +### Условия проверки (expect) + +| Условие | Описание | Пример | +|---|---|---| +| `equals` | Точное совпадение | `equals: 0` | +| `contains` | Подстрока содержится (без учёта регистра) | `contains: "здравствуйте"` | +| `not_contains` | Подстрока отсутствует | `not_contains: "ошибка"` | +| `not_empty` | Поле не пустое | `not_empty: true` | +| `regex` | Регулярное выражение | `regex: "\\d{10,}"` | +| `exists` | Поле существует и не None | `exists` (без значения) | + +### Пути JSON (path) + +Формат: `ключ.подключ[индекс].поле` + +```yaml +# Корневое поле +path: "error" + +# Массив → первый элемент → поле +path: "answer[0].messages[0]" + +# Глубокая вложенность +path: "request.data[0].reply.confidence" +``` + +## Как адаптировать под ваш проект + +1. **Узнайте robot_id** вашего сценария: + ```bash + curl -s http://:6190/ses/robot/listall + ``` + +2. **Пропишите robot_id** в файле сценария (поле `robot_id` в YAML) + +3. **Создайте файл сценария**, описав шаги диалога: + ```bash + cp config/scenarios/example_liveness.yaml config/scenarios/$(hostname).yaml + vim config/scenarios/$(hostname).yaml + ``` + +4. **Протестируйте вручную** шаги через curl: + ```bash + curl -X POST "http://:6190/ses/ask/?text=Привет&channel=monitoring" + ``` + +5. **Запустите демон** — метрики агрегируются автоматически (шаблон Zabbix универсален) + +## Расширение системы + +### Добавление нового сценария + +1. Создать `config/scenarios/my_scenario.yaml` +2. Скопировать структуру из примеров +3. Указать `robot_id` вашего робота (узнать: `GET /ses/robot/listall`) +4. Описать шаги и ожидания +5. Сценарий автоматически подхватится через hot-reload (каждые 60 секунд) + +### Добавление новой метрики + +1. В нужном модуле добавить поле в датакласс-результат +2. В `main.py` вызвать `self._zabbix.add(key, value)` +3. Добавить item в Zabbix шаблон + +### Prometheus/Grafana + +При необходимости можно добавить экспорт метрик в Prometheus: + +```python +from prometheus_client import Gauge, start_http_server + +api_liveness = Gauge("ses_api_liveness", "SES API availability") +``` + +## Мониторинг самого монитора + +- **systemd**: `systemctl is-active ses-monitor` +- **Zabbix Active Agent**: `proc.num[python,root,ses_monitor]` +- **Логи**: `journalctl -u ses-monitor --since "10 min ago" -o cat | grep ERROR` + +## Отказоустойчивость + +- **Автоперезапуск**: systemd Restart=always, RestartSec=10 +- **Graceful shutdown**: по SIGTERM отправляет последний батч метрик +- **Таймауты**: все HTTP-запросы с ограничением по времени (по умолчанию 30с) +- **Изоляция сценариев**: ошибка в одном сценарии не влияет на другие +- **Hot-reload**: сценарии перечитываются без перезапуска демона + +## Лицензия + +MIT diff --git a/config/config.yaml.example b/config/config.yaml.example new file mode 100644 index 0000000..8e29ccb --- /dev/null +++ b/config/config.yaml.example @@ -0,0 +1,89 @@ +# ============================================================================= +# Основная конфигурация системы мониторинга SES (шаблон) +# ============================================================================= +# Скопируйте в config.yaml и заполните свои значения. +# cp config.yaml.example config.yaml +# Формат: YAML 1.2 + +# --------------------------------------------------------------------------- +# Настройки подключения к SES серверу +# --------------------------------------------------------------------------- +ses: + # Базовый URL SES API + base_url: "http://YOUR_SES_HOST:6190" + + # Эндпоинт для получения списка роботов: + # /ses/robot/list — только опубликованные + # /ses/robot/listall — все (включая черновики) + robot_list_endpoint: "/ses/robot/listall" + + # Канал для мониторинговых запросов (не влияет на реальных пользователей) + channel: "monitoring" + + # Таймаут HTTP-запросов в секундах + request_timeout: 30 + +# --------------------------------------------------------------------------- +# Настройки интеграции с Zabbix +# --------------------------------------------------------------------------- +zabbix: + # Включить отправку метрик в Zabbix + enabled: true + + # Режим доставки: "sender" (zabbix_sender) или "userparameter" (Zabbix агент) + # sender — активный push через zabbix_sender (trapper) + # userparameter — пассивный опрос Zabbix агентом (через JSON-файл) + mode: "userparameter" + + # Имя хоста в Zabbix (техническое имя узла сети). + host: "YOUR_ZABBIX_HOST" + + # ---- Настройки режима "sender" (zabbix_sender) ---- + server: "YOUR_ZABBIX_SERVER_IP" + port: 10051 + sender_path: "/usr/bin/zabbix_sender" + + # ---- Настройки режима "userparameter" (Zabbix агент) ---- + metrics_file: "/var/tmp/ses_monitor/metrics.json" + agentd_config_dir: "/opt/ses-monitor/config" + +# --------------------------------------------------------------------------- +# Настройки мониторинга +# --------------------------------------------------------------------------- +monitoring: + # Интервал liveness-проверок (секунды) + liveness_interval: 30 + + # Интервал проверки архива диалогов (секунды) + archive_check_interval: 600 + + # Включить чтение логов через POST /ses/log (ресурсоёмко) + log_enabled: false + + # Каталог с YAML-файлами сценариев + scenarios_dir: "/opt/ses-monitor/config/scenarios" + + # Таймаут выполнения сценария по умолчанию (секунды) + scenario_default_timeout: 30 + + # Мониторинг времени обработки шага диалога из архива + step_time_monitoring: false + +# --------------------------------------------------------------------------- +# Мониторинг внутренних моделей SES (SEE, SMC, TTS, QAS, SPR) +# --------------------------------------------------------------------------- +models_monitoring: + enabled: true + see_confidence_threshold: 0.5 + smc_confidence_threshold: 0.5 + tts_enabled: true + qas_enabled: true + spr_enabled: true + +# --------------------------------------------------------------------------- +# Настройки логирования +# --------------------------------------------------------------------------- +logging: + level: "INFO" + file: "/var/log/ses-monitor/monitor.log" + format: "%(asctime)s [%(levelname)s] %(name)s: %(message)s" diff --git a/config/scenarios/example_dialog.yaml b/config/scenarios/example_dialog.yaml new file mode 100644 index 0000000..2627bb6 --- /dev/null +++ b/config/scenarios/example_dialog.yaml @@ -0,0 +1,43 @@ +# ============================================================================= +# Пример сценария: диалог с несколькими шагами +# ============================================================================= +# Имитирует полноценный диалог пользователя с роботом SES. +# Каждый шаг проверяет конкретные поля ответа. +# +# ИНСТРУКЦИЯ: скопируйте, переименуйте и опишите ваш диалог. +# cp example_dialog.yaml my_project_dialog.yaml +# # замените шаги и ожидания под логику вашего сценария + +name: "example: диалог" +description: > + Пример многошагового диалога с проверкой промежуточных ответов. + Адаптируйте под логику вашего робота. + +enabled: false # Пример выключен по умолчанию +interval: 600 +timeout: 45 +# robot_id: "uuid-вашего-робота" # ОБЯЗАТЕЛЬНО: ID робота из /ses/robot/listall +restart_closed: false # Не переоткрывать закрытые сессии + +steps: + # Шаг 1: инициируем диалог + - send: "Привет" + expect: + - path: "error" + equals: 0 + - path: "answer[0].messages[0]" + not_empty: true + + # Шаг 2: продолжаем диалог + - send: "Как дела?" + expect: + - path: "error" + equals: 0 + - path: "answer[0].messages[0]" + not_empty: true + + # Шаг 3: завершаем + - send: "Спасибо" + expect: + - path: "error" + equals: 0 diff --git a/config/scenarios/example_liveness.yaml b/config/scenarios/example_liveness.yaml new file mode 100644 index 0000000..c87cf41 --- /dev/null +++ b/config/scenarios/example_liveness.yaml @@ -0,0 +1,28 @@ +# ============================================================================= +# Пример сценария: базовая проверка liveness робота +# ============================================================================= +# Проверяет что робот отвечает на приветствие и не возвращает ошибок API. +# robot_id указывается в этом же файле (не наследуется из глобального конфига). +# +# ИНСТРУКЦИЯ: скопируйте этот файл, переименуйте и адаптируйте под ваш проект. +# cp example_liveness.yaml my_project_liveness.yaml +# # отредактируйте: name, robot_id, steps, expect + +name: "example: liveness" +description: > + Базовая проверка: отправляем приветствие роботу + и убеждаемся что API отвечает без ошибок. + Не зависит от конкретного сценария SES. + +enabled: false # Пример выключен по умолчанию +interval: 60 +timeout: 15 +# robot_id: "uuid-вашего-робота" # ОБЯЗАТЕЛЬНО: ID робота из /ses/robot/listall + +steps: + - send: "Здравствуйте" + expect: + - path: "error" + equals: 0 + - path: "answer" + not_empty: true diff --git a/config/zbx_agentd_ses.conf b/config/zbx_agentd_ses.conf new file mode 100644 index 0000000..80f39e3 --- /dev/null +++ b/config/zbx_agentd_ses.conf @@ -0,0 +1,13 @@ +# UserParameter для ses-monitor. +# +# Подключение: добавьте в /etc/zabbix/zabbix_agentd.conf строку: +# Include=/opt/ses-monitor/config/zbx_agentd_ses.conf +# +# После этого перезапустите агента: +# sudo systemctl restart zabbix-agent +# +# Шаблон для Zabbix 5.0: zabbix/zbx_ses_template_5.0_up.xml +# Master-элемент: ses.metrics (JSON) +# Dependent-элементы: ses.* с JSONPath препроцессингом + +UserParameter=ses.metrics,/opt/ses-monitor/scripts/zbx_get.py diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f33e56e --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,52 @@ +[project] +name = "ses-monitor" +version = "1.0.0" +description = "Система мониторинга SES (Service Engine Server) — диалоговой сценарной машины" +readme = "README.md" +requires-python = ">=3.7" +license = { text = "MIT" } +authors = [ + { name = "SES Monitoring Team" } +] +keywords = ["monitoring", "ses", "zabbix", "dialog", "watchdog"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: System Administrators", + "License :: OSI Approved :: MIT License", + "Operating System :: POSIX :: Linux", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Topic :: System :: Monitoring", +] + +dependencies = [ + "httpx>=0.21,<0.24", + "pyyaml>=5.4", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0", + "pytest-asyncio>=0.21", + "pytest-cov>=4.0", + "mypy>=1.0", +] + +[project.scripts] +ses-monitor = "ses_monitor.main:main" + +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +where = ["src"] +include = ["ses_monitor*"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] +python_files = ["test_*.py"] +pythonpath = ["src"] diff --git a/scripts/zbx_get.py b/scripts/zbx_get.py new file mode 100644 index 0000000..8bdc5a8 --- /dev/null +++ b/scripts/zbx_get.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +""" +Скрипт для Zabbix UserParameter. + +Возвращает JSON со всеми метриками ses-monitor. +Используется как один UserParameter: ses.metrics + +Настройка Zabbix агента: + UserParameter=ses.metrics,/opt/ses-monitor/scripts/zbx_get.py + +В шаблоне Zabbix создаётся master-элемент ses.metrics (тип: Zabbix агент, +формат: Текст), а все остальные ses.* элементы — зависимые (dependent) с +препроцессингом JSONPath. +""" + +import json +import sys + +METRICS_FILE = "/var/tmp/ses_monitor/metrics.json" + + +def main() -> int: + try: + with open(METRICS_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + except FileNotFoundError: + # Файла ещё нет — возвращаем пустой JSON + print("{}") + return 0 + except json.JSONDecodeError: + print("{}") + return 0 + + print(json.dumps(data, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/ses_monitor/__init__.py b/src/ses_monitor/__init__.py new file mode 100644 index 0000000..8daa914 --- /dev/null +++ b/src/ses_monitor/__init__.py @@ -0,0 +1,13 @@ +""" +Система мониторинга SES (Service Engine Server). + +Модульный watchdog-демон для непрерывной проверки работоспособности +диалоговой сценарной машины SES через её REST API. + +Основные компоненты: + - checker: быстрые liveness-проверки (доступность API, лицензия) + - scenario_runner: выполнение сценариев диалогов из YAML-файлов + - zabbix: отправка метрик в Zabbix через zabbix_sender +""" + +__version__ = "1.0.0" diff --git a/src/ses_monitor/__main__.py b/src/ses_monitor/__main__.py new file mode 100644 index 0000000..e15249b --- /dev/null +++ b/src/ses_monitor/__main__.py @@ -0,0 +1,8 @@ +"""Точка входа: python -m ses_monitor.""" + +import sys + +from ses_monitor.main import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/ses_monitor/checker.py b/src/ses_monitor/checker.py new file mode 100644 index 0000000..9008b93 --- /dev/null +++ b/src/ses_monitor/checker.py @@ -0,0 +1,183 @@ +""" +Быстрые liveness-проверки: доступность API, лицензия, количество роботов. + +Запускаются с высокой частотой (каждые 30 секунд) и дают первый сигнал +о проблемах с SES сервером. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from ses_monitor.config import SESConfig +from ses_monitor.session_client import SESClient, HostnameResponse, LicenseResponse + +log = logging.getLogger("ses_monitor.checker") + + +@dataclass +class LivenessResult: + """Результат одного цикла liveness-проверок.""" + + # Доступность API + api_ok: bool = True + api_response_time_ms: float = 0.0 + api_error: str = "" + + # Лицензия + license_ok: bool = True + license_error: str = "" + license_days_remaining: int = -1 + + # Количество сценариев (роботов) + robots_count: int = 0 + robots_error: str = "" + + # Общая метка времени + timestamp: float = 0.0 + + @property + def all_ok(self) -> bool: + """Все проверки прошли успешно.""" + return self.api_ok and self.license_ok and not self.robots_error + + +class LivenessChecker: + """Выполняет быстрые проверки работоспособности SES. + + Проверяет: + 1. GET /ses/hostname — жив ли сервер. + 2. GET /license/check — не истекла ли лицензия. + 3. GET /ses/robot/list — не пропали ли сценарии. + """ + + def __init__(self, ses_config: SESConfig, robot_names: Optional[List[str]] = None): + self._ses_config = ses_config + # Эталонный список роботов (для сравнения) + self._expected_robots = robot_names + + async def check(self, client: SESClient) -> LivenessResult: + """Выполняет полный цикл liveness-проверок. + + Args: + client: HTTP-клиент для SES API. + + Returns: + LivenessResult со статусом всех проверок. + """ + import time + + result = LivenessResult(timestamp=time.time()) + + # 1. Проверка hostname (доступность API) + await self._check_hostname(client, result) + + # 2. Проверка лицензии + await self._check_license(client, result) + + # 3. Проверка количества роботов + await self._check_robots(client, result) + + return result + + async def _check_hostname( + self, client: SESClient, result: LivenessResult + ) -> None: + """Проверяет GET /ses/hostname и замеряет время ответа.""" + import time + + try: + t0 = time.monotonic() + resp = await client.get_hostname() + elapsed = (time.monotonic() - t0) * 1000 + + result.api_response_time_ms = round(elapsed, 1) + + if resp.error == 0: + result.api_ok = True + log.debug( + "Liveness hostname OK: %s v%s, время=%dms", + resp.hostname, resp.version, round(elapsed), + ) + else: + result.api_ok = False + result.api_error = f"API ошибка error={resp.error}: {resp.message}" + log.warning("Liveness hostname ошибка: %s", result.api_error) + + except Exception as e: + result.api_ok = False + result.api_error = f"Недоступен: {e}" + log.error("Liveness hostname исключение: %s", e) + + async def _check_license( + self, client: SESClient, result: LivenessResult + ) -> None: + """Проверяет GET /license/check.""" + try: + resp = await client.get_license() + + if resp.error == 0: + # "infinity" или число > 0 — лицензия в порядке + remaining = resp.remaining_licenses + if remaining == "infinity": + result.license_ok = True + result.license_days_remaining = -1 + elif remaining.isdigit(): + days = int(remaining) + result.license_days_remaining = days + if days > 0: + result.license_ok = True + else: + result.license_ok = False + result.license_error = ( + f"Лицензия истекла: remaining={remaining}" + ) + else: + result.license_ok = False + result.license_error = ( + f"Лицензия истекла или исчерпана: remaining={remaining}" + ) + log.debug("Liveness license OK: remaining=%s, days=%d", remaining, result.license_days_remaining) + else: + result.license_ok = False + result.license_error = f"Ошибка лицензии error={resp.error}" + log.warning("Liveness license ошибка: %s", result.license_error) + + except Exception as e: + result.license_ok = False + result.license_error = f"Недоступна: {e}" + log.error("Liveness license исключение: %s", e) + + async def _check_robots( + self, client: SESClient, result: LivenessResult + ) -> None: + """Проверяет GET /ses/robot/list — список сценариев.""" + try: + robots = await client.get_robots() + result.robots_count = len(robots) + + # Если задан эталонный список — проверяем имена + if self._expected_robots is not None: + actual_names = {r.get("name", "") for r in robots} + expected_set = set(self._expected_robots) + missing = expected_set - actual_names + extra = actual_names - expected_set + + if missing: + result.robots_error = f"Отсутствуют роботы: {missing}" + log.warning("Liveness robots: не хватает %s", missing) + if extra: + msg = f"Новые роботы: {extra}" + if result.robots_error: + result.robots_error += "; " + msg + else: + result.robots_error = msg + log.info("Liveness robots: новые %s", extra) + + log.debug("Liveness robots count: %d", result.robots_count) + + except Exception as e: + result.robots_error = f"Недоступен: {e}" + log.error("Liveness robots исключение: %s", e) diff --git a/src/ses_monitor/config.py b/src/ses_monitor/config.py new file mode 100644 index 0000000..245f1a6 --- /dev/null +++ b/src/ses_monitor/config.py @@ -0,0 +1,337 @@ +""" +Загрузка и валидация YAML-конфигурации монитора. + +Основной конфигурационный файл: config/config.yaml +Сценарии загружаются из config/scenarios/*.yaml +""" + +from __future__ import annotations + +import enum +import logging +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional + +import yaml + +log = logging.getLogger("ses_monitor.config") + + +class ZabbixMode(enum.Enum): + """Режим доставки метрик в Zabbix.""" + + SENDER = "sender" # zabbix_sender — активный push + USERPARAMETER = "userparameter" # UserParameter — пассивный опрос агентом + + @classmethod + def from_str(cls, value: str) -> ZabbixMode: + """Создаёт режим из строки (регистронезависимо).""" + for mode in cls: + if mode.value.lower() == value.strip().lower(): + return mode + raise ValueError(f"Неизвестный режим Zabbix: '{value}'. Допустимые: {[m.value for m in cls]}") + + +@dataclass +class SESConfig: + """Настройки подключения к SES серверу.""" + + base_url: str = "http://10.101.60.31:6190" + channel: str = "monitoring" + request_timeout: int = 30 + robot_list_endpoint: str = "/ses/robot/list" # или "/ses/robot/listall" + + +@dataclass +class ZabbixConfig: + """Настройки отправки метрик в Zabbix.""" + + enabled: bool = True + mode: ZabbixMode = ZabbixMode.SENDER # "sender" или "userparameter" + + # ---- Общие поля для обоих режимов ---- + host: str = "" # Задаётся в config.yaml: zabbix.host (техническое имя узла сети) + + # ---- Поля режима sender ---- + server: str = "127.0.0.1" + port: int = 10051 + sender_path: str = "/usr/bin/zabbix_sender" + + # ---- Поля режима userparameter ---- + metrics_file: str = "/var/tmp/ses_monitor/metrics.json" + agentd_config_dir: str = "/opt/ses-monitor/config" + + +@dataclass +class MonitoringConfig: + """Общие настройки мониторинга.""" + + liveness_interval: int = 30 # Интервал liveness-проверок, секунды + archive_check_interval: int = 600 # Интервал проверки архива, секунды + log_enabled: bool = False # Пока отключено — POST /ses/log тяжёлый + scenarios_dir: str = "config/scenarios" + scenario_default_timeout: int = 30 + step_time_monitoring: bool = True # Мониторинг времени ответа из архива + + +@dataclass +class LoggingConfig: + """Настройки логирования.""" + + level: str = "INFO" + file: Optional[str] = None + format: str = "%(asctime)s [%(levelname)s] %(name)s: %(message)s" + + +@dataclass +class ModelsMonitoringConfig: + """Настройки мониторинга внутренних моделей SES (SEE, SMC, TTS, QAS, SPR).""" + + enabled: bool = True + # Порог уверенности SEE ниже которого считаем проблемой (0..1) + see_confidence_threshold: float = 0.5 + # Порог уверенности SMC ниже которого считаем проблемой (0..1) + smc_confidence_threshold: float = 0.5 + # Отслеживать ли голосовые ответы TTS + tts_enabled: bool = True + # Отслеживать ли наполненность QAS + qas_enabled: bool = True + # Отслеживать ли наполненность SPR (распознавание речи: аудио→текст) + spr_enabled: bool = True + + +@dataclass +class AppConfig: + """Корневая конфигурация приложения.""" + + ses: SESConfig = field(default_factory=SESConfig) + zabbix: ZabbixConfig = field(default_factory=ZabbixConfig) + monitoring: MonitoringConfig = field(default_factory=MonitoringConfig) + models: ModelsMonitoringConfig = field(default_factory=ModelsMonitoringConfig) + logging: LoggingConfig = field(default_factory=LoggingConfig) + + +@dataclass +class ScenarioStep: + """Один шаг диалогового сценария.""" + + send: str # Текст сообщения, которое отправляем роботу + expect: List[Dict[str, Any]] = field(default_factory=list) + # Каждое ожидание — dict с ключами: + # path: str (путь в JSON-ответе, например "answer[0].messages[0]") + # contains/equals/not_contains/not_empty/regex: условие + + +@dataclass +class ArchiveCheckConfig: + """Настройки проверки архива для сценария.""" + + enabled: bool = False + lookback_minutes: int = 10 + expect: List[Dict[str, Any]] = field(default_factory=list) + + +@dataclass +class ScenarioConfig: + """Конфигурация одного сценария мониторинга.""" + + name: str = "" + description: str = "" + enabled: bool = True + interval: int = 300 # Интервал запуска, секунды + timeout: int = 30 + robot_id: Optional[str] = None # Если None — берётся из config.yaml + channel: Optional[str] = None + restart_closed: bool = False + steps: List[ScenarioStep] = field(default_factory=list) + archive_check: Optional[ArchiveCheckConfig] = None + + +def _dict_to_scenario_step(data: Dict[str, Any]) -> ScenarioStep: + """Преобразует словарь из YAML в ScenarioStep.""" + return ScenarioStep( + send=data["send"], + expect=data.get("expect", []), + ) + + +def _dict_to_scenario_config(data: Dict[str, Any]) -> ScenarioConfig: + """Преобразует словарь из YAML в ScenarioConfig.""" + steps = [_dict_to_scenario_step(s) for s in data.get("steps", [])] + + archive_data = data.get("archive_check") + archive_check = None + if archive_data: + archive_check = ArchiveCheckConfig( + enabled=archive_data.get("enabled", False), + lookback_minutes=archive_data.get("lookback_minutes", 10), + expect=archive_data.get("expect", []), + ) + + return ScenarioConfig( + name=data.get("name", ""), + description=data.get("description", ""), + enabled=data.get("enabled", True), + interval=data.get("interval", 300), + timeout=data.get("timeout", 30), + robot_id=data.get("robot_id"), + channel=data.get("channel"), + restart_closed=data.get("restart_closed", False), + steps=steps, + archive_check=archive_check, + ) + + +def _build_config(data: Dict[str, Any]) -> AppConfig: + """Собирает AppConfig из распарсенного YAML-словаря.""" + ses_data = data.get("ses", {}) + ses = SESConfig( + base_url=ses_data.get("base_url", SESConfig.base_url), + channel=ses_data.get("channel", SESConfig.channel), + request_timeout=ses_data.get("request_timeout", SESConfig.request_timeout), + robot_list_endpoint=ses_data.get("robot_list_endpoint", SESConfig.robot_list_endpoint), + ) + + zbx_data = data.get("zabbix", {}) + mode_raw = zbx_data.get("mode", "sender") + try: + mode = ZabbixMode.from_str(mode_raw) + except ValueError: + mode = ZabbixMode.SENDER + log.warning("Неизвестный режим Zabbix '%s', используется 'sender'.", mode_raw) + + zabbix = ZabbixConfig( + enabled=zbx_data.get("enabled", ZabbixConfig.enabled), + mode=mode, + host=zbx_data.get("host", ZabbixConfig.host), + server=zbx_data.get("server", ZabbixConfig.server), + port=zbx_data.get("port", ZabbixConfig.port), + sender_path=zbx_data.get("sender_path", ZabbixConfig.sender_path), + metrics_file=zbx_data.get("metrics_file", ZabbixConfig.metrics_file), + agentd_config_dir=zbx_data.get("agentd_config_dir", ZabbixConfig.agentd_config_dir), + ) + + mon_data = data.get("monitoring", {}) + monitoring = MonitoringConfig( + liveness_interval=mon_data.get("liveness_interval", MonitoringConfig.liveness_interval), + archive_check_interval=mon_data.get("archive_check_interval", MonitoringConfig.archive_check_interval), + log_enabled=mon_data.get("log_enabled", MonitoringConfig.log_enabled), + scenarios_dir=mon_data.get("scenarios_dir", MonitoringConfig.scenarios_dir), + scenario_default_timeout=mon_data.get("scenario_default_timeout", MonitoringConfig.scenario_default_timeout), + step_time_monitoring=mon_data.get("step_time_monitoring", MonitoringConfig.step_time_monitoring), + ) + + models_data = data.get("models_monitoring", {}) + models_cfg = ModelsMonitoringConfig( + enabled=models_data.get("enabled", ModelsMonitoringConfig.enabled), + see_confidence_threshold=models_data.get( + "see_confidence_threshold", ModelsMonitoringConfig.see_confidence_threshold + ), + smc_confidence_threshold=models_data.get( + "smc_confidence_threshold", ModelsMonitoringConfig.smc_confidence_threshold + ), + tts_enabled=models_data.get("tts_enabled", ModelsMonitoringConfig.tts_enabled), + qas_enabled=models_data.get("qas_enabled", ModelsMonitoringConfig.qas_enabled), + spr_enabled=models_data.get("spr_enabled", ModelsMonitoringConfig.spr_enabled), + ) + + log_data = data.get("logging", {}) + logging_cfg = LoggingConfig( + level=log_data.get("level", LoggingConfig.level), + file=log_data.get("file"), + format=log_data.get("format", LoggingConfig.format), + ) + + return AppConfig(ses=ses, zabbix=zabbix, monitoring=monitoring, models=models_cfg, logging=logging_cfg) + + +def load_app_config(config_path: Optional[str] = None) -> AppConfig: + """Загружает основную конфигурацию из YAML-файла. + + Args: + config_path: Путь к config.yaml. Если None — ищет + config/config.yaml относительно корня проекта. + + Returns: + AppConfig с заполненными полями. + + Raises: + FileNotFoundError: Файл конфигурации не найден. + yaml.YAMLError: Ошибка парсинга YAML. + """ + if config_path is None: + # Автоопределение пути: от места запуска ищем config/config.yaml + candidates = [ + Path("config/config.yaml"), + Path("/opt/ses-monitor/config/config.yaml"), + ] + # Добавляем относительный путь от этого файла + this_dir = Path(__file__).resolve().parent.parent + candidates.insert(0, this_dir / "config" / "config.yaml") + + found = None + for c in candidates: + if c.exists(): + found = c + break + if found is None: + raise FileNotFoundError( + "Файл конфигурации не найден. Проверьте config/config.yaml " + "или укажите путь явно." + ) + config_path = str(found) + + with open(config_path, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) + + if data is None: + data = {} + + return _build_config(data) + + +def load_scenarios(scenarios_dir: str) -> List[ScenarioConfig]: + """Загружает все YAML-файлы сценариев из указанного каталога. + + Args: + scenarios_dir: Путь к каталогу со сценариями. + + Returns: + Список конфигураций сценариев. + + Notes: + Файлы сценариев читаются из `*.yaml` и `*.yml` в указанном каталоге. + Сценарии с `enabled: false` пропускаются. + """ + dir_path = Path(scenarios_dir) + if not dir_path.exists(): + raise FileNotFoundError(f"Каталог сценариев не найден: {scenarios_dir}") + + scenarios: List[ScenarioConfig] = [] + for yaml_file in sorted(dir_path.glob("*.yaml")): + try: + with open(yaml_file, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) + if data is None: + continue + scenario = _dict_to_scenario_config(data) + if scenario.enabled: + scenarios.append(scenario) + except (yaml.YAMLError, KeyError) as e: + raise RuntimeError(f"Ошибка чтения сценария {yaml_file}: {e}") from e + + for yml_file in sorted(dir_path.glob("*.yml")): + try: + with open(yml_file, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) + if data is None: + continue + scenario = _dict_to_scenario_config(data) + if scenario.enabled: + scenarios.append(scenario) + except (yaml.YAMLError, KeyError) as e: + raise RuntimeError(f"Ошибка чтения сценария {yml_file}: {e}") from e + + return scenarios diff --git a/src/ses_monitor/log_client.py b/src/ses_monitor/log_client.py new file mode 100644 index 0000000..f6b14fa --- /dev/null +++ b/src/ses_monitor/log_client.py @@ -0,0 +1,96 @@ +""" +Чтение логов SES через POST /ses/log. + +Позволяет получить логи за указанную дату для анализа ошибок. +Функциональность пока отключена по умолчанию (config.yaml: monitoring.log_enabled), +так как запрос логов может быть ресурсоёмким. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Dict, List, Optional + +from ses_monitor.session_client import SESClient + +log = logging.getLogger("ses_monitor.log_client") + + +@dataclass +class LogCheckResult: + """Результат проверки логов.""" + + date: str = "" + fetch_error: str = "" + lines_count: int = 0 + # Паттерны для поиска ошибок (расширяются по мере необходимости) + error_patterns: Dict[str, int] = None # type: ignore + + def __post_init__(self): + if self.error_patterns is None: + self.error_patterns = {} + + +class LogChecker: + """Проверяет логи SES на наличие ошибок. + + Ищет паттерны: + - "ERROR" / "CRITICAL" — критические ошибки. + - "HTTPError" / "RequestException" — ошибки внешних API. + - "json.JSONDecodeError" — ошибки парсинга. + - "connection refused" / "timeout" — сетевые проблемы. + """ + + ERROR_PATTERNS = [ + ("error_count", "ERROR"), + ("critical_count", "CRITICAL"), + ("http_error_count", "HTTPError"), + ("request_exception_count", "RequestException"), + ("json_error_count", "JSONDecodeError"), + ("timeout_count", "timeout"), + ("refused_count", "Connection refused"), + ] + + async def check(self, client: SESClient) -> LogCheckResult: + """Запрашивает и анализирует логи за сегодняшний день. + + Args: + client: HTTP-клиент для SES API. + + Returns: + Результат анализа логов. + """ + result = LogCheckResult(date=datetime.now().strftime("%Y-%m-%d")) + + try: + log_data = await client.get_logs(result.date) + except Exception as e: + result.fetch_error = str(e) + log.error("Log check: ошибка получения логов: %s", e) + return result + + # Если ответ — строка (plain text логов) + if isinstance(log_data, str): + result.lines_count = log_data.count("\n") + 1 + for key, pattern in self.ERROR_PATTERNS: + result.error_patterns[key] = log_data.count(pattern) + + # Если ответ — словарь (структурированные данные) + elif isinstance(log_data, dict): + if log_data.get("error") == 0 and "data" in log_data: + data = log_data["data"] + if isinstance(data, str): + result.lines_count = data.count("\n") + 1 + for key, pattern in self.ERROR_PATTERNS: + result.error_patterns[key] = data.count(pattern) + + log.info( + "Log check: %d строк за %s, ошибок: %s", + result.lines_count, + result.date, + {k: v for k, v in result.error_patterns.items() if v > 0}, + ) + + return result diff --git a/src/ses_monitor/main.py b/src/ses_monitor/main.py new file mode 100644 index 0000000..7800be3 --- /dev/null +++ b/src/ses_monitor/main.py @@ -0,0 +1,328 @@ +""" +Главный цикл watchdog-демона для мониторинга SES. + +Архитектура: + - Бесконечный asyncio-цикл с тремя параллельными задачами: + 1. liveness_task: быстрые проверки каждые 30 секунд. + 2. scenarios_task: запуск сценариев по их индивидуальному расписанию. + 3. archive_task: анализ архива каждые 10 минут. + - Hot-reload сценариев из каталога каждые 60 секунд. + - Все метрики отправляются в Zabbix единым батчем после проверок. + - Graceful shutdown по SIGTERM/SIGINT. +""" + +from __future__ import annotations + +import asyncio +import logging +import signal +import sys +import time +from pathlib import Path +from typing import Dict, List, Optional, Set + +from ses_monitor.config import ( + AppConfig, + ScenarioConfig, + ZabbixMode, + load_app_config, + load_scenarios, +) +from ses_monitor.utils import setup_logging +from ses_monitor.session_client import SESClient +from ses_monitor.checker import LivenessChecker +from ses_monitor.scenario_runner import ScenarioRunner, ScenarioResult, aggregate_scenario_results +from ses_monitor.log_client import LogChecker +from ses_monitor.zabbix import ZabbixSender, UserParameterWriter + +log = logging.getLogger("ses_monitor") + + +class SESMonitorApp: + """Главный класс watchdog-приложения.""" + + def __init__(self, config: AppConfig): + self._config = config + + # HTTP-клиент (один на всё приложение) + self._client = SESClient(config.ses) + + # Компоненты — выбираем способ доставки в Zabbix по конфигу + self._zabbix = self._create_zabbix_sender(config) + self._liveness = LivenessChecker(config.ses) + self._log_checker = LogChecker() + + # Сценарии загружаются динамически (hot-reload) + self._scenarios: List[ScenarioConfig] = [] + self._scenario_runner: Optional[ScenarioRunner] = None + self._scenarios_dir = config.monitoring.scenarios_dir + + # Отслеживание времени последнего запуска сценариев + self._last_scenario_run: Dict[str, float] = {} + + # Флаг graceful shutdown — создаётся в start() внутри цикла (Python 3.7 фикс) + + # Эталонный список роботов (заполняется при первом успешном запросе) + self._known_robots: Set[str] = set() + + @staticmethod + def _create_zabbix_sender(config: AppConfig): + """Фабрика: создаёт ZabbixSender или UserParameterWriter по конфигу.""" + if config.zabbix.mode == ZabbixMode.USERPARAMETER: + log.info( + "Режим доставки: UserParameter (файл: %s)", + config.zabbix.metrics_file, + ) + return UserParameterWriter(config.zabbix) + log.info( + "Режим доставки: zabbix_sender (сервер: %s:%d, хост: %s)", + config.zabbix.server, + config.zabbix.port, + config.zabbix.host, + ) + return ZabbixSender(config.zabbix) + + async def start(self) -> None: + """Запускает watchdog-цикл.""" + zbx_cfg = self._config.zabbix + if zbx_cfg.enabled: + if zbx_cfg.mode == ZabbixMode.USERPARAMETER: + log.info( + "SES Monitor v%s запускается (SES: %s, Zabbix UP: %s)", + __import__("ses_monitor").__version__, + self._config.ses.base_url, + zbx_cfg.metrics_file, + ) + else: + log.info( + "SES Monitor v%s запускается (SES: %s, Zabbix: %s:%d, хост: %s)", + __import__("ses_monitor").__version__, + self._config.ses.base_url, + zbx_cfg.server, zbx_cfg.port, zbx_cfg.host, + ) + else: + log.info("SES Monitor v%s запускается (SES: %s, Zabbix отключён)", + __import__("ses_monitor").__version__, + self._config.ses.base_url) + + # Создаём Event УЖЕ внутри цикла (Python 3.7 фик: иначе self._loop + # привязывается к умолчательному циклу, а не к текущему) + self._shutdown_event = asyncio.Event() + + # Привязываем child watcher к текущему циклу (Python 3.7-3.11) + loop = asyncio.get_running_loop() + if sys.platform != "win32" and sys.version_info < (3, 12): + watcher = asyncio.get_child_watcher() + if hasattr(watcher, "attach_loop"): + watcher.attach_loop(loop) + + # Регистрируем обработчики сигналов + for sig in (signal.SIGTERM, signal.SIGINT): + try: + loop.add_signal_handler(sig, self._handle_shutdown) + except NotImplementedError: + # Сигналы не поддерживаются (например, Windows) + pass + + try: + # Основной цикл + await self._main_loop() + except asyncio.CancelledError: + pass + finally: + await self._shutdown() + + def _handle_shutdown(self) -> None: + """Обработчик SIGTERM/SIGINT.""" + log.info("Получен сигнал завершения, останавливаемся...") + self._shutdown_event.set() + + async def _shutdown(self) -> None: + """Graceful shutdown: закрываем клиент, сбрасываем метрики.""" + log.info("Завершение работы...") + if self._zabbix.enabled: + # Отправляем последний батч метрик если есть + await self._zabbix.flush() + await self._client.close() + log.info("SES Monitor остановлен.") + + async def _load_scenarios(self) -> None: + """Загружает или перезагружает сценарии (hot-reload).""" + try: + scenarios = load_scenarios(self._scenarios_dir) + self._scenarios = scenarios + if self._scenario_runner is None: + self._scenario_runner = ScenarioRunner( + self._config.ses, scenarios + ) + else: + self._scenario_runner.update_scenarios(scenarios) + log.debug("Загружено %d сценариев", len(scenarios)) + except Exception as e: + log.error("Ошибка загрузки сценариев: %s", e) + + async def _main_loop(self) -> None: + """Основной цикл: liveness, сценарии, архив.""" + # Первичная загрузка сценариев + await self._load_scenarios() + + # Временные метки для планирования + last_hot_reload = time.monotonic() + + while not self._shutdown_event.is_set(): + now = time.monotonic() + + # ---------- 1. Liveness-проверка ---------- + await self._run_liveness() + + # ---------- 2. Сценарии по расписанию ---------- + await self._run_due_scenarios(now) + + # ---------- 3. Hot-reload сценариев ---------- + if now - last_hot_reload >= 60.0: + await self._load_scenarios() + last_hot_reload = now + + # ---------- 4. Логи (если включено) ---------- + if self._config.monitoring.log_enabled: + # Логи проверяем реже — раз в 30 минут + if int(now) % 1800 < self._config.monitoring.liveness_interval: + await self._run_log_check() + + # ---------- 6. Отправка метрик в Zabbix ---------- + await self._zabbix.flush() + + # Ждём или до следующего цикла liveness, или до shutdown + sleep_time = self._config.monitoring.liveness_interval + try: + await asyncio.wait_for( + self._shutdown_event.wait(), timeout=sleep_time + ) + except asyncio.TimeoutError: + pass # Нормально: просто прошёл интервал + + async def _run_liveness(self) -> None: + """Выполняет быстрые liveness-проверки.""" + try: + result = await self._liveness.check(self._client) + + if self._zabbix.enabled: + self._zabbix.add("ses.api.liveness", int(result.api_ok)) + self._zabbix.add("ses.api.response_time", round(result.api_response_time_ms)) + self._zabbix.add("ses.license.ok", int(result.license_ok)) + self._zabbix.add("ses.license.days_remaining", result.license_days_remaining) + + if result.robots_count > 0: + self._zabbix.add("ses.robots.count", result.robots_count) + + if result.robots_count > 0 and not self._known_robots: + log.info( + "Liveness: %d сценариев, API=%s, license=%s, days_left=%d, время=%0.0fms", + result.robots_count, + "OK" if result.api_ok else "FAIL", + "OK" if result.license_ok else "FAIL", + result.license_days_remaining, + result.api_response_time_ms, + ) + + if not result.api_ok: + log.error("Liveness FAIL: %s", result.api_error) + if not result.license_ok: + log.error("License FAIL: %s", result.license_error) + + except Exception as e: + log.exception("Liveness: неожиданная ошибка: %s", e) + + async def _run_due_scenarios(self, now: float) -> None: + """Запускает сценарии, у которых подошло время. + + Все результаты собираются и агрегируются в фиксированный набор + Zabbix-метрик (ses.scenario.ok, .total, .failed, .avg_time, .max_time, + .error). Это делает шаблон Zabbix универсальным — не нужно добавлять + отдельный item под каждый сценарий. + """ + if not self._scenario_runner: + return + + due_results: List[ScenarioResult] = [] + + for scenario in self._scenarios: + name = scenario.name or "unknown" + last_run = self._last_scenario_run.get(name, 0) + + if now - last_run >= scenario.interval: + try: + result = await self._scenario_runner.run_one( + self._client, scenario + ) + due_results.append(result) + self._last_scenario_run[name] = now + + if not result.passed: + log.warning( + "Сценарий '%s': ПРОВАЛЕН за %dms (ошибок: %d шагов)", + name, + round(result.total_time_ms), + len(result.failed_steps), + ) + except Exception as e: + log.exception( + "Сценарий '%s': неожиданное исключение: %s", name, e + ) + + # Агрегируем результаты и отправляем фиксированные ключи + if due_results and self._zabbix.enabled: + agg = aggregate_scenario_results(due_results) + self._zabbix.add("ses.scenario.ok", agg["ok"]) + self._zabbix.add("ses.scenario.total", agg["total"]) + self._zabbix.add("ses.scenario.failed", agg["failed"]) + self._zabbix.add("ses.scenario.avg_time", agg["avg_time"]) + self._zabbix.add("ses.scenario.max_time", agg["max_time"]) + if agg["error"]: + self._zabbix.add("ses.scenario.error", agg["error"]) + + async def _run_log_check(self) -> None: + """Выполняет проверку логов (если включено).""" + try: + result = await self._log_checker.check(self._client) + + if self._zabbix.enabled: + for key, count in result.error_patterns.items(): + self._zabbix.add(f"ses.log.{key}", count) + + except Exception as e: + log.exception("Log check: неожиданная ошибка: %s", e) + + +def main() -> int: + """Точка входа в приложение. + + Returns: + 0 при успешном завершении, 1 при ошибке. + """ + try: + # Загружаем конфигурацию + config = load_app_config() + except FileNotFoundError as e: + print(f"ОШИБКА: {e}", file=sys.stderr) + return 1 + except Exception as e: + print(f"ОШИБКА загрузки конфигурации: {e}", file=sys.stderr) + return 1 + + # Настраиваем логирование + logger = setup_logging(config.logging) + + try: + # Python 3.7: инициализируем child watcher ДО создания цикла + if sys.platform != "win32": + asyncio.get_child_watcher() + app = SESMonitorApp(config) + asyncio.run(app.start()) + return 0 + except KeyboardInterrupt: + logger.info("Остановлено пользователем (Ctrl+C)") + return 0 + except Exception as e: + logger.exception("Критическая ошибка: %s", e) + return 1 diff --git a/src/ses_monitor/scenario_runner.py b/src/ses_monitor/scenario_runner.py new file mode 100644 index 0000000..436ac47 --- /dev/null +++ b/src/ses_monitor/scenario_runner.py @@ -0,0 +1,240 @@ +""" +Движок выполнения диалоговых сценариев из YAML-файлов. + +Каждый сценарий — это последовательность шагов: + 1. Отправить сообщение роботу (POST /ses/ask) + 2. Проверить ответ по условиям (expect) + 3. Перейти к следующему шагу + +Сценарии загружаются из config/scenarios/ во время старта и автоматически +перечитываются каждые 60 секунд (hot-reload). +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from ses_monitor.config import ScenarioConfig, ScenarioStep, SESConfig +from ses_monitor.session_client import SESClient, AskResponse +from ses_monitor.utils import check_all_expectations, timer + +log = logging.getLogger("ses_monitor.scenario") + + +@dataclass +class StepResult: + """Результат выполнения одного шага сценария.""" + + step_index: int + text_sent: str + response: Optional[AskResponse] = None + passed: bool = True + errors: List[str] = field(default_factory=list) + response_time_ms: float = 0.0 + + +@dataclass +class ScenarioResult: + """Результат выполнения полного сценария.""" + + scenario_name: str + passed: bool = True + total_time_ms: float = 0.0 + step_results: List[StepResult] = field(default_factory=list) + error_message: str = "" + + @property + def failed_steps(self) -> List[StepResult]: + """Шаги, которые не прошли проверку.""" + return [s for s in self.step_results if not s.passed] + + +def aggregate_scenario_results( + results: List[ScenarioResult], +) -> Dict[str, Any]: + """Агрегирует результаты всех сценариев в фиксированный набор Zabbix-метрик. + + Возвращает словарь с ключами: + ok (int): 1 если все прошли, 0 если есть упавшие. + total (int): общее количество выполненных сценариев. + failed (int): количество упавших сценариев. + avg_time (int): среднее время выполнения (мс), округлённое. + max_time (int): максимальное время выполнения (мс), округлённое. + error (str): текст ошибки ("N сценариев упали") или пустая строка. + """ + if not results: + return { + "ok": 1, + "total": 0, + "failed": 0, + "avg_time": 0, + "max_time": 0, + "error": "", + } + + failed = sum(1 for r in results if not r.passed) + total = len(results) + times = [r.total_time_ms for r in results] + + return { + "ok": 0 if failed else 1, + "total": total, + "failed": failed, + "avg_time": round(sum(times) / total), + "max_time": round(max(times)), + "error": f"{failed} сценариев упали" if failed else "", + } + + +class ScenarioRunner: + """Выполняет диалоговые сценарии мониторинга. + + Поддерживает: + - Переиспользование HTTP-клиента между проверками. + - Hot-reload сценариев из каталога. + - Автоматическое создание и отслеживание сессий. + """ + + def __init__(self, ses_config: SESConfig, scenarios: List[ScenarioConfig]): + self._ses_config = ses_config + self._scenarios = scenarios + + def update_scenarios(self, scenarios: List[ScenarioConfig]) -> None: + """Обновляет список сценариев (hot-reload).""" + self._scenarios = scenarios + + async def run_all( + self, client: SESClient + ) -> List[ScenarioResult]: + """Запускает все сценарии последовательно. + + Args: + client: HTTP-клиент для SES API. + + Returns: + Список результатов выполнения сценариев. + """ + results: List[ScenarioResult] = [] + for scenario in self._scenarios: + result = await self.run_one(client, scenario) + results.append(result) + return results + + async def run_one( + self, client: SESClient, scenario: ScenarioConfig + ) -> ScenarioResult: + """Выполняет один сценарий. + + Args: + client: HTTP-клиент для SES API. + scenario: Конфигурация сценария. + + Returns: + Результат выполнения. + """ + result = ScenarioResult( + scenario_name=scenario.name, + ) + + # Уникальный ID мониторинговой сессии + session_id = f"monitoring_{scenario.name.replace(' ', '_')}_{int(time.time())}" + log.info("Запуск сценария '%s' (сессия %s)", scenario.name, session_id) + + with timer() as total_time: + try: + for i, step in enumerate(scenario.steps): + step_result = await self._execute_step( + client, scenario, session_id, i, step + ) + result.step_results.append(step_result) + + if not step_result.passed: + result.passed = False + log.warning( + "Сценарий '%s': шаг %d НЕ ПРОЙДЕН: %s", + scenario.name, i + 1, step_result.errors, + ) + break # Прерываем сценарий при первой ошибке + + if result.passed and result.step_results: + log.info( + "Сценарий '%s': УСПЕХ (%d шагов)", + scenario.name, len(result.step_results), + ) + + except Exception as e: + result.passed = False + result.error_message = f"Исключение: {e}" + log.exception("Сценарий '%s': исключение", scenario.name) + + result.total_time_ms = total_time.get("elapsed", 0) * 1000 + return result + + async def _execute_step( + self, + client: SESClient, + scenario: ScenarioConfig, + session_id: str, + step_index: int, + step: ScenarioStep, + ) -> StepResult: + """Выполняет один шаг сценария: отправка → проверка ответа. + + Args: + client: HTTP-клиент. + scenario: Конфигурация сценария. + session_id: ID текущей сессии. + step_index: Номер шага (0-based). + step: Определение шага. + + Returns: + Результат шага. + """ + result = StepResult( + step_index=step_index, + text_sent=step.send, + ) + + try: + # Отправляем сообщение роботу + with timer() as step_time: + response = await client.ask( + text=step.send, + session_id=session_id, + robot_id=scenario.robot_id, + channel=scenario.channel, + restart_closed=scenario.restart_closed, + ) + result.response = response + result.response_time_ms = step_time.get("elapsed", 0) * 1000 + + # Проверяем базовую ошибку API + if response.error != 0: + result.passed = False + result.errors.append( + f"SES API error={response.error} на сообщение '{step.send[:80]}'" + ) + return result + + # Проверяем ожидания + if step.expect: + passed, errors = check_all_expectations(response.raw, step.expect) + result.passed = passed + result.errors = errors + + if not result.passed and not step.expect: + # Если ожиданий не было — считаем что ответ error=0 это успех + result.passed = True + + except Exception as e: + result.passed = False + result.errors.append(f"HTTP ошибка: {e}") + log.exception( + "Сценарий '%s', шаг %d: исключение", + scenario.name, step_index + 1, + ) + + return result diff --git a/src/ses_monitor/session_client.py b/src/ses_monitor/session_client.py new file mode 100644 index 0000000..98dcfbc --- /dev/null +++ b/src/ses_monitor/session_client.py @@ -0,0 +1,302 @@ +""" +HTTP-клиент для взаимодействия с SES REST API. + +Только чтение (GET). Для POST /ses/ask используется безопасный подход: +каждый сценарий открывает новую сессию, тестовые запросы идут с +префиксом 'monitoring_' в номере сессии. +""" + +from __future__ import annotations + +import json +import logging +import time +from dataclasses import dataclass +from typing import Any, Dict, List, Optional +from urllib.parse import urlencode, urljoin + +import httpx + +from ses_monitor.config import SESConfig + +log = logging.getLogger("ses_monitor.client") + + +@dataclass +class AskResponse: + """Ответ на POST /ses/ask.""" + + error: int + question: str + answer: List[Dict[str, Any]] + session: str + uuid: str + raw: Dict[str, Any] + + @classmethod + def from_json(cls, data: Dict[str, Any]) -> "AskResponse": + """Создаёт объект из JSON-ответа SES.""" + return cls( + error=data.get("error", -1), + question=data.get("question", ""), + answer=data.get("answer", []), + session=data.get("session", ""), + uuid=data.get("uuid", ""), + raw=data, + ) + + +@dataclass +class HostnameResponse: + """Ответ на GET /ses/hostname.""" + + error: int + message: str + hostname: str + version: str + + @classmethod + def from_json(cls, data: Dict[str, Any]) -> "HostnameResponse": + return cls( + error=data.get("error", -1), + message=data.get("message", ""), + hostname=data.get("hostname", ""), + version=data.get("version", ""), + ) + + +@dataclass +class LicenseResponse: + """Ответ на GET /license/check.""" + + error: int + uuid: str + remaining_licenses: str + threshold: int + + @classmethod + def from_json(cls, data: Dict[str, Any]) -> "LicenseResponse": + return cls( + error=data.get("error", -1), + uuid=data.get("uuid", ""), + remaining_licenses=data.get("remaining_licenses", ""), + threshold=data.get("threshold", 0), + ) + + +@dataclass +class ArchiveEntry: + """Одна запись из архива диалогов.""" + + id: str + session: str + robot: str + hostname: str + channel: str + unknown: bool + closed: Any + endpoint: Optional[str] + answered: bool + conversion: bool + models: List[str] + events: List[str] + request_message: str + last_reply: str + raw: Dict[str, Any] + + @classmethod + def from_json(cls, data: Dict[str, Any]) -> "ArchiveEntry": + robot_raw = data.get("robot", "") + if isinstance(robot_raw, list): + robot_raw = robot_raw[0] if robot_raw else "" + return cls( + id=data.get("id", ""), + session=data.get("session", ""), + robot=robot_raw, + hostname=data.get("hostname", ""), + channel=data.get("channel", ""), + unknown=data.get("unknown", False), + closed=data.get("closed", False), + endpoint=data.get("endpoint"), + answered=data.get("answered", False), + conversion=data.get("conversion", False), + models=data.get("models", []), + events=data.get("events", []), + request_message=data.get("request", {}).get("message", ""), + last_reply=data.get("current", {}).get("lastreply", ""), + raw=data, + ) + + +class SESClient: + """HTTP-клиент для SES REST API. + + Выполняет только GET и POST /ses/ask запросы. + Все методы асинхронные для работы в asyncio-цикле. + """ + + def __init__(self, config: SESConfig): + self._base_url = config.base_url.rstrip("/") + self._channel = config.channel + self._timeout = config.request_timeout + self._robot_list_endpoint = config.robot_list_endpoint + + self._client: Optional[httpx.AsyncClient] = None + + async def _ensure_client(self) -> httpx.AsyncClient: + """Ленивое создание HTTP-клиента.""" + if self._client is None: + self._client = httpx.AsyncClient( + timeout=httpx.Timeout(self._timeout), + limits=httpx.Limits(max_keepalive_connections=5), + ) + return self._client + + async def close(self) -> None: + """Закрывает HTTP-клиент.""" + if self._client is not None: + await self._client.aclose() + self._client = None + + async def _get(self, path: str) -> Dict[str, Any]: + """Выполняет GET-запрос к SES API.""" + url = urljoin(self._base_url, path) + client = await self._ensure_client() + + try: + response = await client.get(url) + response.raise_for_status() + data = response.json() + return data + except httpx.HTTPError as e: + log.error("HTTP ошибка GET %s: %s", path, e) + raise + except json.JSONDecodeError as e: + log.error("Ошибка парсинга JSON от %s: %s", path, e) + raise + + async def _post(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """Выполняет POST-запрос к SES API.""" + url = urljoin(self._base_url, path) + client = await self._ensure_client() + + try: + response = await client.post(url, params=params) + response.raise_for_status() + data = response.json() + return data + except httpx.HTTPError as e: + log.error("HTTP ошибка POST %s: %s", path, e) + raise + except json.JSONDecodeError as e: + log.error("Ошибка парсинга JSON от %s: %s", path, e) + raise + + # --------------- Liveness-проверки --------------- + + async def get_hostname(self) -> HostnameResponse: + """GET /ses/hostname — проверка доступности сервера.""" + data = await self._get("/ses/hostname") + return HostnameResponse.from_json(data) + + async def get_license(self) -> LicenseResponse: + """GET /license/check — проверка лицензии.""" + data = await self._get("/license/check") + return LicenseResponse.from_json(data) + + async def get_robots(self) -> List[Dict[str, Any]]: + """GET robot list — список сценариев (роботов). + + Эндпоинт задаётся в config.yaml (robot_list_endpoint). + /ses/robot/list — только опубликованные. + /ses/robot/listall — все (включая черновики). + """ + data = await self._get(self._robot_list_endpoint) + return data if isinstance(data, list) else [] + + # --------------- Диалоговые сценарии --------------- + + async def ask( + self, + text: str, + *, + robot_id: str, + session_id: Optional[str] = None, + channel: Optional[str] = None, + restart_closed: bool = False, + ) -> AskResponse: + """POST /ses/ask/{robot_id} — отправка сообщения роботу. + + Args: + text: Текст сообщения. + robot_id: ID робота (обязательно, не наследуется из конфига). + session_id: UUID сессии. Если None — генерируется новый. + channel: Канал. Если None — из конфигурации. + restart_closed: Переоткрывать закрытую сессию. + + Returns: + AskResponse с ответом робота. + """ + if not robot_id: + raise ValueError("robot_id обязателен для POST /ses/ask") + ch = channel or self._channel + + if session_id is None: + session_id = f"monitoring_{int(time.time())}" + + params = { + "text": text, + "channel": ch, + "session": session_id, + "restartClosed": int(restart_closed), + } + + data = await self._post(f"/ses/ask/{robot_id}", params=params) + return AskResponse.from_json(data) + + # --------------- Архив диалогов --------------- + + async def get_archive( + self, + start_date: str, + end_date: str, + **filters: Any, + ) -> List[ArchiveEntry]: + """GET /ses/archive/list — получение истории диалогов. + + Args: + start_date: Начало периода, формат "YYYY-MM-DD HH:MM". + end_date: Конец периода, формат "YYYY-MM-DD HH:MM". + **filters: Дополнительные параметры (channel, robot, text, etc.). + + Returns: + Список записей архива. + """ + params: Dict[str, Any] = { + "startdate": start_date, + "enddate": end_date, + **filters, + } + + data = await self._get(f"/ses/archive/list?{urlencode(params)}") + + if isinstance(data, list): + return [ArchiveEntry.from_json(entry) for entry in data] + return [] + + async def get_dialog(self, dialog_uuid: str) -> Dict[str, Any]: + """GET /ses/archive/dialog/{uuid} — получение диалога по UUID.""" + return await self._get(f"/ses/archive/dialog/{dialog_uuid}") + + # --------------- Логи --------------- + + async def get_logs(self, date: str) -> Dict[str, Any]: + """POST /ses/log — получение логов за дату.""" + params = {"date": date} + return await self._post("/ses/log", params=params) + + # --------------- TestAPI --------------- + + async def call_test_api(self, name: str) -> Any: + """GET /ses/testapi/{name} — вызов тестового API.""" + return await self._get(f"/ses/testapi/{name}") diff --git a/src/ses_monitor/utils.py b/src/ses_monitor/utils.py new file mode 100644 index 0000000..b522640 --- /dev/null +++ b/src/ses_monitor/utils.py @@ -0,0 +1,233 @@ +""" +Вспомогательные утилиты: логирование, работа с JSON-путями, тайминг. +""" + +from __future__ import annotations + +import logging +import re +import time +from contextlib import contextmanager +from typing import Any, Dict, Generator, List, Optional, Tuple, Union + +from ses_monitor.config import LoggingConfig + + +def setup_logging(cfg: LoggingConfig) -> logging.Logger: + """Настраивает и возвращает корневой логгер проекта. + + Args: + cfg: Конфигурация логирования. + + Returns: + Настроенный логгер 'ses_monitor'. + """ + logger = logging.getLogger("ses_monitor") + logger.setLevel(getattr(logging, cfg.level.upper(), logging.INFO)) + + # Убираем существующие обработчики, чтобы не дублировать + logger.handlers.clear() + + formatter = logging.Formatter(cfg.format, datefmt="%Y-%m-%d %H:%M:%S") + + # Всегда пишем в stdout для systemd + console = logging.StreamHandler() + console.setFormatter(formatter) + logger.addHandler(console) + + # Опционально: файловый лог + if cfg.file: + from logging.handlers import RotatingFileHandler + + file_handler = RotatingFileHandler( + cfg.file, maxBytes=10 * 1024 * 1024, backupCount=5, + ) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + return logger + + +def get_json_value(data: Union[Dict[str, Any], List[Any]], path: str) -> Any: + """Извлекает значение из вложенной JSON-структуры по пути. + + Args: + data: Словарь или список с данными. + path: Путь в формате "key.subkey[0].field" или "items[0].name". + + Returns: + Значение по указанному пути, или None если путь не существует. + + Examples: + >>> get_json_value({"a": {"b": [1, 2]}}, "a.b[0]") + 1 + >>> get_json_value({"answer": [{"text": "да"}]}, "answer[0].text") + 'да' + """ + try: + # Разбираем путь: key, [0], .subkey + tokens = re.findall(r"(?:^|\.)(\w+)|\[(\d+)\]", path) + current: Any = data + + for key, index in tokens: + if key: + if isinstance(current, dict): + current = current.get(key) + else: + return None + elif index is not None: + idx = int(index) + if isinstance(current, list) and idx < len(current): + current = current[idx] + else: + return None + + return current + except (KeyError, IndexError, TypeError, AttributeError): + return None + + +def check_expectation( + data: Dict[str, Any], + path: str, + condition: str, + value: Any = None, +) -> Tuple[bool, str]: + """Проверяет одно условие ожидания для JSON-ответа. + + Args: + data: JSON-ответ от SES API. + path: Путь к полю в JSON (формат "answer[0].messages[0]"). + condition: Тип проверки: + - "equals": поле == value + - "contains": поле содержит value + - "not_contains": поле не содержит value + - "not_empty": поле не пустое + - "regex": поле соответствует регулярному выражению в value + - "exists": поле существует и не None + value: Значение для сравнения (для equals/contains/not_contains/regex). + + Returns: + (успех, сообщение_об_ошибке). При успехе сообщение пустое. + """ + actual = get_json_value(data, path) + + if condition == "exists": + if actual is not None: + return True, "" + return False, f"Поле '{path}' не найдено в ответе" + + if condition == "not_empty": + if actual is not None and actual != "" and actual != [] and actual != {}: + return True, "" + return False, f"Поле '{path}' пустое: {actual!r}" + + if actual is None: + return False, f"Поле '{path}' не найдено в ответе" + + if isinstance(actual, str) and value is not None: + actual_str = actual.lower() + if not isinstance(value, str): + value = str(value) + value_str = value.lower() + + if condition == "contains": + if value_str in actual_str: + return True, "" + return False, ( + f"Поле '{path}' = {actual_str[:100]!r} " + f"не содержит {value_str[:100]!r}" + ) + + if condition == "not_contains": + if value_str not in actual_str: + return True, "" + return False, ( + f"Поле '{path}' = {actual_str[:100]!r} " + f"содержит запрещённое {value_str[:100]!r}" + ) + + if condition == "regex": + try: + if re.search(str(value), actual): + return True, "" + return False, ( + f"Поле '{path}' = {actual[:100]!r} " + f"не соответствует regex {value!r}" + ) + except re.error as e: + return False, f"Ошибка regex '{value}': {e}" + + if condition == "equals": + # Приводим к строке для сравнения + actual_conv = actual + value_conv = value + if not isinstance(actual_conv, type(value_conv)): + # Пробуем сравнить через строковое представление + if str(actual_conv) == str(value_conv): + return True, "" + return False, ( + f"Поле '{path}' = {actual_conv!r} " + f"не равно {value_conv!r}" + ) + if actual_conv == value_conv: + return True, "" + return False, ( + f"Поле '{path}' = {actual_conv!r} " + f"не равно {value_conv!r}" + ) + + return False, f"Неизвестное условие проверки: {condition}" + + +def check_all_expectations( + data: Dict[str, Any], + expectations: List[Dict[str, Any]], +) -> Tuple[bool, List[str]]: + """Проверяет все ожидания для одного шага сценария. + + Args: + data: JSON-ответ от SES API. + expectations: Список условий — каждый dict с ключами + path, condition_type_1, condition_type_2, ... + + Returns: + (все_прошли, список_сообщений_об_ошибках). + """ + errors: List[str] = [] + + for exp in expectations: + path = exp.get("path", "") + if not path: + errors.append("Ожидание без указания path") + continue + + for condition, value in exp.items(): + if condition == "path": + continue # path — не условие + + ok, msg = check_expectation(data, path, condition, value) + if not ok: + errors.append(msg) + + return len(errors) == 0, errors + + +@contextmanager +def timer() -> Generator[Any, None, None]: + """Контекстный менеджер для измерения времени выполнения блока. + + Yields: + Словарь, в который будет записано прошедшее время (в секундах) + по ключам 'start' и 'elapsed'. + + Usage: + >>> with timer() as t: + ... time.sleep(0.1) + >>> print(f"{t['elapsed']:.2f}s") + """ + result: Dict[str, float] = {"start": time.monotonic()} + try: + yield result + finally: + result["elapsed"] = time.monotonic() - result["start"] diff --git a/src/ses_monitor/zabbix.py b/src/ses_monitor/zabbix.py new file mode 100644 index 0000000..c9323f8 --- /dev/null +++ b/src/ses_monitor/zabbix.py @@ -0,0 +1,225 @@ +""" +Отправка метрик в Zabbix. + +Два режима: + - sender: zabbix_sender (активный push). + - userparameter: запись JSON-файла + конфиг для Zabbix агента (пассивный опрос). +""" + +from __future__ import annotations + +import asyncio +import concurrent.futures +import json +import logging +import os +import subprocess +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, List, Optional + +from ses_monitor.config import ZabbixConfig, ZabbixMode + +log = logging.getLogger("ses_monitor.zabbix") + + +@dataclass +class ZabbixMetric: + """Одна метрика для Zabbix.""" + + host: str + key: str + value: str + + def to_line(self) -> str: + """Форматирует метрику в строку для zabbix_sender.""" + return f"{self.host} {self.key} {self.value}" + + +class ZabbixSender: + """Асинхронная обёртка над zabbix_sender. + + Отправляет метрики через stdin утилиты zabbix_sender. + Использует run_in_executor + subprocess.run (синхронный), чтобы избежать + проблем с asyncio child watcher на Python 3.7. + """ + + def __init__(self, config: ZabbixConfig): + self._config = config + self._metrics: List[ZabbixMetric] = [] + self._executor: Optional[concurrent.futures.ThreadPoolExecutor] = None + + @property + def enabled(self) -> bool: + """Включена ли отправка в Zabbix.""" + return self._config.enabled + + def add(self, key: str, value: Any) -> None: + """Добавляет метрику в буфер отправки. + + Args: + key: Ключ метрики (например, "ses.api.liveness"). + value: Значение. Числа и булевы конвертируются в строку. + """ + str_value = str(value) if not isinstance(value, bool) else str(int(value)) + metric = ZabbixMetric( + host=self._config.host, + key=key, + value=str_value, + ) + self._metrics.append(metric) + + def _run_sender(self, input_data: str) -> bool: + """Синхронный запуск zabbix_sender (выполняется в thread pool).""" + try: + proc = subprocess.run( + [ + self._config.sender_path, + "-z", self._config.server, + "-p", str(self._config.port), + "-s", self._config.host, + "-i", "-", + ], + input=input_data.encode("utf-8"), + capture_output=True, + timeout=30, + ) + + if proc.returncode != 0: + log.error( + "zabbix_sender завершился с кодом %d: %s", + proc.returncode, + proc.stderr.decode("utf-8", errors="replace").strip(), + ) + return False + + output = proc.stdout.decode("utf-8", errors="replace") + log.debug( + "Zabbix: отправлено метрик. Ответ: %s", + output.strip(), + ) + return True + + except FileNotFoundError: + log.error( + "zabbix_sender не найден по пути %s. Установите пакет.", + self._config.sender_path, + ) + return False + except subprocess.TimeoutExpired: + log.error("zabbix_sender: таймаут отправки (30с)") + return False + except Exception as e: + log.error("zabbix_sender: ошибка отправки: %s", e) + return False + + async def flush(self) -> bool: + """Отправляет накопленные метрики в Zabbix. + + Returns: + True если отправка успешна, False при ошибке. + """ + if not self._metrics: + return True + + if not self._config.enabled: + self._metrics.clear() + return True + + lines = "\n".join(m.to_line() for m in self._metrics) + "\n" + self._metrics.clear() + + loop = asyncio.get_running_loop() + return await loop.run_in_executor(self._executor, self._run_sender, lines) + + +class UserParameterWriter: + """Запись метрик в JSON для Zabbix UserParameter. + + Пишет атомарный JSON во временный файл с последующей заменой. + Zabbix агент читает файл через UserParameter-скрипт zbx_get.py. + """ + + def __init__(self, config: ZabbixConfig): + self._config = config + self._metrics: Dict[str, str] = {} + self._executor: Optional[concurrent.futures.ThreadPoolExecutor] = None + + @property + def enabled(self) -> bool: + """Включена ли запись метрик.""" + return self._config.enabled + + def add(self, key: str, value: Any) -> None: + """Добавляет метрику в буфер. + + Args: + key: Ключ метрики (например, "ses.api.liveness"). + value: Значение. Конвертируется в строку. + """ + str_value = str(value) if not isinstance(value, bool) else str(int(value)) + self._metrics[key] = str_value + + async def flush(self) -> bool: + """Сбрасывает буфер метрик в JSON-файл и обновляет конфиг агента. + + Returns: + True если запись успешна, False при ошибке. + """ + if not self._metrics: + return True + + if not self._config.enabled: + self._metrics.clear() + return True + + data = dict(self._metrics) + self._metrics.clear() + + loop = asyncio.get_running_loop() + ok = await loop.run_in_executor( + self._executor, self._write_json, data + ) + return ok + + def _write_json(self, data: Dict[str, str]) -> bool: + """Атомарная запись JSON-файла с метриками.""" + metrics_path = Path(self._config.metrics_file) + try: + metrics_path.parent.mkdir(parents=True, exist_ok=True) + tmp = tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=str(metrics_path.parent), + prefix=".metrics_", + suffix=".tmp", + delete=False, + ) + tmp_path = tmp.name + try: + json.dump(data, tmp, ensure_ascii=False, indent=2) + tmp.write("\n") + tmp.close() + os.replace(tmp_path, str(metrics_path)) + os.chmod(str(metrics_path), 0o644) + except Exception: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + raise + return True + except Exception as e: + log.error("Ошибка записи JSON-файла метрик: %s", e) + return False + + def generate_agentd_config(self) -> str: + """Генерирует содержимое конфигурационного файла UserParameter. + + Returns: + Строка с конфигурацией для Zabbix агента. + """ + return ( + "# Автоматически сгенерировано ses-monitor.\n" + "# Требуется: Include=/opt/ses-monitor/config/zbx_agentd_ses.conf\n" + f"UserParameter=ses.metrics,/opt/ses-monitor/scripts/zbx_get.py\n" + ) diff --git a/systemd/ses-monitor.service b/systemd/ses-monitor.service new file mode 100644 index 0000000..c5afd69 --- /dev/null +++ b/systemd/ses-monitor.service @@ -0,0 +1,41 @@ +[Unit] +Description=SES Monitor — непрервыный мониторинг Service Engine Server +Documentation=https://github.com/your-org/ses-monitor +After=network.target network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=root +Group=root + +# Рабочий каталог проекта +WorkingDirectory=/opt/ses-monitor + +# Запуск через venv +ExecStart=/opt/ses-monitor/venv/bin/python3 -m ses_monitor + +# Перезапуск при падении +Restart=always +RestartSec=10 + +# Лимиты +LimitNOFILE=65536 +LimitNPROC=4096 + +# Логирование — systemd journal +StandardOutput=journal +StandardError=journal +SyslogIdentifier=ses-monitor + +# Окружение +Environment=PYTHONUNBUFFERED=1 +Environment=PYTHONPATH=/opt/ses-monitor + +# Защита (раскомментировать после тестирования) +# ProtectSystem=full +# ProtectHome=true +# NoNewPrivileges=true + +[Install] +WantedBy=multi-user.target diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_scenario_runner.py b/tests/test_scenario_runner.py new file mode 100644 index 0000000..e4f0de9 --- /dev/null +++ b/tests/test_scenario_runner.py @@ -0,0 +1,694 @@ +""" +Тесты для основных компонентов системы мониторинга SES. + +Проверяют: + - Загрузку конфигурации из YAML + - Извлечение JSON-значений по пути + - Проверку условий ожидания (expect) + - Парсинг ответов SES API + - Выполнение шагов сценария (с мок-клиентом) +""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest +import yaml + +from ses_monitor.config import ( + ScenarioConfig, + ScenarioStep, + load_scenarios, +) +from ses_monitor.utils import ( + check_all_expectations, + check_expectation, + get_json_value, + setup_logging, + timer, +) +from ses_monitor.session_client import AskResponse, HostnameResponse, LicenseResponse + + +# ============================================================================= +# Тесты: config.py +# ============================================================================= + +class TestConfigLoading: + """Тесты загрузки конфигурации.""" + + def test_load_scenarios(self): + """Проверяет загрузку YAML-сценариев из каталога.""" + with tempfile.TemporaryDirectory() as tmpdir: + # Создаём тестовый сценарий + scenario_yaml = { + "name": "Тестовый сценарий", + "description": "Проверка загрузки", + "enabled": True, + "interval": 120, + "steps": [ + { + "send": "привет", + "expect": [ + {"path": "error", "equals": 0}, + {"path": "answer[0].messages[0]", "contains": "при"}, + ], + } + ], + } + with open(Path(tmpdir) / "test_scenario.yaml", "w") as f: + yaml.dump(scenario_yaml, f) + + scenarios = load_scenarios(tmpdir) + assert len(scenarios) == 1 + s = scenarios[0] + assert s.name == "Тестовый сценарий" + assert s.interval == 120 + assert len(s.steps) == 1 + assert s.steps[0].send == "привет" + assert len(s.steps[0].expect) == 2 + + def test_load_scenarios_disabled(self): + """Проверяет что отключённые сценарии не загружаются.""" + with tempfile.TemporaryDirectory() as tmpdir: + scenario_yaml = { + "name": "Отключённый", + "enabled": False, + "steps": [], + } + with open(Path(tmpdir) / "disabled.yaml", "w") as f: + yaml.dump(scenario_yaml, f) + + scenarios = load_scenarios(tmpdir) + assert len(scenarios) == 0 + + def test_load_scenarios_empty_dir(self): + """Проверяет загрузку из пустого каталога.""" + with tempfile.TemporaryDirectory() as tmpdir: + scenarios = load_scenarios(tmpdir) + assert scenarios == [] + + +# ============================================================================= +# Тесты: utils.py — get_json_value +# ============================================================================= + +class TestGetJsonValue: + """Тесты извлечения значений из JSON по пути.""" + + def test_simple_key(self): + data = {"error": 0, "message": "ok"} + assert get_json_value(data, "error") == 0 + assert get_json_value(data, "message") == "ok" + + def test_nested_key(self): + data = {"answer": [{"messages": ["здравствуйте"]}]} + assert get_json_value(data, "answer[0].messages[0]") == "здравствуйте" + + def test_missing_key_returns_none(self): + data = {"error": 0} + assert get_json_value(data, "nonexistent") is None + assert get_json_value(data, "answer[0].text") is None + + def test_array_index_out_of_bounds(self): + data = {"items": [1, 2]} + assert get_json_value(data, "items[5]") is None + + def test_deeply_nested(self): + data = { + "request": { + "data": [ + { + "reply": { + "confidence": 0.95, + } + } + ] + } + } + assert get_json_value(data, "request.data[0].reply.confidence") == 0.95 + + def test_non_dict_data(self): + assert get_json_value([1, 2, 3], "items[0]") is None + + +# ============================================================================= +# Тесты: utils.py — check_expectation +# ============================================================================= + +class TestCheckExpectation: + """Тесты проверки условий ожидания.""" + + def test_equals_int(self): + ok, msg = check_expectation({"error": 0}, "error", "equals", 0) + assert ok + assert msg == "" + + def test_equals_fail(self): + ok, msg = check_expectation({"error": 1}, "error", "equals", 0) + assert not ok + assert "не равно" in msg + + def test_contains_str(self): + data = {"answer": [{"messages": ["введите фамилию"]}]} + ok, msg = check_expectation( + data, "answer[0].messages[0]", "contains", "фамилию" + ) + assert ok + + def test_contains_case_insensitive(self): + data = {"answer": [{"messages": ["Введите ФАМИЛИЮ"]}]} + ok, msg = check_expectation( + data, "answer[0].messages[0]", "contains", "фамилию" + ) + assert ok + + def test_not_contains(self): + data = {"answer": [{"messages": ["всё хорошо"]}]} + ok, msg = check_expectation( + data, "answer[0].messages[0]", "not_contains", "ошибка" + ) + assert ok + + def test_not_contains_fail(self): + data = {"answer": [{"messages": ["произошла ошибка"]}]} + ok, msg = check_expectation( + data, "answer[0].messages[0]", "not_contains", "ошибка" + ) + assert not ok + + def test_not_empty(self): + ok, msg = check_expectation( + {"text": "привет"}, "text", "not_empty", True + ) + assert ok + + def test_not_empty_fail(self): + ok, msg = check_expectation({}, "text", "not_empty", True) + assert not ok + + def test_not_empty_empty_string(self): + ok, msg = check_expectation({"text": ""}, "text", "not_empty", True) + assert not ok + + def test_regex_match(self): + data = {"phone": "+7-913-123-45-67"} + ok, msg = check_expectation(data, "phone", "regex", r"\+7[\d\-]+") + assert ok + + def test_regex_no_match(self): + data = {"phone": "abc"} + ok, msg = check_expectation(data, "phone", "regex", r"^\+\d+$") + assert not ok + + def test_exists(self): + ok, msg = check_expectation({"name": "test"}, "name", "exists") + assert ok + + def test_exists_fail(self): + ok, msg = check_expectation({}, "name", "exists") + assert not ok + + def test_missing_path(self): + ok, msg = check_expectation({}, "nonexistent[0].field", "equals", 0) + assert not ok + assert "не найдено" in msg or "не найд" in msg + + +# ============================================================================= +# Тесты: utils.py — check_all_expectations +# ============================================================================= + +class TestCheckAllExpectations: + """Тесты групповой проверки ожиданий.""" + + def test_all_pass(self): + data = {"error": 0, "answer": [{"messages": ["здравствуйте"]}]} + expectations = [ + {"path": "error", "equals": 0}, + {"path": "answer[0].messages[0]", "contains": "здрав"}, + ] + passed, errors = check_all_expectations(data, expectations) + assert passed + assert errors == [] + + def test_one_fails(self): + data = {"error": 1} + expectations = [ + {"path": "error", "equals": 0}, + {"path": "answer", "not_empty": True}, + ] + passed, errors = check_all_expectations(data, expectations) + assert not passed + assert len(errors) >= 1 # Первое или оба могут упасть + + def test_no_expectations(self): + passed, errors = check_all_expectations({}, []) + assert passed + assert errors == [] + + def test_missing_path_in_expectation(self): + passed, errors = check_all_expectations({}, [{"contains": "test"}]) + assert not passed + assert any("path" in e for e in errors) + + +# ============================================================================= +# Тесты: utils.py — timer +# ============================================================================= + +class TestTimer: + """Тесты контекстного менеджера timer.""" + + def test_measures_time(self): + import time + + with timer() as t: + time.sleep(0.05) + + assert "start" in t + assert "elapsed" in t + assert t["elapsed"] >= 0.04 # Допускаем небольшую погрешность + + +# ============================================================================= +# Тесты: session_client.py — парсинг ответов +# ============================================================================= + +class TestAskResponse: + """Тесты парсинга ответа POST /ses/ask.""" + + def test_from_valid_json(self): + data = { + "error": 0, + "question": "привет", + "answer": [{"messages": ["здравствуйте"], "voice": "Коля"}], + "session": "monitoring_test_123", + "uuid": "abc-def-123", + } + resp = AskResponse.from_json(data) + assert resp.error == 0 + assert resp.question == "привет" + assert len(resp.answer) == 1 + assert resp.session == "monitoring_test_123" + assert resp.uuid == "abc-def-123" + assert resp.raw == data + + def test_from_error_json(self): + data = {"error": 1, "message": "robot not found"} + resp = AskResponse.from_json(data) + assert resp.error == 1 + assert resp.question == "" + assert resp.answer == [] + + +class TestHostnameResponse: + """Тесты парсинга ответа GET /ses/hostname.""" + + def test_from_valid_json(self): + data = { + "error": 0, + "message": "Success", + "hostname": "ses-host.example.com", + "version": "5.046 release", + } + resp = HostnameResponse.from_json(data) + assert resp.error == 0 + assert resp.hostname == "ses-host.example.com" + assert "5.046" in resp.version + + +class TestLicenseResponse: + """Тесты парсинга ответа GET /license/check.""" + + def test_from_valid_json(self): + data = { + "error": 0, + "uuid": "398e8f18a83a6535d2bade81e9ee52a3", + "remaining_licenses": "infinity", + "threshold": 0, + } + resp = LicenseResponse.from_json(data) + assert resp.error == 0 + assert resp.remaining_licenses == "infinity" + + +# ============================================================================= +# Тесты: scenario_runner.py — с мок-клиентом +# ============================================================================= + +class TestScenarioRunner: + """Тесты выполнения сценариев с мок-клиентом.""" + + @pytest.mark.asyncio + async def test_run_simple_scenario(self): + """Проверяет выполнение простого сценария с успешным ответом.""" + from ses_monitor.config import SESConfig + from ses_monitor.scenario_runner import ScenarioRunner + + scenario = ScenarioConfig( + name="test", + interval=10, + steps=[ + ScenarioStep( + send="привет", + expect=[{"path": "error", "equals": 0}], + ), + ], + ) + + runner = ScenarioRunner(SESConfig(), [scenario]) + + # Мок HTTP-клиента + mock_client = AsyncMock() + mock_client.ask.return_value = AskResponse.from_json({ + "error": 0, + "question": "привет", + "answer": [{"messages": ["здравствуйте"], "voice": "Коля"}], + "session": "monitoring_test_1", + "uuid": "test-uuid", + }) + + results = await runner.run_all(mock_client) + assert len(results) == 1 + assert results[0].passed + assert len(results[0].step_results) == 1 + assert results[0].step_results[0].passed + + @pytest.mark.asyncio + async def test_run_scenario_api_error(self): + """Проверяет что сценарий фейлится при error != 0.""" + from ses_monitor.config import SESConfig + from ses_monitor.scenario_runner import ScenarioRunner + + scenario = ScenarioConfig( + name="test", + interval=10, + steps=[ + ScenarioStep( + send="привет", + expect=[{"path": "error", "equals": 0}], + ), + ], + ) + + runner = ScenarioRunner(SESConfig(), [scenario]) + + mock_client = AsyncMock() + mock_client.ask.return_value = AskResponse.from_json({ + "error": 1, + "question": "привет", + "answer": [], + "session": "", + "uuid": "", + }) + + results = await runner.run_all(mock_client) + assert len(results) == 1 + assert not results[0].passed + + @pytest.mark.asyncio + async def test_run_scenario_expectation_fail(self): + """Проверяет что сценарий фейлится при несовпадении ожиданий.""" + from ses_monitor.config import SESConfig + from ses_monitor.scenario_runner import ScenarioRunner + + scenario = ScenarioConfig( + name="test", + interval=10, + steps=[ + ScenarioStep( + send="привет", + expect=[{"path": "answer[0].messages[0]", "contains": "ОЖИДАЕМЫЙ_ТЕКСТ"}], + ), + ], + ) + + runner = ScenarioRunner(SESConfig(), [scenario]) + + mock_client = AsyncMock() + mock_client.ask.return_value = AskResponse.from_json({ + "error": 0, + "question": "привет", + "answer": [{"messages": ["другой текст"], "voice": "Коля"}], + "session": "test_1", + "uuid": "test-uuid", + }) + + results = await runner.run_all(mock_client) + assert len(results) == 1 + assert not results[0].passed + + +# ============================================================================= +# Тесты: aggregate_scenario_results +# ============================================================================= + +class TestAggregateScenarioResults: + """Тесты агрегации результатов сценариев.""" + + def _make_passed(self, name: str = "test", time_ms: float = 100.0): + """Создаёт успешный результат.""" + from ses_monitor.scenario_runner import ScenarioResult + return ScenarioResult( + scenario_name=name, + passed=True, + total_time_ms=time_ms, + ) + + def _make_failed(self, name: str = "test", time_ms: float = 100.0): + """Создаёт упавший результат.""" + from ses_monitor.scenario_runner import ScenarioResult + return ScenarioResult( + scenario_name=name, + passed=False, + total_time_ms=time_ms, + error_message="тестовая ошибка", + ) + + def test_empty_list_returns_zeroes(self): + """Пустой список — ok=1, всё нули.""" + from ses_monitor.scenario_runner import aggregate_scenario_results + agg = aggregate_scenario_results([]) + assert agg["ok"] == 1 + assert agg["total"] == 0 + assert agg["failed"] == 0 + assert agg["avg_time"] == 0 + assert agg["max_time"] == 0 + assert agg["error"] == "" + + def test_all_pass(self): + """Все сценарии прошли — ok=1, failed=0, error пуст.""" + from ses_monitor.scenario_runner import aggregate_scenario_results + results = [ + self._make_passed("a", 100.0), + self._make_passed("b", 200.0), + self._make_passed("c", 300.0), + ] + agg = aggregate_scenario_results(results) + assert agg["ok"] == 1 + assert agg["total"] == 3 + assert agg["failed"] == 0 + assert agg["error"] == "" + assert agg["avg_time"] == 200 # (100+200+300)/3 + assert agg["max_time"] == 300 + + def test_some_fail(self): + """Часть сценариев упала — ok=0, failed=N, error содержит счётчик.""" + from ses_monitor.scenario_runner import aggregate_scenario_results + results = [ + self._make_passed("a", 100.0), + self._make_failed("b", 200.0), + self._make_failed("c", 300.0), + ] + agg = aggregate_scenario_results(results) + assert agg["ok"] == 0 + assert agg["total"] == 3 + assert agg["failed"] == 2 + assert agg["error"] == "2 сценариев упали" + assert agg["avg_time"] == 200 + assert agg["max_time"] == 300 + + def test_all_fail(self): + """Все сценарии упали — ok=0, failed=total.""" + from ses_monitor.scenario_runner import aggregate_scenario_results + results = [ + self._make_failed("a", 50.0), + self._make_failed("b", 150.0), + ] + agg = aggregate_scenario_results(results) + assert agg["ok"] == 0 + assert agg["total"] == 2 + assert agg["failed"] == 2 + assert agg["error"] == "2 сценариев упали" + + def test_avg_time_calculation(self): + """Среднее время считается корректно (включая упавшие).""" + from ses_monitor.scenario_runner import aggregate_scenario_results + results = [ + self._make_passed("fast", 50.0), + self._make_failed("slow", 450.0), + ] + agg = aggregate_scenario_results(results) + assert agg["avg_time"] == 250 # (50+450)/2 + assert agg["max_time"] == 450 + + def test_max_time_with_single_result(self): + """Один сценарий — avg_time == max_time.""" + from ses_monitor.scenario_runner import aggregate_scenario_results + results = [self._make_passed("x", 123.0)] + agg = aggregate_scenario_results(results) + assert agg["ok"] == 1 + assert agg["total"] == 1 + assert agg["failed"] == 0 + assert agg["avg_time"] == 123 + assert agg["max_time"] == 123 + + def test_rounding_behaviour(self): + """Время округляется до целых миллисекунд.""" + from ses_monitor.scenario_runner import aggregate_scenario_results + results = [ + self._make_passed("a", 10.6), + self._make_passed("b", 10.3), + ] + agg = aggregate_scenario_results(results) + assert agg["avg_time"] == 10 # round(20.9/2) = round(10.45) = 10 + assert agg["max_time"] == 11 # round(10.6) = 11 + + +@pytest.mark.asyncio +class TestUserParameterWriter: + """Тесты для UserParameterWriter.""" + + async def test_write_json_file(self, tmp_path): + """Проверяет что JSON-файл пишется атомарно и читается.""" + from ses_monitor.zabbix import UserParameterWriter + from ses_monitor.config import ZabbixConfig, ZabbixMode + + metrics_file = str(tmp_path / "metrics.json") + config = ZabbixConfig( + enabled=True, + mode=ZabbixMode.USERPARAMETER, + metrics_file=metrics_file, + ) + writer = UserParameterWriter(config) + + writer.add("ses.api.liveness", 1) + writer.add("ses.api.response_time", 42) + writer.add("ses.license.ok", True) + + ok = await writer.flush() + assert ok, "flush должен вернуть True" + + import json + with open(metrics_file, "r") as f: + data = json.load(f) + + assert data["ses.api.liveness"] == "1" + assert data["ses.api.response_time"] == "42" + assert data["ses.license.ok"] == "1" + + async def test_disabled_no_write(self, tmp_path): + """Проверяет что при отключённом Zabbix файл не пишется.""" + from ses_monitor.zabbix import UserParameterWriter + from ses_monitor.config import ZabbixConfig, ZabbixMode + + metrics_file = str(tmp_path / "metrics.json") + config = ZabbixConfig( + enabled=False, + mode=ZabbixMode.USERPARAMETER, + metrics_file=metrics_file, + ) + writer = UserParameterWriter(config) + + writer.add("ses.test", 1) + ok = await writer.flush() + assert ok + + import os + assert not os.path.exists(metrics_file), "файл не должен создаваться" + + async def test_empty_no_write(self, tmp_path): + """Проверяет что при пустом буфере файл не пишется.""" + from ses_monitor.zabbix import UserParameterWriter + from ses_monitor.config import ZabbixConfig, ZabbixMode + + metrics_file = str(tmp_path / "metrics.json") + config = ZabbixConfig( + enabled=True, + mode=ZabbixMode.USERPARAMETER, + metrics_file=metrics_file, + ) + writer = UserParameterWriter(config) + + ok = await writer.flush() + assert ok + + import os + assert not os.path.exists(metrics_file), "файл не должен создаваться" + + async def test_accumulates_in_single_cycle(self, tmp_path): + """Проверяет что все метрики одного цикла попадают в файл.""" + from ses_monitor.zabbix import UserParameterWriter + from ses_monitor.config import ZabbixConfig, ZabbixMode + + metrics_file = str(tmp_path / "metrics.json") + config = ZabbixConfig( + enabled=True, + mode=ZabbixMode.USERPARAMETER, + metrics_file=metrics_file, + ) + writer = UserParameterWriter(config) + + writer.add("ses.metric.one", 1) + writer.add("ses.metric.two", 2) + await writer.flush() + + import json + with open(metrics_file, "r") as f: + data = json.load(f) + + assert data.get("ses.metric.one") == "1" + assert data.get("ses.metric.two") == "2" + + async def test_generate_agentd_config(self): + """Проверяет генерацию конфига Zabbix агента.""" + from ses_monitor.zabbix import UserParameterWriter + from ses_monitor.config import ZabbixConfig, ZabbixMode + + config = ZabbixConfig(mode=ZabbixMode.USERPARAMETER) + writer = UserParameterWriter(config) + cfg = writer.generate_agentd_config() + + assert "UserParameter=ses.metrics," in cfg + assert "zbx_get.py" in cfg + + async def test_atomic_write(self, tmp_path): + """Проверяет что запись атомарна (файл не остаётся в повреждённом состоянии).""" + from ses_monitor.zabbix import UserParameterWriter + from ses_monitor.config import ZabbixConfig, ZabbixMode + + metrics_file = str(tmp_path / "metrics.json") + config = ZabbixConfig( + enabled=True, + mode=ZabbixMode.USERPARAMETER, + metrics_file=metrics_file, + ) + writer = UserParameterWriter(config) + + for i in range(100): + writer.add(f"ses.test.{i}", i) + + ok = await writer.flush() + assert ok + + import json + with open(metrics_file, "r") as f: + data = json.load(f) + + assert len(data) == 100 + assert data["ses.test.99"] == "99" diff --git a/zabbix/zbx_ses_monitor.conf b/zabbix/zbx_ses_monitor.conf new file mode 100644 index 0000000..260a8b9 --- /dev/null +++ b/zabbix/zbx_ses_monitor.conf @@ -0,0 +1,10 @@ +# ============================================================================= +# Резервный вариант: UserParameter для Zabbix Agent (пассивный режим). +# ============================================================================= +# Используется если zabbix_sender недоступен. +# Поместить в /etc/zabbix/zabbix_agentd.d/ses_monitor.conf +# и перезапустить zabbix-agent: systemctl restart zabbix-agent +# +# Скрипт ses_monitor_zabbix_agent.py должен быть установлен в +# /opt/ses-monitor/scripts/ses_monitor_zabbix_agent.py +# ============================================================================= diff --git a/zabbix/zbx_ses_template.xml b/zabbix/zbx_ses_template.xml new file mode 100644 index 0000000..0dfd294 --- /dev/null +++ b/zabbix/zbx_ses_template.xml @@ -0,0 +1,465 @@ + + + 6.0 + 2026-06-19T12:00:00Z + + + Templates + + + + + + diff --git a/zabbix/zbx_ses_template_5.0.xml b/zabbix/zbx_ses_template_5.0.xml new file mode 100644 index 0000000..510ea67 --- /dev/null +++ b/zabbix/zbx_ses_template_5.0.xml @@ -0,0 +1,212 @@ + + + 5.0 + 2026-06-19T12:00:00Z + + + Templates + + + + + + diff --git a/zabbix/zbx_ses_template_5.0_up.xml b/zabbix/zbx_ses_template_5.0_up.xml new file mode 100644 index 0000000..563eb3e --- /dev/null +++ b/zabbix/zbx_ses_template_5.0_up.xml @@ -0,0 +1,261 @@ + + + 5.0 + 2026-06-19T12:00:00Z + + + Templates + + + + + +