本文目录导读：
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'
最佳实践建议
-
定期执行策略：
- 关键指标：每分钟检查一次
- 常规检查：每小时检查一次
- 详细巡检：每天凌晨执行一次
-
监控指标：
- 连接数
- 查询性能
- 磁盘使用率
- 复制延迟
- 错误日志
- 慢查询日志
-
告警级别：
- 严重：服务不可用
- 警告：性能下降
- 信息：日常状态报告
-
日志管理：
- 设置日志轮转
- 保留最近30天日志
- 日志压缩存档
选择哪种方案取决于您的具体需求：
- 简单场景：Shell脚本 + crontab
- 复杂场景：Python + systemd
- 大规模集群：Prometheus + Grafana
- 自动化运维：Ansible
