# 步骤1：配置Jenkins流水线
$ cat > Jenkinsfile << EOF pipeline { agent any stages { stage('Build') { steps { sh 'mvn clean package' } } stage('Test') { steps { sh 'mvn test' } } stage('Deploy') { steps { sh 'ansible-playbook deploy.yml' } } stage('DR Test') { steps { sh './run_dr_test.sh' } } } post { success { echo 'Deployment successful' } failure { echo 'Deployment failed' } } } EOF # 步骤2：配置GitLab CI/CD $ cat > .gitlab-ci.yml << EOF stages: - build - test - deploy - dr_test build: stage: build script: - mvn clean package test: stage: test script: - mvn test deploy: stage: deploy script: - ansible-playbook deploy.yml only: - master dr_test: stage: dr_test script: - ./run_dr_test.sh only: - master EOF

# 步骤1：创建容灾测试脚本
$ cat > run_dr_test.sh << EOF #!/bin/bash # 记录测试开始时间 echo "[$(date)] 开始容灾测试" # 模拟故障 echo "[$(date)] 模拟主系统故障" systemctl stop mysql # 执行故障转移 echo "[$(date)] 执行故障转移" /usr/local/bin/failover.sh # 验证备用系统状态 echo "[$(date)] 验证备用系统状态" sleep 30 if systemctl is-active mysql; then echo "[$(date)] 备用系统启动成功" else echo "[$(date)] 备用系统启动失败" exit 1 fi # 验证数据一致性 echo "[$(date)] 验证数据一致性" mysql -u root -p -e "SELECT COUNT(*) FROM test_db.test_table;" # 执行回切操作 echo "[$(date)] 执行回切操作" /usr/local/bin/failback.sh # 记录测试结束时间 echo "[$(date)] 容灾测试完成" EOF # 步骤2：配置测试报告 $ cat > generate_test_report.sh << EOF #!/bin/bash # 生成测试报告 echo "容灾测试报告" > dr_test_report.txt
echo “测试时间: $(date)” >> dr_test_report.txt
echo “测试结果: 成功” >> dr_test_report.txt
echo “RTO: 30秒” >> dr_test_report.txt
echo “RPO: 0秒” >> dr_test_report.txt

# 发送测试报告
mail -s “容灾测试报告” admin@fgedu.net.cn < dr_test_report.txt EOF

# 步骤1：创建Terraform配置文件
$ cat > main.tf << EOF provider "aws" { region = "us-east-1" } # 创建S3存储桶 resource "aws_s3_bucket" "dr_backup" { bucket = "fgedu-dr-backup" acl = "private" versioning { enabled = true } lifecycle_rule { id = "transition-to-ia" enabled = true transition { days = 30 storage_class = "STANDARD_IA" } } } # 创建EC2实例 resource "aws_instance" "dr_instance" { ami = "ami-0c55b159cbfafe1f0" instance_type = "t2.micro" key_name = "my-key-pair" security_groups = [aws_security_group.dr_sg.id] user_data = <<-EOF #!/bin/bash yum update -y yum install -y mysql-server systemctl start mysqld systemctl enable mysqld EOF } # 创建安全组 resource "aws_security_group" "dr_sg" { name = "dr-security-group" description = "Security group for DR instances" ingress { from_port = 22 to_port = 22 protocol = "tcp" cidr_blocks = ["0.0.0.0/0"] } ingress { from_port = 3306 to_port = 3306 protocol = "tcp" cidr_blocks = ["0.0.0.0/0"] } } EOF # 步骤2：执行Terraform部署 $ terraform init $ terraform plan $ terraform apply

# 步骤1：创建Ansible playbook
$ cat > dr-setup.yml << EOF --- - hosts: dr_servers become: yes tasks: - name: 安装必要软件 yum: name: - mysql-server - keepalived - rsync state: present - name: 配置MySQL template: src: templates/my.cnf.j2 dest: /etc/my.cnf - name: 配置Keepalived template: src: templates/keepalived.conf.j2 dest: /etc/keepalived/keepalived.conf - name: 启动服务 service: name: "{{ item }}" state: started enabled: yes with_items: - mysqld - keepalived - name: 配置数据复制 shell: | mysql -u root -p{{ mysql_password }} -e "CHANGE MASTER TO MASTER_HOST='{{ master_ip }}', MASTER_USER='repl', MASTER_PASSWORD='{{ replication_password }}', MASTER_LOG_FILE='mysql-bin.000001', MASTER_LOG_POS=107; START SLAVE;" EOF # 步骤2：执行Ansible部署 $ ansible-playbook dr-setup.yml -i inventory.ini

# 步骤1：创建Prometheus配置文件
$ cat > prometheus.yml << EOF global: scrape_interval: 15s evaluation_interval: 15s alerting: alertmanagers: - static_configs: - targets: - fgedudb:9093 rule_files: - "dr_alerts.yml" scrape_configs: - job_name: 'disaster_recovery' static_configs: - targets: ['dr-monitor:9100'] metrics_path: '/metrics' EOF # 步骤2：创建告警规则 $ cat > dr_alerts.yml << EOF groups: - name: dr_alerts rules: - alert: ReplicationFailed expr: mysql_slave_running == 0 for: 5m labels: severity: critical annotations: summary: "复制失败" description: "MySQL复制已失败超过5分钟" - alert: BackupFailed expr: backup_success == 0 for: 1h labels: severity: warning annotations: summary: "备份失败" description: "备份已失败超过1小时" - alert: FailoverTriggered expr: failover_triggered == 1 labels: severity: critical annotations: summary: "故障转移已触发" description: "容灾系统已触发故障转移" EOF

# 步骤1：创建Grafana仪表板
$ cat > dr-dashboard.json << EOF { "dashboard": { "id": null, "title": "Disaster Recovery Status", "panels": [ { "title": "Replication Status", "type": "gauge", "targets": [ { "expr": "mysql_slave_running" } ], "options": { "maxValue": 1, "minValue": 0, "thresholds": [ { "colorMode": "critical", "fill": true, "line": true, "op": "lt", "value": 1 } ] } }, { "title": "Backup Status", "type": "gauge", "targets": [ { "expr": "backup_success" } ], "options": { "maxValue": 1, "minValue": 0, "thresholds": [ { "colorMode": "critical", "fill": true, "line": true, "op": "lt", "value": 1 } ] } }, { "title": "Failover Events", "type": "graph", "targets": [ { "expr": "failover_triggered" } ] } ] } } EOF # 步骤2：导入Grafana仪表板 $ curl -X POST -H "Content-Type: application/json" -d @dr-dashboard.json http://grafana:3000/api/dashboards/db

# 步骤1：创建自动化测试脚本
$ cat > auto_dr_test.py << EOF #!/usr/bin/env python3 import subprocess import time import smtplib from email.mime.text import MIMEText # 执行故障转移测试 def test_failover(): print("开始故障转移测试...") # 模拟主系统故障 subprocess.run(["systemctl", "stop", "mysql"]) # 执行故障转移 subprocess.run(["/usr/local/bin/failover.sh"]) # 等待备用系统启动 time.sleep(30) # 验证备用系统状态 result = subprocess.run(["systemctl", "is-active", "mysql"], capture_output=True, text=True) if result.stdout.strip() == "active": print("备用系统启动成功") return True else: print("备用系统启动失败") return False # 执行回切测试 def test_failback(): print("开始回切测试...") # 执行回切操作 subprocess.run(["/usr/local/bin/failback.sh"]) # 等待主系统启动 time.sleep(30) # 验证主系统状态 result = subprocess.run(["systemctl", "is-active", "mysql"], capture_output=True, text=True) if result.stdout.strip() == "active": print("主系统启动成功") return True else: print("主系统启动失败") return False # 发送测试报告 def send_report(success): msg = MIMEText(f"容灾测试{'成功' if success else '失败'}") msg['Subject'] = "容灾测试报告" msg['From'] = "dr-test@fgedu.net.cn" msg['To'] = "admin@fgedu.net.cn" with smtplib.SMTP('smtp.fgedu.net.cn') as server: server.login('username', 'password') server.send_message(msg) if __name__ == "__main__": failover_success = test_failover() failback_success = test_failback() if failover_success and failback_success: print("容灾测试成功") send_report(True) else: print("容灾测试失败") send_report(False) EOF # 步骤2：配置Cron定时执行 $ crontab -e 0 2 * * 0 /usr/local/bin/auto_dr_test.py

# 步骤1：创建GitHub Actions workflow
$ cat > .github/workflows/dr-test.yml << EOF name: Disaster Recovery Test on: push: branches: [ master ] pull_request: branches: [ master ] schedule: - cron: '0 2 * * 0' jobs: dr-test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.8' - name: Install dependencies run: | python -m pip install --upgrade pip if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - name: Run DR test run: | python auto_dr_test.py - name: Send notification if: failure() run: | curl -X POST -H "Content-Type: application/json" -d '{"text": "容灾测试失败"}' ${{ secrets.SLACK_WEBHOOK }} EOF