如何用Python导出指定时间范围的Prometheus Alertmanager告警至Excel

我在AWS上部署了基于EKS的监控平台，配置了Prometheus数据源与Alertmanager，当前Alertmanager触发的告警会发送至Slack频道。我尝试通过Alertmanager API获取指定时间段内触发的告警，并编写了Python代码。

目前代码可成功将告警导出至Excel，但导出的告警包含多个随机日期（如2024年2月20日、21日、26日、27日）的内容，而非仅指定日期（如2024年2月27日）的告警。需要修改代码，使其仅导出指定日期范围内触发的告警。

import requests
import pandas as pd
from datetime import datetime
import os  # Import the os module

def export_alerts(prometheus_url, start_time, end_time):
    # Construct the URL to query the alerts
    alerts_url = f"{prometheus_url}/api/v1/alerts"

    # Initialize an empty list to store all alerts
    all_alerts = []

    # Define query parameters for the initial request
    params = {
        "start": start_time,
        "end": end_time,
        "limit": 100  # Adjust as needed based on your expected number of alerts per query
    }

    try:
        while True:
            # Send a GET request to fetch alerts with pagination
            response = requests.get(alerts_url, params=params)
            response.raise_for_status()  # Raise an exception for any error response

            # Extract alerts from the response
            alerts_data = response.json()["data"]["alerts"]

            # Append alerts to the list of all alerts
            all_alerts.extend(alerts_data)

            # Check if there are more alerts to fetch
            if "next" in response.json()["data"]:
                params["start"] = response.json()["data"]["next"]
            else:
                break  # Exit loop if no more alerts to fetch

        # Return all alerts
        return all_alerts

    except requests.RequestException as e:
        # Handle any HTTP request errors
        print(f"Error fetching alerts: {e}")
        return None

# Example usage
prometheus_url = "https://prometheus.dev.spencecom.eu.spence.cloud"
start_time = "2024-02-26T00:00:00Z"
end_time = "2024-02-27T23:59:59Z"

alerts = export_alerts(prometheus_url, start_time, end_time)
if alerts:

    # Convert alerts to a DataFrame
    df = pd.DataFrame(alerts)

    # Print column names for verification
    print("Column names in DataFrame:")
    print(df.columns)

    # Generate timestamp for filename
    timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")

    # dir to save excel
    # Example: "/Users/your_username/Documents/alerts/"
    output_directory = "/Users/spenceebuka/Documents/PYTHON_SCRIPT"

    # Ensure that the output directory exists
    os.makedirs(output_directory, exist_ok=True)

    # Export DataFrame to Excel file with specified column names and timestamp in filename
    excel_file_path = os.path.join(output_directory, f"alerts_{timestamp}.xlsx")
    df.to_excel(excel_file_path, index=False)
    print(f"Alerts exported to {excel_file_path} successfully.")
else:
    print("Failed to fetch alerts.")

你当前调用的/api/v1/alerts是Prometheus的接口，它仅返回当前处于活跃状态（firing）的告警，并不支持按时间段过滤历史告警。传入的start和end参数对这个接口无效，所以会导出所有当前活跃的告警，不管它们的触发时间。

要获取指定时间段内触发的所有告警（包括已恢复的），需要改用Prometheus的/api/v1/query_range接口查询ALERTS指标，该指标记录了所有告警的触发与恢复事件。

import requests
import pandas as pd
from datetime import datetime
import os
from dateutil import parser

def export_alerts(prometheus_url, start_time, end_time):
    # 使用Prometheus的query_range接口查询ALERTS指标，获取指定时间段内的所有告警事件
    query_url = f"{prometheus_url}/api/v1/query_range"
    
    # 将时间字符串转换为时间戳（秒）
    start_ts = int(parser.isoparse(start_time).timestamp())
    end_ts = int(parser.isoparse(end_time).timestamp())
    
    params = {
        "query": "ALERTS",
        "start": start_ts,
        "end": end_ts,
        "step": "30s"  # 步长，根据告警频率调整，越小越精准但数据量越大
    }

    try:
        response = requests.get(query_url, params=params)
        response.raise_for_status()
        data = response.json()["data"]["result"]
        
        all_alerts = []
        for series in data:
            labels = series["metric"]
            # 遍历每个告警的时间点数据
            for timestamp, value in series["values"]:
                alert_time = datetime.utcfromtimestamp(int(timestamp)).isoformat() + "Z"
                alert_status = "firing" if value == "1" else "resolved"
                
                # 提取告警关键信息
                alert_info = {
                    "alertname": labels.get("alertname"),
                    "status": alert_status,
                    "instance": labels.get("instance"),
                    "job": labels.get("job"),
                    "severity": labels.get("severity"),
                    "trigger_time": alert_time,
                    "labels": str(labels)  # 保留完整标签信息
                }
                all_alerts.append(alert_info)
        
        return all_alerts

    except requests.RequestException as e:
        print(f"Error fetching alerts: {e}")
        return None

# Example usage
prometheus_url = "https://prometheus.dev.spencecom.eu.spence.cloud"
start_time = "2024-02-26T00:00:00Z"
end_time = "2024-02-27T23:59:59Z"

alerts = export_alerts(prometheus_url, start_time, end_time)
if alerts:
    df = pd.DataFrame(alerts)
    
    # 进一步过滤触发时间在指定范围内的告警（确保数据准确性）
    df["trigger_time"] = pd.to_datetime(df["trigger_time"])
    start_dt = pd.to_datetime(start_time)
    end_dt = pd.to_datetime(end_time)
    df_filtered = df[(df["trigger_time"] >= start_dt) & (df["trigger_time"] <= end_dt)]
    
    print(f"Filtered {len(df_filtered)} alerts within the specified time range")
    
    timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    output_directory = "/Users/spenceebuka/Documents/PYTHON_SCRIPT"
    os.makedirs(output_directory, exist_ok=True)
    
    excel_file_path = os.path.join(output_directory, f"alerts_{timestamp}.xlsx")
    df_filtered.to_excel(excel_file_path, index=False)
    print(f"Alerts exported to {excel_file_path} successfully.")
else:
    print("Failed to fetch alerts.")

• 接口替换：改用/api/v1/query_range查询ALERTS指标，该指标会返回指定时间段内所有告警的触发（value=1）和恢复（value=0）事件。
• 时间处理：将ISO格式的时间字符串转换为时间戳，符合Prometheus接口要求；同时解析返回的时间戳为可读格式。
• 数据过滤：在生成DataFrame后，额外添加了时间范围过滤，确保仅保留指定时间段内的告警。
• 字段提取：从指标标签中提取告警名称、实例、级别等关键信息，方便Excel查看。

1. 确保Prometheus服务器允许该API请求，没有权限限制。
2. 调整step参数：如果告警频率高，可减小步长（如10s）；如果告警少，可增大步长（如1m）以减少数据量。
3. 若需要获取告警的注释（annotations），可以查询ALERTS_ANNOTATIONS指标，结合ALERTS数据关联获取。

内容的提问来源于stack exchange，提问作者Emmanuel Spencer Egbuniwe