本文目录导读：
基于最后执行时间的清理脚本
#!/usr/bin/env python3
"""
清理未使用触发器脚本 - 基于最后执行时间
"""
import json
import requests
import datetime
from typing import List, Dict
class TriggerCleaner:
def __init__(self, zabbix_url, username, password):
self.zabbix_url = zabbix_url
self.auth_token = None
self.login(username, password)
def login(self, username, password):
"""登录Zabbix并获取认证token"""
data = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"user": username,
"password": password
},
"id": 1
}
response = requests.post(self.zabbix_url, json=data)
self.auth_token = response.json()['result']
def get_unused_triggers(self, days_threshold=90, status='DISABLED'):
"""获取未使用的触发器"""
triggers = []
# 获取所有触发器
data = {
"jsonrpc": "2.0",
"method": "trigger.get",
"params": {
"output": ["triggerid", "description", "status", "lastchange", "priority"],
"filter": {
"status": status
},
"selectHosts": ["hostid", "host"]
},
"auth": self.auth_token,
"id": 2
}
response = requests.post(self.zabbix_url, json=data)
all_triggers = response.json().get('result', [])
current_time = datetime.datetime.now()
for trigger in all_triggers:
# 检查触发器最后修改时间
last_change = int(trigger.get('lastchange', '0'))
if last_change > 0:
last_change_date = datetime.datetime.fromtimestamp(last_change)
days_since_change = (current_time - last_change_date).days
# 如果超过阈值天数未更改，标记为未使用
if days_since_change > days_threshold:
triggers.append({
'triggerid': trigger['triggerid'],
'description': trigger['description'],
'host': trigger.get('hosts', [{}])[0].get('host', 'Unknown'),
'last_change': last_change_date.strftime('%Y-%m-%d %H:%M:%S'),
'days_since_change': days_since_change,
'priority': trigger.get('priority', '0')
})
return triggers
def delete_triggers(self, trigger_ids: List[str]):
"""删除指定的触发器"""
data = {
"jsonrpc": "2.0",
"method": "trigger.delete",
"params": trigger_ids,
"auth": self.auth_token,
"id": 3
}
response = requests.post(self.zabbix_url, json=data)
return response.json()
# 使用示例
if __name__ == "__main__":
cleaner = TriggerCleaner(
zabbix_url="http://your-zabbix-server/api_jsonrpc.php",
username="Admin",
password="your-password"
)
# 获取90天未更改的禁用触发器
unused_triggers = cleaner.get_unused_triggers(days_threshold=90)
# 打印未使用的触发器
print("未使用的触发器列表：")
for trigger in unused_triggers:
print(f"- {trigger['description']} on {trigger['host']}")
# 确认删除
if unused_triggers:
confirm = input("是否删除这些触发器？(yes/no): ")
if confirm.lower() == 'yes':
trigger_ids = [t['triggerid'] for t in unused_triggers]
result = cleaner.delete_triggers(trigger_ids)
print(f"已删除 {len(trigger_ids)} 个触发器")
基于事件历史记录的清理脚本（更精确）
#!/usr/bin/env python3
"""
基于事件历史记录的未使用触发器清理
"""
import json
import requests
from datetime import datetime, timedelta
class SmartTriggerCleaner:
def __init__(self, api_url, api_token):
self.api_url = api_url
self.api_token = api_token
self.headers = {
'Authorization': f'Bearer {api_token}',
'Content-Type': 'application/json'
}
def get_triggers_with_event_history(self, lookback_days=30):
"""
获取在指定时间内没有触发事件的触发器
"""
# 计算时间范围
end_date = datetime.now()
start_date = end_date - timedelta(days=lookback_days)
query = """
query GetTriggerEvents($startDate: DateTime!, $endDate: DateTime!) {
events(
where: {
time: {_gte: $startDate, _lte: $endDate}
}
) {
trigger_id
count
}
triggers(
where: {
enabled: {_eq: true}
}
) {
id
name
description
host
last_event
enabled
}
}
"""
variables = {
"startDate": start_date.isoformat(),
"endDate": end_date.isoformat()
}
response = requests.post(
self.api_url,
json={"query": query, "variables": variables},
headers=self.headers
)
result = response.json().get('data', {})
# 获取有事件的触发器
trigger_with_events = set()
for event in result.get('events', []):
trigger_with_events.add(event['trigger_id'])
# 筛选未触发的触发器
unused_triggers = []
for trigger in result.get('triggers', []):
if trigger['id'] not in trigger_with_events:
unused_triggers.append(trigger)
return unused_triggers
def analyze_trigger_usage(self, trigger_id, historical_months=3):
"""
分析单个触发器的使用情况
"""
# 获取历史事件数据
events_query = """
query GetTriggerHistory($triggerId: ID!, $months: Int!) {
trigger(id: $triggerId) {
events(
where: {
time: {_gte: "-${months}months"}
}
order_by: {time: desc}
) {
time
status
value
}
}
}
"""
# 这里实现更详细的分析逻辑
pass
def cleanup_unused_triggers(self, unused_triggers, dry_run=True):
"""
清理未使用的触发器，支持dry-run模式
"""
if dry_run:
print("\n=== Dry Run - 即将清理的触发器 ===")
for trigger in unused_triggers:
print(f" · {trigger['name']} (ID: {trigger['id']})")
print(f" 主机: {trigger['host']}")
print(f"\n总计: {len(unused_triggers)} 个触发器将被清理")
return
# 实际清理逻辑
mutation = """
mutation DeleteTriggers($triggerIds: [ID!]!) {
delete_triggers(where: {id: {_in: $triggerIds}}) {
affected_rows
}
}
"""
trigger_ids = [t['id'] for t in unused_triggers]
response = requests.post(
self.api_url,
json={"query": mutation, "variables": {"triggerIds": trigger_ids}},
headers=self.headers
)
return response.json()
# 使用示例
if __name__ == "__main__":
# 配置
cleaner = SmartTriggerCleaner(
api_url="http://your-graphql-api/graphql",
api_token="your-api-token"
)
# 获取未使用的触发器
unused = cleaner.get_triggers_with_event_history(lookback_days=30)
# 先执行dry-run检查
cleaner.cleanup_unused_triggers(unused, dry_run=True)
# 确认后实际删除
confirm = input("\n是否执行实际清理？(yes/no): ")
if confirm.lower() == 'yes':
result = cleaner.cleanup_unused_triggers(unused, dry_run=False)
print(f"清理完成，已删除 {len(unused)} 个触发器")
Shell脚本版本（适用于Linux系统）
#!/bin/bash
# 清理未使用的Zabbix触发器脚本
# 配置
ZABBIX_URL="http://your-zabbix-server/api_jsonrpc.php"
USERNAME="Admin"
PASSWORD="your-password"
# 获取认证token
get_token() {
response=$(curl -s -X POST \
-H "Content-Type: application/json" \
-d "{
\"jsonrpc\": \"2.0\",
\"method\": \"user.login\",
\"params\": {
\"user\": \"$USERNAME\",
\"password\": \"$PASSWORD\"
},
\"id\": 1
}" \
"$ZABBIX_URL")
echo $response | python3 -c "import sys, json; print(json.load(sys.stdin)['result'])"
}
# 获取未使用的触发器
get_unused_triggers() {
local auth_token="$1"
local days_threshold="${2:-90}"
response=$(curl -s -X POST \
-H "Content-Type: application/json" \
-d "{
\"jsonrpc\": \"2.0\",
\"method\": \"trigger.get\",
\"params\": {
\"output\": [\"triggerid\", \"description\", \"status\", \"lastchange\"],
\"filter\": {
\"status\": \"DISABLED\"
},
\"selectHosts\": [\"host\"]
},
\"auth\": \"$auth_token\",
\"id\": 2
}" \
"$ZABBIX_URL")
echo $response | python3 -c "
import sys, json, datetime
data = json.load(sys.stdin)
triggers = data.get('result', [])
current_time = datetime.datetime.now()
for trigger in triggers:
last_change = int(trigger.get('lastchange', '0'))
if last_change:
last_date = datetime.datetime.fromtimestamp(last_change)
days_since = (current_time - last_date).days
if days_since > $days_threshold:
host = trigger.get('hosts', [{}])[0].get('host', 'Unknown')
print(f\"{trigger['triggerid']}:{trigger['description']}:{host}:{days_since}\")
"
}
# 删除触发器
delete_triggers() {
local auth_token="$1"
shift
local trigger_ids=("$@")
# 转换为JSON数组
json_ids=$(printf '%s\n' "${trigger_ids[@]}" | python3 -c "
import sys, json
ids = [line.strip() for line in sys.stdin if line.strip()]
print(json.dumps(ids))
")
response=$(curl -s -X POST \
-H "Content-Type: application/json" \
-d "{
\"jsonrpc\": \"2.0\",
\"method\": \"trigger.delete\",
\"params\": $json_ids,
\"auth\": \"$auth_token\",
\"id\": 3
}" \
"$ZABBIX_URL")
echo "删除结果: $response"
}
# 主程序
main() {
echo "开始获取认证..."
AUTH_TOKEN=$(get_token)
if [ -z "$AUTH_TOKEN" ]; then
echo "认证失败！"
exit 1
fi
echo "获取超过90天未使用的已禁用触发器..."
UNUSED_TRIGGERS=$(get_unused_triggers "$AUTH_TOKEN" 90)
if [ -z "$UNUSED_TRIGGERS" ]; then
echo "没有找到未使用的触发器"
exit 0
fi
echo "找到以下未使用的触发器："
echo "$UNUSED_TRIGGERS" | while IFS=: read -r id desc host days; do
echo " - [$id] $desc on $host (已${days}天未使用)"
TRIGGER_IDS+=("$id")
done
echo ""
read -p "是否删除这些触发器？(yes/no): " CONFIRM
if [ "$CONFIRM" = "yes" ]; then
echo "正在删除..."
delete_triggers "$AUTH_TOKEN" "${TRIGGER_IDS[@]}"
echo "删除完成"
else
echo "操作取消"
fi
}
# 执行主程序
main
Prometheus/Grafana监控环境清理
# prometheus-alert-cleaner.yml
# 用于清理Prometheus Alertmanager中的未使用trigger
apiVersion: v1
kind: ConfigMap
metadata:
name: trigger-cleaner
data:
cleanup.sh: |
#!/bin/sh
# 配置
AM_URL="http://alertmanager:9093"
PROM_URL="http://prometheus:9090"
# 获取所有已失效的警报规则
get_inactive_rules() {
curl -s "$PROM_URL/api/v1/rules" | \
jq -r '.data.groups[].rules[] | select(.state == "inactive") | .name'
}
# 获取最近没有触发的规则
get_untriggered_rules() {
local days_threshold=${1:-90}
curl -s "$AM_URL/api/v2/alerts/groups" | \
jq -r '.[] | .alerts[] | select(.status.state == "suppressed") | .labels.alertname' | \
sort -u
}
# 清理未使用的规则
cleanup_rules() {
local rules_file="$1"
while IFS= read -r rule; do
echo "清理规则: $rule"
# 这里实现具体的清理逻辑
# curl -X DELETE "$PROM_URL/api/v1/rules/$rule"
done < "$rules_file"
}
# 主程序
main() {
echo "获取未使用的警报规则..."
# 获取超过90天未触发的规则
UNUSED_RULES=$(get_untriggered_rules 90)
if [ -z "$UNUSED_RULES" ]; then
echo "没有未使用的规则"
exit 0
fi
echo "找到以下未使用的规则："
echo "$UNUSED_RULES"
# 执行清理
echo "$UNUSED_RULES" > /tmp/unused_rules.txt
cleanup_rules /tmp/unused_rules.txt
rm /tmp/unused_rules.txt
}
main
使用建议
- 先执行dry-run模式：在正式清理前，先模拟执行查看结果
- 备份重要规则：清理前备份触发器配置
- 分批次清理：避免一次性删除大量关键触发器
- 设置合理的阈值：根据业务情况调整时间阈值
- 日志记录：记录所有清理操作以便审计
选择适合你环境的方案,并根据实际需求进行修改，建议先在测试环境验证后再在生产环境使用。
