qtile — тайловый оконный менеджер для Linux, целиком написанный на Python. Конфиг — тоже Python, с asyncio, доступом к procfs и вообще ко всему, что есть в системе. Я сижу на qtile уже почти 15 лет с одним и тем же конфигом, который потихоньку допиливаю, и что мне в нём нравится: панель оконного менеджера — удобное место для визуализации метрик, собранных из произвольных Python-скриптов.
А дело было так: у меня /tmp смонтирован как ext2 на zram — 16 гигов (да, я знаю, что это спорное решение, но для меня работает отлично — пока память не кончается). Гонял Maestro-тесты на Android-девайсе через Claude Code, и в какой-то момент Bash tool начал возвращать ошибку на любую команду, даже echo test. Минут пять я пытался понять, что происходит, пока не догадался проверить df — оказалось, из за неудачной фоновой команды запущенной Claude Code, Maestro спамил логами в /tmp, и тот забился подчистую. Никакого алерта на это у меня, естественно, не было. GNOME/KDE и прочие показывают «мало места» при фиксированном пороге вроде 5%, а 5% от 16 ГБ — это 800 МБ, у меня к тому моменту и этого не осталось.
В общем, я подумал: хватит это терпеть — хочу видеть заранее, если что-то идёт не так. И не просто по порогу, «свободно < 10%» — это грубый инструмент, на двухтерабайтном диске 10% это 200 ГБ, проблемы ещё далеко, а на моём zram в 16 ГБ всё может кончиться за пару минут, причём в момент проверки может быть свободно ещё 60%. Нужно именно предсказание по тренду: «при текущей скорости через 2 часа настанет ******».
Серверный мониторинг давно так умеет — Prometheus predict_linear(), в Zabbix timeleft() с 3.0, Netdata из коробки алертит если диск заполнится менее чем через 48 часов. На десктопе — ничего подобного. Ну и раз конфиг на Python, собственно, в моем config.py и появился новый виджет — дополнительные 100 строк без внешних зависимостей.
У меня, кстати, на этой машине стоит node_exporter и пара кастомных Prometheus-экспортеров — но это для других задач, и на остальных моих рабочих и личных компах их нет. А виджет в панели работает везде, где есть qtile.
Как устроен виджет
Наследуем base.InLoopPollText — стандартный базовый класс для текстовых виджетов с периодическим опросом. Каждые 5 секунд:
Читаем
/proc/mounts, фильтруем виртуальные ФС.Для каждой реальной точки монтирования —
os.statvfs().Пишем значение свободного места в скользящий буфер (
deque).Строим линейную регрессию по накопленной истории.
Если прямая пересекает ноль в пределах 24 часов — показываем предупреждение.
Если всё хорошо — виджет возвращает пустую строку и не занимает место на панели.
Фильтрация точек монтирования
В /proc/mounts на типичной Linux-системе десятки записей: sysfs, proc, cgroup, tmpfs, overlay и прочие виртуальные ФС, которые нет смысла мониторить. Фильтруем по типу ФС и по префиксу пути:
DISK_SPACE_SKIP_FS = { 'sysfs', 'proc', 'devtmpfs', 'devpts', 'tmpfs', 'securityfs', 'cgroup', 'cgroup2', 'pstore', 'debugfs', 'hugetlbfs', 'mqueue', 'configfs', 'fusectl', 'efivarfs', 'binfmt_misc', 'autofs', 'fuse.portal', 'nsfs', 'tracefs', 'bpf', 'ramfs', 'rpc_pipefs', 'nfsd', 'overlay', 'fuse.gvfsd-fuse', } DISK_SPACE_SKIP_PREFIXES = ('/sys/', '/proc/', '/dev/', '/run/') def get_mountpoints(self): mounts = [] try: with open('/proc/mounts') as f: for line in f: parts = line.split() if len(parts) < 3: continue mountpoint, fstype = parts[1], parts[2] if fstype in DISK_SPACE_SKIP_FS: continue if any(mountpoint.startswith(p) for p in DISK_SPACE_SKIP_PREFIXES): continue mounts.append(mountpoint) except OSError: pass return mounts
Кстати, tmpfs в списке исключений, а вот ext2 на zram — нет. Собственно, на нём-то виджет и сработал первый раз — на том самом /tmp, из-за которого всё и началось. Теперь я хотя бы вижу, что место тает, а не узнаю об этом по упавшему Bash tool.
Линейная регрессия без numpy
Для предсказания — OLS, метод наименьших квадратов. Тот же алгоритм, что стоит за predict_linear() в Prometheus (ну, по сути — пять сумм и два деления, ничего сложного). Аппроксимируем зависимость «свободное место от времени» прямой 
и ищем, когда 
.
Формулы классические:
В коде:
@staticmethod def _predict_exhaustion(history, now): n = len(history) t = [h[0] - history[0][0] for h in history] y = [h[1] for h in history] sum_t = sum(t) sum_y = sum(y) sum_tt = sum(ti * ti for ti in t) sum_ty = sum(ti * yi for ti, yi in zip(t, y)) denom = n * sum_tt - sum_t * sum_t if denom == 0: return None slope = (n * sum_ty - sum_t * sum_y) / denom intercept = (sum_y - slope * sum_t) / n if slope >= 0: return None time_to_zero = -intercept / slope elapsed = now - history[0][0] seconds_left = time_to_zero - elapsed if seconds_left > 0: return seconds_left / 3600 return None
Можно было бы обойтись попроще: взять разницу между первым и последним значением, поделить на время — вот и скорость. Но я попробовал, и это постоянно триггерилось — стоит какому-нибудь процессу записать и удалить временный файл аккурат в момент замера, и предсказание улетает. OLS по всем точкам окна ведёт себя заметно спокойнее.
Красный и оранжевый
Тут есть нюанс, который меня в первой реалзиации не устроил. Предсказание может оставаться активным, даже когда потребление уже прекратилось — в скользящем окне ещё остались старые точки, с нисходящим трендом (когда свободное место заканчивалось). Убил процесс, место больше не утекает, а виджет всё равно красный.
Поэтому два цвета:
Красный — тренд активен: место продолжает убывать прямо сейчас (за последние 30 секунд), или уже ниже порога.
Оранжевый — предсказание ещё показывает исчерпание, но потребление остановилось. Ситуация стабилизировалась, алерт скоро уйдёт сам.
COLOR_RED = '#FF0000' COLOR_ORANGE = '#FF8800' self._set_color( self.COLOR_RED if any_active else self.COLOR_ORANGE )
Надо учитывать: метод update(), который вызывается после poll(), обновляет только текст — цвет layout’а он не трогает. Есть set_font(foreground=...), но он заодно вызывает bar.draw(), что нам не нужно (перерисовка и так произойдёт). Поэтому проще напрямую выставить self.layout.colour.
def _set_color(self, color): self.foreground = color if self.layout: self.layout.colour = color
Полный код виджета
DiskSpaceAlert — полный код
DISK_SPACE_CHECK_INTERVAL = 5 DISK_SPACE_WARN_PERCENT = 10 DISK_SPACE_PREDICT_HOURS = 24 DISK_SPACE_SKIP_FS = { 'sysfs', 'proc', 'devtmpfs', 'devpts', 'tmpfs', 'securityfs', 'cgroup', 'cgroup2', 'pstore', 'debugfs', 'hugetlbfs', 'mqueue', 'configfs', 'fusectl', 'efivarfs', 'binfmt_misc', 'autofs', 'fuse.portal', 'nsfs', 'tracefs', 'bpf', 'ramfs', 'rpc_pipefs', 'nfsd', 'overlay', 'fuse.gvfsd-fuse', } DISK_SPACE_SKIP_PREFIXES = ('/sys/', '/proc/', '/dev/', '/run/') class DiskSpaceAlert(base.InLoopPollText): COLOR_RED = '#FF0000' COLOR_ORANGE = '#FF8800' def __init__(self, **config): config.setdefault('foreground', self.COLOR_RED) config.setdefault('update_interval', DISK_SPACE_CHECK_INTERVAL) super().__init__(**config) self.space_history: dict[str, deque] = {} def get_mountpoints(self): mounts = [] try: with open('/proc/mounts') as f: for line in f: parts = line.split() if len(parts) < 3: continue mountpoint, fstype = parts[1], parts[2] if fstype in DISK_SPACE_SKIP_FS: continue if any(mountpoint.startswith(p) for p in DISK_SPACE_SKIP_PREFIXES): continue mounts.append(mountpoint) except OSError: pass return mounts @staticmethod def _predict_exhaustion(history, now): n = len(history) t = [h[0] - history[0][0] for h in history] y = [h[1] for h in history] sum_t = sum(t) sum_y = sum(y) sum_tt = sum(ti * ti for ti in t) sum_ty = sum(ti * yi for ti, yi in zip(t, y)) denom = n * sum_tt - sum_t * sum_t if denom == 0: return None slope = (n * sum_ty - sum_t * sum_y) / denom intercept = (sum_y - slope * sum_t) / n if slope >= 0: return None time_to_zero = -intercept / slope elapsed = now - history[0][0] seconds_left = time_to_zero - elapsed if seconds_left > 0: return seconds_left / 3600 return None def _set_color(self, color): self.foreground = color if self.layout: self.layout.colour = color def poll(self): now = time() alerts = [] any_active = False for mp in self.get_mountpoints(): try: st = os.statvfs(mp) except OSError: continue if st.f_blocks == 0: continue free_pct = st.f_bfree / st.f_blocks * 100 free_bytes = st.f_bfree * st.f_frsize history = self.space_history.setdefault( mp, deque(maxlen=120)) history.append((now, free_bytes)) reasons = [] if free_pct < DISK_SPACE_WARN_PERCENT: reasons.append(f'{free_pct:.0f}%free') any_active = True if len(history) >= 2 and now - history[0][0] > 60: hours_left = self._predict_exhaustion( history, now) if (hours_left is not None and hours_left < DISK_SPACE_PREDICT_HOURS): recent = [h for h in history if now - h[0] < 30] if (len(recent) >= 2 and recent[-1][1] <= recent[0][1]): any_active = True reasons.append(f'{hours_left:.0f}h left') if reasons: label = mp if mp != '/' else '/' alerts.append( f'DISK {label}: {", ".join(reasons)}') self._set_color( self.COLOR_RED if any_active else self.COLOR_ORANGE) current = set(self.get_mountpoints()) for mp in list(self.space_history): if mp not in current: del self.space_history[mp] return ' | '.join(alerts)
Добавление в панель:
DiskSpaceAlert(font=font, fontsize=fontsize, foreground='#FF0000')
Что можно подкрутить
DISK_SPACE_CHECK_INTERVAL — по умолчанию 5 секунд. Если использовать на сервере, можно поставить 30.
DISK_SPACE_WARN_PERCENT — порог для красного алерта без предсказания, по умолчанию 10%.
DISK_SPACE_PREDICT_HOURS — горизонт предсказания, 24 часа. Можно уменьшить, если не хочется видеть алерт «12h left» на корневом разделе.
maxlen=120 — размер скользящего окна. При интервале 5 секунд это 10 минут истории. Окно — компромисс: слишком короткое — дёргается на каждый всплеск записи, слишком длинное — медленно замечает новые тренды. 10 минут на десктопе работают нормально.
Тестирование
Можно имитировать потребление:
mkdir -p /tmp/disk_space_test for i in $(seq 0 99); do dd if=/dev/zero of=/tmp/disk_space_test/fill_$i \ bs=1M count=200 2>/dev/null sleep 2 done
Через минуту после начала заполнения виджет покажет красный DISK /tmp: 0h left. После rm -rf /tmp/disk_space_test — сменится на оранжевый, а потом исчезнет по мере обновления окна.
Ограничения
Линейная модель. Если потребление нелинейное (экспоненциальный рост логов, например), предсказание будет оптимистичным. Zabbix предлагает на выбор полиномиальную, экспоненциальную и логарифмическую аппроксимацию, коммерческие платформы используют Holt-Winters и Prophet. Для десктопного виджета с 10-минутным окном линейной модели хватает.
Потеря истории при перезагрузке. История в памяти, при перезапуске qtile сбрасывается. Можно было бы сохранять в файл, но, честно говоря, 60 секунд на набор минимальной истории — не та проблема, которую хочется решать.
Дедупликация. Если один блочный девайс смонтирован в нескольких местах (bind mounts), виджет покажет одинаковый алерт для каждой точки. Можно фильтровать по (major, minor) устройства, но у меня такой ситуации нет, так что не стал.
Дисклеймер: и код виджета, и текст этой статьи написаны в соавторстве с Claude Code. Я старался сохранить разговорный стиль и донести ценность, а не просто сгенерировать текст — но считаю правильным об этом сказать.
