批量运维管理平台配置示例

#!/usr/bin/env python3
# batch_operation_platform.py
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn

from flask import Flask, render_template, request, jsonify
import psycopg2
import paramiko
import time

app = Flask(__name__)

# 数据库连接配置
DB_CONFIG = {
    'host': 'localhost',
    'port': 5432,
    'user': 'fgedu',
    'password': 'Fgedu@123',
    'database': 'batch_platform'
}

# 服务器列表
SERVERS = [
    {'host': '192.168.1.100', 'port': 22, 'user': 'root', 'password': 'password123'},
    {'host': '192.168.1.101', 'port': 22, 'user': 'root', 'password': 'password123'},
    {'host': '192.168.1.102', 'port': 22, 'user': 'root', 'password': 'password123'}
]

# 数据库实例列表
DB_INSTANCES = [
    {'name': 'fgedudb01', 'host': '192.168.1.100', 'port': 5432, 'user': 'fgedu', 'password': 'Fgedu@123'},
    {'name': 'fgedudb02', 'host': '192.168.1.101', 'port': 5432, 'user': 'fgedu', 'password': 'Fgedu@123'},
    {'name': 'fgedudb03', 'host': '192.168.1.102', 'port': 5432, 'user': 'fgedu', 'password': 'Fgedu@123'}
]

# 连接数据库
def connect_db():
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        return conn
    except Exception as e:
        print(f"数据库连接失败: {e}")
        return None

# 执行Shell命令
def execute_shell_command(host, port, user, password, command):
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(host, port, user, password)
        stdin, stdout, stderr = ssh.exec_command(command)
        output = stdout.read().decode('utf-8')
        error = stderr.read().decode('utf-8')
        ssh.close()
        return output, error
    except Exception as e:
        return "", f"执行失败: {e}"

# 执行SQL语句
def execute_sql(host, port, user, password, database, sql):
    try:
        conn = psycopg2.connect(
            host=host,
            port=port,
            user=user,
            password=password,
            database=database
        )
        cur = conn.cursor()
        cur.execute(sql)
        if sql.strip().lower().startswith('select'):
            result = cur.fetchall()
        else:
            conn.commit()
            result = f"执行成功，影响行数: {cur.rowcount}"
        cur.close()
        conn.close()
        return result, ""
    except Exception as e:
        return "", f"执行失败: {e}"

# 批量执行Shell命令
@app.route('/batch/shell', methods=['POST'])
def batch_shell():
    command = request.form.get('command')
    if not command:
        return jsonify({'status': 'error', 'message': '命令不能为空'})
    
    results = []
    for server in SERVERS:
        output, error = execute_shell_command(
            server['host'],
            server['port'],
            server['user'],
            server['password'],
            command
        )
        results.append({
            'server': server['host'],
            'output': output,
            'error': error
        })
    
    return jsonify({'status': 'success', 'results': results})

# 批量执行SQL语句
@app.route('/batch/sql', methods=['POST'])
def batch_sql():
    sql = request.form.get('sql')
    if not sql:
        return jsonify({'status': 'error', 'message': 'SQL语句不能为空'})
    
    results = []
    for instance in DB_INSTANCES:
        result, error = execute_sql(
            instance['host'],
            instance['port'],
            instance['user'],
            instance['password'],
            instance['name'],
            sql
        )
        results.append({
            'instance': instance['name'],
            'result': result,
            'error': error
        })
    
    return jsonify({'status': 'success', 'results': results})

# 获取实例状态
@app.route('/instances/status', methods=['GET'])
def get_instances_status():
    results = []
    for instance in DB_INSTANCES:
        result, error = execute_sql(
            instance['host'],
            instance['port'],
            instance['user'],
            instance['password'],
            instance['name'],
            'SELECT pg_is_in_recovery() as is_slave, now() as current_time'
        )
        if not error:
            status = '备库' if result[0][0] else '主库'
        else:
            status = '异常'
        results.append({
            'instance': instance['name'],
            'host': instance['host'],
            'status': status,
            'error': error
        })
    
    return jsonify({'status': 'success', 'results': results})

# 主页面
@app.route('/')
def index():
    return render_template('index.html')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

3.3 批量操作脚本开发

批量操作脚本开发示例：

#!/bin/bash
# batch_backup.sh
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn

# 定义变量
BACKUP_DIR="/opengauss/backup"
DATE=$(date +'%Y%m%d_%H%M%S')
LOG_FILE="/opengauss/logs/batch_backup.log"

# 数据库实例列表
INSTANCES=("192.168.1.100:5432:fgedu:Fgedu@123:fgedudb01" "192.168.1.101:5432:fgedu:Fgedu@123:fgedudb02" "192.168.1.102:5432:fgedu:Fgedu@123:fgedudb03")

# 日志函数
log() {
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1" >> $LOG_FILE
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
}

# 检查备份目录
check_backup_dir() {
    if [ ! -d $BACKUP_DIR ]; then
        log "创建备份目录：$BACKUP_DIR"
        mkdir -p $BACKUP_DIR
    fi
}

# 执行备份
execute_backup() {
    for instance in "${INSTANCES[@]}"; do
        IFS=":" read -r host port user password dbname <<< "$instance"
        log "开始备份数据库：$dbname@$host:$port"
        
        # 执行备份
        backup_file="$BACKUP_DIR/${dbname}_backup_$DATE.sql"
        gs_dump -h $host -p $port -U $user -d $dbname -f $backup_file -F p
        
        if [ $? -eq 0 ]; then
            log "备份成功，备份文件：$backup_file"
            # 压缩备份文件
            gzip $backup_file
            log "备份文件已压缩：$backup_file.gz"
        else
            log "备份失败：$dbname@$host:$port"
        fi
    done
}

# 清理过期备份
cleanup_backup() {
    log "清理7天前的备份文件"
    find $BACKUP_DIR -name "*.sql.gz" -mtime +7 -delete
    log "清理完成"
}

# 主流程
log "=== 批量备份开始 ==="

check_backup_dir
execute_backup
cleanup_backup

log "=== 批量备份完成 ==="

#!/bin/bash
# platform_troubleshoot.sh
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn

# 定义变量
LOG_FILE="/opengauss/logs/platform_troubleshoot.log"
PLATFORM_DIR="/opengauss/batch_platform"

# 日志函数
log() {
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1" >> $LOG_FILE
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
}

# 检查平台状态
check_platform_status() {
    log "检查平台状态..."
    
    # 检查进程是否运行
    ps aux | grep batch_operation_platform.py | grep -v grep > /dev/null 2>&1
    if [ $? -eq 0 ]; then
        log "平台进程运行正常"
    else
        log "平台进程未运行"
    fi
    
    # 检查端口是否监听
    netstat -tlnp | grep 5000 > /dev/null 2>&1
    if [ $? -eq 0 ]; then
        log "平台端口监听正常"
    else
        log "平台端口未监听"
    fi
    
    # 检查数据库连接
    psql -h localhost -p 5432 -U fgedu -d batch_platform -c "SELECT 1; 
" > /dev/null 2>&1
    if [ $? -eq 0 ]; then
        log "数据库连接正常"
    else
        log "数据库连接失败"
    fi
}

# 检查日志
check_logs() {
    log "检查平台日志..."
    if [ -f "$PLATFORM_DIR/error.log" ]; then
        log "平台错误日志："
        tail -n 10 "$PLATFORM_DIR/error.log" >> $LOG_FILE
    else
        log "平台错误日志不存在"
    fi
}

# 检查配置
check_config() {
    log "检查平台配置..."
    if [ -f "$PLATFORM_DIR/config.py" ]; then
        log "平台配置文件存在"
    else
        log "平台配置文件不存在"
    fi
}

# 重启平台
restart_platform() {
    log "重启平台..."
    
    # 停止平台
    pkill -f batch_operation_platform.py
    sleep 5
    
    # 启动平台
    cd $PLATFORM_DIR
    nohup python3 batch_operation_platform.py > platform.log 2>&1 &
    sleep 5
    
    # 检查平台状态
    check_platform_status
}

# 主流程
log "=== 平台故障处理开始 ==="

check_platform_status
check_logs
check_config

# 询问是否重启平台
read -p "是否重启平台？(y/n): " answer
if [ "$answer" = "y" ]; then
    restart_platform
fi

log "=== 平台故障处理完成 ==="