短生命周期任务可能在 Prometheus 下一次抓取之前就结束。Pushgateway 提供中间层，保留任务推送的指标以供后续抓取，但它不是每个短进程的默认方案。Prometheus 主要建议把它用于适当的服务级批任务结果。保留数据的生命周期与任务自身独立，因此抓取成功不能证明任务最近运行过，也不能证明业务工作已经成功完成。

为稳定的批任务结果建立身份

夜间对账这样的服务级任务，可以把最近结果发布到稳定分组键下。如果把新的运行编号、时间戳或机器身份放进键中，每次运行都可能创建额外分组，而不是更新同一份结果。Pushgateway 没有自动清理旧分组的过期时间机制，负责人需要明确替换或删除相关数据，不能假定退出任务后数据也会跟着消失。

与具体机器相关的批任务，往往可以使用节点导出器的文本文件采集器。共享网关本身也增加了可用性依赖。它的 up 表示 Prometheus 能否抓取网关，不表示每个原始推送任务都健康。应根据观测事件与清理责任选择采集方式，而不是把推送当成绕过服务发现或网络策略的通用办法。任务语义没有确定时，先增加网关并不会自动解决监控定义。

替换范围由 HTTP 方法决定

URL 路径定义分组键，包括任务名称与其他标签对。HTTP PUT 替换这个精确分组中的指标；POST 则替换本次请求包含的同名指标族，同时保留组内其他指标族。POST 不表示给原值递增，也不是向持久事件日志追加一条记录。如果一次报告把数值从七写成九，结果是九，而不是自动累加成十六。

DELETE 删除 URL 标识的精确分组。仅删除任务名构成的分组键，不会删除所有同时包含该任务名和额外标签的分组。网关文档规定的方法顺序允许后续成功推送排在先前已入队的删除之后，这可以用于验证清理结果。不能只看到删除请求被接受，就认定所有名称相似的序列都已经被处理，尤其不能把分组键误当成标签子集查询。

抓取时间不等于最近成功时间

网关通过 push_time_seconds 导出某组最近成功推送的时间。读取指标端点不会推进这个值。Prometheus 为采集样本赋予抓取时间戳，所以刚刚抓到的样本仍可能描述很久以前的任务结果。如果真正关心任务何时成功完成，应把最近成功时间作为独立的仪表值导出，并按任务计划检查其新鲜度。

不要给推送的指标样本附带显式样本时间戳，Pushgateway 会拒绝这种输入。把 Unix 时间戳作为仪表指标的数值是另一回事，可以正常使用。推送成功也不能证明业务成功或已经立即持久化：任务可以成功报告自己的失败，而没有配置持久化的网关在重启后可能丢失保留数据。各项完成语义应分别核对。

同一分组键还需要明确并发写入责任。两个运行可能互相覆盖，较早开始但延迟完成的任务，也可能在较新任务之后推送。HTTP 接口不会替业务建立运行先后关系。应协调重叠任务，或在合适的记录中保留完成身份；不能只凭最后抵达网关，就断定它代表最新一轮。稳定键控制基数，不自动解决写入顺序。

坏掉的样子

任务失败或被停用之后，旧的成功指标可能一直存在。只检查序列是否存在的告警，就会漏掉没有发生的新执行。应把最近成功值与预期调度比较，并让任务移除与分组清理同步。若分组标签不断改变，新旧运行还可能永久并列显示，给面板制造重复任务或持续健康的错觉，而实际已经没有新的成功结果。

本来想发布完整快照，却使用 POST，可能保留本次任务已不再输出的指标；只准备更新一个指标，却使用 PUT，又可能删除其他面板仍需要的结果。应明确哪个组件拥有整个分组，以及它使用哪种方法。不存在能够代替这些 HTTP 语义的 --replace 选项，不能把一个未经核实的命令参数写进操作流程。

快速验证

保存为 pushgateway.py 后运行 python3 pushgateway.py。需要 Docker 与 Pushgateway 1.11.1 镜像，未缓存时可能下载。一次性服务器只通过容器回环地址接收请求，不发布宿主机端口，不挂载本地目录，退出清理也仅针对本次唯一名称的容器。实验数据是固定的示意批任务结果，不会接触真实任务。

import http.client
import io
import json
import subprocess
import time
import uuid

name = "xbcatm-monitoring-" + uuid.uuid4().hex[:12]


class ResponseSocket:
    def __init__(self, data):
        self.data = data

    def makefile(self, mode):
        return io.BytesIO(self.data)


def request(method, path, body="", content_type="application/json"):
    body = body.encode()
    headers = (f"{method} {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n"
               f"Connection: close\r\nContent-Type: {content_type}\r\n"
               f"Content-Length: {len(body)}\r\n\r\n").encode()
    result = subprocess.run(
        ["docker", "exec", "-i", name, "/bin/nc", "-w", "3", "127.0.0.1", str(port)],
        input=headers + body, capture_output=True, timeout=5,
    )
    if result.returncode:
        raise RuntimeError(result.stderr.decode())
    response = http.client.HTTPResponse(ResponseSocket(result.stdout))
    response.begin()
    return response.status, response.read().decode()


def get_json(path):
    status, body = request("GET", path)
    assert status == 200, (status, body)
    return json.loads(body)


def eventually(check):
    for attempt in range(150):
        try:
            value = check()
            if value:
                return value
        except (RuntimeError, AssertionError, http.client.HTTPException):
            pass
        time.sleep(0.1)
    raise AssertionError("Expected local server state did not appear")


port = 9091
script = r"""exec /bin/pushgateway --web.listen-address=127.0.0.1:9091
"""
try:
    subprocess.run(
        ["docker", "run", "-d", "--rm", "--name", name, "--network", "none",
         "--entrypoint", "/bin/sh", "prom/pushgateway:v1.11.1", "-c", script],
        check=True, capture_output=True, text=True, timeout=20,
    )
    eventually(lambda: request("GET", "/metrics")[0] == 200)
    group = "/metrics/job/nightly_sync/team/data"
    first = "# TYPE items_processed gauge\nitems_processed 7\n# TYPE last_success_unixtime gauge\nlast_success_unixtime 1234\n"
    assert request("PUT", group, first, "text/plain; version=0.0.4")[0] == 200
    def snapshot():
        status, body = request("GET", "/metrics")
        assert status == 200
        return {line.split("{", 1)[0]: float(line.rsplit(" ", 1)[1])
                for line in body.splitlines()
                if not line.startswith("#") and 'job="nightly_sync"' in line}
    def first_snapshot():
        data = snapshot()
        return data if data.get("items_processed") == 7 else None
    initial = eventually(first_snapshot)
    assert initial["last_success_unixtime"] == 1234
    time.sleep(0.2)
    assert snapshot()["push_time_seconds"] == initial["push_time_seconds"]
    partial = "# TYPE items_processed gauge\nitems_processed 9\n"
    assert request("POST", group, partial, "text/plain; version=0.0.4")[0] == 200
    after_post = snapshot()
    assert after_post["items_processed"] == 9 and after_post["last_success_unixtime"] == 1234
    replacement = "# TYPE items_processed gauge\nitems_processed 11\n"
    assert request("PUT", group, replacement, "text/plain; version=0.0.4")[0] == 200
    after_put = snapshot()
    assert after_put["items_processed"] == 11 and "last_success_unixtime" not in after_put
    assert request("DELETE", "/metrics/job/nightly_sync")[0] == 202
    # A successful later POST orders the preceding queued deletion before this check.
    assert request("POST", group, "", "text/plain; version=0.0.4")[0] == 200
    assert snapshot()["items_processed"] == 11
    assert request("DELETE", group)[0] == 202
    eventually(lambda: not snapshot())
    print(json.dumps({"initial_items": 7, "post_items": 9, "put_items": 11,
                      "post_preserved_other_metric": True, "put_removed_other_metric": True,
                      "scrapes_did_not_refresh_push_time": True,
                      "parent_group_delete_did_not_match": True, "group_deleted": True}, indent=2))
finally:
    subprocess.run(["docker", "rm", "-f", name],
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

断言显示，最初 PUT 后处理数量为七；POST 后变成九，同时保留独立的最近成功仪表值；再次 PUT 后变成十一，并移除另一指标。重复读取不会刷新推送时间。删除仅包含任务名的分组，不影响带团队标签的分组；删除完整分组键后，其指标才消失。实验验证网关接口行为，不验证任务调度、Prometheus 接入或持久恢复。

本轮没有修改此运行实验，也没有重新执行。文中的预期输出和此前记录的观察，与本轮执行证据分别保留，不能据此声称真实集成已经再次通过。

来源