如何编写定时巡检数据库运行状态

本文目录导读：

1. Shell脚本 + crontab（最常用）
2. Python脚本 + systemd timer
3. 使用监控工具（Prometheus + Grafana）
4. 使用Ansible进行自动化巡检
5. 最佳实践建议

Shell脚本 + crontab（最常用）

检查MySQL/MariaDB

#!/bin/bash
# db_check.sh
# 数据库连接配置
DB_HOST="localhost"
DB_PORT="3306"
DB_USER="root"
DB_PASS="your_password"
DB_NAME="mysql"
LOG_FILE="/var/log/db_check.log"
DATE=$(date '+%Y-%m-%d %H:%M:%S')
# 发送告警函数
send_alert() {
    local message="$1"
    echo "$DATE - ALERT: $message" >> $LOG_FILE
    # 这里可以添加邮件、短信等告警方式
    # echo "$message" | mail -s "Database Alert" admin@example.com
}
# 检查数据库连接
check_connection() {
    if mysql -h$DB_HOST -P$DB_PORT -u$DB_USER -p$DB_PASS -e "SELECT 1" &>/dev/null; then
        echo "$DATE - OK: 数据库连接正常" >> $LOG_FILE
        return 0
    else
        echo "$DATE - ERROR: 数据库连接失败" >> $LOG_FILE
        send_alert "数据库连接失败: $DB_HOST:$DB_PORT"
        return 1
    fi
}
# 检查表空间使用
check_tablespace() {
    local threshold=80
    local usage=$(mysql -h$DB_HOST -P$DB_PORT -u$DB_USER -p$DB_PASS \
        -e "SELECT ROUND(SUM(data_length+index_length)/1024/1024, 2) as 'size_mb' \
            FROM information_schema.tables WHERE table_schema='$DB_NAME'" 2>/dev/null | tail -1)
    echo "$DATE - 表空间使用: ${usage}MB" >> $LOG_FILE
    if [ $usage -gt $threshold ]; then
        send_alert "数据库空间使用率过高: ${usage}MB"
    fi
}
# 检查复制状态（主从架构）
check_replication() {
    local status=$(mysql -h$DB_HOST -P$DB_PORT -u$DB_USER -p$DB_PASS \
        -e "SHOW SLAVE STATUS\G" 2>/dev/null | grep -E "Slave_IO_Running|Slave_SQL_Running")
    if echo "$status" | grep -q "No"; then
        send_alert "数据库复制异常: $status"
    else
        echo "$DATE - OK: 数据库复制正常" >> $LOG_FILE
    fi
}
# 主函数
main() {
    echo "========== 数据库巡检开始: $DATE ==========" >> $LOG_FILE
    check_connection
    check_tablespace
    check_replication
    echo "========== 数据库巡检结束 ==========" >> $LOG_FILE
}
main

设置crontab定时任务

# 编辑crontab
crontab -e
# 每小时运行一次
0 * * * * /path/to/db_check.sh
# 每天凌晨2点运行
0 2 * * * /path/to/db_check.sh
# 每30分钟运行一次
*/30 * * * * /path/to/db_check.sh

Python脚本 + systemd timer

Python巡检脚本

#!/usr/bin/env python3
# db_checker.py
import subprocess
import json
import logging
import datetime
import smtplib
from email.mime.text import MIMEText
# 配置
DB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': 'your_password'
}
LOG_FILE = '/var/log/db_checker.log'
class DatabaseChecker:
    def __init__(self):
        self.logger = self.setup_logger()
    def setup_logger(self):
        logger = logging.getLogger('DB_Checker')
        logger.setLevel(logging.INFO)
        handler = logging.FileHandler(LOG_FILE)
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        return logger
    def run_query(self, query):
        """执行SQL查询"""
        cmd = f"mysql -h{DB_CONFIG['host']} -P{DB_CONFIG['port']} -u{DB_CONFIG['user']} -p{DB_CONFIG['password']} -e '{query}'"
        try:
            result = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT)
            return result.decode().strip()
        except subprocess.CalledProcessError as e:
            self.logger.error(f"查询失败: {e.output.decode()}")
            return None
    def check_connection(self):
        """检查数据库连接"""
        result = self.run_query("SELECT 1 as test")
        if result and '1' in result:
            self.logger.info("数据库连接正常")
            return True
        self.logger.error("数据库连接失败")
        return False
    def check_performance(self):
        """检查数据库性能指标"""
        checks = {
            '连接数': "SHOW VARIABLES LIKE 'max_connections'",
            '当前连接': "SHOW STATUS LIKE 'Threads_connected'",
            '慢查询': "SHOW STATUS LIKE 'Slow_queries'",
            '缓存命中': "SHOW STATUS LIKE 'Qcache_hits'"
        }
        results = {}
        for name, query in checks.items():
            output = self.run_query(query)
            if output:
                results[name] = output.split('\n')[1] if '\n' in output else output
        return results
    def check_tables(self):
        """检查表状态"""
        query = """
            SELECT TABLE_SCHEMA, TABLE_NAME, ENGINE, 
                   ROUND((DATA_LENGTH + INDEX_LENGTH) / 1024 / 1024, 2) as SIZE_MB
            FROM information_schema.tables
            WHERE TABLE_SCHEMA NOT IN ('information_schema', 'mysql', 'performance_schema')
        """
        return self.run_query(query)
    def send_alert(self, subject, message):
        """发送告警"""
        # 实现邮件或其他告警方式
        pass
    def run_checks(self):
        """执行所有检查"""
        self.logger.info("开始数据库巡检")
        checks_passed = 0
        checks_failed = 0
        if self.check_connection():
            checks_passed += 1
        else:
            checks_failed += 1
            self.send_alert("数据库连接失败", "无法连接到数据库服务器")
        # 性能检查
        perf_results = self.check_performance()
        if perf_results:
            self.logger.info(f"性能指标: {json.dumps(perf_results, indent=2)}")
            checks_passed += 1
        else:
            checks_failed += 1
        # 表状态检查
        table_info = self.check_tables()
        if table_info:
            self.logger.info(f"表状态: \n{table_info}")
            checks_passed += 1
        else:
            checks_failed += 1
        self.logger.info(f"巡检完成: 通过 {checks_passed}, 失败 {checks_failed}")
        return {
            'passed': checks_passed,
            'failed': checks_failed,
            'timestamp': datetime.datetime.now().isoformat()
        }
if __name__ == "__main__":
    checker = DatabaseChecker()
    result = checker.run_checks()
    print(json.dumps(result, indent=2))

配置systemd timer

# /etc/systemd/system/db-checker.service
[Unit]
Description=Database Checker Service
After=network.target
[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /opt/scripts/db_checker.py
[Install]
WantedBy=multi-user.target

# /etc/systemd/system/db-checker.timer
[Unit]
Description=Run database checker every hour
Requires=db-checker.service
[Timer]
OnCalendar=hourly
Persistent=true
[Install]
WantedBy=timers.target

使用监控工具（Prometheus + Grafana）

Prometheus exporter配置

# mysqld_exporter配置
[client]
user = exporter
password = exporter_password
# 启动mysqld_exporter
mysqld_exporter --config.my-cnf=/etc/mysql_exporter/.my.cnf

Prometheus告警规则

groups:
- name: mysql_alerts
  rules:
  - alert: MySQLDown
    expr: mysql_up == 0
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "MySQL instance {{ $labels.instance }} is down"
  - alert: MySQLHighThreadsRunning
    expr: mysql_global_status_threads_running > 100
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "MySQL high threads running"
  - alert: MySQLSlowQueries
    expr: rate(mysql_global_status_slow_queries[5m]) > 5
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "MySQL slow queries detected"

使用Ansible进行自动化巡检

# db_check.yml
---
- name: Database Health Check
  hosts: database_servers
  gather_facts: yes
  vars:
    alert_email: "admin@example.com"
  tasks:
    - name: Check MySQL service
      service:
        name: mysql
        state: started
      register: mysql_service_status
    - name: Check MySQL connectivity
      command: mysql -h localhost -u root -p{{ mysql_root_password }} -e "SELECT 1"
      register: mysql_connection
      ignore_errors: yes
    - name: Get database size
      command: >
        mysql -h localhost -u root -p{{ mysql_root_password }} 
        -e "SELECT table_schema, 
                   ROUND(SUM(data_length+index_length)/1024/1024,2) as size_mb 
            FROM information_schema.tables 
            GROUP BY table_schema"
      register: db_size
    - name: Check replication status
      command: mysql -h localhost -u root -p{{ mysql_root_password }} -e "SHOW SLAVE STATUS\G"
      register: slave_status
      when: "'replica' in group_names"
    - name: Generate report
      template:
        src: db_report.j2
        dest: /tmp/db_check_report_{{ ansible_date_time.date }}.txt
    - name: Send report
      mail:
        to: "{{ alert_email }}"
        subject: "Database Health Check Report - {{ ansible_date_time.date }}"
        body: "{{ lookup('file', '/tmp/db_check_report_' + ansible_date_time.date + '.txt') }}"
      when: mysql_connection is failed or mysql_service_status.state != 'started'

最佳实践建议

1. 定期执行策略：

   • 关键指标：每分钟检查一次
   • 常规检查：每小时检查一次
   • 详细巡检：每天凌晨执行一次

2. 监控指标：

   • 连接数
   • 查询性能
   • 磁盘使用率
   • 复制延迟
   • 错误日志
   • 慢查询日志

3. 告警级别：

   • 严重：服务不可用
   • 警告：性能下降
   • 信息：日常状态报告

4. 日志管理：

   • 设置日志轮转
   • 保留最近30天日志
   • 日志压缩存档

选择哪种方案取决于您的具体需求：

• 简单场景：Shell脚本 + crontab
• 复杂场景：Python + systemd
• 大规模集群：Prometheus + Grafana
• 自动化运维：Ansible