zont/zont-exporter
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix
Build and Push / maintence (push) Successful in 1s
Details
Build and Push / buildApp (push) Failing after 8s
Details
Build and Push / pushApp (push) Skipped
Details

Browse Source
This commit is contained in:
energys
2026-09-25 20:58:42 +09:00
parent ef6356781d
commit eb9b49b3e1
2 changed files with 139 additions and 606 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zont-exporter/Dockerfile
+9 -27
View File
@@ -1,33 +1,15 @@
# syntax=docker/dockerfile:1.7
FROM python:3.12-slim AS base
# Меньше мусора в логах python
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1
# Непривилегированный пользователь
RUN groupadd --system --gid 1001 app \
&& useradd --system --uid 1001 --gid app --create-home app
FROM python:3.11-slim
WORKDIR /app
# Ставим зависимости отдельным слоем (кэшируется)
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
# Установка зависимостей
RUN pip install --no-cache-dir requests prometheus-client
# Копируем код
COPY --chown=app:app zont_exporter.py ./
# Копирование кода
COPY exporter.py .
USER app
# Экспорт порта Prometheus
EXPOSE 9101
EXPOSE 9877
# Healthcheck: сам /metrics должен отвечать 200
HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \
CMD python -c "import urllib.request,sys; \
sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:9877/metrics', timeout=4).status==200 else 1)"
ENTRYPOINT ["python", "zont_exporter.py"]
# Запуск
CMD ["python", "exporter.py"]
zont-exporter/zont_exporter.py
+130 -579
View File
@@ -1,4 +1,3 @@
#!/usr/bin/env python3
"""
ZONT API Prometheus Exporter
@@ -17,608 +16,160 @@ Mega SX, ZTC и др.) в формате Prometheus.
EXPORTER_PORT — порт HTTP-сервера метрик (по умолчанию 9877)
LOG_LEVEL — DEBUG/INFO/WARNING/ERROR (по умолчанию INFO)
"""
import os
import sys
import time
import signal
import logging
from typing import Any, Optional
import requests
from prometheus_client import start_http_server, Gauge, Info
from prometheus_client import start_http_server, Gauge
# ---------------------------------------------------------------------------
# Настройки (через переменные окружения)
# ---------------------------------------------------------------------------
ZONT_LOGIN = os.environ.get("ZONT_LOGIN", "demo")
ZONT_PASSWORD = os.environ.get("ZONT_PASSWORD", "demo")
ZONT_DEVICE_ID = int(os.environ.get("ZONT_DEVICE_ID", "0")) # 0 = авто-выбор
ZONT_EMAIL = os.environ.get("ZONT_EMAIL", "exporter@example.com")
ZONT_TOKEN = os.environ.get("ZONT_TOKEN") or None
ZONT_POLL_INTERVAL = int(os.environ.get("ZONT_POLL_INTERVAL", "60"))
EXPORTER_PORT = int(os.environ.get("EXPORTER_PORT", "9877"))
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")
# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Конфигурация из переменных окружения
ZONT_LOGIN = os.getenv('ZONT_LOGIN')
ZONT_PASSWORD = os.getenv('ZONT_PASSWORD')
ZONT_CLIENT = os.getenv('ZONT_CLIENT', 'my_exporter@example.com')
DEVICE_ID = int(os.getenv('ZONT_DEVICE_ID'))
API_BASE = "https://my.zont.online/api"
SCRAPE_INTERVAL = int(os.getenv('SCRAPE_INTERVAL', 60))
logging.basicConfig(
level=getattr(logging, LOG_LEVEL.upper(), logging.INFO),
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
log = logging.getLogger("zont-exporter")
# Определение метрик Prometheus
TEMP_GAUGE = Gauge('zont_temperature_celsius', 'Current temperature', ['device_id', 'sensor_name', 'sensor_type'])
BOILER_WORK = Gauge('zont_boiler_work_seconds', 'Boiler work time in last minute', ['device_id'])
TARGET_TEMP = Gauge('zont_target_temperature_celsius', 'Target temperature', ['device_id'])
# Глобальная переменная для токена
auth_token = None
# ---------------------------------------------------------------------------
# Клиент ZONT API
# ---------------------------------------------------------------------------
class ZontClient:
"""Тонкая обёртка над REST API ZONT с автоматическим переполучением токена."""
def __init__(
self,
login: str,
password: str,
client_email: str,
token: Optional[str] = None,
):
self.login = login
self.password = password
self.session = requests.Session()
self.session.headers.update({
"X-ZONT-Client": client_email,
"Content-Type": "application/json",
"Accept": "application/json",
})
self.session.auth = (login, password)
self.token: Optional[str] = token
# -- Аутентификация ----------------------------------------------------
def get_authtoken(self, client_name: str = "prometheus-exporter") -> str:
"""POST /get_authtoken — получаем токен по логину/паролю."""
r = self.session.post(
def get_auth_token():
"""Получает токен аутентификации ZONT API."""
global auth_token
try:
resp = requests.post(
f"{API_BASE}/get_authtoken",
json={"client_name": client_name},
timeout=20,
auth=(ZONT_LOGIN, ZONT_PASSWORD),
headers={'X-ZONT-Client': ZONT_CLIENT},
json={"client_name": "Prometheus Exporter"},
timeout=10
)
r.raise_for_status()
data = r.json()
if not data.get("ok"):
raise RuntimeError(f"get_authtoken failed: {data}")
self.token = data["token"]
log.info("Получен новый auth-токен")
return self.token
resp.raise_for_status()
data = resp.json()
if data.get('ok'):
auth_token = data.get('token')
logging.info("Successfully obtained ZONT auth token.")
else:
logging.error(f"Failed to get auth token: {data}")
except Exception as e:
logging.error(f"Exception while getting auth token: {e}")
def _headers(self) -> dict:
h = {}
if self.token:
h["X-ZONT-Token"] = self.token
return h
# -- Универсальный POST -------------------------------------------------
def _post(self, method: str, payload: dict) -> dict:
"""POST с автоматическим переполучением токена при 403."""
if not self.token:
self.get_authtoken()
url = f"{API_BASE}/{method}"
for attempt in (1, 2):
r = self.session.post(
url, json=payload, headers=self._headers(), timeout=20
)
if r.status_code == 403 and attempt == 1:
log.warning("Токен отозван, получаю новый...")
self.get_authtoken()
continue
r.raise_for_status()
data = r.json()
if not data.get("ok"):
raise RuntimeError(f"{method} error: {data}")
return data
raise RuntimeError(f"{method}: не удалось выполнить запрос")
# -- Высокоуровневые методы --------------------------------------------
def devices(self, load_io: bool = True) -> list:
"""POST /devices — список устройств с состояниями."""
return self._post("devices", {"load_io": load_io}).get("devices", [])
def load_data(
self,
device_id: int,
data_types: list,
mintime: Optional[int] = None,
maxtime: Optional[int] = None,
) -> dict:
"""POST /load_data — история данных."""
req: dict = {"device_id": device_id, "data_types": data_types}
if mintime is not None:
req["mintime"] = mintime
if maxtime is not None:
req["maxtime"] = maxtime
return self._post("load_data", {"requests": [req]})["responses"][0]
# ---------------------------------------------------------------------------
# Метрики Prometheus
# ---------------------------------------------------------------------------
# --- Общие ---
g_device_info = Info(
"zont_device",
"Информация об устройстве ZONT",
)
# --- Термостат / отопление ---
g_target_temp = Gauge(
"zont_target_temp_celsius",
"Целевая температура термостата, °C",
)
g_pza_temp = Gauge(
"zont_pza_temp_celsius",
"Расчётная температура ПЗА, °C",
)
g_boiler_work = Gauge(
"zont_boiler_work_seconds",
"Время работы котла за последнюю минуту, сек",
)
g_power = Gauge(
"zont_power_on",
"Наличие основного питания (1=есть, 0=нет)",
)
g_boiler_fail = Gauge(
"zont_boiler_fail",
"Авария котла (1=есть, 0=нет)",
)
g_gate = Gauge(
"zont_gate_on",
"Режим комнатного термостата активен (1/0)",
)
# --- Температурные датчики (H-серия) ---
g_thermo_temp = Gauge(
"zont_thermometer_temperature_celsius",
"Показание термодатчика, °C",
["uuid", "name", "type"],
)
g_thermo_state = Gauge(
"zont_thermometer_state",
"Состояние термодатчика (1=ok, 0=malfunction)",
["uuid", "name", "type"],
)
g_thermo_last_update = Gauge(
"zont_thermometer_last_update_timestamp",
"Unix-время последнего обновления показаний датчика",
["uuid", "name"],
)
# --- OpenTherm ---
g_ot_flag = Gauge(
"zont_opentherm_flag",
"Флаги состояния OpenTherm (1=активен)",
["flag"],
)
g_ot_cs = Gauge(
"zont_opentherm_ch_setpoint_celsius",
"Расчётная температура теплоносителя отопления (OpenTherm), °C",
)
g_ot_bt = Gauge(
"zont_opentherm_boiler_temp_celsius",
"Фактическая температура теплоносителя (OpenTherm), °C",
)
g_ot_dt = Gauge(
"zont_opentherm_dhw_temp_celsius",
"Фактическая температура ГВС (OpenTherm), °C",
)
g_ot_rwt = Gauge(
"zont_opentherm_return_temp_celsius",
"Температура обратного потока (OpenTherm), °C",
)
g_ot_outside = Gauge(
"zont_opentherm_outside_temp_celsius",
"Уличная температура (OpenTherm), °C",
)
g_ot_rml = Gauge(
"zont_opentherm_flame_modulation_percent",
"Уровень модуляции горелки (OpenTherm), %",
)
g_ot_wp = Gauge(
"zont_opentherm_water_pressure_bar",
"Давление теплоносителя (OpenTherm), бар",
)
g_ot_flow = Gauge(
"zont_opentherm_dhw_flow_lpm",
"Скорость потока ГВС (OpenTherm), л/мин",
)
g_ot_fail_code = Gauge(
"zont_opentherm_fail_code",
"OEM-код аварии котла (OpenTherm)",
)
# --- Связь / контроллер ---
g_gsm_level = Gauge(
"zont_gsm_level",
"Уровень сигнала GSM [0..31]",
)
g_gsm_state = Gauge(
"zont_gsm_state",
"Состояние регистрации в GSM-сети (см. документацию)",
)
g_gsm_balance = Gauge(
"zont_gsm_balance",
"Баланс SIM-карты (в копейках/центах)",
)
g_voltage = Gauge(
"zont_supply_voltage_volts",
"Напряжение основного питания, В",
)
g_wifi_rssi = Gauge(
"zont_wifi_rssi",
"Уровень сигнала Wi-Fi (RSSI)",
)
g_status_flags = Gauge(
"zont_status_flags",
"Битовая маска состояния контроллера (см. документацию)",
)
g_internal_temp = Gauge(
"zont_internal_temperature_celsius",
"Показания встроенных температурных входов контроллера, °C",
["channel"],
)
# --- Охрана / авто (ZTC) ---
g_guard_state = Gauge(
"zont_guard_enabled",
"Охрана активна (1=enabled, 0=disabled)",
)
g_siren = Gauge(
"zont_siren_on",
"Сирена активна (1/0)",
)
g_engine_block = Gauge(
"zont_engine_block_on",
"Блокировка двигателя активна (1/0)",
)
g_webasto = Gauge(
"zont_webasto_on",
"Предпусковой подогреватель включён (1/0)",
)
g_auto_ignition = Gauge(
"zont_auto_ignition_state",
"Состояние автозапуска (0=disabled,1=enabling,2=enabled,3=webasto,4=webasto-confirmed)",
)
g_ignition = Gauge(
"zont_ignition_on",
"Зажигание включено (1/0)",
)
g_engine = Gauge(
"zont_engine_on",
"Двигатель запущен (1/0)",
)
# --- Служебные ---
g_scrape_success = Gauge(
"zont_scrape_success",
"1 если последний опрос прошёл успешно, 0 иначе",
)
g_scrape_timestamp = Gauge(
"zont_scrape_timestamp_seconds",
"Unix-время последнего успешного опроса",
)
# ---------------------------------------------------------------------------
# Хелперы
# ---------------------------------------------------------------------------
def _b(v: Any) -> Optional[int]:
"""Bool → 0/1, None → None."""
if v is None:
def fetch_last_value(dta_list):
"""Извлекает последнее значение из Delta-time array."""
if not dta_list or not isinstance(dta_list, list):
return None
return 1 if v else 0
def _guard_state_to_int(v: Optional[str]) -> Optional[int]:
if v is None:
return None
if v == "enabled":
return 1
if v in ("disabled", "enabling"):
return 0
# Последний элемент массива имеет актуальное значение
last_entry = dta_list[-1]
if isinstance(last_entry, list) and len(last_entry) >= 2:
return last_entry[1]
return None
def collect_metrics():
"""Основная функция сбора метрик."""
if not auth_token:
logging.warning("No auth token available, trying to authenticate...")
get_auth_token()
if not auth_token:
return
def _auto_ign_state_to_int(v: Optional[str]) -> Optional[int]:
"""
Кодируем строку автозапуска в число:
0 — disabled
1 — enabling
2 — enabled
3 — webasto
4 — webasto-confirmed
"""
mapping = {
"disabled": 0,
"enabling": 1,
"enabled": 2,
"webasto": 3,
"webasto-confirmed": 4,
}
if v is None:
return None
return mapping.get(v)
def _set(gauge: Gauge, value: Any, **labels: str) -> None:
"""Безопасная установка метрики."""
if value is None:
return
try:
if labels:
gauge.labels(**labels).set(float(value))
else:
gauge.set(float(value))
except Exception as e:
log.debug("Не удалось установить метрику: %s", e)
def _dta_last_value(dta: list) -> Optional[float]:
"""
Возвращает последнее значение из Delta-time Array.
Формат: [[t1, v1], [-dt, v2], ...]
"""
if not dta:
return None
try:
return dta[-1][1]
except (IndexError, TypeError):
return None
# ---------------------------------------------------------------------------
# Обновление метрик из /devices (state = поле io)
# ---------------------------------------------------------------------------
def update_from_devices(device: dict) -> None:
io = device.get("io") or {}
# --- Общая информация ---
g_device_info.info({
"id": str(device.get("id", "")),
"serial": str(device.get("serial", "")),
"name": str(device.get("name", "")),
"devtype": str(device.get("devtype", device.get("type", ""))),
})
# --- Термостат (last-boiler-state) ---
lbs = io.get("last-boiler-state")
if isinstance(lbs, dict):
_set(g_target_temp, lbs.get("target_temp"))
_set(g_pza_temp, lbs.get("pza_t"))
_set(g_boiler_work, lbs.get("boiler_work_time"))
_set(g_power, _b(lbs.get("power")))
_set(g_boiler_fail, _b(lbs.get("fail")))
_set(g_gate, _b(lbs.get("gate")))
ot = lbs.get("ot")
if isinstance(ot, dict):
# Флаги OpenTherm — сбрасываем все известные и ставим активные
known_flags = ["f", "ch", "dhw", "fl", "cl", "ch2", "di"]
active = set(ot.get("s") or [])
for flag in known_flags:
_set(g_ot_flag, 1 if flag in active else 0, flag=flag)
_set(g_ot_cs, ot.get("cs"))
_set(g_ot_bt, ot.get("bt"))
_set(g_ot_dt, ot.get("dt"))
_set(g_ot_rwt, ot.get("rwt"))
_set(g_ot_outside, ot.get("ot"))
_set(g_ot_rml, ot.get("rml"))
_set(g_ot_wp, ot.get("wp"))
_set(g_ot_flow, ot.get("fr"))
ff = ot.get("ff")
if isinstance(ff, dict):
_set(g_ot_fail_code, ff.get("c"))
# --- Температурные датчики (в настройке thermometers) ---
thermometers = device.get("thermometers") or []
# Удаляем старые серии, чтобы не тянуть мёртвые датчики
for metric in (g_thermo_temp, g_thermo_state, g_thermo_last_update):
try:
metric.clear()
except Exception:
pass
for t in thermometers:
if not t.get("is_assigned_to_slot", True):
continue
uuid = str(t.get("uuid", ""))
name = str(t.get("name", ""))
ttype = str(t.get("type", ""))
last_state = t.get("last_state")
state_val = 1 if last_state == "ok" else 0
_set(g_thermo_temp, t.get("last_value"), uuid=uuid, name=name, type=ttype)
_set(g_thermo_state, state_val, uuid=uuid, name=name, type=ttype)
_set(g_thermo_last_update, t.get("last_value_time"), uuid=uuid, name=name)
# --- Связь / контроллер ---
_set(g_gsm_level, io.get("gsm-level"))
gsm_state = io.get("gsm-state")
if isinstance(gsm_state, dict):
_set(g_gsm_level, gsm_state.get("level"))
# state приходит строкой — приводим к int по справочнику
st_map = {
"not-registered": 0,
"home-network": 1,
"searching": 2,
"rejected": 3,
"roaming": 5,
"unknown": 255,
}
st = gsm_state.get("state")
if st in st_map:
_set(g_gsm_state, st_map[st])
_set(g_voltage, io.get("voltage"))
_set(g_wifi_rssi, io.get("wifi-rssi"))
_set(g_status_flags, io.get("status-flags"))
# --- Охрана ---
_set(g_guard_state, _guard_state_to_int(io.get("guard-state")))
_set(g_siren, _b(io.get("siren")))
# --- Автомобиль (ZTC) ---
_set(g_engine_block, _b(io.get("engine-block")))
_set(g_webasto, _b(io.get("webasto")))
_set(g_ignition, _b(io.get("ignition-state")))
_set(g_engine, _b(io.get("engine-state")))
auto = io.get("auto-ignition")
if isinstance(auto, dict):
_set(g_auto_ignition, _auto_ign_state_to_int(auto.get("state")))
# ---------------------------------------------------------------------------
# Обновление метрик из /load_data (история: ztc_state, temperature, ...)
# ---------------------------------------------------------------------------
def update_from_load_data(client: ZontClient, device_id: int) -> None:
"""
Догружаем некоторые исторические метрики,
которых нет в /devices: ztc_state, temperature (для H-серии).
Берём окно за последние 10 минут.
"""
# Запрос данных thermostat_work за последний час (3600 секунд)
now = int(time.time())
mintime = now - 600
mintime = now - 3600
payload = {
"requests": [{
"device_id": DEVICE_ID,
"data_types": ["thermostat_work"],
"mintime": mintime,
"maxtime": now
}]
}
# ztc_state — есть не у всех устройств, ошибку игнорируем
try:
resp = client.load_data(
device_id=device_id,
data_types=["ztc_state"],
mintime=mintime,
maxtime=now,
resp = requests.post(
f"{API_BASE}/load_data",
headers={
'X-ZONT-Client': ZONT_CLIENT,
'X-ZONT-Token': auth_token,
'Content-Type': 'application/json'
},
json=payload,
timeout=15
)
st = resp.get("ztc_state")
if isinstance(st, dict):
_set(g_gsm_level, _dta_last_value(st.get("gsm_level") or []))
_set(g_gsm_state, _dta_last_value(st.get("gsm_state") or []))
_set(g_gsm_balance, _dta_last_value(st.get("gsm_balance") or []))
_set(g_voltage, _dta_last_value(st.get("voltage") or []))
_set(g_wifi_rssi, _dta_last_value(st.get("wifi_rssi") or []))
_set(g_status_flags, _dta_last_value(st.get("status_flags") or []))
resp.raise_for_status()
data = resp.json()
if not data.get('ok'):
logging.error(f"API returned error: {data.get('error_ui', data.get('error'))}")
# Если токен протух — сбрасываем его
if data.get('error') in ['unauthorized', 'token_expired']:
global auth_token
auth_token = None
return
responses = data.get('responses', [])
if not responses:
logging.warning("Empty responses from API.")
return
work_data = responses[0].get('thermostat_work')
if not work_data:
logging.info("No thermostat_work data in response.")
return
# 1. Время работы котла за последнюю минуту
boiler_work = fetch_last_value(work_data.get('boiler_work_time'))
if boiler_work is not None:
BOILER_WORK.labels(device_id=str(DEVICE_ID)).set(boiler_work)
# 2. Целевая температура
target_temp = fetch_last_value(work_data.get('target_temp'))
if target_temp is not None:
TARGET_TEMP.labels(device_id=str(DEVICE_ID)).set(target_temp)
# 3. Температура с датчиков (если есть в ответе)
# В thermostat_work датчики могут быть в отдельном поле или не быть вовсе.
# Здесь пример для поля temperature, если оно придёт.
temps = work_data.get('temperature')
if temps and isinstance(temps, dict):
for sensor_id, sensor_info in temps.items():
if not isinstance(sensor_info, dict): continue
name = sensor_info.get('name', sensor_id)
val = fetch_last_value(sensor_info.get('temperature'))
if val is not None:
TEMP_GAUGE.labels(
device_id=str(DEVICE_ID),
sensor_name=name,
sensor_type='thermostat_sensor'
).set(val)
logging.info(f"Metrics collected successfully for device {DEVICE_ID}")
temps = st.get("temperature") or []
if isinstance(temps, list):
for idx, series in enumerate(temps):
_set(
g_internal_temp,
_dta_last_value(series or []),
channel=str(idx),
)
except Exception as e:
log.debug("ztc_state недоступен: %s", e)
logging.error(f"Error during metric collection: {e}")
# temperature — есть у H-серии
try:
resp = client.load_data(
device_id=device_id,
data_types=["temperature"],
mintime=mintime,
maxtime=now,
)
temp = resp.get("temperature")
if isinstance(temp, dict):
# НЕ очищаем здесь — иначе затрём данные из /devices.
# Просто дописываем исторические серии с суффиксом type=history.
for uuid, item in temp.items():
if not isinstance(item, dict):
continue
_set(
g_thermo_temp,
_dta_last_value(item.get("temperature") or []),
uuid=str(uuid),
name=str(item.get("name", "")),
type="history",
)
except Exception as e:
log.debug("temperature недоступен: %s", e)
if __name__ == '__main__':
# Проверка обязательных переменных
if not all([ZONT_LOGIN, ZONT_PASSWORD, DEVICE_ID]):
logging.critical("Missing required environment variables: ZONT_LOGIN, ZONT_PASSWORD, ZONT_DEVICE_ID")
exit(1)
# Первичная аутентификация
get_auth_token()
# ---------------------------------------------------------------------------
# Основной цикл
# ---------------------------------------------------------------------------
# Запуск HTTP-сервера Prometheus на порту 9101
start_http_server(9101)
logging.info("Prometheus exporter started on port 9101")
def pick_device(client: ZontClient, preferred_id: int) -> Optional[dict]:
"""Возвращает объект устройства (первое подходящее или с заданным ID)."""
devices = client.devices(load_io=True)
if not devices:
log.warning("У пользователя нет устройств")
return None
if preferred_id:
for d in devices:
if d.get("id") == preferred_id:
return d
log.warning("Устройство id=%s не найдено, беру первое", preferred_id)
return devices[0]
def scrape_once(client: ZontClient) -> None:
device = pick_device(client, ZONT_DEVICE_ID)
if device is None:
g_scrape_success.set(0)
return
update_from_devices(device)
update_from_load_data(client, device["id"])
g_scrape_success.set(1)
g_scrape_timestamp.set(time.time())
log.info(
"Метрики обновлены (device_id=%s, name=%s)",
device.get("id"),
device.get("name"),
)
def _handle_sigterm(signum, frame): # noqa: ARG001
log.info("Получен сигнал %s, завершаюсь...", signum)
sys.exit(0)
def main() -> None:
# Корректная остановка в контейнере (docker stop → SIGTERM)
signal.signal(signal.SIGTERM, _handle_sigterm)
signal.signal(signal.SIGINT, _handle_sigterm)
log.info("Запуск ZONT exporter на порту %d", EXPORTER_PORT)
log.info(
"Логин: %s, устройство: %s, интервал: %ds",
ZONT_LOGIN,
ZONT_DEVICE_ID or "auto",
ZONT_POLL_INTERVAL,
)
client = ZontClient(ZONT_LOGIN, ZONT_PASSWORD, ZONT_EMAIL, token=ZONT_TOKEN)
# Первый опрос синхронно, чтобы /metrics сразу отдавал данные
try:
scrape_once(client)
except Exception as e:
log.exception("Первый опрос не удался: %s", e)
g_scrape_success.set(0)
start_http_server(EXPORTER_PORT)
try:
while True:
time.sleep(ZONT_POLL_INTERVAL)
try:
scrape_once(client)
except Exception as e:
log.exception("Ошибка опроса: %s", e)
g_scrape_success.set(0)
except KeyboardInterrupt:
log.info("Остановка по сигналу")
if __name__ == "__main__":
main()
# Основной цикл
while True:
collect_metrics()
time.sleep(SCRAPE_INTERVAL)