将数据发送到 Zabbix 服务器或 proxy

zabbix_utils 允许你将监控项值发送到 Zabbix 服务器或 proxy 上的 trapper 监控项（类似于 Zabbix sender）。

你可以发送单个值、多个值，甚至将数据发送到多个 Zabbix 集群。

数据可以通过同步或异步模式发送：

• 在同步模式下，Python 脚本会发送值并在继续执行前等待响应；这适用于简单、顺序且可预测的操作。
• 在异步模式下，脚本发送值后不会等待每个响应，从而允许其他操作并行进行；这对于较慢的请求或大批量数据更高效。

本页中的示例主要聚焦于同步模式，不过 异步模式 采用类似的模式。 更多示例可在 zabbix_utils GitHub 仓库中找到。

导入

要使用 zabbix_utils 发送监控项值，请在脚本中导入 Sender 类：

from zabbix_utils import Sender

如需发送多个值，您也可以导入 ItemValue 类：

from zabbix_utils import Sender, ItemValue

发送单个值

要发送一个监控项值：

1. 创建一个 Sender 实例，指定 Zabbix 服务器或 proxy 的 IP 地址和端口。
2. 使用以下格式在 Sender 实例上调用 send_value() 方法：

sender_instance.send_value('host', 'item.key', 'value', optional_timestamp, optional_nanoseconds)

例如，要向 Linux server 主机上的 service.status trapper 监控项发送 1：

sender = Sender(server='127.0.0.1', port=10051)
response = sender.send_value('Linux server', 'service.status', 1)

使用非默认 IP

如果运行脚本的服务器有多个 IP 地址，您可以为 Sender 指定一个 source_ip，以便在向 Zabbix 服务器或 proxy 发送值时使用：

sender = Sender(
    server='127.0.0.1',
    port=10051,
    source_ip='10.10.7.1'
)

使用超时

您可以为 Sender 设置响应 timeout，以控制脚本在放弃之前应等待 Zabbix 服务器或 proxy 响应的时长：

sender = Sender(
    server='127.0.0.1',
    port=10051,
    timeout=30
)

使用 agent 配置文件

您可以让 zabbix_utils 从本地 Zabbix agent 或 agent 2 配置文件中读取 Server 或 ServerActive 参数。 在这种情况下，创建 Sender 实例时无需指定连接参数：

sender = Sender(
    use_config=True,
    config_path='/etc/zabbix/zabbix_agent2.conf'
)

使用加密

Sender 不包含内置加密支持，但你可以通过使用第三方库创建一个包装器来提供它：

def psk_wrapper(sock, tls):
    # ...
    # socket 的 TLS PSK 包装器实现
    # ...

sender = Sender(
    server='127.0.0.1',
    port=10051,
    socket_wrapper=psk_wrapper
)

单个值的响应

Zabbix 服务器或 proxy 返回的响应会由库进行处理，并作为 TrapperResponse 对象返回：

print(response)
# {"processed": 1, "failed": 0, "total": 1, "time": "0.000123", "chunk": 1}

print(response.processed)
# 1

print(response.failed)
# 0

print(response.total)
# 1

异步模式

异步模式允许你的 Python 脚本在不等待 Zabbix 服务器或 proxy 响应的情况下发送值。 当脚本需要发送大量值，或者某些值发送耗时较长时，这可以提高脚本的效率。

使用异步模式时，与同步模式相比有一些重要区别：

• 导入 Python 的 asyncio 模块（你必须先安装所需的依赖项）。
• 导入 AsyncSender，而不是 Sender。
• 将代码写在 async 函数中。
• 调用 send_value() 方法时使用 await。

例如，使用异步模式发送单个值：

# 1. 导入 asyncio 以使用异步模式，并从 zabbix_utils 导入 AsyncSender：
import asyncio
from zabbix_utils import AsyncSender

# 2. 定义主 async 函数，所有数据发送操作（必须使用 await）都将在其中执行：
async def main():
    sender = AsyncSender(server='127.0.0.1', port=10051)
    response = await sender.send_value('Linux server', 'service.status', 1)

    # 3. 打印 Zabbix 服务器或 proxy 返回的响应：
    print(response)

# 4. 使用 asyncio 的事件循环运行 async main() 函数：
asyncio.run(main())

发送多个值

要发送多个值：

1. 准备一个 ItemValue 对象数组，每个对象都使用与 send_value() 方法相同的格式。
2. 创建一个 Sender 实例，指定 Zabbix 服务器或 proxy 的 IP 地址和端口。
3. 在 Sender 实例上调用 send() 方法（而不是 send_value()），并指定要发送值的对象数组。

例如，要向不同的主机发送五个值：

items = [
    ItemValue('server-de', 'service.status', 'up', 1770887205, 100),
    ItemValue('server-fr', 'service.status', 'up', 1770887205, 100),
    ItemValue('server-uk', 'service.status', 'up', 1770887205, 100),
    ItemValue('server-nl', 'service.status', 'up', 1770887205, 100),
    ItemValue('server-pl', 'service.status', 'up', 1770887205, 100),
]

sender = Sender(server='127.0.0.1', port=10051)
response = sender.send(items)

使用自定义块大小

如果你需要发送的值数量超过单个请求中 trapper 监控项可接受的数量，可以将它们拆分为多个块。

默认情况下，块大小为 250 个值。 你可以在创建 Sender 实例时设置 chunk_size 参数来更改它。

例如，要将 5 个值分成 3 个块发送（2-2-1），请将 chunk_size 参数设置为 2：

items = [
    ItemValue('server-de', 'service.status', 'up'),
    ItemValue('server-fr', 'service.status', 'up'),
    ItemValue('server-uk', 'service.status', 'up'),
    ItemValue('server-nl', 'service.status', 'up'),
    ItemValue('server-pl', 'service.status', 'up'),
]

sender = Sender(server='127.0.0.1', port=10051, chunk_size=2)
response = sender.send(items)

向多个 Zabbix 集群发送值

要向多个 Zabbix 集群发送值：

1. 准备一个 Zabbix 集群数组。如果某个集群包含多个节点，则该值将发送到每个集群中第一个可用的节点。
2. 创建一个 Sender，并指定您的 Zabbix 集群数组。
3. 使用与 send_value() 方法相同的格式，在 Sender 实例上调用 send_value() 方法。

例如，要向每个集群中第一个可用的节点发送一个值：

zabbix_clusters = [
    ['zabbix.cluster1.node1', 'zabbix.cluster1.node2:10051'],
    ['zabbix.cluster2.node1:10051', 'zabbix.cluster2.node2', 'zabbix.cluster2.node3']
]

sender = Sender(clusters=zabbix_clusters)
response = sender.send_value('Linux server', 'service.status', 1)

多个值的响应

默认情况下，Sender 会返回跨所有主机或集群发送值的汇总结果：

print(response)
# {"processed": 2, "failed": 0, "total": 2, "time": "0.000108", "chunk": 2}

如果您需要更详细的信息，可以使用 response.details 属性查看每个集群和每个分块的结果：

print(response)
# {"processed": 2, "failed": 0, "total": 2, "time": "0.000108", "chunk": 2}

if response.failed == 0:
    print(f"Value sent successfully in {response.time}")
else:
    print(response.details)
    # {
    #     127.0.0.1:10051: [
    #         {
    #             "processed": 1,
    #             "failed": 0,
    #             "total": 1,
    #             "time": "0.000051",
    #             "chunk": 1
    #         }
    #     ],
    #     zabbix.example.local:10051: [
    #         {
    #             "processed": 1,
    #             "failed": 0,
    #             "total": 1,
    #             "time": "0.000057",
    #             "chunk": 1
    #         }
    #     ]
    # }

    for node, chunks in response.details.items():
        for resp in chunks:
            print(f"Processed {resp.processed} of {resp.total} at {node.address}:{node.port}")
            # Processed 1 of 1 at 127.0.0.1:10051
            # Processed 1 of 1 at zabbix.example.local:10051