结构化日志让查询约定明确可见

import asyncio
import io
import json
import logging
from collections import Counter
from contextvars import ContextVar

request_id = ContextVar("request_id", default=None)
stream = io.StringIO()


class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({"schema_version": 1, "event": record.event,
                           "request_id": request_id.get(), "worker": record.worker,
                           "duration_ms": record.duration_ms,
                           "level": record.levelname, "message": record.getMessage()},
                          ensure_ascii=False, separators=(",", ":"))


logger = logging.getLogger("structured-log-lab")
logger.setLevel(logging.INFO)
logger.propagate = False
handler = logging.StreamHandler(stream)
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)


async def request(name, ready, release):
    token = request_id.set(name)
    try:
        logger.info("starting %s", name, extra={"event": "request.started",
                    "worker": name, "duration_ms": 0})
        ready.set()
        await release.wait()
        await asyncio.sleep(0)
        logger.info('untrusted line\n{"event":"forged"}',
                    extra={"event": "request.completed", "worker": name, "duration_ms": 4})
    finally:
        request_id.reset(token)


async def main():
    ready = [asyncio.Event(), asyncio.Event()]
    release = asyncio.Event()
    tasks = [asyncio.create_task(request(name, signal, release))
             for name, signal in zip(("req-a", "req-b"), ready)]
    await asyncio.gather(*(signal.wait() for signal in ready))
    release.set()
    await asyncio.gather(*tasks)
    assert request_id.get() is None


try:
    asyncio.run(main())
finally:
    logger.removeHandler(handler)
    handler.close()
lines = stream.getvalue().splitlines()
records = [json.loads(line) for line in lines]
assert len(lines) == 4
assert Counter(record["request_id"] for record in records) == {"req-a": 2, "req-b": 2}
assert all(record["request_id"] == record["worker"] for record in records)
assert all(type(record["duration_ms"]) is int for record in records)
completed = [record for record in records if record["event"] == "request.completed"]
assert len(completed) == 2 and all("\n" in record["message"] for record in completed)
assert not any(record["event"] == "forged" for record in records)
print(json.dumps({"physical_lines": len(lines), "parsed_records": len(records),
                  "requests": 2, "context_mixing": False,
                  "numeric_duration_preserved": True, "forged_records": 0}, indent=2))