zabbix-deploy/bootstrap/zbx-bootstrap.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Zabbix 「自监控-only」bootstrap 脚本（幂等，可重复执行）

背景
----
Zabbix 服务端首次启动会自动创建名为 "Zabbix server" 的自身监控主机，并默认挂上
"Linux by Zabbix agent" + "Zabbix server health" 两个模板。
但本机并不安装 zabbix-agent，于是：
  - 那批 agent 监控项永远采集不到数据（空采）；
  - 还会周期性尝试连接本机 10050（agent 端口）而产生 "agent 不可用" 之类的无效告警/报错。

本脚本在每次 `docker compose up` 后运行一次，把「服务器自身主机」上的
"Linux by Zabbix agent" 模板摘除（连同其监控项/触发器），只保留
"Zabbix server health" 自监控模板 —— 即“自己监控自己、不连 agent”。

幂等：每次先查询再决定是否操作；主机已干净则跳过。

认证：优先使用环境变量 ZBX_API_TOKEN（推荐，API Token）；否则用
ZBX_API_USER / ZBX_API_PASSWORD 登录获取会话令牌。
"""
import os
import sys
import time
import json
import urllib.request
import urllib.error

URL = os.environ.get("ZBX_API_URL", "http://zabbix-web-nginx-mysql:8080/api_jsonrpc.php")
USER = os.environ.get("ZBX_API_USER", "Admin")
PWD = os.environ.get("ZBX_API_PASSWORD", "zabbix")
TOKEN = os.environ.get("ZBX_API_TOKEN", "").strip()
HEALTH = os.environ.get("ZBX_HEALTH_TEMPLATE", "Zabbix server health")
AGENT = os.environ.get("ZBX_AGENT_TEMPLATE", "Linux by Zabbix agent")
WAIT = int(os.environ.get("ZBX_WAIT_SECONDS", "300"))


def call(method, params, token=None):
    payload = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1}
    headers = {"Content-Type": "application/json-rpc"}
    if token:
        headers["Authorization"] = "Bearer " + token
    req = urllib.request.Request(URL, data=json.dumps(payload).encode(), headers=headers)
    with urllib.request.urlopen(req, timeout=30) as resp:
        j = json.loads(resp.read().decode())
    if "error" in j:
        raise RuntimeError("%s -> %s" % (method, j["error"]))
    return j["result"]


def wait_api():
    deadline = time.time() + WAIT
    while time.time() < deadline:
        try:
            call("apiinfo.version", {})
            print("[bootstrap] API is up")
            return
        except Exception as e:  # noqa: BLE001
            print("[bootstrap] waiting for API ... (%s)" % e)
            time.sleep(5)
    print("[bootstrap] ERROR: API not reachable within %ss" % WAIT)
    sys.exit(1)


def main():
    wait_api()

    if TOKEN:
        token = TOKEN
        print("[bootstrap] using provided API token")
    else:
        token = call("user.login", {"username": USER, "password": PWD})
        print("[bootstrap] logged in as %s" % USER)

    tpls = call(
        "template.get",
        {"output": ["templateid", "host", "name"], "filter": {"host": [AGENT, HEALTH]}},
        token,
    )
    ids = {t["host"]: t["templateid"] for t in tpls}
    if AGENT not in ids:
        print("[bootstrap] agent template %r not found -> nothing to do" % AGENT)
        return
    if HEALTH not in ids:
        print("[bootstrap] health template %r not found -> cannot identify server host, skip" % HEALTH)
        return

    agent_id = ids[AGENT]
    health_id = ids[HEALTH]

    hosts = call(
        "host.get",
        {
            "output": ["hostid", "host"],
            "templateids": [health_id],
            "selectParentTemplates": ["templateid", "host"],
        },
        token,
    )

    changed = 0
    for h in hosts:
        tids = [p["templateid"] for p in h.get("parentTemplates", [])]
        if agent_id in tids:
            keep = [{"templateid": x} for x in tids if x != agent_id]
            call(
                "host.update",
                {
                    "hostid": h["hostid"],
                    "templates": keep,
                    "templates_clear": [{"templateid": agent_id}],
                },
                token,
            )
            print("[bootstrap] removed %r from host %s (%s)" % (AGENT, h["host"], h["hostid"]))
            changed += 1
        else:
            print("[bootstrap] host %s (%s) already clean" % (h["host"], h["hostid"]))

    print("[bootstrap] done, changed %d host(s)" % changed)


if __name__ == "__main__":
    main()