Logrotate 在执行时应用轮转策略

from pathlib import Path
from tempfile import mkdtemp
import gzip
import os
import select
import shutil
import stat
import subprocess
import sys
import time


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
    replacement=os.open(path,os.O_WRONLY|os.O_APPEND)
    os.close(fd);fd=replacement
    print('reopened',flush=True)
signal.signal(signal.SIGHUP,reopen)
try:
    print('ready',flush=True)
    for line in sys.stdin:
        if line=='QUIT\n':break
        data=line.encode()
        if os.write(fd,data)!=len(data):raise OSError('short log write')
        print('wrote',flush=True)
finally:os.close(fd)
'''


def expect(process, expected):
    # Raw nonblocking reads: readiness alone does not mean a complete line.
    descriptor = process.stdout.fileno()
    deadline = time.monotonic() + 5
    answer = bytearray()
    while time.monotonic() < deadline:
        remaining = max(0, deadline - time.monotonic())
        if not select.select([descriptor], [], [], remaining)[0]:
            continue
        try:
            byte = os.read(descriptor, 1)
        except BlockingIOError:
            continue
        if not byte:
            raise RuntimeError(f'logger EOF waiting for {expected}')
        answer.extend(byte)
        if byte == b'\n':
            if answer != (expected + '\n').encode():
                raise RuntimeError(f'unexpected logger acknowledgment: {answer!r}')
            return
        if len(answer) > 64:
            raise RuntimeError('logger acknowledgment is too long')
    raise TimeoutError(f'logger did not acknowledge {expected}')


def send_line(process, text):
    descriptor = process.stdin.fileno()
    pending = (text + '\n').encode()
    deadline = time.monotonic() + 5
    while pending and time.monotonic() < deadline:
        remaining = max(0, deadline - time.monotonic())
        if not select.select([], [descriptor], [], remaining)[1]:
            continue
        try:
            written = os.write(descriptor, pending)
        except BlockingIOError:
            continue
        if written == 0:
            raise RuntimeError('logger input made no progress')
        pending = pending[written:]
    if pending:
        raise TimeoutError('logger input deadline expired')


def attempt(errors, name, operation):
    try:
        operation()
    except Exception as error:
        errors.append(f'{name}: {error!r}')


def main():
    if sys.argv[1:] != ['--isolated-disposable-lab']:
        raise SystemExit('requires a separately verified disposable Linux lab')
    if os.geteuid() != 0:
        raise RuntimeError('root is required, but does not prove isolation')
    root = Path(mkdtemp(prefix='xbcatm-logrotate-', dir='/tmp'))
    active = root / 'app.log'
    configuration = root / 'logrotate.conf'
    state = root / 'state'
    process = error_file = None
    rotation_uncertain = False
    try:
        # A regular error file cannot fill an unread stderr pipe.
        error_file = (root / 'logger.stderr').open('wb')
        process = subprocess.Popen([sys.executable, '-c', worker, str(active)],
                                   stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                   stderr=error_file, bufsize=0)
        os.set_blocking(process.stdin.fileno(), False)
        os.set_blocking(process.stdout.fileno(), False)
        expect(process, '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):
            send_line(process, text)
            expect(process, 'wrote')
        def run_rotation(arguments):
            nonlocal rotation_uncertain
            rotation_uncertain = True
            result = subprocess.run(arguments, capture_output=True, text=True, timeout=5)
            rotation_uncertain = False
            return result
        def rotate(*options):
            return run_rotation(['logrotate', *options, '--state', str(state), str(configuration)])
        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
            expect(process, '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 = run_rotation(['logrotate', '--state', str(root / 'failed.state'), str(bad)])
        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')
        send_line(process, 'QUIT')
        assert process.wait(timeout=5) == 0, (root / 'logger.stderr').read_text()
    finally:
        original = sys.exc_info()[1]
        errors = []
        if rotation_uncertain:
            errors.append('logrotate did not return; hook/compressor descendants are unverified')
        if process is not None:
            # Popen.kill is a no-op when this owned child has already completed.
            attempt(errors, 'kill owned logger', process.kill)
            attempt(errors, 'reap logger', lambda: process.wait(timeout=5))
            for name in ('stdin', 'stdout'):
                stream = getattr(process, name)
                if stream is not None:
                    attempt(errors, 'close ' + name, stream.close)
        if error_file is not None:
            attempt(errors, 'close logger error file', error_file.close)
        if not errors:
            attempt(errors, 'remove private files', lambda: shutil.rmtree(root))
        if errors:
            message = 'cleanup incomplete; discard lab: ' + '; '.join(errors)
            if original is not None:
                original.add_note(message)
            else:
                raise RuntimeError(message)


if __name__ == '__main__':
    main()