systemd 管理进程的启动、失败与停止

from pathlib import Path
from tempfile import TemporaryDirectory
import subprocess,time

unit='xbcatm-supervise-lab.service';bad='xbcatm-exec-failure-lab.service'
unit_path=Path('/run/systemd/system')/unit;bad_path=unit_path.with_name(bad)
assert not unit_path.exists() and not bad_path.exists()
def run(args,success=True):
    result=subprocess.run(args,text=True,capture_output=True,timeout=10)
    assert (result.returncode==0)==success,(args,result.returncode,result.stderr)
    return result.stdout.strip()
def value(name,key):return run(['systemctl','show',name,'-p',key,'--value'])
def wait_for(check):
    deadline=time.monotonic()+5
    while not check():
        assert time.monotonic()<deadline,'state transition timed out'
        time.sleep(0.02)
with TemporaryDirectory(prefix='xbcatm-supervise-',dir='/run') as directory:
    root=Path(directory);worker=root/'worker.py';ready=root/'ready'
    worker.write_text("""from pathlib import Path
import os,signal,sys
signal.signal(signal.SIGTERM,lambda *_: sys.exit(0))
Path(__file__).with_name('ready').write_text(str(os.getpid()))
print('worker ready',flush=True)
while True: signal.pause()
""")
    unit_path.write_text(f'''[Unit]
Description=Disposable supervision exercise
StartLimitIntervalSec=10s
StartLimitBurst=3
[Service]
Type=exec
ExecStart=/usr/bin/python3 -u {worker}
Restart=on-failure
RestartSec=200ms
TimeoutStopSec=2s
StandardOutput=journal
StandardError=journal
''')
    bad_path.write_text(f'[Service]\nType=exec\nExecStart={root}/missing-program\n')
    try:
        run(['systemd-analyze','verify',str(unit_path)])
        run(['systemctl','daemon-reload']);run(['systemctl','start',unit])
        wait_for(lambda: ready.exists() and ready.read_text()==value(unit,'MainPID'))
        first=value(unit,'MainPID');assert first!='0'
        first_invocation=value(unit,'InvocationID')
        run(['systemctl','kill','--kill-who=main','--signal=KILL',unit])
        wait_for(lambda: value(unit,'MainPID') not in ('0',first) and ready.read_text()==value(unit,'MainPID'))
        assert int(value(unit,'NRestarts'))==1
        print('SIGKILL: replacement PID confirmed; NRestarts=1')
        run(['journalctl','--sync'])
        for invocation in (first_invocation,value(unit,'InvocationID')):
            logs=run(['journalctl','_SYSTEMD_INVOCATION_ID='+invocation,'--no-pager','-o','cat'])
            assert 'worker ready' in logs
        print('journal contains both application startups')
        run(['systemctl','stop',unit]);time.sleep(0.3)
        assert value(unit,'ActiveState')=='inactive' and value(unit,'MainPID')=='0'
        print('manager stop leaves service inactive')
        run(['systemctl','daemon-reload']);run(['systemctl','start',bad],success=False)
        assert value(bad,'Result')=='exit-code' and value(bad,'ExecMainStatus')=='203'
        print('Type=exec reports missing executable: status 203/EXEC')
    finally:
        run(['systemctl','stop',unit,bad])
        for name in (unit,bad):
            if value(name,'ActiveState')=='failed':
                run(['systemctl','reset-failed',name])
        unit_path.unlink();bad_path.unlink(missing_ok=True)
        run(['systemctl','daemon-reload'])