trim virtual machine disk images

PKGBUILD

pkgname=virttrim
pkgver=0.20250109.1
pkgrel=1
pkgdesc='trim libvirt disk images automatically'
arch=(any)
license=(Unlicense)

makedepends=(
	sed
)

source=(
	virttrim.py
	virttrim.service.in
)
cksums=(
	SKIP
	SKIP
)

build() {
	sed 's!@bin@!/usr/lib/'"$pkgname"'!' < "$srcdir/virttrim.service.in" > "$srcdir/virttrim.service"
}

package() {
	depends=(
		guestfs-tools
		libvirt-python
		python
		systemd
	)
	install -Dm644 -t "$pkgdir/usr/lib/systemd/system" "$srcdir/virttrim.service"
	install -Dm755 -t "$pkgdir/usr/lib/$pkgname" "$srcdir/virttrim.py"
}

virttrim.py

#!/usr/bin/env python3

PROGRAM = 'virttrim'
UNIT = PROGRAM + '.service'

from argparse import ArgumentParser
from contextlib import suppress
from xml.etree import ElementTree as ET
import libvirt
import logging
import os

logging.basicConfig(level=logging.INFO)
logger = logging

def findvols(conn):
    r = []
    for pool in conn.listAllStoragePools():
        pdsc = ET.fromstring(pool.XMLDesc())
        if pdsc.get('type') != 'dir':
            continue
        for v in pool.listAllVolumes():
            p = v.path()
            if p in r:
                continue
            vdsc = ET.fromstring(v.XMLDesc())
            if vdsc.get('type') != 'file':
                continue
            fmt = vdsc.find('target/format')
            if fmt is None or fmt.get('type') not in {'raw', 'qcow2'}:
                continue
            if p.rsplit('.', 1)[-1] not in {'raw', 'qcow2'}:
                continue
            r.append((p, fmt.get('type')))
    return r

def volactive(conn, v):
    for dom in conn.listAllDomains():
        if dom.state()[0] == libvirt.VIR_DOMAIN_SHUTOFF:
            continue
        ddsc = ET.fromstring(dom.XMLDesc())
        for src in ddsc.findall('devices/disk//source[@file]'):
            if v == src.get('file'):
                return True
    for pool in conn.listAllStoragePools():
        for vol in pool.listAllVolumes():
            vdsc = ET.fromstring(vol.XMLDesc())
            for path in vdsc.findall('backingStore//path'):
                if v == path.text:
                    return True
    return False

def trimvol(conn, v, fmt):
    logger.debug('found volume %s', v)
    mtime = os.stat(v).st_mtime
    if volactive(conn, v):
        logger.info('volume %s is active, skip', v)
        return
    tmpimg = os.path.join(os.path.dirname(v), '.' + PROGRAM + '-' + os.path.basename(v))
    with suppress(FileNotFoundError):
        os.unlink(tmpimg)
    logger.info('trimming %s', v)
    finished = False
    sp.run(['cp', '-aL', '--reflink=always', v, tmpimg], check=True)
    try:
        sp.run(['virt-sparsify', '--format', fmt, '--in-place', tmpimg], check=True)
        if os.stat(v).st_mtime != mtime:
            raise RuntimeError(f'volume {v} modified during trimming')
        os.replace(tmpimg, v)
        finished = True
        logger.info('finished trimming %s', v)
    finally:
        if not finished:
            os.unlink(tmpimg)

def trimall(uri):
    with libvirt.open(uri) as conn:
        for volpath, fmt in sorted(findvols(conn)):
            try:
                trimvol(conn, os.path.realpath(volpath), fmt)
            except Exception as e:
                logger.warning('error trimming %s: %r', volpath, e)

def main():
    a = ArgumentParser(description='Trim virtual machine disk images')
    a.add_argument('--connect', '-c', help='libvirt URI')
    a = a.parse_args()

    trimall(a.connect)

main()

virttrim.service.in

[Unit]
Before=fstrim.service

[Service]
Type=oneshot
ExecStart=@bin@/virttrim.py

[Install]
WantedBy=fstrim.service