suspend kvm guests before host sleep

PKGBUILD

pkgname=virtsleep
pkgver=0.20230916.0
pkgrel=1
pkgdesc='suspend libvirt guests before host sleeps'
arch=(any)
license=(Unlicense)

depends=(
	libvirt-python
	python
	python-systemd
	systemd
)
makedepends=(
	sed
)

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

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

package() {
	install -Dm644 -t "$pkgdir/usr/lib/systemd/system" "$srcdir/virtsleep.service"
	install -Dm755 -t "$pkgdir/usr/lib" "$srcdir/virtsleep.py"
}

virtsleep.py

#!/usr/bin/env python3

from argparse import ArgumentParser
from contextlib import suppress
from signal import signal, pause, SIGHUP, SIGINT, SIGTERM
from systemd.daemon import notify
from time import time, sleep
from xml.etree import ElementTree as ET
import libvirt
import logging
import sys

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

class Virtsleep:
    SHUTDOWN_TIMEOUT = 30
    SUSPEND_TIMEOUT = 10

    def __init__(self, uri):
        self.conn = libvirt.open(uri)
        self.stopped = []

    def __enter__(self):
        return self

    @staticmethod
    def notify(**kw):
        notify('\n'.join(x[0].upper() + '=' + str(x[1]) for x in kw.items()))

    def __exit__(self, *a):
        self.notify(stopping=1)
        sleeping = []
        halted = []

        for vm in self.stopped:
            s = vm.state()[0]
            if s == libvirt.VIR_DOMAIN_PAUSED:
                sleeping.append(vm)
            elif s == libvirt.VIR_DOMAIN_SHUTOFF:
                halted.append(vm)

        for vm in sleeping:
            s = f'Resuming {vm.name()}'
            logger.info(s)
            self.notify(status=s, extend_timeout_usec=50000000)
            vm.resume()
            sleep(1)

        for vm in halted:
            s = f'Restarting {vm.name()}'
            logger.info(s)
            self.notify(status=s, extend_timeout_usec=15000000)
            vm.create()
            sleep(10)

        for vm in sleeping:
            s = f'Setting time of {vm.name()}'
            logger.info(s)
            self.notify(status=s, extend_timeout_usec=5000000)
            t = time()
            vm.setTime({
                'seconds': int(t),
                'nseconds': int(t * 1e9) % 1000000000,
                })

        self.conn.close()

    def main(self):
        self.setsig()
        self.stop_vm()
        self.notify(ready=1)
        while True:
            pause()

    def setsig(self):
        def handle(*_):
            sys.exit(0)
        for s in (SIGHUP, SIGINT, SIGTERM):
            signal(s, handle)

    @staticmethod
    def hashostdev(d):
        x = ET.fromstring(d.XMLDesc())
        if x.find('.//hostdev') is not None:
            return True
        return False

    @staticmethod
    def isrunning(d):
        return d.state()[0] == libvirt.VIR_DOMAIN_RUNNING

    def stop_vm(self):
        for d in self.conn.listAllDomains():
            if not self.isrunning(d):
                continue

            self.stopped.append(d)

            if not self.hashostdev(d):
                with suppress(libvirt.libvirtError):
                    logger.info(f'Suspending {d.name()}')
                    self.notify(extend_timeout_usec=5000000)
                    d.suspend()
                    t = 0
                    while t < self.SUSPEND_TIMEOUT and self.isrunning(d):
                        self.notify(status=f'Suspending {d.name()} [{t}/{self.SUSPEND_TIMEOUT}]', extend_timeout_usec=2000000)
                        t += 1
                        sleep(1)

            if self.isrunning(d):
                logger.info(f'Shutting down {d.name()}')
                self.notify(extend_timeout_usec=5000000)
                d.shutdown()
                t = 0
                while t < self.SHUTDOWN_TIMEOUT and self.isrunning(d):
                    self.notify(status=f'Shutting down {d.name()} [{t}/{self.SHUTDOWN_TIMEOUT}]', extend_timeout_usec=2000000)
                    t += 1
                    sleep(1)
                if self.isrunning(d):
                    with suppress(libvirt.libvirtError):
                        logger.info(f'Forcing off {d.name()}')
                        self.notify(extend_timeout_usec=5000000)
                        d.destroy()

def main():
    a = ArgumentParser(description='Suspend KVM guests on host sleep')
    a.add_argument('--connect', '-c', help='libvirt URI')
    a = a.parse_args()

    with Virtsleep(a.connect) as s:
        s.main()

main()

virtsleep.service.in

[Unit]
Before=sleep.target
StopWhenUnneeded=yes

[Service]
Type=notify
NotifyAccess=main
ExecStart=@libexec@/virtsleep.py

[Install]
WantedBy=sleep.target