Dynamic Alerting with Python and Zabbix API: Ditch Fixed Thresholds

Why static thresholds fail

Fixed CPU thresholds (e.g., >80%) generate alerts during idle periods on low‑load hosts and miss alerts on high‑load hosts. Host baselines differ widely (some run at 20% idle, others at 70%). Business traffic varies (night low, day high, promotion spikes). Each change requires manual trigger edits, incurring high maintenance cost.

AIOps dynamic‑threshold core idea

Alert threshold = max(current CPU usage + buffer offset, minimum protection threshold)

Concrete examples:

Current CPU = 22%, buffer = 10% → threshold = 32%.

Current CPU = 3%, buffer = 10% → 13% < minimum 15% → threshold = 15%.

Benefits

Idle machines keep a higher threshold, avoiding noisy alerts.

When load rises, the threshold lifts automatically to match the real baseline.

The whole process is automated; no manual trigger adjustments are needed.

Environment preparation

Zabbix Server 5.0/6.0/7.0 with API reachable

Python 3.6 or newer

Python libraries: requests, paramiko Install the libraries:

pip3 install requests paramiko

Core code implementation

1. Configuration block

# ==========配置区域==========
ZABBIX_URL = "http://192.168.40.160/zabbix/api_jsonrpc.php"
ZABBIX_USER = "Admin"
ZABBIX_PASSWORD = "zabbix"
HOST_NAME = "agent-161"
TRIGGER_DESC = "CPU使用率过高"
LINUX_HOST = "192.168.40.161"
LINUX_USER = "root"
LINUX_PASS = "111111"
CPU_BUFFER = 10   # baseline buffer percent
MIN_THRESHOLD = 15   # minimum protection threshold
CHECK_INTERVAL = 60   # seconds

2. Zabbix API login

def zabbix_login():
    payload = {
        "jsonrpc": "2.0",
        "method": "user.login",
        "params": {"username": ZABBIX_USER, "password": ZABBIX_PASSWORD},
        "id": 1
    }
    resp = requests.post(ZABBIX_URL, json=payload, timeout=10)
    data = resp.json()
    if "result" in data:
        print("✅ Zabbix登录成功")
        return data["result"]
    return None

3. SSH remote CPU collection

def get_linux_cpu():
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(LINUX_HOST, username=LINUX_USER, password=LINUX_PASS, timeout=10)
    cmd = """
    prev=$(grep '^cpu ' /proc/stat)
    sleep 1
    curr=$(grep '^cpu ' /proc/stat)
    prev_idle=${prev[4]}
    prev_total=0
    for v in "${prev[@]:1}"; do prev_total=$((prev_total+v)); done
    curr_idle=${curr[4]}
    curr_total=0
    for v in "${curr[@]:1}"; do curr_total=$((curr_total+v)); done
    idle=$((curr_idle-prev_idle))
    total=$((curr_total-prev_total))
    usage=$((100*(total-idle)/total))
    echo $usage
    """
    _, stdout, _ = ssh.exec_command(cmd)
    cpu = float(stdout.read().decode().strip())
    ssh.close()
    return cpu

4. Compute threshold and update trigger

def update_trigger(auth, trigger_id, threshold):
    expr = f"last(/{{HOST_NAME}}/system.cpu.util[,user])>{{threshold}}"
    payload = {
        "jsonrpc": "2.0",
        "method": "trigger.update",
        "params": {"triggerid": trigger_id, "expression": expr},
        "auth": auth,
        "id": 3
    }
    requests.post(ZABBIX_URL, json=payload)
    print(f"🔔触发器更新完成，新阈值：{{threshold}}%")

cpu = get_linux_cpu()
threshold = max(cpu + CPU_BUFFER, MIN_THRESHOLD)
threshold = int(threshold)

5. Main loop

while True:
    auth = zabbix_login()
    trigger_info = get_trigger(auth)
    cpu_val = get_linux_cpu()
    new_threshold = max(cpu_val + CPU_BUFFER, MIN_THRESHOLD)
    update_trigger(auth, trigger_info["triggerid"], int(new_threshold))
    time.sleep(CHECK_INTERVAL)

Execution flow

SSH into the target host and read /proc/stat to calculate real CPU usage.

Apply the dynamic‑threshold formula.

Authenticate to Zabbix via JSON‑RPC API.

Query the existing trigger, then call trigger.update to rewrite its expression.

Sleep for the configured interval and repeat.

Running result

🚀 启动 Zabbix CPU 自适应告警引擎
============================================================
开始新一轮检测
✅ Zabbix 登录成功
当前触发器表达式: last(/agent‑161/system.cpu.util[,user])>25
当前 CPU 使用率:23.0%
计算得到的新阈值:33%
✅ 触发器已更新为: last(/agent‑161/system.cpu.util[,user])>33
本轮完成，60 秒后再次执行

After the script starts, the trigger expression in Zabbix changes automatically with the host's CPU load, eliminating manual adjustments.

Conclusion

Static monitoring relies on manually set thresholds that cannot cope with complex, fluctuating workloads. A lightweight Python script that calls Zabbix’s native API can implement a dynamic‑threshold solution, reducing alert storms and false positives for small‑to‑medium enterprises.