> Как реализовать джобы с разной частотой выгрузки данных из Zabbix (Python)

Короткий ответ

Для выгрузки данных из Zabbix с разной частотой оптимально использовать планировщик задач (Celery beat, APScheduler) с отдельными периодическими задачами под каждый интервал. Внутри задачи - запрос к Zabbix API (history.get) за последний период, фильтрация по itemid и запись в хранилище. Ключевой момент - не плодить задачи на каждый item, а группировать выгрузки по частоте и использовать один batch-запрос. Для надёжности - идемпотентность записи и обработка частичных ошибок.

Подробное объяснение

Zabbix API отдаёт данные через метод history.get с параметрами itemids, time_from, time_till, history (тип данных: 0 - float, 1 - string, 3 - integer, 4 - text). Основная сложность - разные типы метрик требуют разных интервалов опроса: например, CPU load каждые 30 секунд, а статус сервиса - раз в час.

Архитектурно задача разбивается на три уровня:

1. Конфигурация - маппинг itemid → частота. Хранится в БД или конфиг-файле, чтобы менять без деплоя.

2. Планировщик - Celery beat с расписанием {'every': 30, 'period': 'seconds'} и т.п. Каждая задача получает список itemid для своей частоты.

3. Исполнитель - функция, которая за один вызов API тянет данные по всем itemid группы за последний интервал. Интервал вычисляется как now - max(frequency * 2, 60s) - чтобы перекрыть возможные задержки Zabbix.

Важный trade-off: чем больше интервал между выгрузками, тем больше данных за один запрос. Для history.get есть лимит на количество записей (обычно 100k), поэтому для частых метрик нужно разбивать запрос по времени на под-окна (например, по 5 минут внутри 30-минутного интервала).

Для хранения данных стоит использовать append-only таблицу с уникальным ключом (itemid, clock) - это даёт идемпотентность при повторных запусках. Если нужна агрегация - делать её отдельной задачей, не смешивая с выгрузкой.

На практике

Типичный сценарий: есть Zabbix с 5000 метрик, из них 200 требуют частоты 30 секунд, 800 - 5 минут, остальные - час. Нельзя создавать 5000 отдельных задач - это убьёт планировщик. Правильно: три задачи (по одной на частоту), каждая получает свой список itemid из конфигурации.

При реализации важно:

• Использовать time_from = now - interval * 2 - Zabbix может задерживать данные на 1-2 интервала.
• Обрабатывать ZabbixAPIException (недоступен хост, таймаут) - задача должна падать с retry, а не молча терять данные.
• Для больших объёмов - пагинация через limit и offset или разбиение по времени.
• Логировать количество полученных и записанных записей - расхождение сигнализирует о проблеме.
• Если данные нужны в реальном времени - использовать Zabbix trapper или webhook, но это уже другая архитектура.

Пример кода

PYTHON
# tasks.py (Celery)
from celery import shared_task
from celery.schedules import crontab
from zabbix_api import ZabbixAPI

# Конфигурация: itemid -> частота (секунды)
FREQ_MAP = {
    30: [1001, 1002, 1003],   # каждые 30 сек
    300: [2001, 2002],        # каждые 5 мин
    3600: [3001, 3002, 3003], # каждый час
}

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def export_zabbix_data(self, frequency: int):
    itemids = FREQ_MAP.get(frequency)
    if not itemids:
        return

    zapi = ZabbixAPI(url=settings.ZABBIX_URL, user=..., password=...)
    now = int(time.time())
    # Перекрываем двойной интервал для учёта задержек Zabbix
    time_from = now - frequency * 2

    # Для больших интервалов бьём на под-окна по 300 сек
    window = min(frequency, 300)
    for start in range(time_from, now, window):
        end = min(start + window, now)
        history = zapi.history.get({
            "itemids": itemids,
            "time_from": start,
            "time_till": end,
            "output": "extend",
            "history": 0,  # float
            "sortfield": "clock",
            "sortorder": "ASC",
            "limit": 100000,
        })
        if not history:
            continue

        # Идемпотентная запись: INSERT ... ON CONFLICT DO NOTHING
        rows = [
            (h["itemid"], int(h["clock"]), float(h["value"]))
            for h in history
        ]
        bulk_insert_ignore_conflict(rows)

# beat_schedule.py
CELERY_BEAT_SCHEDULE = {
    "export-30s": {
        "task": "tasks.export_zabbix_data",
        "schedule": 30.0,
        "args": (30,),
    },
    "export-5m": {
        "task": "tasks.export_zabbix_data",
        "schedule": 300.0,
        "args": (300,),
    },
    "export-1h": {
        "task": "tasks.export_zabbix_data",
        "schedule": crontab(minute=0),
        "args": (3600,),
    },
}

Как отвечать на собеседовании

Начни с планировщика и группировки по частоте - это главное. Затем упомяни, что данные из Zabbix API тянутся batch-запросом, а не по одному item. Обязательно скажи про идемпотентность и перекрытие интервалов. Если спросят про масштабирование - расскажи про разбиение на под-окна и лимиты API. Хорошо добавить, что для реального времени нужен другой механизм (trapper), а периодическая выгрузка - это компромисс между нагрузкой на Zabbix и свежестью данных.

Что проверяет интервьюер

• Понимание работы с внешними API: пагинация, лимиты, обработка ошибок.
• Умение проектировать фоновые задачи: планировщик, retry, идемпотентность.
• Знание trade-off между частотой опроса и нагрузкой на источник.
• Способность группировать данные и не плодить сущности без необходимости.
• Практический опыт с Celery/APScheduler и понимание, как работает Zabbix API.

Типичные ошибки

• Создание отдельной задачи на каждый item - убивает планировщик и создаёт тысячи лишних запросов к API.
• Запрос данных ровно за интервал (time_from = now - frequency) - из-за задержек Zabbix часть данных теряется.
• Отсутствие идемпотентности - повторный запуск задачи создаёт дубликаты.
• Игнорирование лимита history.get - при большом объёме данных запрос падает с ошибкой.
• Смешивание выгрузки и агрегации в одной задаче - усложняет отладку и масштабирование.
• Необработанные исключения API - задача молча падает, данные теряются без логирования.