Predicting Disk Space Exhaustion in qtile with a Custom Widget
qtile developers can integrate the DiskSpaceAlert widget to forecast disk space exhaustion using linear regression. It analyzes trends in free space on real filesystems—excluding virtual ones—and triggers when the zero threshold is crossed within 24 hours. The widget inherits from InLoopPollText, polling every 5 seconds via /proc/mounts and os.statvfs, using a deque to maintain historical data.
The code runs without external dependencies, making it ideal for desktop setups without Prometheus. The first trigger occurred on /tmp (ext2 over zram, 16 GB): background logs filled the partition, causing a Bash tool to crash silently.
Filtering Mount Points
The widget ignores virtual FS and system paths, focusing only on physical disks. Constants define exclusions:
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/')
The get_mountpoints function parses /proc/mounts, skipping entries by filesystem type or path prefix. For tmpfs on zram, the filter doesn’t apply—allowing detection of critical scenarios.
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
Linear Regression for Prediction
Prediction uses ordinary least squares (OLS) without NumPy. It models free space as y(t) = a·t + b, calculating time until y=0. Formulas:
a = (n∑(t_i y_i) - ∑t_i ∑y_i) / (n∑t_i² - (∑t_i)²)b = (∑y_i - a ∑t_i) / n
Time to exhaustion: -b / a minus elapsed window time. If slope ≥ 0 or seconds_left ≤ 0, returns 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
A sliding window of maxlen=120 (10 minutes at 5-second intervals) smooths noise from temporary files.
Color Indication and Alert Logic
Two alert levels:
- Red (#FF0000): Active consumption (free space dropping in last 30 seconds) or less than 10% free.
- Orange (#FF8800): Exhaustion predicted within 24 hours, but trend has stabilized.
The _set_color method updates self.foreground and self.layout.colour efficiently, avoiding unnecessary redraws. Label text: DISK /tmp: 8%free, 2h left.
def _set_color(self, color):
self.foreground = color
if self.layout:
self.layout.colour = color
History is cleared for unmounted filesystems.
Full DiskSpaceAlert Class Code
DISK_SPACE_CHECK_INTERVAL = 5
DISK_SPACE_WARN_PERCENT = 10
DISK_SPACE_PREDICT_HOURS = 24
# ... (SKIP_FS and PREFIXES as above)
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() as above
# _predict_exhaustion() as above
# _set_color() as above
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
No comments yet.