按明确的字段与优先级规则查询 journal

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

name='xbcatm-query-'+uuid.uuid4().hex[:8];unit=name+'.service'
path=Path('/run/systemd/system')/unit
assert not path.exists()
def run(*args):
    result=subprocess.run(args,text=True,capture_output=True,timeout=10)
    assert result.returncode==0,(args,result.stderr)
    return result.stdout.strip()
def query(*matches):
    output=run('journalctl','--no-pager','-o','json','-u',unit,*matches)
    return [json.loads(line) for line in output.splitlines() if line]
def labels(rows):return sorted(row['MESSAGE'] for row in rows)
with TemporaryDirectory(prefix=name+'-',dir='/run') as directory:
    root=Path(directory);worker=root/'worker.py'
    worker.write_text(r'''from pathlib import Path
import socket
phase=Path(__file__).with_name('phase').read_text()
with socket.socket(socket.AF_UNIX,socket.SOCK_DGRAM) as log:
    log.connect('/run/systemd/journal/socket')
    for component,priority in [('api',2),('api',3),('api',6),('db',3)]:
        fields=[f'MESSAGE={phase}-{component}-{priority}',f'PRIORITY={priority}',
                f'LAB_RUN={phase}',f'LAB_COMPONENT={component}',
                '_SYSTEMD_UNIT=forged.service']
        log.sendall(('\n'.join(fields)+'\n').encode())
''')
    path.write_text(f'[Service]\nType=oneshot\nExecStart=/usr/bin/python3 {worker}\n')
    try:
        run('systemd-analyze','verify',str(path));run('systemctl','daemon-reload')
        for phase in ('first','second'):
            (root/'phase').write_text(phase);run('systemctl','start',unit)
        run('journalctl','--sync')
        first=query('LAB_RUN=first')
        assert len(first)==4 and all(row['_SYSTEMD_UNIT']==unit for row in first)
        second=query('LAB_RUN=second');assert len(second)==4
        invocation=first[0]['_SYSTEMD_INVOCATION_ID']
        assert {row['_SYSTEMD_INVOCATION_ID'] for row in first}=={invocation}
        assert {row['_SYSTEMD_INVOCATION_ID'] for row in second}!={invocation}
        assert labels(query('LAB_RUN=first','LAB_COMPONENT=api','-p','err'))==['first-api-2','first-api-3']
        print('different fields AND; -p err includes critical priority 2 and error priority 3')
        assert labels(query('LAB_RUN=first','LAB_COMPONENT=api','PRIORITY=3'))==['first-api-3']
        assert labels(query('LAB_RUN=first','LAB_COMPONENT=api','LAB_COMPONENT=db','PRIORITY=3'))==['first-api-3','first-db-3']
        print('exact PRIORITY=3 differs from threshold; repeated component fields OR')
        union=query('LAB_RUN=first','LAB_COMPONENT=db','+','LAB_RUN=second','LAB_COMPONENT=db')
        assert labels(union)==['first-db-3','second-db-3']
        assert labels(query('_SYSTEMD_INVOCATION_ID='+invocation,'LAB_COMPONENT=db'))==['first-db-3']
        print('explicit + joins match groups; invocation filter separates repeated starts')
        assert query('LAB_RUN=absent')==[]
        print('empty query still succeeds; underscore unit field comes from journal metadata')
    finally:
        run('systemctl','stop',unit)
        if run('systemctl','show',unit,'-p','ActiveState','--value')=='failed':run('systemctl','reset-failed',unit)
        path.unlink();run('systemctl','daemon-reload')