Backup web bằng Restic và Rclone với Google Drive lưu trữ

Ở phần trước, mình đã hướng dẫn các bạn khởi tạo Restic với Rclone và Google Drive để backup và lưu trữ backup trên đó, trong phần này mình sẽ tạo script backup bằng python. Script này dùng để backup user trên DirectAdmin chạy OS AlmaLinux, chi tiết backup:

#!/usr/bin/env python3
import logging
import os
import subprocess
import json
import concurrent.futures
import smtplib
import requests
from pathlib import Path
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.utils import formatdate
import time

CONFIG = {
'restic_bin': '/usr/local/bin/restic',
'rclone_bin': '/usr/local/bin/rclone',
'repository': 'rclone:gd:/restic-backups',
'password': 'i2bdi2oa(23J',
'exclude_patterns': [
'*.tmp',
'*.cache',
'*.log',
'*.temp',
'/home/*/tmp',
'/home/*/cache'
],
'file_types': {
'config': ['.conf', '.ini', '.cfg'],
'website': ['.html', '.php', '.css', '.js'],
'database': ['.sql', '.dmp'],
'media': ['.jpg', '.png', '.mp4']
},
'rclone_config': '/root/.config/rclone/rclone.conf',
'mysql': {
'enabled': True,
'user': 'da_admin',
'password': 'qcqw832hjasUancis',
'host': 'localhost',
'port': 3306,
'dump_dir': '/tmp/mysql_dumps',
'retention_days': 3,
'excluded_databases': [
'information_schema', 
'performance_schema', 
'mysql', 
'sys'
],
'options': [
'--single-transaction',
'--quick',
'--skip-lock-tables'
]
},
'telegram': {
'enabled': True,
'bot_token': 'CHATBOT_TOKEN',
'chat_id': 'CHAT_ID',
'api_timeout': 10
},
'parallel_workers': 4,
'compression': 'auto',
'upload_limit': 204800,
'cache_dir': '/tmp/restic-cache',
'retention_policy': {
'keep_last': 7,
'keep_daily': 14,
'keep_weekly': 8,
'keep_monthly': 12,
'enable_prune': True
},
'date_tag_format': '%Y%m%d'
}

logging.basicConfig(
filename='/var/log/file_level_backup.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filemode='a'
)

class EnhancedBackup:
def __init__(self):
self.start_time = datetime.now()
self.env = os.environ.copy()
self.current_date_tag = datetime.now().strftime(CONFIG['date_tag_format'])
self.env.update({
'RESTIC_PASSWORD': CONFIG['password'],
'RCLONE_CONFIG': CONFIG['rclone_config']
})
self.db_success_count = 0
self.db_failure_count = 0
self.setup_path()
self.clean_old_dumps()

def setup_path(self):
essential_paths = ['/usr/bin', '/usr/local/bin', '/bin']
self.env['PATH'] = ":".join(essential_paths + self.env.get('PATH', '').split(':'))

def clean_old_dumps(self):
if CONFIG['mysql']['enabled']:
dump_dir = Path(CONFIG['mysql']['dump_dir'])
cutoff = time.time() - (CONFIG['mysql']['retention_days'] * 86400)
for f in dump_dir.glob("*.sql.gz"):
if f.stat().st_mtime < cutoff:
try:
f.unlink()
logging.info(f"Deleted old dump: {f}")
except Exception as e:
logging.error(f"Failed to delete {f}: {str(e)}")

def mysql_backup(self):
if not CONFIG['mysql']['enabled']:
return True

try:
databases = self.get_mysql_databases()
if not databases:
logging.warning("No databases found for backup")
return True

logging.info(f"Starting MySQL backup for {len(databases)} databases")

with concurrent.futures.ThreadPoolExecutor(max_workers=CONFIG['parallel_workers']) as executor:
futures = {executor.submit(self.backup_single_database, db): db for db in databases}

for future in concurrent.futures.as_completed(futures):
db_name = futures[future]
try:
future.result()
except Exception as e:
logging.error(f"Database backup error: {db_name} - {str(e)}")

return self.db_failure_count == 0

except Exception as e:
logging.error(f"MySQL backup failed: {str(e)}")
return False

def get_mysql_databases(self):
cmd = [
'mysql',
f"--user={CONFIG['mysql']['user']}",
f"--password={CONFIG['mysql']['password']}",
f"--host={CONFIG['mysql']['host']}",
f"--port={CONFIG['mysql']['port']}",
"-NBe", 
"SHOW DATABASES;"
]

try:
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
check=True,
env=self.env
)
databases = result.stdout.splitlines()
return [db for db in databases if db not in CONFIG['mysql']['excluded_databases']]
except subprocess.CalledProcessError as e:
logging.error(f"Failed to get database list: {e.stderr}")
raise Exception("MySQL connection failed")

def backup_single_database(self, db_name):
dump_dir = Path(CONFIG['mysql']['dump_dir'])
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
dump_file = dump_dir / f"{db_name}_{timestamp}.sql.gz"

try:
cmd = [
'mysqldump',
f"--user={CONFIG['mysql']['user']}",
f"--password={CONFIG['mysql']['password']}",
f"--host={CONFIG['mysql']['host']}",
f"--port={CONFIG['mysql']['port']}",
*CONFIG['mysql']['options'],
db_name,
'| gzip'
]

with open(dump_file, 'wb') as f:
process = subprocess.run(
' '.join(cmd),
shell=True,
stdout=f,
stderr=subprocess.PIPE,
env=self.env
)

if process.returncode != 0:
raise Exception(f"mysqldump error: {process.stderr.decode()}")

backup_cmd = [
CONFIG['restic_bin'],
'backup',
str(dump_file),
'--repo', CONFIG['repository'],
'--tag', f"mysql,database:{db_name},date:{self.current_date_tag}",
'--option', f"rclone.program={CONFIG['rclone_bin']}",
'--option', f"rclone.args=serve restic --stdio --config {CONFIG['rclone_config']}",
'--compression', CONFIG['compression']
]

if self.run_cmd(backup_cmd):
logging.info(f"Backup successful for database: {db_name}")
dump_file.unlink()
self.db_success_count += 1
return True

except Exception as e:
logging.error(f"Failed to backup database {db_name}: {str(e)}")
if dump_file.exists():
dump_file.unlink()
self.db_failure_count += 1
return False

def backup_user(self, user):
home_dir = f"/home/{user}"
cmd = [
CONFIG['restic_bin'],
'backup',
home_dir,
'--repo', CONFIG['repository'],
'--exclude-file', self.generate_exclude_file(user),
'--tag', f"user:{user},type:full,date:{self.current_date_tag}",
'--option', f"rclone.program={CONFIG['rclone_bin']}",
'--option', f"rclone.args=serve restic --stdio --config {CONFIG['rclone_config']}",
'--compression', CONFIG['compression'],
'--limit-upload', str(CONFIG['upload_limit'])
]
return self.run_cmd(cmd)

def generate_exclude_file(self, user):
exclude_path = f"/tmp/exclude_{user}.txt"
with open(exclude_path, 'w') as f:
for pattern in CONFIG['exclude_patterns']:
f.write(pattern.replace('*', user) + '\n')
return exclude_path

def run_cmd(self, cmd, context="Command"):
logging.info(f"Executing: {' '.join(cmd)}")
try:
result = subprocess.run(
cmd,
env=self.env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
check=True
)
logging.info(f"{context} succeeded: {result.stdout[:200]}...")
return True
except subprocess.CalledProcessError as e:
error_msg = f"""
{context} failed: {' '.join(e.cmd)}
Exit code: {e.returncode}
Error output: {e.stderr[:1000]}
"""
logging.error(error_msg)
return False

def process_users(self):
users_dir = '/usr/local/directadmin/data/users'
users = [u for u in os.listdir(users_dir) if os.path.isdir(os.path.join(users_dir, u))]

with concurrent.futures.ThreadPoolExecutor(max_workers=CONFIG['parallel_workers']) as executor:
futures = {executor.submit(self.backup_user, user): user for user in users}

for future in concurrent.futures.as_completed(futures):
user = futures[future]
try:
if future.result():
logging.info(f"User {user} backup completed")
else:
logging.warning(f"User {user} backup had errors")
except Exception as e:
logging.error(f"User {user} backup failed: {str(e)}")

def apply_retention_policy(self):
try:
cmd = [
CONFIG['restic_bin'],
'forget',
'--repo', CONFIG['repository'],
'--keep-last', str(CONFIG['retention_policy']['keep_last']),
'--keep-daily', str(CONFIG['retention_policy']['keep_daily']),
'--keep-weekly', str(CONFIG['retention_policy']['keep_weekly']),
'--keep-monthly', str(CONFIG['retention_policy']['keep_monthly']),
'--option', f"rclone.program={CONFIG['rclone_bin']}",
'--option', f"rclone.args=serve restic --stdio --config {CONFIG['rclone_config']}"
]

if CONFIG['retention_policy']['enable_prune']:
cmd.append('--prune')

return self.run_cmd(cmd, "Retention Policy")
except Exception as e:
logging.error(f"Retention policy failed: {str(e)}")
return False

def send_telegram_report(self, success=True, error_msg=None):
if not CONFIG['telegram']['enabled']:
return

try:
report = self.generate_report(success, error_msg)
bot_token = CONFIG['telegram']['bot_token']
chat_id = CONFIG['telegram']['chat_id']

url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
payload = {
'chat_id': chat_id,
'text': report,
'parse_mode': 'HTML'
}

response = requests.post(
url,
data=payload,
timeout=CONFIG['telegram']['api_timeout']
)

if response.status_code != 200:
logging.error(f"Telegram API error: {response.text}")

except Exception as e:
logging.error(f"Telegram notification failed: {str(e)}")

def generate_report(self, success, error_msg=None):
status_emoji = "✅" if success else "❌"
status_text = "THÀNH CÔNG" if success else "THẤT BẠI"
duration = datetime.now() - self.start_time

total_seconds = duration.total_seconds()
hours = int(total_seconds // 3600)
minutes = int((total_seconds % 3600) // 60)
seconds = int(total_seconds % 60)
duration_str = f"{hours}h {minutes}m {seconds}s"

report = f"""
<b>{status_emoji} BACKUP REPORT {status_emoji}</b>
━━━━━━━━━━━━━━━━━━━
<b>• Trạng thái:</b> {status_text}
<b>• Thời gian bắt đầu:</b> {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}
<b>• Thời lượng:</b> {duration_str}
<b>• Repository:</b> {CONFIG['repository']}
<b>• Tag ngày:</b> {self.current_date_tag}
<b>• Databases:</b> {self.db_success_count} thành công, {self.db_failure_count} thất bại
"""

if error_msg:
report += f"\n<b>• Lỗi chính:</b> <code>{error_msg[:500]}</code>"

return report

def main():
start_time = datetime.now()
error_message = None
success = False

try:
backup = EnhancedBackup()

# Backup MySQL
if not backup.mysql_backup():
error_message = "MySQL backup failed"
raise Exception(error_message)

# Backup users
backup.process_users()

# Áp dụng retention policy
if not backup.apply_retention_policy():
error_message = "Retention policy application failed"
raise Exception(error_message)

# Kiểm tra repository
check_cmd = [
CONFIG['restic_bin'],
'check',
'--repo', CONFIG['repository'],
'--option', f"rclone.program={CONFIG['rclone_bin']}",
'--option', f"rclone.args=serve restic --stdio --config {CONFIG['rclone_config']}"
]
backup.run_cmd(check_cmd, "Repository Check")

success = True

except Exception as e:
error_message = str(e)
logging.critical(f"Critical failure: {error_message}")

finally:
# Gửi thông báo
report = backup.generate_report(success, error_message)

# Gửi qua Telegram
backup.send_telegram_report(success, error_message)

# Dọn dẹp file tạm
for f in Path('/tmp').glob('exclude_*.txt'):
try:
f.unlink()
except Exception as e:
logging.error(f"Cleanup failed for {f}: {str(e)}")

return 0 if success else 1

if __name__ == "__main__":
exit(main())