systemd 重启策略要明确失败与限流规则

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

unit='xbcatm-restart-lab-'+uuid.uuid4().hex[:8]+'.service'
path=Path('/run/systemd/system')/unit
assert not path.exists()
def ctl(*args):
    result=subprocess.run(['systemctl',*args],text=True,capture_output=True,timeout=10)
    assert result.returncode==0,(args,result.stderr)
    return result.stdout.strip()
def value(key):return ctl('show',unit,'-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-restart-',dir='/run') as directory:
    root=Path(directory);worker=root/'worker.py';calls=root/'calls';healthy=root/'healthy'
    worker.write_text("""from pathlib import Path
import sys,time
root=Path(__file__).parent
with (root/'calls').open('a') as output: output.write('started\\n')
if not (root/'healthy').exists(): sys.exit(7)
time.sleep(60)
""")
    path.write_text(f'''[Unit]
StartLimitIntervalSec=10s
StartLimitBurst=3
[Service]
Type=exec
ExecStart=/usr/bin/python3 {worker}
Restart=on-failure
RestartSec=100ms
''')
    try:
        ctl('daemon-reload');ctl('start',unit)
        wait_for(lambda: value('ActiveState')=='failed')
        assert calls.read_text().splitlines()==['started']*3
        assert value('Result')=='exit-code'
        subprocess.run(['journalctl','--sync'],check=True)
        logs=subprocess.run(['journalctl','-u',unit,'--no-pager','-o','cat'],capture_output=True,text=True,check=True).stdout
        assert 'Start request repeated too quickly' in logs
        print('three starts; failed/exit-code with start-rate-limit journal message')
        healthy.touch();ctl('reset-failed',unit);ctl('start',unit)
        wait_for(lambda: len(calls.read_text().splitlines())==4)
        first=value('MainPID');assert first!='0'
        ctl('kill','--kill-who=main','--signal=TERM',unit)
        wait_for(lambda: value('ActiveState')=='inactive')
        assert value('NRestarts')=='0'
        print('on-failure treats ordinary SIGTERM as clean; no restart')
        ctl('start',unit);first=value('MainPID')
        ctl('kill','--kill-who=main','--signal=KILL',unit)
        wait_for(lambda: value('MainPID') not in ('0',first))
        assert value('NRestarts')=='1'
        print('SIGKILL triggers one restart')
        ctl('stop',unit);time.sleep(0.2)
        assert value('ActiveState')=='inactive'
        print('explicit stop remains stopped')
    finally:
        ctl('stop',unit)
        if value('ActiveState')=='failed':ctl('reset-failed',unit)
        path.unlink();ctl('daemon-reload')