zabbix-api

Zabbix 自动化技能

概述

此技能提供通过 API 和官方 Python 库 zabbix_utils 自动执行 Zabbix 监控操作的指导。

快速开始

安装

pip install zabbix-utils --break-system-packages

身份验证

from zabbix_utils import ZabbixAPI

# Option 1: Username/password
api = ZabbixAPI(url="https://zabbix.example.com")
api.login(user="Admin", password="zabbix")

# Option 2: API token (Zabbix 5.4+, preferred)
api = ZabbixAPI(url="https://zabbix.example.com")
api.login(token="your_api_token")

# Verify connection
print(api.api_version())

环境变量模式

import os
from zabbix_utils import ZabbixAPI

api = ZabbixAPI(url=os.environ.get("ZABBIX_URL", "http://localhost/zabbix"))
api.login(token=os.environ["ZABBIX_TOKEN"])

核心 API 方法

所有 API 都遵循 api.<object>.<method>() 模式，可用方法包括：get、create、update、delete。

主机操作

# Get hosts
hosts = api.host.get(output=["hostid", "host", "name"],
                     selectInterfaces=["ip"])

# Create host
api.host.create(
    host="server01",
    groups=[{"groupid": "2"}],  # Linux servers
    interfaces=[{
        "type": 1,  # 1=agent, 2=SNMP, 3=IPMI, 4=JMX
        "main": 1,
        "useip": 1,
        "ip": "192.168.1.100",
        "dns": "",
        "port": "10050"
    }],
    templates=[{"templateid": "10001"}]
)

# Update host
api.host.update(hostid="10084", status=0)  # 0=enabled, 1=disabled

# Delete host
api.host.delete("10084")

模板操作

# Get templates
templates = api.template.get(output=["templateid", "host", "name"],
                             selectHosts=["hostid", "name"])

# Link template to host
api.host.update(hostid="10084",
                templates=[{"templateid": "10001"}])

# Import template from XML
with open("template.xml") as f:
    api.configuration.import_(
        source=f.read(),
        format="xml",
        rules={
            "templates": {"createMissing": True, "updateExisting": True},
            "items": {"createMissing": True, "updateExisting": True},
            "triggers": {"createMissing": True, "updateExisting": True}
        }
    )

监控项操作

# Get items
items = api.item.get(hostids="10084",
                     output=["itemid", "name", "key_"],
                     search={"key_": "system.cpu"})

# Create item
api.item.create(
    name="CPU Load",
    key_="system.cpu.load[percpu,avg1]",
    hostid="10084",
    type=0,  # 0=Zabbix agent
    value_type=0,  # 0=float, 3=integer, 4=text
    delay="30s",
    interfaceid="1"
)

触发器操作

# Get triggers
triggers = api.trigger.get(hostids="10084",
                          output=["triggerid", "description", "priority"],
                          selectFunctions="extend")

# Create trigger
api.trigger.create(
    description="High CPU on {HOST.NAME}",
    expression="last(/server01/system.cpu.load[percpu,avg1])>5",
    priority=3  # 0=not classified, 1=info, 2=warning, 3=average, 4=high, 5=disaster
)

主机群组操作

# Get groups
groups = api.hostgroup.get(output=["groupid", "name"])

# Create group
api.hostgroup.create(name="Production/Web Servers")

# Add hosts to group
api.hostgroup.massadd(groups=[{"groupid": "5"}],
                      hosts=[{"hostid": "10084"}])

维护时段

import time

# Create maintenance
api.maintenance.create(
    name="Server Maintenance",
    active_since=int(time.time()),
    active_till=int(time.time()) + 3600,  # 1 hour
    hostids=["10084"],
    timeperiods=[{
        "timeperiod_type": 0,  # One-time
        "period": 3600
    }]
)

事件和问题

# Get current problems
problems = api.problem.get(output=["eventid", "name", "severity"],
                          recent=True)

# Get events
events = api.event.get(hostids="10084",
                       time_from=int(time.time()) - 86400,
                       output="extend")

历史数据

# Get history (value_type must match item's value_type)
# 0=float, 1=character, 2=log, 3=integer, 4=text
history = api.history.get(
    itemids="28269",
    history=0,  # float
    time_from=int(time.time()) - 3600,
    output="extend",
    sortfield="clock",
    sortorder="DESC"
)

Zabbix Sender（Trapper 监控项）

from zabbix_utils import Sender

sender = Sender(server="zabbix.example.com", port=10051)

# Send single value
response = sender.send_value("hostname", "trap.key", "value123")
print(response)  # {"processed": 1, "failed": 0, "total": 1}

# Send multiple values
from zabbix_utils import ItemValue
values = [
    ItemValue("host1", "key1", "value1"),
    ItemValue("host2", "key2", 42),
]
response = sender.send(values)

Zabbix Getter（Agent 查询）

from zabbix_utils import Getter

agent = Getter(host="192.168.1.100", port=10050)
response = agent.get("system.uname")
print(response.value)

常见模式

从 CSV 批量创建主机

import csv
from zabbix_utils import ZabbixAPI

api = ZabbixAPI(url="https://zabbix.example.com")
api.login(token="your_token")

with open("hosts.csv") as f:
    for row in csv.DictReader(f):
        try:
            api.host.create(
                host=row["hostname"],
                groups=[{"groupid": row["groupid"]}],
                interfaces=[{
                    "type": 1, "main": 1, "useip": 1,
                    "ip": row["ip"], "dns": "", "port": "10050"
                }]
            )
            print(f"Created: {row['hostname']}")
        except Exception as e:
            print(f"Failed {row['hostname']}: {e}")

查找未关联模板的主机

# Get all hosts
all_hosts = api.host.get(output=["hostid", "host"],
                         selectParentTemplates=["templateid"])

# Filter hosts without specific template
template_id = "10001"
hosts_without = [h for h in all_hosts
                 if not any(t["templateid"] == template_id
                           for t in h.get("parentTemplates", []))]

按模式禁用触发器

triggers = api.trigger.get(
    search={"description": "test"},
    output=["triggerid"]
)
for t in triggers:
    api.trigger.update(triggerid=t["triggerid"], status=1)  # 1=disabled

监控项类型参考

类型	值	描述
Zabbix Agent	0	主动检查
Zabbix trapper	2	被动检查，通过 sender 推送数据
简单检查	3	ICMP、TCP 等
Zabbix 内部	5	服务端内部指标
Zabbix Agent（主动）	7	由 Agent 发起
HTTP Agent	19	HTTP/REST API 监控
依赖监控项	18	从主监控项派生
脚本	21	自定义脚本

值类型参考

类型	值	描述
浮点数	0	数值（浮点数）
字符	1	字符串
日志	2	日志文件
无符号整数	3	数值（整数）
文本	4	文本

触发器严重性参考

严重性	值	颜色
未分类	0	灰色
信息	1	浅蓝色
警告	2	黄色
一般	3	橙色
高	4	浅红色
灾难	5	红色

错误处理

from zabbix_utils import ZabbixAPI
from zabbix_utils.exceptions import APIRequestError

try:
    api.host.create(host="duplicate_host", groups=[{"groupid": "2"}])
except APIRequestError as e:
    print(f"API Error: {e.message}")
    print(f"Code: {e.code}")

调试

import logging
logging.basicConfig(level=logging.DEBUG)
# Now all API calls will be logged

脚本参考

可在 scripts/ 目录中找到可直接使用的自动化脚本：

• zabbix-bulk-hosts.py - 从 CSV 批量管理主机
• zabbix-maintenance.py - 创建/管理维护时段
• zabbix-export.py - 将主机/模板导出为 JSON/XML

最佳实践

1. 尽可能使用 API 令牌，而不是用户名/密码
2. 限制输出字段 - 始终指定 output=["field1", "field2"]，而不是 output="extend"
3. 使用搜索/筛选 - 不要获取所有对象后再用 Python 筛选
4. 处理分页 - 大型结果集可能需要使用 limit 和 offset
5. 批量操作 - 使用 massadd、massupdate 进行批量更改
6. 错误处理 - 始终将 API 调用包装在 try/except 中
7. 幂等性 - 创建对象前检查其是否已存在

---

注意事项

• api.history.get 需要正确的 history value_type，否则会返回空结果： 对整数型项目传入 history=0（浮点数），会静默返回 []，而不是报错。请务必先读取项目的 value_type；四种类型不匹配（0/1/3/4）是最常见的“无数据”原因。
• 大型查询使用 output="extend" 会让服务不堪重负： 获取数千台主机或项目的所有字段，会导致响应耗时数秒，并使 API 前端内存溢出。请务必传入 output=["field1","field2"]，只包含所需的列——API 没有隐式分页保护。
• 触发器表达式引用的是主机和键，而非项目 ID： 重命名主机或键会破坏所有引用它们的触发器表达式，且在求值前不会发出警告。请在表达式中使用模板和宏（{HOST.HOST}），而不是硬编码名称。
• zabbix_utils 中容易混淆 API 令牌和会话令牌： api.login(token=...) 使用永久 API 令牌（Zabbix 5.4+）；api.login(user=..., password=...) 会签发会话令牌。将令牌身份验证与 api.logout() 混用，会使所有使用该永久令牌的人都无法再使用它。
• massadd 不会去重，而 massupdate 会替换： 在某些版本中，hostgroup.massadd 会直接将同一台主机重复添加到组中；hostgroup.massupdate 则会静默移除不在请求数据中的主机。批量操作前请仔细确认操作动词，否则可能会移除你原本想保留的主机。
• maintenance.create 接受过去的 active_since，但会忽略它： 将维护窗口设为过去的时间，不会追溯性地抑制已触发的警报。请将 active_since 设为略晚于当前的时间，并通过 maintenance.get 验证，然后再假定警报已静默。