自动化脚本如何清理未使用触发器

本文目录导读：

1. 方案一：基于最后执行时间的清理脚本
2. 方案二：基于事件历史记录的清理脚本（更精确）
3. 方案三：Shell脚本版本（适用于Linux系统）
4. 方案四：Prometheus/Grafana监控环境清理
5. 使用建议

基于最后执行时间的清理脚本

#!/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

使用建议

1. 先执行dry-run模式：在正式清理前，先模拟执行查看结果
2. 备份重要规则：清理前备份触发器配置
3. 分批次清理：避免一次性删除大量关键触发器
4. 设置合理的阈值：根据业务情况调整时间阈值
5. 日志记录：记录所有清理操作以便审计

选择适合你环境的方案,并根据实际需求进行修改，建议先在测试环境验证后再在生产环境使用。