Diskplatz-Auslastung vorhersagen mit einem benutzerdefinierten qtile-Widget
qtile-Entwickler können das DiskSpaceAlert-Widget integrieren, um eine bevorstehende Erschöpfung des Speicherplatzes mithilfe linearer Regression vorherzusagen. Es analysiert Trends im freien Speicherplatz auf echten Dateisystemen – virtuelle Systeme werden dabei ausgeschlossen – und löst aus, sobald der Nullschwellenwert innerhalb von 24 Stunden unterschritten wird. Das Widget erbt von InLoopPollText, aktualisiert alle 5 Sekunden über /proc/mounts und os.statvfs und verwendet eine deque, um historische Daten zu speichern.
Der Code läuft ohne externe Abhängigkeiten und ist ideal für Desktop-Umgebungen ohne Prometheus. Der erste Auslöser ereignete sich auf /tmp (ext2 über zram, 16 GB): Hintergrund-Logs füllten die Partition vollständig und ließen ein Bash-Tool stumm abstürzen.
Mount-Punkte filtern
Das Widget ignoriert virtuelle Dateisysteme und Systempfade und konzentriert sich ausschließlich auf physische Laufwerke. Konstanten definieren die Ausschlusskriterien:
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/')
Die Funktion get_mountpoints liest /proc/mounts aus und überspringt Einträge nach Dateisystemtyp oder Pfadpräfix. Für tmpfs auf zram gilt dieser Filter nicht – so können kritische Szenarien erkannt werden.
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
Lineare Regression zur Vorhersage
Die Vorhersage basiert auf der Methode der kleinsten Quadrate (OLS) ohne NumPy. Sie modelliert den freien Speicherplatz als y(t) = a·t + b und berechnet die Zeit bis y=0. Die Formeln lauten:
a = (n∑(t_i y_i) - ∑t_i ∑y_i) / (n∑t_i² - (∑t_i)²)b = (∑y_i - a ∑t_i) / n
Zeit bis zur Erschöpfung: -b / a abzüglich der verstrichenen Fensterzeit. Wenn die Steigung ≥ 0 oder seconds_left ≤ 0, wird None zurückgegeben.
@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
Ein gleitendes Fenster mit maxlen=120 (10 Minuten bei 5-Sekunden-Intervallen) glättet Rauschen durch temporäre Dateien.
Farbliche Anzeige und Alarmlogik
Zwei Alarmstufen:
- Rot (#FF0000): Aktive Verbrauchstendenz (freier Speicher sinkt in den letzten 30 Sekunden) oder weniger als 10 % frei.
- Orange (#FF8800): Erschöpfung innerhalb von 24 Stunden vorhergesagt, aber die Tendenz hat sich stabilisiert.
Die Methode _set_color aktualisiert self.foreground und self.layout.colour effizient und vermeidet unnötige Neumalungen. Die Beschriftung lautet: DISK /tmp: 8%frei, 2h übrig.
def _set_color(self, color):
self.foreground = color
if self.layout:
self.layout.colour = color
Die Historie wird für abgemeldete Dateisysteme gelöscht.
Vollständiger Code der DiskSpaceAlert-Klasse
DISK_SPACE_CHECK_INTERVAL = 5
DISK_SPACE_WARN_PERCENT = 10
DISK_SPACE_PREDICT_HOURS = 24
# ... (SKIP_FS und PREFIXES wie oben)
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() wie oben
# _predict_exhaustion() wie oben
# _set_color() wie oben
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}%frei')
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 übrig')
if reasons:
label = mp if mp != '/' else '/'
alerts.append(
f'DISK {label}: {',
— Editorial Team
Noch keine Kommentare.