自动化

由于它依据一段时间序列的平均值而不是单个尖峰来决定动作，你往往还要配一个最大副本数，避免扩张失控。 把平滑后的均值当作决策依据 HPA 不会拿一个瞬时读数当依据;它先按窗口对观测值取平均、做平滑,再拿去和期望值比较。这样是为了避免一次一秒的尖峰就重画副本数:先聚合,等到平滑后的数值真正越过了目标,再去动副本。因此真正要调的是三件事:指标源、聚合口径(按 Pod 平均还是集群总量),以及取样的窗口。窗口太短会放大噪声,太长又追赶不上需求。动手改 HPA 之前,先把文档里「怎么聚合、多久算一次」那段读明白,再做任何别的。 选一个你真正能改进的指标 选指标才是设计核心,因为伸缩的上限由目标定。CPU 在任何集群上都好取、对 CPU 密集的负载也顺,可它是个代理:真正排队等锁、等数据库的服务,CPU 往往是闲着的,变坏的是延迟。请求速率、或一个自定义的 queue_depth 这类指标测起来更麻烦,反映的却是用户真正感受到的东西。选一个你能对得上的指标:当它越线时,真实的体验确实变了;并且让 HPA 用和应用同样的口径去定义负载,于是它收敛的是那个让体验变差的数字,而不是那个恰好便宜的抄近道数。 自动化与初始值无关 replicas 的初值是起点,不是 HPA 的依据。一旦 HPA 接管,副本数归它管,初值只是下限或出发点。二者在最好的意义上互不相关:关掉控制器,你仍然有一个稳定、明码的陈述基数;开着一个无人调整的 HPA,它也只管自己那部分。所以下限要现实(它才是那个地板),上限要挡住成本风险的峰值,而初值只负责让最开始的几秒体面一点——在控制器的头几次采样落地之前,它撑起第一段安全期。 坏掉的样子 自定义指标不可用时不会自动回退到中央处理器。接口返回错误或空序列时，控制器把对应指标标为无效，把扩缩条件置为假，然后保持当前副本数而不是猜一个数。只配了队列深度的工作负载，会在最需要扩容时原地冻结。正确做法是把中央处理器与自定义指标并列配置，取算出副本数更多的一方，这样队列信号丢失时仍有降级可用的输入，至少不会把服务钉在原地。 适配器链路还会引入延迟。抓取间隔与缓存叠加后，读到的可能是几分钟前的队列深度，于是对已排空的工作继续加副本，造成超调。先看序列时间戳与查询延迟，再决定是否放宽目标值，每个工作负载只暴露一条预聚合序列，不要把原始高基数序列直接喂给扩缩器。 基数爆炸会从另一侧压垮同一条链路。按用户或按请求打标签，会把一个信号膨胀成数万条序列，拖慢接口，让拉取超时，读数翻转为未知。应先在监控侧聚合，只暴露工作负载级的平均值或总量，收敛序列规模后再对接自动扩缩。 目标类型配错同样致命。自定义指标只支持数值与平均值，利用率只适用于配好资源申请的资源指标。没有资源申请就用中央处理器利用率，或拿内存利用率当延迟代理，都会长期未知。先确认哪一路报未知，再把类型改成链路真正提供的那种，并补齐缺失的资源申请。 快速验证 把下面的代码存为水平扩缩模拟脚本并运行。它只用标准库，不碰任何集群，复刻两条核心规则：多指标取算出副本数更多的一方，缺失指标直接跳过，全部缺失时保持副本数不变，便于本地复现线上行为。 import math import sys def desired_for(current, value, target): """单个指标算出的期望副本数；返回空表示该路指标不可用。""" if value is None or target is None or target <= 0: return None return math.ceil(current * value / target) def decide(current, metrics, min_replicas, max_replicas): """多指标取最大值；全部缺失时保持当前副本数不变。""" wants = [] for _, value, target in metrics: want = desired_for(current, value, target) if want is not None: wants.append(want) if not wants: return current return max(min_replicas, min(max_replicas, max(wants))) def check(actual, expected, message): if actual != expected: raise AssertionError(f"{message}：实际 {actual}，期望 {expected}") print(f"OK: {message} -> {actual}") def main(): current = 4 cpu = ("cpu", 80, 50) # 4 * 80/50 = 6.4 -> 7 queue = ("queue", 900, 300) # 4 * 900/300 = 12 check(decide(current, [cpu, queue], 1, 20), 12, "多指标取最大值") check(decide(current, [cpu, ("queue", None, 300)], 1, 20), 7, "一路缺失仍按可用指标扩容") check(decide(current, [("cpu", None, 50), ("queue", None, 300)], 1, 20), 4, "全部缺失时保持副本数") check(decide(current, [cpu, queue], 1, 10), 10, "最大副本数依然限幅") print("PASS: 4 scenarios; max-wins and hold-on-missing verified") print(sys.version.split()[0]) if __name__ == "__main__": main() 四个场景锁定关键行为：合并场景证明中央处理器要七个而队列要十二个时取十二个；单个缺失证明一路失败仍由中央处理器接管；全部缺失证明保持不动而非缩零；限幅场景证明算完后最大副本数依然生效。预期输出以通过信息结尾，说明四组场景成立，最大值优先与缺失保持均已验证，并打印解释器版本。 ...

Cron 按日历字段匹配时间，并在匹配时启动命令。它不会天然等待同一个命令的上一次执行结束。如果每分钟触发的任务要运行七十秒，下一分钟就可能再启动一份。因此，触发规则、并发规则和成功完成状态必须分别设计，不能因为写好了一个周期表达式，就认为重复工作的所有行为都已经确定。 把字段理解为日历选择 传统用户 crontab 一行依次包含分钟、小时、月内日期、月份、星期，然后是命令。星号表示该字段的全部取值。分钟位置的 */15 是在每个小时内选择相应分钟值，不是从上次任务结束之后再计时十五分钟。步长如果不能均匀划分字段范围，跨过边界时就可能出现不同长度的间隔，不能把它当作连续经过时长。 Debian/Vixie 格式对于两个日期字段有特殊规则：当月内日期和星期都受到限制时，满足其中一个日期条件就可能运行，而分钟、小时和月份限制仍然必须满足。这不是月份与星期之间的“或”关系。安装简短表达式之前，应列出具有代表性的实际日期核对意图，避免以为“每月某日恰好是周五”却写成两个日期条件各自都能触发。 时区、夏令时变化和停机都会影响日历调度，不同实现对跳过或重复时段的处理并不完全相同。Debian 文档为部分小幅时钟调整和固定时间任务定义了特殊行为，从其他调度器复制结论可能不适用。本实验没有修改时钟，也没有覆盖这些切换情况。日历触发同样不等于持久队列，不能自动保证停机期间错过的每次任务都会补跑。 坏掉的样子 重叠任务可能重复产生副作用、争夺资源或相互覆盖文件。如果要求同一时刻只有一个实例，就要明确实现这一政策。非阻塞建议锁可以让后来的实例记录跳过；阻塞锁则可能留下等待进程。两种选择都不自动提供业务效果恰好一次，也不能修复已经执行了一半的操作。是否允许跳过，应来自任务自己的业务要求。 本地文件锁只协调使用同一相关锁机制的参与者，不会自动串行化无关主机上的工作。网络文件系统的锁语义也需要单独确认。锁必须覆盖实际工作生命周期，跳过结果则应能被观察。如果一个实例永久卡住，后来每次触发都静默丢弃，表面上调度器还在工作，但业务可能早已停止推进，因此还要监控最近成功完成距今多久。 Cron 也并非一律没有日志。Debian 默认通过 syslog 记录任务启动，并提供更细的日志选项。命令输出可能在可用邮件链路存在时被投递，但极简容器经常没有这样的链路。应把启动、完成、失败和跳过结果写到明确目的地。启动记录只能证明任务被派发，不能证明真正的应用工作已经成功完成。 快速验证 使用全新的临时 Debian 12 Docker 容器，以 root 运行，确保没有既有 root 任务表或 cron 进程。通过 apt 安装 python3、cron 和 procps，将下面代码保存为 cron-overlap.py，再执行 python3 cron-overlap.py。需要预留约两分钟，因为程序等待的是实际日历触发，没有用模拟调度器替代 cron，也没有加速或调整宿主时钟。 from pathlib import Path from tempfile import TemporaryDirectory import json import os import subprocess import time assert os.geteuid() == 0 and Path('/.dockerenv').exists(), 'use a disposable root Linux container' assert subprocess.run(['pgrep', '-x', 'cron'], capture_output=True).returncode == 1 assert subprocess.run(['crontab', '-l'], capture_output=True).returncode != 0 with TemporaryDirectory(prefix='cron-overlap-') as directory: root = Path(directory) events_path = root / 'events.jsonl' stop = root / 'stop' worker = root / 'worker.py' worker.write_text('''from pathlib import Path import fcntl,json,os,sys,time root=Path(__file__).parent kind=sys.argv[1] def record(event): data=json.dumps({'event':event,'kind':kind,'pid':os.getpid(),'monotonic':time.monotonic()})+'\\n' fd=os.open(root/'events.jsonl',os.O_WRONLY|os.O_CREAT|os.O_APPEND,0o600) try: os.write(fd,data.encode()) finally: os.close(fd) lock=None if kind=='locked': lock=open(root/'job.lock','a') try: fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB) except BlockingIOError: record('skip') raise SystemExit(0) record('start') try: until=time.monotonic()+70 while time.monotonic()<until and not (root/'stop').exists(): time.sleep(0.1) finally: record('end') if lock: lock.close() ''') table = ''.join(f'* * * * * /usr/bin/python3 {worker} {kind} >>{root}/worker.log 2>&1\n' for kind in ('plain', 'locked')) subprocess.run(['crontab', '-'], input=table, text=True, check=True) daemon = subprocess.Popen(['cron', '-f'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def read_events(): if not events_path.exists(): return [] return [json.loads(line) for line in events_path.read_text().splitlines()] try: deadline = time.monotonic() + 140 while time.monotonic() < deadline: assert daemon.poll() is None, 'cron exited' events = read_events() plain = [e for e in events if (e['kind'], e['event']) == ('plain', 'start')] locked = [e for e in events if (e['kind'], e['event']) == ('locked', 'start')] skipped = [e for e in events if (e['kind'], e['event']) == ('locked', 'skip')] if len(plain) >= 2 and skipped: break time.sleep(0.2) else: raise AssertionError('two real cron ticks were not observed') assert len(plain) == 2 and len(locked) == 1 and len(skipped) == 1 assert plain[1]['monotonic'] - plain[0]['monotonic'] < 70 assert not any(e['event'] == 'end' and e['pid'] == plain[0]['pid'] for e in events) print(json.dumps({'actual_cron_starts': len(plain), 'overlap_observed': True, 'seconds_between_starts': round(plain[1]['monotonic']-plain[0]['monotonic'], 3), 'locked_starts': len(locked), 'locked_skips': len(skipped)}), flush=True) finally: daemon.terminate() stop.touch() daemon.communicate(timeout=5) subprocess.run(['crontab', '-r'], check=True) deadline = time.monotonic() + 5 while time.monotonic() < deadline: events = read_events() started = {e['pid'] for e in events if e['event'] == 'start'} ended = {e['pid'] for e in events if e['event'] == 'end'} if started <= ended: break time.sleep(0.1) else: raise AssertionError('test workers did not finish') 两条每分钟执行的记录会运行原本持续七十秒的工作进程。plain 分支没有锁；locked 分支使用共享文件和非阻塞独占 flock，未取得锁时记录 skip 并退出。每个事件包含进程标识和单调时间，能够直接区分不同实例，并检查第二次启动发生时第一次是否仍在执行，而不是仅根据配置推测并发。 ...

每个 PR 都叠加 CI 结果，会让 review 更诚实。那份 YAML 与要检查的代码一起版本化，就是这层契约。 YAML 就是契约 工作流 YAML 同时声明了「何时运行」和「以什么权限运行」,所以它就是整份契约,而不只是信任的起点:on: 列出触发器,permissions: 钉住它能碰的东西,jobs: 声明作业和步骤。因为 on 在任何 job 跑之前就被解析,它也正是你决定「push 必须验证通过才允许合并」或「release 标签可以无人值守触发」的地方。把制定文件跟代码一起入库,让一个分支的 CI 和该分支自己的契约始终同版本、同步调。 job 与 step 有序执行 job 默认并行,step 在 job 内顺序执行;两张规则叠在一起,才组成整张图。一个失败的 step 会让它所在 job 后面的步骤取决于条件;能否继续后面 job,由 needs、if 和 job 级并发决定。这些声明把工作流变成一座真的闸门——红了就拦住合并,而不是跑完就撒手。实际工程里把必检的和快的放在前面、贵重的放在后面,让一次失败在廉价处就显形,而不是耗半天才喊停。 ...

systemd timer 把调度规则与执行工作的服务分开。定时器决定何时请求激活，服务则定义命令、执行身份、环境、超时和结果。因此，维护任务可以使用普通单元工具检查，但一个按时发出的启动请求，并不能证明备份、清理或报表已经正确完成。调度是否运转与工作是否成功，需要不同证据；只有两者都符合要求，周期任务才真正达到目的。 坏掉的样子 活动的定时器可以与失败的服务同时存在。如果监控只检查 timer 是否活动，就会漏掉真正执行工作的失败。同样，最近触发时间只说明调度发生过，不代表业务输出成功。应检查服务结果，以及任务本来要生成的产物或状态变化，并保留足够上下文区分前后两次激活，避免把昨天留下的文件误当成今天任务已经完成。 除非通过 Unit 指向别的目标，定时器通常激活同名的 service。如果目标服务已经活动，就不会再启动另一份独立副本。这阻止的是同一个单元重叠激活，不是分布式锁，也不能阻止另一个服务执行相同命令。oneshot 如果使用 RemainAfterExit=yes 保持活动，后续计划执行就可能一直不再运行命令；保留状态是否合适，取决于任务真正需要表达什么。 停止定时器会停止后续调度，但不会自动停止已经开始的服务调用。事故期间暂停维护工作时，这个区别尤其重要：需要检查目标服务，再有意决定让当前工作结束还是中断它。即使停止命令成功，先前已经发生的文件或数据库修改也不会因此自动回滚，因此不能把“停了 timer”当成系统已经没有维护副作用。 快速验证 在 PID 1 为 systemd 的一次性 Linux 系统中，以管理员身份运行下面的 Python 脚本。实测环境是 Debian 12、systemd 252.39。它创建唯一命名的定时器与服务，以及临时工作程序。工作程序记录真实进程进入和完成时刻，成功任务则通过显式释放文件等待，以便直接观察调度行为： from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,time,uuid name='xbcatm-timer-'+uuid.uuid4().hex[:8];service=name+'.service';timer=name+'.timer' paths={unit:Path('/run/systemd/system')/unit for unit in (service,timer)} assert all(not p.exists() for p in paths.values()) 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 value(unit,key):return run('systemctl','show',unit,'-p',key,'--value') def wait_for(check): deadline=time.monotonic()+5 while not check(): assert time.monotonic()<deadline,'transition timed out' time.sleep(0.02) with TemporaryDirectory(prefix=name+'-',dir='/run') as directory: root=Path(directory);worker=root/'worker.py' worker.write_text(r'''from pathlib import Path import json,sys,time root=Path(__file__).parent;number=len(list(root.glob('start-*.json')))+1 def record(kind): temp=root/'record.tmp';temp.write_text(json.dumps(time.monotonic())) temp.replace(root/(kind+'-'+str(number)+'.json')) record('start') if (root/'fail').exists():sys.exit(7) while not (root/('release-'+str(number))).exists():time.sleep(0.01) record('end') ''') paths[service].write_text(f'[Service]\nType=oneshot\nTimeoutStartSec=10s\nExecStart=/usr/bin/python3 {worker}\n') definition='[Timer]\nOnActiveSec=100ms\nOnUnitInactiveSec=10s\nAccuracySec=1ms\n' paths[timer].write_text(definition);(root/'fail').touch() try: run('systemd-analyze','verify',*[str(p) for p in paths.values()]) run('systemctl','daemon-reload');run('systemctl','start',timer) wait_for(lambda:value(service,'ActiveState')=='failed') assert value(service,'ExecMainStatus')=='7' and value(timer,'ActiveState')=='active' run('systemctl','stop',timer) print('timer active while its first job failed with exit status 7') (root/'fail').unlink();run('systemctl','reset-failed',service) paths[timer].write_text(definition.replace('OnUnitInactiveSec=10s','OnUnitInactiveSec=250ms')) run('systemctl','daemon-reload');run('systemctl','start',timer) wait_for(lambda:(root/'start-2.json').exists()) time.sleep(0.65) assert len(list(root.glob('start-*.json')))==2 (root/'release-2').touch();wait_for(lambda:(root/'start-3.json').exists()) gap=json.loads((root/'start-3.json').read_text())-json.loads((root/'end-2.json').read_text()) assert gap>=0.25,gap print('no overlapping launch; next job starts at least 250ms after prior deactivation') pid=value(service,'MainPID');assert pid!='0' run('systemctl','stop',timer) assert value(service,'MainPID')==pid and value(service,'ActiveState')=='activating' print('stopping timer leaves its already running oneshot alive') (root/'release-3').touch();wait_for(lambda:value(service,'ActiveState')=='inactive') time.sleep(0.4) assert len(list(root.glob('start-*.json')))==3 and value(service,'Result')=='success' print('released job completes; stopped timer schedules no further run') finally: run('systemctl','stop',timer,service) for unit,path in paths.items(): if value(unit,'ActiveState')=='failed':run('systemctl','reset-failed',unit) path.unlink() run('systemctl','daemon-reload') 第一次触发的工作进程故意以状态七退出，服务变为失败，定时器却继续活动。此时相对服务结束的下一次间隔先设置为十秒，让脚本有时间检查这个状态，并在再次尝试前停止定时器。结果证明调度器完全可能正常工作，而它安排的业务任务一直失败；仅凭 timer 状态不能得出任务可靠的结论。 ...

Ansible 的 playbook 记录的是机器应该拥有的状态。部署中途崩溃后，下一次运行会先观察真实现状，只补上缺少的部分，而不是从头盲目重放每条命令。它是一份可以执行的契约：inventory 决定对象，模块实现比对，操作者审阅即将发生的变化。 只写终点，不写每一步 命令式 shell 脚本会说“按顺序执行这些命令”；playbook 则说“这个包应该已安装、这个文件应该是这些字节、这个服务应该启用并运行”。模块在动作前比较目标主机与目标状态，已经满足的任务报告 ok，只有不一致才报告 changed。同一份声明可以用于全新的机器、只配置了一半的机器，也可以用于被人手工改过的机器。 这里的目标不能只是模糊意图。state: present、模板的精确内容、所有者和权限、服务的 enabled/running 都是契约的一部分。把多台机器共用的值放到变量或 role 没问题，但最终值应该对审阅者可见。隐藏的默认值，或者绕过模块直接改变状态的命令，都会让这份契约难以推理。每一个 task 都应回答“运行完成后应该看到什么”，而不是“我习惯先敲哪一条命令”。 模块让重跑逐渐收敛 多数内置模块都围绕幂等性设计。ansible.builtin.copy 会先比较文件内容和元数据，只有不同才替换；ansible.builtin.package 向包管理器询问目标状态；ansible.builtin.service 只在启用状态或运行状态不符时调整。第二次运行应该是一份安静的确认，而不是再次重启、追加重复行或再写一条记录。幂等不是“什么都不做”，而是“已经达到目标，所以没有必要再做”。 并非所有动作天然幂等。直接用 command 创建用户、轮换密钥或调用外部 API，若没有条件限制就会重复副作用。优先找专用模块；没有时先 register 查询结果，再用 when 判断是否执行，或者使用精确的 creates/removes 文件条件。changed_when 只决定变更报告和 handler 通知，不控制任务是否执行；设成 false 不能阻止重复变更，还可能隐瞒真实副作用。对外部接口仍要定义超时、重试与部分成功后的恢复方法，不能把网络调用假装成普通文件复制。 Inventory 是地图 playbook 描述结果，inventory 选择主机，变量提供上下文。ansible_user、SSH agent 或 Vault 可以提供凭据而不把秘密写进仓库。组变量让同一份声明覆盖整支机器队伍，同时保留端口、区域等主机差异。在 CI 中使用显式 inventory，避免“localhost”意外指向执行器本身。执行前打印解析后的主机集合，确认别名、组继承和变量优先级都与审阅时一致。 ...

一份计划必须先说清楚“时间”是什么意思。“纽约每天早上九点”指向民用日历，“任务结束十五分钟后”则指向经过时长和一个参考事件。systemd timer 同时支持这两类表达，但它们使用的时钟与起算点不同，在时区变化、系统休眠和定时器停用期间会表现出不同结果。只把它们统称为周期调度，会隐藏本来应该由需求决定的差别。 坏掉的样子 OnCalendar 按墙钟日历表达式触发，省略时区通常使用系统本地时区，因此同一份配置在不同机器上可能代表不同实际时刻。夏令时规则变化时，每天相同的本地时间也可能相对 UTC 移动。应使用固定参考日期预览预期表达式，而不是把今天的 UTC 偏移当成全年不变的事实。调度语法能解析，只证明表达式有效，不证明它符合业务要求的地区时间。 单调计时选项明确命名参考事件。OnActiveSec 从 timer 本身激活起算，OnStartupSec 从服务管理器启动起算，OnUnitInactiveSec 从目标单元结束活动起算。这些并不是“隔几分钟运行”的可互换写法。普通单调计时在系统休眠期间暂停，WakeSystem 会改变时钟选择，并依赖平台支持。因此是否计入休眠时间，也应该属于调度需求，而不是部署后的偶然发现。 Persistent=yes 对日历计时器生效。它记录触发时刻，在定时器恢复活动且至少错过一个事件时请求一次补触发；不会逐次重放所有错过的日历时刻，也不能证明上次业务执行成功。如果配置了随机延迟，补触发仍受其影响。触发历史不是一条持久业务队列，不能据此推定停机期间应做的所有工作都已完整保存。 快速验证 在 PID 1 为 systemd、安装了时区数据库的一次性 Linux 系统中，以管理员身份运行下面的 Python 脚本。记录中的环境为 Debian 12、systemd 252.39。脚本先预览日历表达式，再安装两个唯一命名的定时器与服务组合，双方使用同样三个即将到来的日历时刻，全程不会调整机器时钟： from pathlib import Path from tempfile import TemporaryDirectory from datetime import datetime,timezone import subprocess,time,uuid name='xbcatm-calendar-'+uuid.uuid4().hex[:8] names=[name,name+'-plain'];paths=[] 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() preview=run('systemd-analyze','calendar','--base-time=2026-03-07 00:00:00 UTC', '--iterations=3','*-*-* 09:00:00 America/New_York') for expected in ('2026-03-07 14:00:00 UTC','2026-03-08 13:00:00 UTC','2026-03-09 13:00:00 UTC'): assert expected in preview,preview print('09:00 New York: 14:00 UTC before spring DST, 13:00 UTC after it') with TemporaryDirectory(prefix=name+'-',dir='/run') as directory: root=Path(directory);worker=root/'worker.py' worker.write_text("from pathlib import Path\nimport sys\np=Path(sys.argv[1])\nwith p.open('a') as output:output.write('run\\n')\n") now=int(time.time());events=[now+3,now+4,now+5] calendar=''.join('OnCalendar='+datetime.fromtimestamp(t,timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')+'\n' for t in events) try: for index,prefix in enumerate(names): service=Path('/run/systemd/system')/(prefix+'.service');timer=service.with_suffix('.timer') assert not service.exists() and not timer.exists() service.write_text(f'[Service]\nType=oneshot\nExecStart=/usr/bin/python3 {worker} {root}/{index}.log\n');paths.append(service) timer.write_text('[Timer]\n'+calendar+'AccuracySec=1ms\nRandomizedDelaySec=0\nPersistent='+('yes' if index==0 else 'no')+'\n');paths.append(timer) run('systemd-analyze','verify',*[str(p) for p in paths]);run('systemctl','daemon-reload') run('systemctl','start',name+'.timer');run('systemctl','stop',name+'.timer') assert not (root/'0.log').exists() and time.time()<events[0] while time.time()<events[-1]+0.15:time.sleep(0.05) run('systemctl','start',name+'.timer',name+'-plain.timer') deadline=time.monotonic()+3 while not (root/'0.log').exists(): assert time.monotonic()<deadline,'catch-up did not run';time.sleep(0.02) time.sleep(0.15) assert (root/'0.log').read_text()=='run\n' and not (root/'1.log').exists() print('three elapsed calendar events: persistent timer catches up once; plain timer does not') run('systemctl','stop',name+'.timer');run('systemctl','start',name+'.timer') time.sleep(0.2);assert (root/'0.log').read_text()=='run\n' print('reactivating with no newly missed event does not replay the job') print('timer inactivity was tested without changing the system clock or rebooting') finally: if paths: run('systemctl','stop',*[p.name for p in paths]) run('systemctl','clean','--what=state',name+'.timer') for path in paths:path.unlink(missing_ok=True) run('systemctl','daemon-reload') 预览从二〇二六年三月夏令时切换前的固定日期开始，询问纽约早上九点的后续发生时刻。第一次对应 UTC 十四点，后两次对应 UTC 十三点，本地钟表时间却始终是九点。这检查的是日历解释，并不是修改主机时区，也不是把系统时间跳到那些日期。对指定地区的每日任务而言，这种偏移变化可能正是正确行为。 ...

调度器启动的是一次尝试，业务关心的通常是某个逻辑任务是否完成。一笔每日额度可能因为报错、重启或人工补跑而尝试多次。如果每次尝试都生成新的身份，重复执行就无法与新工作区分。应使用能够代表预期任务的键，例如任务名称与名义计划时间，并在修改业务状态之前定义重复键的含义。尝试日志中的唯一编号可以保留，但不能代替业务层面的去重身份。 调度触发与业务执行是两件事 不能假设所有调度器都会补跑遗漏工作。普通 cron 的具体行为随实现变化，没有统一补跑承诺。启用 Persistent=true 的 systemd 日历定时器，在停用期间至少错过一次触发时，可以在重新启用后补触发；这不等于把所有错过的时间点逐个重放。Kubernetes 文档也明确说明，某些情况下 CronJob 可能重复创建 Job，或者没有创建对应任务。 因此要先确定业务规则：过时任务直接跳过、每个缺失周期都处理，还是对当前状态做一次校正。去重键应使用名义上的逻辑周期，并明确时区。实际启动时间会随重试变化，不能用于识别同一次计划工作。诊断需要的尝试编号与业务身份应分开记录，这样才能同时看见执行了几次尝试，以及到底完成了几个逻辑周期。 坏掉的样子 先查询有没有处理过，再单独修改数据，存在竞态。两个工作进程可能都看到键不存在，然后都应用业务效果。进程锁也许能阻止一台机器上的重叠运行，却不能让业务更新与完成标记文件自动形成原子操作。如果更新成功后、写标记之前失败，重试就会重复生效；先写标记再执行业务，则可能把尚未发生的工作误记为完成。调整两个步骤的先后顺序不能消除中间窗口。 当业务状态与完成记录位于同一个事务数据库时，应在唯一约束下共同提交。提交之前失败，两者都不应留下；提交之后重试，应识别已经完成的逻辑键。SQLite 的 BEGIN IMMEDIATE 会提前开始写事务，其他写入者可能等待，也可能因为繁忙而失败。超时限制等待长度，并不保证任何并发量下都不会冲突；调用方仍需正确区分重试、失败和重复完成。 这个原子性边界不覆盖外部邮件、支付或 HTTP 调用。本地回滚无法撤销远端已经接受的效果。此类流程需要远端提供合适的幂等契约，或采用持久投递设计并在接收方去重。同一个键如果可能携带不同负载，还需要明确拒绝这种冲突的规则。下面示例始终增加固定额度，因此只验证同一逻辑操作的重复尝试，没有把变更参数的重用键视为新请求。 快速验证 保存为 idempotent-jobs.py 后运行 python3 idempotent-jobs.py。脚本使用临时 SQLite 数据库和八个工作线程，每个线程建立独立数据库连接。业务效果是给余额增加十个单位；非空主键记录已经完成的每日逻辑额度发放，并与余额修改处于同一个事务中。实验不需要实际修改系统定时器，也不会访问外部服务。 from concurrent.futures import ThreadPoolExecutor from pathlib import Path from tempfile import TemporaryDirectory from threading import Barrier import json import sqlite3 with TemporaryDirectory() as directory: database = str(Path(directory) / 'jobs.db') with sqlite3.connect(database) as connection: connection.executescript(''' CREATE TABLE balance (id INTEGER PRIMARY KEY, units INTEGER NOT NULL); INSERT INTO balance VALUES (1, 0); CREATE TABLE completed (logical_key TEXT PRIMARY KEY NOT NULL); ''') def run(key, fail=False, gate=None): connection = sqlite3.connect(database, timeout=10, isolation_level=None) try: if gate: gate.wait(timeout=5) connection.execute('BEGIN IMMEDIATE') inserted = connection.execute( 'INSERT INTO completed VALUES (?) ON CONFLICT(logical_key) DO NOTHING', (key,)) if not inserted.rowcount: connection.execute('COMMIT') return 'duplicate' connection.execute('UPDATE balance SET units = units + 10 WHERE id = 1') if fail: raise RuntimeError('injected failure before commit') connection.execute('COMMIT') return 'applied' except Exception: if connection.in_transaction: connection.execute('ROLLBACK') raise finally: connection.close() key = 'daily-credit:2026-09-13T00:00:00Z' try: run(key, fail=True) except RuntimeError as error: assert str(error) == 'injected failure before commit' with sqlite3.connect(database) as connection: assert connection.execute('SELECT units FROM balance').fetchone() == (0,) assert connection.execute('SELECT count(*) FROM completed').fetchone() == (0,) gate = Barrier(8) with ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda _: run(key, gate=gate), range(8))) assert results.count('applied') == 1 and results.count('duplicate') == 7 assert run(key) == 'duplicate' assert run('daily-credit:2026-09-14T00:00:00Z') == 'applied' with sqlite3.connect(database) as connection: units = connection.execute('SELECT units FROM balance').fetchone()[0] completed = connection.execute('SELECT count(*) FROM completed').fetchone()[0] assert (units, completed) == (20, 2) print(json.dumps({'rolled_back_attempt_left_no_state': True, 'concurrent_applied': results.count('applied'), 'concurrent_duplicates': results.count('duplicate'), 'final_units': units, 'completed_logical_jobs': completed})) 第一次尝试在完成插入和余额更新之后、提交之前故意抛出异常。脚本验证回滚之后余额仍为零，也不存在完成键。随后八个工作线程同时处理同一个逻辑时间点，必须只有一个返回 applied，另七个返回 duplicate。唯一键判断与业务修改依靠真实数据库串行化，不是通过假的内存锁预先安排好结果。并发开始屏障让竞争条件实际存在。 ...