> Как реализовать сложные триггеры для мониторинга, например отправку SMS при загрузке CPU и низком потреблении памяти (Python)

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

Сложный триггер - это правило, которое срабатывает на основе комбинации метрик, а не одной. Для CPU и памяти нужно использовать оконное агрегирование (например, среднее за 5 минут) и пороговое условие вида cpu > 80% AND mem < 20%. В Python это реализуется через polling метрик, буферизацию значений и проверку условия в отдельном потоке или через асинхронный цикл. Важно учитывать дедупликацию событий и backoff, чтобы не спамить SMS.

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

Сложный триггер отличается от простого тем, что условие включает несколько метрик и временных окон. В вашем случае: "CPU загружен выше 80% в течение 5 минут И свободная память ниже 20% в течение того же окна". Это защищает от ложных срабатываний на кратковременных пиках.

Архитектурно решение состоит из трёх частей:

1. Сбор метрик - периодический опрос источника (psutil, /proc, внешний API). Интервал обычно 10-30 секунд.
2. Оконное агрегирование - хранение последних N значений каждой метрики и вычисление среднего/медианы за окно. Можно использовать collections.deque(maxlen=N) или кольцевой буфер.
3. Проверка условия и эскалация - после каждого обновления буфера проверяется комбинированное условие. Если оно истинно и прошло достаточно времени с последнего уведомления (cooldown), отправляется SMS.

Ключевой момент - состояние триггера. Нужно различать состояния: normal, firing, cooldown. Переход в firing происходит только при выполнении условия, а выход - когда условие перестало выполняться в течение определённого времени (hysteresis). Это предотвращает флаппинг.

Для отправки SMS используйте внешний сервис (Twilio, SMS.ru) через HTTP API. Сам триггер не должен блокировать сбор метрик - отправка выполняется асинхронно или в отдельном потоке.

На практике

В реальном продакшене такие триггеры обычно не пишут с нуля - используют Prometheus + Alertmanager или Zabbix. Но на собеседовании важно показать понимание механики: окна, состояния, дедупликация.

Если пишете сами, структура такая:

• Класс MetricBuffer - хранит историю значений и умеет считать среднее за окно.
• Класс Trigger - принимает буферы, условие (callable) и callback для уведомления.
• Планировщик - while True цикл с time.sleep(interval) или asyncio задача.

Для production-grade решения добавьте: персистентность состояния (чтобы перезапуск не сбросил cooldown), конфигурацию через YAML/JSON, метрики самого мониторинга (сколько раз сработал триггер).

Пример кода

PYTHON
import time
import threading
from collections import deque
from dataclasses import dataclass
from typing import Callable

@dataclass
class MetricBuffer:
    maxlen: int
    values: deque = None
    
    def __post_init__(self):
        self.values = deque(maxlen=self.maxlen)
    
    def push(self, value: float):
        self.values.append(value)
    
    def avg(self) -> float:
        if not self.values:
            return 0.0
        return sum(self.values) / len(self.values)

class ComplexTrigger:
    def __init__(
        self,
        cpu_buffer: MetricBuffer,
        mem_buffer: MetricBuffer,
        cpu_threshold: float = 80.0,
        mem_threshold: float = 20.0,
        cooldown_sec: int = 300,
        notify: Callable[[str], None] = None
    ):
        self.cpu = cpu_buffer
        self.mem = mem_buffer
        self.cpu_threshold = cpu_threshold
        self.mem_threshold = mem_threshold
        self.cooldown = cooldown_sec
        self.notify = notify or (lambda msg: print(msg))
        self._last_fire = 0.0
        self._firing = False
    
    def check(self) -> bool:
        cpu_avg = self.cpu.avg()
        mem_avg = self.mem.avg()
        condition = cpu_avg > self.cpu_threshold and mem_avg < self.mem_threshold
        
        now = time.time()
        if condition and not self._firing and (now - self._last_fire) > self.cooldown:
            self._firing = True
            self._last_fire = now
            self.notify(f"CPU={cpu_avg:.1f}% MEM={mem_avg:.1f}%")
            return True
        
        if not condition:
            self._firing = False
        
        return False

def monitor_loop(trigger: ComplexTrigger, interval: int = 10):
    # Здесь вместо заглушек - реальный сбор через psutil
    while True:
        cpu_val = get_cpu_percent()   # например, psutil.cpu_percent()
        mem_val = get_mem_percent()   # psutil.virtual_memory().percent
        trigger.cpu.push(cpu_val)
        trigger.mem.push(mem_val)
        trigger.check()
        time.sleep(interval)

# Использование
cpu_buf = MetricBuffer(maxlen=30)  # 30 * 10 сек = 5 минут
mem_buf = MetricBuffer(maxlen=30)

trigger = ComplexTrigger(
    cpu_buffer=cpu_buf,
    mem_buffer=mem_buf,
    notify=lambda msg: send_sms_via_api(msg)
)

thread = threading.Thread(target=monitor_loop, args=(trigger,), daemon=True)
thread.start()

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

Начните с формулировки проблемы: "Сложный триггер - это комбинация нескольких условий с временным окном". Затем опишите архитектуру: сбор метрик, буферизация, проверка состояния. Обязательно упомяните cooldown и hysteresis - это показывает понимание реальных проблем мониторинга.

Если спросят про масштабирование - скажите, что для многих серверов лучше использовать готовые решения (Prometheus), а самописный вариант оправдан только для специфических условий или обучения.

Не углубляйтесь в детали реализации SMS-шлюза - это не главное. Главное - логика триггера.

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

• Понимание разницы между простым и сложным условием.
• Умение работать с временными рядами и окнами.
• Знание паттернов state machine для избежания повторных срабатываний.
• Практическое знание Python: deque, потоки, асинхронность.
• Умение проектировать отказоустойчивые системы (cooldown, дедупликация).

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

• Проверка мгновенных значений вместо средних за окно - приводит к ложным срабатываниям.
• Отсутствие cooldown - SMS уходят каждые 10 секунд.
• Блокирующая отправка SMS внутри цикла сбора метрик - задерживает следующий опрос.
• Не учитывается состояние "уже уведомили" - триггер срабатывает повторно, пока условие истинно.
• Использование list вместо deque для буфера - неэффективно при добавлении и удалении.
• Игнорирование случая, когда метрики не приходят (пустой буфер) - нужно обрабатывать как отсутствие данных, а не как ноль.