python脚本：pushgateway+mysql自定查询

#!/usr/bin/python3
import mysql.connector
import requests

MYSQL_CONFIG = {
    'user': 'your_user',
    'password': 'your_password',
    'host': 'your_host',
    'database': 'your_database',
    'port': 3306
}

# Pushgateway 配置
PUSHGATEWAY_URL = 'http://your_pushgateway_url:9091/metrics/job/mysql_custom_queries'

# 自定义查询
CUSTOM_QUERIES = [
    {
        'query': "SELECT COUNT(1) FROM table_1 WHERE condition_col = 'value' AND timestamp_col > DATE_ADD(NOW(), INTERVAL -1 HOUR);",
        'metrics_name': 'metric_1_count',
        'description': 'Description of metric 1 for the last hour'
    },
    {
        'query': "SELECT COUNT(1) FROM table_2 WHERE condition_col = 'value' AND timestamp_col > DATE_ADD(NOW(), INTERVAL -1 HOUR);",
        'metrics_name': 'metric_2_count',
        'description': 'Description of metric 2 for the last hour'
    },
    {
        'query': "SELECT COUNT(1) FROM table_3 WHERE timestamp_col > DATE_ADD(NOW(), INTERVAL -1 HOUR);",
        'metrics_name': 'metric_3_count',
        'description': 'Description of metric 3 for the last hour'
    },
    {
        'query': "SELECT COUNT(1) FROM table_4 WHERE timestamp_col > DATE_ADD(NOW(), INTERVAL -1 HOUR);",
        'metrics_name': 'metric_4_count',
        'description': 'Description of metric 4 for the last hour'
    }
]

# 执行查询并将数据发送到 Pushgateway
def send_metrics_to_pushgateway():
    try:
        # 连接到 MySQL 数据库
        connection = mysql.connector.connect(**MYSQL_CONFIG)
        cursor = connection.cursor()

        # 用于存储所有指标的结果
        metrics_data = ""

        # 遍历每个查询
        for query_info in CUSTOM_QUERIES:
            cursor.execute(query_info['query'])
            result = cursor.fetchone()
            value = result[0] if result else 0

            # 准备 metrics 数据
            metrics_data += f"# HELP {query_info['metrics_name']} {query_info['description']}\n"
            metrics_data += f"# TYPE {query_info['metrics_name']} gauge\n"
            metrics_data += f"{query_info['metrics_name']} {value}\n"

        # 发送到 Pushgateway
        response = requests.post(PUSHGATEWAY_URL, data=metrics_data)
  
    except mysql.connector.Error as err:
        print(f"Error: {err}")
  
    finally:
        cursor.close()
        connection.close()

# 执行主函数
if __name__ == "__main__":
    send_metrics_to_pushgateway()