Logrotate 在执行时应用轮转策略

from pathlib import Path
from tempfile import TemporaryDirectory
import gzip,os,stat,subprocess,sys

worker=r'''
import os,signal,sys
path=sys.argv[1];fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_APPEND,0o600)
def reopen(signum,frame):
    global fd
    os.close(fd);fd=os.open(path,os.O_WRONLY|os.O_APPEND)
    print('reopened',flush=True)
signal.signal(signal.SIGHUP,reopen)
try:
    print('ready',flush=True)
    for line in sys.stdin:
        if line=='QUIT\n':break
        os.write(fd,line.encode());print('wrote',flush=True)
finally:os.close(fd)
'''
assert os.geteuid()==0,'use a disposable container for the root-owned fixture'
with TemporaryDirectory(prefix='xbcatm-logrotate-') as directory:
    root=Path(directory);active=root/'app.log';configuration=root/'logrotate.conf';state=root/'state'
    process=subprocess.Popen([sys.executable,'-c',worker,str(active)],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
    try:
        assert process.stdout.readline().strip()=='ready'
        configuration.write_text(f'''"{active}" {{
    size 64
    rotate 2
    compress
    delaycompress
    missingok
    notifempty
    create 0600 root root
    postrotate
        /bin/kill -HUP {process.pid}
    endscript
}}
''')
        def append(text):
            process.stdin.write(text+'\n');process.stdin.flush();assert process.stdout.readline().strip()=='wrote'
        def rotate(*options):return subprocess.run(['logrotate',*options,'--state',str(state),str(configuration)],capture_output=True,text=True,timeout=5)
        append('short');initial=active.read_bytes()
        result=rotate('--debug');assert result.returncode==0,result.stderr
        assert active.read_bytes()==initial and not state.exists() and not (root/'app.log.1').exists()
        result=rotate();assert result.returncode==0,result.stderr
        assert state.exists() and not (root/'app.log.1').exists()
        print('debug changes neither logs nor state; a normal below-threshold run records state without rotating')
        first=initial+b'first-'+b'a'*80+b'\n';second=b'second-'+b'b'*80+b'\n';third=b'third-'+b'c'*80+b'\n'
        for number,payload in enumerate([first,second,third],1):
            append((payload[len(initial):] if number==1 else payload).decode().rstrip('\n'))
            result=rotate();assert result.returncode==0,result.stderr
            assert process.stdout.readline().strip()=='reopened'
            assert active.read_bytes()==b'' and stat.S_IMODE(active.stat().st_mode)==0o600
            assert (root/'app.log.1').read_bytes()==payload
            if number>=2:
                with gzip.open(root/'app.log.2.gz','rb') as handle:assert handle.read()==(first if number==2 else second)
            assert not (root/'app.log.3.gz').exists()
        append('new-active');assert active.read_bytes()==b'new-active\n'
        print('three actual rotations reopen the writer, delay compression one generation, retain two archives, and preserve the expected bytes')
        failed=root/'failed.log';failed.write_bytes(b'failure-fixture')
        bad=root/'failed.conf';bad.write_text(f'''"{failed}" {{
    size 1
    rotate 1
    create 0600 root root
    postrotate
        /bin/false
    endscript
}}
''')
        result=subprocess.run(['logrotate','--state',str(root/'failed.state'),str(bad)],capture_output=True,text=True,timeout=5)
        assert result.returncode!=0
        assert (root/'failed.log.1').read_bytes()==b'failure-fixture' and failed.read_bytes()==b''
        print('a failed postrotate command returns failure after rename/create side effects have already happened')
        output,error=process.communicate('QUIT\n',timeout=5);assert process.returncode==0,error
    finally:
        if process.poll() is None:process.kill();process.wait()