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（トラッパーアイテム）

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（エージェントへの問い合わせ）

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 エージェント	0	アクティブチェック
Zabbix トラッパー	2	パッシブ。sender 経由でデータを送信
シンプルチェック	3	ICMP、TCP など
Zabbix 内部	5	サーバー内部のメトリクス
Zabbix エージェント（アクティブ）	7	エージェント側から開始
HTTP エージェント	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="extend" ではなく、常に output=["field1", "field2"] を指定します
3. 検索／フィルターを使用します - すべてのオブジェクトを取得して Python で絞り込むことは避けてください
4. ページネーションを処理します - 大きな結果セットでは limit と offset が必要になる場合があります
5. 操作を一括処理します - 一括変更には massadd、massupdate を使用します
6. エラーを処理します - API 呼び出しは必ず try/except で囲みます
7. 冪等性を確保します - 作成前にオブジェクトが存在するか確認します

---

注意点

• api.history.get では正しい history の value_type が必要です。そうでない場合、空の結果が返されます。 整数型のアイテムに history=0（浮動小数点数）を渡すと、エラーではなく [] が返されます。必ず最初にアイテムの value_type を確認してください。4 種類の不一致（0/1/3/4）が、「データなし」の最も一般的な原因です。
• 大規模なクエリで output="extend" を使うと、サーバーに大きな負荷がかかります。 数千のホストやアイテムについてすべてのフィールドを取得すると、応答に数秒かかり、API フロントエンドで OOM が発生します。必要な列だけを指定した 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 で確認してください。