向 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)

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

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):
    # ...
    # 套接字的 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 参数来更改它。

例如，要将五个值分三块发送（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. 在 Sender 实例上调用 send_value() 方法，格式与 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