当前写入的 journal 文件与归档文件适用不同的删除规则。

systemd journal 的保留范围需要明确约定

systemd journal 把消息与结构化元数据一起保存在日志文件中,因此可以查询服务输出,不必先为每种应用文本设计一套解析器。但让这些消息保留足够长时间,是另一项需要明确的要求。存储位置、空间策略、轮转方式与转发路径,决定了事故发生一段时间后还有哪些证据可查。查询方便,不等于历史永远完整,也不等于数据已经离开这台机器安全保存。 坏掉的样子 默认命名空间通常使用 Storage=auto:存在 /var/log/journal 时选择持久存储,否则使用运行时存储。Storage=volatile 把文件放在 /run/log/journal 下,并不是简单借用内核的文本环形缓冲区。这些运行时文件通常会在整台机器重新启动后消失,但重启 journald 服务是不同的事件,并不必然清空文件。排查“重启后日志去哪了”之前,必须明确重启的是哪个对象。 Storage=persistent 优先使用 /var/log/journal,早期启动阶段或持久位置不可写时仍可能使用运行时存储。满足文档规定条件时,flush 操作可以转移运行时数据。因此应检查真实文件和最终生效配置,不能只因为某份配置片段里出现了 persistent,就推定所有消息已经写在预期磁盘上。路径、挂载状态与执行时点都需要对应到实际记录。 journald 本身有默认空间限制和轮转行为,不会仅仅因为管理员没有写一个轮转任务就无限增长。不过,SystemMaxUse 也不是保证文件系统永不写满的精确硬配额。运行时存储有另一组设置,可用空间限制同时起作用,其他程序也会占用磁盘,而清理时只能删除已经归档的 journal 文件。把这些条件压缩成“设置一个上限就绝对安全”,会掩盖真正的磁盘压力来源。 快速验证 在由 systemd 管理的一次性 Linux 系统中,以管理员身份运行下面的 Python 脚本。记录环境为 Debian 12、systemd 252.39。脚本创建唯一日志命名空间,显式配置持久存储,通过临时服务写入两条无害消息,最后只删除该命名空间的数据。实验中的 vacuum 会有意删除自己刚写入的旧消息: from pathlib import Path import json,shutil,subprocess,uuid name='xbcatm-journal-'+uuid.uuid4().hex[:8] config=Path('/etc/systemd')/('journald@'+name+'.conf') directory=Path('/var/log/journal')/(Path('/etc/machine-id').read_text().strip()+'.'+name) assert not config.exists() and not directory.exists() def run(*args): result=subprocess.run(args,text=True,capture_output=True,timeout=15) assert result.returncode==0,(args,result.stderr) return result.stdout def journal(*args):return run('journalctl','--namespace='+name,*args) def emit(message): run('systemd-run','--quiet','--wait','--collect','--unit='+name+'-'+message, '--property=LogNamespace='+name,'--property=StandardOutput=journal', '--property=SyslogIdentifier='+name,'/usr/bin/printf',message+'\n') journal('--sync') def messages(): rows=journal('--no-pager','-o','json','SYSLOG_IDENTIFIER='+name) return [json.loads(line)['MESSAGE'] for line in rows.splitlines()] config.write_text('[Journal]\nStorage=persistent\nSystemMaxUse=16M\nRuntimeMaxUse=8M\n') try: emit('old');journal('--flush');journal('--sync') assert messages()==['old'] and list(directory.glob('*.journal')) run('systemctl','restart','systemd-journald@'+name+'.service') assert messages()==['old'] print('persistent namespace: message survives journald service restart') journal('--vacuum-size=1') assert messages()==['old'] print('vacuum cannot remove the active journal file to satisfy a one-byte target') journal('--rotate');emit('new') assert messages()==['old','new'] assert list(directory.glob('*@*.journal')) journal('--vacuum-size=1') assert messages()==['new'] and not list(directory.glob('*@*.journal')) print('after rotation: archived old message removed; active new message remains') print('persistent files observed under /var/log/journal; no reboot durability claim') finally: run('systemctl','stop','systemd-journald@'+name+'.service', 'systemd-journald@'+name+'.socket','systemd-journald-varlink@'+name+'.socket') config.unlink(missing_ok=True) for path in (directory,Path('/run/log/journal')/directory.name,Path('/run/systemd')/('journal.'+name)): if path.exists():shutil.rmtree(path) 第一条消息位于该命名空间的 /var/log/journal 目录中,重启对应 journald 服务后仍然能够查到。这验证了选中的存储位置,以及消息跨越日志守护进程重启仍然存在;它没有模拟突然断电,也没有证明存储设备的持久性,更没有执行整台机器重启。说明验证范围,可以避免把一个真实但有限的结果扩大成并未做过的耐久性测试。 ...

十月 26, 2027 · 1 分钟 · XB Field Notes
通过 journal 字段把同一服务的两次运行缩小为精确结果集。

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

有效的 journal 查询应该选出能够回答具体事故问题的记录。单元身份、运行实例、时间、优先级和应用字段提供了不同维度;正确组合它们,比在整台机器的历史里找到一个熟悉词语更重要。journald 可以附加进程元数据,但不会自动收集应用写出的每个文件,也不会自动理解每条消息的业务含义。查询条件和日志实际采集路径必须相互对应。 坏掉的样子 字段匹配有明确的布尔组合规则:不同字段之间是 AND,同一字段的多个取值之间是 OR,单独一个 + 则把两组条件按 OR 连接。因此,再添加一个字段值,有时会扩大结果集,而不是进一步缩小。应把查询看成一个明确的条件表达式,先用已知记录验证它,再放进事故脚本,避免因为“参数更多了”就误认为结果必然更精确。 -p err 选择错误及更严重的级别,并不是只选择数字三。如果真正需要精确的三级记录,应使用 PRIORITY=3 字段匹配。应用在文本中写了“error”,本身也不等于设置了错误优先级;标准输出的采集使用配置好的优先级规则,除非应用通过受支持的表示方式提供级别。消息文字、日志字段和筛选阈值不能互相混同。 命令成功但没有记录,不证明服务从未运行。所选启动周期、时间窗口、命名空间、读取权限、保留策略或输出路由,都可能解释空结果。反过来,查询一个单元的全部历史,又可能找到以前的成功启动,然后错误归因到当前进程。应区分服务长期身份与某一次具体运行,不能只因为单元名称相同就把证据拼接成一次执行。 快速验证 在 PID 1 为 systemd 的一次性 Linux 系统中,以管理员身份运行下面的 Python 脚本。实测环境是 Debian 12、systemd 252.39。脚本安装一个唯一命名的临时服务,并激活两次。每个工作进程向原生日志套接字发送四条较短的结构化消息,分别包含两个应用组件和几个明确的优先级: 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') 第一次激活生成 API 严重消息、API 错误消息、API 普通信息和数据库错误消息;第二次生成相同类别,但带另一个应用运行标签。按第一次运行、API 组件和 -p err 过滤,会返回两条记录,优先级分别为二和三。改成精确匹配三级,只剩 API 错误。这组对照直接展示阈值与精确值的区别,而不是仅解释命令名称。 ...

七月 30, 2027 · 2 分钟 · XB Field Notes
JSON 转义保护字符串内容,可选字段加引号则在值缺失时仍能保留有效记录。

结构化 nginx 日志需要同时约定转义和字段类型

看起来像 JSON 的 log_format,不一定能可靠地产生结构化事件。nginx 把变量值代入文本,而变量可能包含引号、反斜杠、控制字符,也可能完全没有数据。正确字符串转义与正确字段类型是两个独立要求。用真实解析器检查输出,能发现肉眼查看配置或 nginx -t 成功无法证明的问题,避免上线后才发现部分日志无法进入查询系统。 坏掉的样子 把 User-Agent 原样插入带引号的 JSON 字段,遇到请求头中的引号就会失败。关闭转义不会更安全,只会让请求字符改变外层记录语法。nginx 默认日志转义则有另一种问题:它可能使用十六进制 \xXX 序列,而这不是合法 JSON 字符串转义。日志适合人阅读,不代表它符合另一种数据格式的语法。 加入 escape=json 可以处理 JSON 字符串转义,但不会自动添加引号、提供 null,或把变量转换成数字。对于本地生成的响应,上游耗时可能根本不存在。如果没有定义可选值的表示方式,就直接不加引号输出,字段可能在语法上缺值;即使实际存在的字符全部转义正确,整个 JSON 仍然无效。 缺失值表示也取决于实测版本的转义路径。nginx 1.28.0 默认转义会为找不到的变量写入短横线,JSON 转义则不写变量文本,所以外面的双引号最终成为空字符串。解析器还必须把这种情况与变量本身确实包含短横线、或包含多次上游尝试列表的情况区分开,不能把所有形式都硬当成同一个数字。 快速验证 使用宿主 Python、Docker 和固定 nginx 镜像运行脚本。实测为 Linux arm64 上的 nginx 1.28.0、Docker 29.3.1。它构建包含四种日志格式的临时配置,验证语法,发送两次真实 HTTP 请求,解析生成文件,最后清理自己的容器、镜像标签和网络。请求头都是公开测试字符串,不读取真实凭据或业务日志: ...

七月 15, 2027 · 2 分钟 · XB Field Notes
三个相连的方块,依次标注观察、决策与行动。

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

结构化日志为字段赋予稳定含义,使查询不必每次都从说明文字中推断全部信息。一条记录可以包含事件名称、请求编号、数值时长、严重级别,以及供人阅读的消息。自由文本仍然有价值,也可以被搜索或解析。结构化的实际收益,是让常用问题依赖一份明确的数据约定,即使说明措辞改变,字段的语义仍然保持稳定,而不是简单地给原来的一段话套上花括号。 这份约定必须涉及类型和单位。作为数字保存的时长可以直接进行数值比较,作为字符串保存时,索引和比较方式却可能不同。毫秒与秒不能共用一个含义模糊的字段名称,事件名称也应跨版本稳定描述所发生的动作。模式版本能够标识有意进行的格式变化,但它不会自动转换旧记录,也不会自动保证全部查询消费者继续兼容。字段设计需要围绕真实查询,而不是追求看起来丰富的输出。 把上下文放到正确的记录上 请求编号只有属于当前发出日志的操作时才有帮助。多个请求并发执行时,使用一个进程级可变全局变量保存当前编号并不合适。Python 上下文变量可以在异步任务范围内保持局部状态,但前提是在正确作用域设置并恢复值。这并不意味着所有线程或自定义执行机制都采用同一种继承方式;经过真正的并发边界时,仍需验证上下文怎样到达执行位置。 需要在上下文尚可取得的位置捕获它。本例在发出日志的任务中同步格式化记录,因此格式化器能够读取该任务的上下文变量。如果改成通过队列交给另一线程或进程处理,就应在交接前把所需字段附加到记录上。让监听线程临时读取它自己的当前请求上下文,并不能可靠恢复原发出者的身份。数据最终都进入同一个日志文件,不代表它们在跨线程之后仍然拥有正确的关联关系。 坏掉的样子 手工拼接 JSON 时,消息中的引号或换行可能破坏格式。按行读取的采集器随后可能把一个应用事件拆成多个看似独立的事件。应使用 JSON 序列化器,并保证每个物理行包含一个完整对象。消息的逻辑字符串可以保留换行,但在线上表示中它会被转义,不会产生新的记录边界。后续界面展示解码后的消息时,也应继续清楚区分记录,避免展示方式重新造成误读。 结构化输出不会自动建立索引、降低查询账单或保证投递成功。不受约束的字段、意外变化的类型、缺失的编号和不该记录的敏感内容,都可能使一份合法 JSON 流难以使用。需要依据实际问题选择字段,保持模式稳定,并核对采集器的解析规则和数据保留方式。可查询的结构是生产者与消费者之间的约定,不是文件扩展名或者一次序列化成功就能单方面建立的能力。 快速验证 将脚本保存为 structured-logging.py 并用 Python 执行;本次实际版本为 Python 3.14.4,只使用标准库。两个异步请求各自写入开始事件,在共同的放行点等待,再写入包含刻意构造消息的完成事件。这会让两个请求的存活区间重叠,便于检查编号是否混淆。时长是用于检查 JSON 类型的固定合成值,并不是测得的请求性能,输出也只保存在内存里供断言读取。 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)) 断言要求最终恰好有四个物理行、四条可解析记录,每个请求编号对应两条记录。即使请求并发,记录中的编号也必须与实际发出日志的工作任务一致。时长在解析后仍然是数字,完成消息中的换行保留在字符串内部,却不会多出一个名为 forged 的伪造事件。每个任务最后恢复自己的上下文令牌,外层上下文仍然没有请求编号,从而验证本例的设置和清理作用域。 ...

三月 9, 2027 · 2 分钟 · XB Field Notes
一个客户端响应可能包含多次上游尝试,原始和最终 URI 则描述不同处理阶段。

访问日志应区分客户端结果与上游尝试

nginx 访问日志在请求处理末尾执行记录。选择合适字段,可以区分客户端请求了什么、最终由哪个 URI 处理、发送了什么响应,以及上游尝试期间发生了什么。这些是不同观察。有用格式会保留区别,而不是假定一个状态、耗时或路径就能独立解释整个请求。字段越准确地对应事件,事故期间越容易沿实际处理链定位问题。 坏掉的样子 最终客户端状态可以隐藏较早上游失败。如果 nginx 成功重试其他服务器,客户端可能得到 200,但尝试列表仍记录第一次连接失败。反过来,本地生成的响应没有上游操作可计时。把每个请求都当成恰好一次后端响应,会丢失解释重试和本地处理所需的信息,也可能把成功率与上游健康情况混在一起。 内部跳转后,原始和当前 URI 字段也不同。$request 保留请求行,$request_uri 保留原 URI 和查询,$uri 则可能反映后来的规范化或重写路径。只记录最终 URI,调查者可能不知道哪个公共地址触发它;只记录原始路径,又可能看不到最终处理器上下文。两类字段需要按各自含义使用。 时间字段不是总能直接相减并得到精确诊断的独立延迟组件。请求时间覆盖到 nginx 日志采用的客户端侧完成点,上游时间字段则描述自己的阶段,还可能包含多次尝试。分辨率、缺失值和边界都重要;毫秒显示看起来为零,不证明没有执行工作。粗粒度测量与真实无耗时不能混为一谈。 快速验证 使用宿主 Python、Docker,以及固定 nginx 和 Python 镜像运行脚本。实测为 Linux arm64 上的 nginx 1.28.0、Python 3.11.16、Docker 29.3.1。它配置真实后端,以及代理容器内一个没有监听者的端口,检查 nginx 配置,只发布宿主回环代理端口,读取日志后清理自身资源,不向外部未知服务发送探测: from contextlib import contextmanager from pathlib import Path from tempfile import TemporaryDirectory import http.client,json,socket,subprocess,time,uuid NGINX='nginx@sha256:552e7481ca93ffccd046aa658dbbed22caefbc09c66fa7cd247cbb90b8a5c609' PYTHON='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' def docker(*args,check=True): result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=120) if check:assert result.returncode==0,result.stderr return result def wait(predicate): deadline=time.monotonic()+15 while time.monotonic()<deadline: try: if predicate():return except OSError:pass time.sleep(.1) raise AssertionError('fixture did not become ready') @contextmanager def lab(configuration,files=None,backend=None): name='xbcatm-nginx-'+uuid.uuid4().hex[:10];tags=[name+':proxy',name+':backend'];containers=[name+'-proxy',name+'-backend'] try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'files').mkdir();(root/'nginx.conf').write_text(configuration) for path,content in (files or {}).items(): target=root/'files'/path;target.parent.mkdir(parents=True,exist_ok=True);target.write_text(content) docker('network','create',name) if backend is not None: (root/'backend.py').write_text(backend) (root/'backend.Dockerfile').write_text(f'FROM {PYTHON}\nCOPY backend.py /backend.py\nCMD ["python", "/backend.py"]\n') docker('build','--network','none','-f',str(root/'backend.Dockerfile'),'-t',tags[1],str(root)) docker('run','-d','--name',containers[1],'--network',name,'--network-alias','backend',tags[1]) wait(lambda:docker('exec',containers[1],'python','-c','import socket;s=socket.create_connection(("127.0.0.1",8081),timeout=.5);s.close()',check=False).returncode==0) (root/'Dockerfile').write_text(f'FROM {NGINX}\nCOPY nginx.conf /etc/nginx/nginx.conf\nCOPY files /srv\n') docker('build','--network','none','-t',tags[0],str(root)) docker('run','--rm','--network',name,tags[0],'nginx','-t') docker('run','-d','--name',containers[0],'--network',name,'-p','127.0.0.1::8080',tags[0]) details=json.loads(docker('inspect',containers[0]).stdout)[0] binding=details['NetworkSettings']['Ports']['8080/tcp'][0];assert binding['HostIp']=='127.0.0.1';port=int(binding['HostPort']) def ready(): connection=socket.create_connection(('127.0.0.1',port),timeout=.5);connection.close();return True wait(ready) def request(path,method='GET',headers=None,body=None): connection=http.client.HTTPConnection('127.0.0.1',port,timeout=5) try: connection.request(method,path,body=body,headers=headers or {}) response=connection.getresponse();return response.status,{k.lower():v for k,v in response.getheaders()},response.read() finally:connection.close() yield containers[0],request finally: docker('rm','-f',*containers,check=False);docker('image','rm',*tags,check=False);docker('network','rm',name,check=False) def config(locations,extra=''): return 'worker_processes 1;\nerror_log /tmp/error.log notice;\nevents {}\nhttp {\naccess_log off;\n'+extra+'\nserver { listen 8080; root /srv;\n'+locations+'\n}\n}\n' backend=r'''from http.server import BaseHTTPRequestHandler,HTTPServer import json class Handler(BaseHTTPRequestHandler): def do_GET(self): body=self.rfile.read(int(self.headers.get('Content-Length','0'))) document={'path':self.path,'method':self.command,'body':body.decode(),'headers':{key:self.headers.get(key) for key in ['Host','X-Forwarded-For','X-Real-IP','X-Forwarded-Proto','X-Forwarded-Host']}} data=json.dumps(document).encode() self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data)));self.send_header('X-Backend-Info','fixture-node');self.end_headers();self.wfile.write(data) do_POST=do_GET def log_message(self,*args):pass HTTPServer(('0.0.0.0',8081),Handler).serve_forever() ''' fields={'request':'$request','request_uri':'$request_uri','uri':'$uri','status':'$status','upstream_status':'$upstream_status','upstream_addr':'$upstream_addr','request_time':'$request_time','upstream_connect_time':'$upstream_connect_time','upstream_header_time':'$upstream_header_time','upstream_response_time':'$upstream_response_time','body_bytes_sent':'$body_bytes_sent','bytes_sent':'$bytes_sent'} extra="log_format details escape=json '"+json.dumps(fields,separators=(',',':'))+"';\nupstream service_pool { server 127.0.0.1:1 max_fails=1; server backend:8081 backup; }" locations=r''' access_log /tmp/access.json details; location = /local { return 204; } location = /rewrite { rewrite ^ /normalized last; } location = /normalized { return 200 "normalized"; } location /proxy { proxy_next_upstream error timeout; proxy_next_upstream_tries 2; proxy_pass http://service_pool; } ''' with lab(config(locations,extra),backend=backend) as (container,request): assert request('/local')[0]==204 assert request('/rewrite?x=1')[2]==b'normalized' status,headers,body=request('/proxy?x=1');assert status==200 def lines():return docker('exec',container,'cat','/tmp/access.json').stdout.splitlines() wait(lambda:len(lines())>=3) records={record['request_uri']:record for record in map(json.loads,lines())} print(json.dumps(records,indent=2),flush=True) local=records['/local'];assert local['status']=='204' and local['upstream_status']=='' and local['upstream_response_time']=='' assert local['body_bytes_sent']=='0' and int(local['bytes_sent'])>0 changed=records['/rewrite?x=1'];assert changed['request']=='GET /rewrite?x=1 HTTP/1.1' and changed['uri']=='/normalized' proxy=records['/proxy?x=1'];assert proxy['status']=='200' and proxy['upstream_status']=='502, 200' assert len(proxy['upstream_addr'].split(', '))==2 and int(proxy['body_bytes_sent'])==len(body) assert float(proxy['request_time'])>=0 and len(proxy['upstream_response_time'].split(', '))==2 print('a local 204 has no upstream and no body bytes, but still sends response-header bytes') print('the final uri reflects the internal redirect while request and request_uri retain the original request') print('one client-visible 200 records upstream attempts 502, 200 and separate attempt-address/timing lists') /local 不访问上游,直接返回 204。日志正文计数为零,bytes_sent 却为正,因为发送了响应头。在本例 JSON 转义格式中,缺失上游字段是空字符串。这不同于实际测得零秒的上游操作,也不同于实测版本默认转义路径使用的短横线。正确解析需要知道字段是否有测量,而不是仅看它能否转成数字。 ...

三月 1, 2027 · 2 分钟 · XB Field Notes
同一瞬间有不同偏移表示,重复的本地时刻则可能对应两个瞬间。

时间戳需要明确的时间语义

像 2026-11-01 01:30 这样的钟表标签,单独存在时不能确定唯一瞬间。解释它需要知道所在时区,在时钟回拨期间还可能需要说明是哪一次出现。反过来,已经明确纪元和单位的 Unix 时间戳,即使没有显示时区名称,也可以具有完整的瞬间语义。选择字符串格式之前,应先明确系统保存的究竟是哪一种时间值。 偏移不是完整时区规则 带数值偏移的时间戳说明本地字段如何对应 UTC。同一天的 12:00Z 和 20:00+08:00 可以表示同一瞬间。IANA 命名时区还包含历史及未来的民用时间规则,包括夏令时切换。安排下个月的洛杉矶本地约会时,固定的负七小时偏移不能替代 America/Los_Angeles,因为到时可能适用不同偏移。 对于已经发生的事件,可以保留无歧义瞬间,再按照读者需要的时区显示。但“每个工作日早上九点”是重复的民用时间意图,还需要保留本地日程及其适用时区。政策变化可能改变未来规则,需要可复现性时也应记录所用时区数据库版本。把今天的偏移固化成永远使用的日程,会悄悄改变原始业务含义。 UTC 格式并不证明时钟准确,也不证明因果顺序。两台机器都能打印 Z 后缀,但各自系统时钟仍可能有偏差;有限精度还会让多个事件得到相同时间戳。需要排序依据时,应保留事件标识及适当的因果上下文。本机经过时长则应选用合适的单调时钟,不能期待时区转换修复最初就测错的瞬间。 数值接口还必须注明采用秒还是毫秒,以及从哪个纪元开始计数。一个较大的整数本身不能完整说明单位;接收方把单位解释错了,之后再切换显示时区也无法恢复正确含义。单位应属于接口契约,而不是靠字段长度临时猜测。 坏掉的样子 任意时间戳字符串直接排序,可能得到与时间先后相反的结果。RFC 3339 对适合文字排序的条件有明确说明,包括统一的时区表示和小数精度。面对混合偏移,应先解析并归一化到瞬间,再进行比较。视觉上本地小时较晚的记录,换算到 UTC 之后可能反而更早,因此字符串字符比较不能替代时间语义解释。 秋季回拨可能让同一个本地小时出现两次,Python 的 ZoneInfo 通过 fold 区分这两个实例。春季前拨则可能让某个本地时刻根本不存在。仅仅把时区对象附到日期上,并不会自动拒绝缺失时间。接受本地约会输入的程序必须定义拒绝或解决歧义的策略,不能悄悄替用户猜测他想选择哪一个瞬间。 日期运算也要与目标一致。两个 datetime 共享同一个 tzinfo 时,相关运算可能使用钟表字段语义,忽略 fold 所表达的瞬间区别。如果需要比较实际经过时间,应先分别转换到 UTC。这个选择需要明确,因为跨越时钟切换时,本地表盘读数之差可能与真正经过的秒数不同,二者都不能无条件替代另一种含义。 快速验证 将下面程序保存为 timezone.py,使用 Python 3.11 或更新版本执行 python3 timezone.py,并确保环境中有 IANA 时区数据。实测环境是 Python 3.11.2 和 Debian tzdata 2026b。程序全部使用固定日期,不会修改机器的时钟,也不会更改系统时区;不同数据库版本需要重新检查相应规则。 ...

十一月 29, 2026 · 2 分钟 · XB Field Notes
Logrotate 创建新活动文件,明确实现的重开钩子再把写者切换过去。

Logrotate 在执行时应用轮转策略

Logrotate 在被调用时读取配置,轮转符合条件的日志,并执行选定的保留与压缩动作。何时检查由定时器或其他调度器决定,它不会持续限制文件大小,改名也不会自动重定向应用的打开描述符。完整设置需要把检查频率、文件策略和写者实际重开行为结合起来,才能判断日志确实按预期流转。 坏掉的样子 两次执行之间可能超过 size 阈值,突发写入或调度停止时尤其明显。rotate 限制的是保留代数,不是历史时间,也不覆盖全部临时磁盘使用。因此,阈值乘代数不是硬容量保证;活动文件、检查间隔内增长,以及压缩或复制工作,都需要空间。只看到配置里有两个有限数字,不能证明磁盘不会耗尽。 create 指定替代文件的模式和所有权,不会向应用发送信号。postrotate 可以调用受支持的重开机制,但 HUP 并不普遍表示重载。delaycompress 只把压缩推迟一个轮转周期,无法修复从不放弃旧 inode 的写者。每项指令负责不同动作,需要核对它们是否共同完成实际交接。 脚本可能在文件操作已经成功后失败。Logrotate 返回非零,不会回滚全部改名和创建;未检查文件及写者状态就反复强制轮转,可能再次切分或删除需要的代数。应诊断已经发生的部分结果,而不是把失败理解成没有变化。错误恢复也必须依据当前事实,而不能只看最后退出码。 快速验证 把脚本保存为 logrotate-cli.py,在装有 Python、logrotate 和 gzip 的一次性 Linux 容器中,以 root 运行 python3 logrotate-cli.py。实测使用 Linux 6.12.76、Python 3.11.2、logrotate 3.21.0 和 ext4。样本使用私有配置与状态文件,只有一个日志子进程,其 HUP 处理函数明确重开自己的临时日志: 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() 配置选择六十四字节阈值、两代保留、延迟一个周期的压缩,以及 root 所有、0600 模式的新文件。钩子只向受控子进程已知 PID 发信号。写者对写入和重开都发出确认,让控制进程验证交接,而不是仅凭发送命令成功就推测新路径已经被使用。 ...

十一月 21, 2026 · 2 分钟 · XB Field Notes
三个相连的方块,依次标注观察、决策与行动。

指标与日志都需要明确统计总体

指标与日志是对已观测行为的不同表示,并不是互相竞争的真相定义。指标通常在一组维度下汇总测量结果,日志记录则描述一个事件,并能够携带更具体的上下文。只要数据覆盖与语义支持,两者都可以回答聚合问题。不能因为某张仪表盘更整齐,就认为它天然更权威;也不能因为日志字段很多,就默认其中包含完整且可直接比较的业务总体。 对于结账服务,一个已完成请求计数器可以把每个最终结果归为成功或错误。路由、结果等少量有界维度,方便持续比较流量与失败比例。诊断日志则可以利用关联编号、错误类型和依赖响应,解释某一次失败请求的具体原因。这两类信息互相帮助,但它们采用的分母仍然需要分别确定。共享服务名称或时间图形,并不足以证明两份数据统计了相同的请求集合。 先决定测量包含什么 累计计数器本身不是每秒请求数。实际速率查询需要跨时间观测,并考虑进程重启等导致的计数重置。不过,一个起点、终点和成员都明确的完整合成批次,可以直接计数。不能悄悄把生产中的时间窗口速率换成两个无关快照的比例,也不能把经过选择的事件流与覆盖全部请求的计数器混为同一个统计总体。计算形式相似,并不代表时间口径已经一致。 日志也不是只能靠人逐条阅读。只要终态事件结构稳定且记录完整,就能聚合出与请求计数器相同的总量。选择时需要考虑事件体积、索引、查询延迟、保留期限和维护复杂度。指标经常适合反复执行的汇总查询,但如果每个请求都生成一个新的标签值,指标存储也可能迅速变得昂贵。究竟哪种方式成本较低,取决于实际维度和使用方式,而不是单凭指标或日志这个名称判断。 坏掉的样子 假设一百个已完成请求里有四个失败。为了排查问题,日志系统或人工查询特意保留其中五个请求,里面包含两个失败。完整总体的失败比例是百分之四,被选日志的失败比例则是百分之四十。两次算术运算都没有错,但把后者宣称为整个服务的失败率就错了,因为这五条记录既不是全部请求,也没有证明是能够代表总体的样本。排障数据的选择目的必须与统计推断分开。 把重试过程中的失败次数当成最终失败请求数,也会造成类似误差:同一个请求的一次尝试可能失败,随后却成功完成。日志缺失、重复投递、时间窗口不同和路由归类不一致,都可能令看起来相关的图表互相矛盾。用于可靠性计算时,应明确每个请求只有一个最终结果,同时保留尝试级证据用于诊断。发现不一致后,不能只改图表标题,或者简单平均两个比例来掩盖口径差异。 快速验证 将代码保存为 metrics-logs.py,在具备 Python 3.14.4 和 Docker 的环境中执行。脚本使用 prom/prometheus:v3.5.1 镜像中的真实 promtool 检查指标文本格式。临时容器不接入网络、不挂载宿主机目录,结束后会删除。工作负载是人为构造的完整集合,没有抓取生产请求,也没有声称这些选择规则代表真实随机采样;实验只把两个不同总体的差异明确呈现出来。 import json import subprocess import uuid from collections import Counter # A complete synthetic population, with exactly one terminal outcome per request. requests = [{"request_id": f"r{i}", "route": "/checkout", "outcome": "error" if i in {7, 18, 53, 94} else "success"} for i in range(1, 101)] counter = Counter(request["outcome"] for request in requests) selected_logs = [request for i, request in enumerate(requests, 1) if i in {1, 7, 18, 33, 55}] logged_errors = sum(record["outcome"] == "error" for record in selected_logs) text = "# HELP demo_requests_total Completed synthetic requests.\n# TYPE demo_requests_total counter\n" for outcome, count in sorted(counter.items()): text += f'demo_requests_total{{route="/checkout",outcome="{outcome}"}} {count}\n' name = "xbcatm-metric-format-" + uuid.uuid4().hex[:12] try: result = subprocess.run( ["docker", "run", "--rm", "-i", "--name", name, "--network", "none", "--entrypoint", "/bin/promtool", "prom/prometheus:v3.5.1", "check", "metrics"], input=text, text=True, capture_output=True, timeout=30, ) assert result.returncode == 0, result.stdout + result.stderr finally: subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) population_rate = counter["error"] / len(requests) selected_rate = logged_errors / len(selected_logs) assert population_rate == 0.04 and selected_rate == 0.4 assert "request_id" not in text and len(counter) == 2 # Full event coverage would recover the same aggregate; these five logs cannot. assert sum(r["outcome"] == "error" for r in requests) == counter["error"] print(json.dumps({"requests": len(requests), "errors": counter["error"], "population_error_rate": population_rate, "exposed_metric_series": len(counter), "selected_logs": len(selected_logs), "selected_log_errors": logged_errors, "selected_log_error_rate": selected_rate, "exposition_validated_by": "promtool 3.5.1"}, indent=2)) 输出应包含一百个请求、四个错误及零点零四的总体比例。筛选后的五条日志中有两个错误,得到零点四。导出文本只有两条按结果区分的计数序列,标签中不包含请求编号。Promtool 负责验证格式,Python 断言负责验证算术结果和事先定义的成员关系。这不是实际抓取流程、计数器重置处理或存储费用基准测试,因此不能把格式检查通过扩大为整套监控链路已经验证。 ...

十月 16, 2026 · 2 分钟 · XB Field Notes