qtile 小部件：磁盘空间预测

用自定义小部件预测 qtile 中磁盘空间耗尽

qtile 开发者可集成 DiskSpaceAlert 小部件，通过线性回归预测磁盘空间耗尽。它分析真实文件系统的可用空间趋势（排除虚拟文件系统），在 24 小时内即将达到零容量时触发警报。该小部件继承自 InLoopPollText，每 5 秒通过 /proc/mounts 和 os.statvfs 检测，使用 deque 保存历史数据。

代码无需外部依赖，特别适合未部署 Prometheus 的桌面环境。首次触发发生在 /tmp 分区（ext2 挂载于 zram，16 GB）：后台日志持续写入填满分区，导致一个 Bash 工具无声崩溃。

排除挂载点

小部件自动忽略虚拟文件系统和系统路径，仅监控物理磁盘。以下常量定义了过滤规则：

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/')

get_mountpoints 函数解析 /proc/mounts，根据文件系统类型或路径前缀跳过无效项。对于 zram 上的 tmpfs，此过滤不生效，从而能及时发现关键异常场景。

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

线性回归预测机制

预测采用普通最小二乘法（OLS），不依赖 NumPy。模型将可用空间建模为 y(t) = a·t + b，计算 y=0 的时间点。公式如下：

• a = (n∑(t_i y_i) - ∑t_i ∑y_i) / (n∑t_i² - (∑t_i)²)
• b = (∑y_i - a ∑t_i) / n

耗尽剩余时间：-b / a 减去窗口已过去的时间。若斜率 ≥ 0 或 seconds_left ≤ 0，则返回 None。

@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

滑动窗口大小为 maxlen=120（5 秒间隔下共 10 分钟），有效平滑临时文件带来的噪声干扰。

颜色提示与告警逻辑

设置两级告警：

• 红色 (#FF0000)：当前仍在持续消耗空间（最近 30 秒内可用空间下降）或可用空间低于 10%。
• 橙色 (#FF8800)：预计 24 小时内耗尽，但趋势已趋于稳定。

_set_color 方法高效更新 self.foreground 和 self.layout.colour，避免无谓重绘。标签显示格式为：DISK /tmp: 8%free, 2h left。

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

# ... (SKIP_FS 和 PREFIXES 如上定义)

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] = {}

    # get_mountpoints() 如上实现

    # _predict_exhaustion() 如上实现

    # _set_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}: {

— Editorial Team