Data Management：Cadangkan database lokal atau database cloud dari penyedia pihak ketiga

Ganti dengan parameter sumber data aktual sesuai kebutuhan.

boot_backup.sh: Mensimulasikan proses pencadangan penuh menggunakan alat xtrabackup.

#!/bin/bash
# Pasangan AccessKey dari akun Alibaba Cloud memiliki izin untuk semua operasi API. Menggunakan kredensial ini untuk melakukan operasi adalah operasi berisiko tinggi. Kami sarankan Anda membuat dan menggunakan pengguna RAM untuk akses API atau O&M rutin.
## Jangan simpan pasangan AccessKey di dalam kode. Ini dapat menyebabkan kebocoran kunci. Anda dapat menyimpan pasangan AccessKey dalam file konfigurasi berdasarkan kebutuhan bisnis Anda.

AK=<ALIBABA_CLOUD_ACCESS_KEY_ID>
SK=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>
DBS_API=dbs-api.<ID wilayah tempat sumber data Anda berada, misalnya, cn-hangzhou>.aliyuncs.com
DATASOURCE_ID=<ID sumber data Anda>

## Dapatkan jalur saat ini dari skrip.
BASE_PATH=$(cd `dirname $0`; pwd)

TS=`date +%Y%m%d_%H%M%S`
XTRA_DATA_DIR=~/tool/xtrabackup_data/$TS

mkdir -p $XTRA_DATA_DIR

## Jalankan perintah pencadangan xtrabackup untuk mengeluarkan log kesalahan ke file xtrabackup.log.
~/innobackupex --defaults-file=/etc/my.cnf --user='root' --password='root' --host='localhost' --port=3306 --socket='/var/lib/mysql/mysql.sock' --parallel=4 $XTRA_DATA_DIR/ 2>$XTRA_DATA_DIR/xtrabackup.log

## Dapatkan direktori pencadangan xtrabackup saat ini.
backup_dir=`ls $XTRA_DATA_DIR | grep -v xtrabackup.log | head -n1`
echo -e "\033[33mexecute innobackupex success, backup_dir: $backup_dir" && echo -n -e "\033[0m" && chmod 755 $XTRA_DATA_DIR/$backup_dir

## Kemas data dalam file tar.gz.
cd $XTRA_DATA_DIR/$backup_dir && tar -czvf ../$backup_dir.tar.gz . && file ../$backup_dir.tar.gz
echo -e "\033[33mpackage to $backup_dir.tar.gz" && echo -n -e "\033[0m" && sleep 2

## Gunakan skrip Python untuk mengunggah data pencadangan dan mendaftarkan metadata set pencadangan.
python3 $BASE_PATH/upload_and_register_backup_set.py --access_key_id $AK --access_key_secret $SK --endpoint $DBS_API --datasource_id $DATASOURCE_ID --region_code=cn-hangzhou --data_type=FullBackup --file_path=$XTRA_DATA_DIR/$backup_dir.tar.gz --xtrabackup_log_path=$XTRA_DATA_DIR/xtrabackup.log

• upload_and_register_backup_set.py: Mengunggah data pencadangan penuh dan log, mengurai metadata yang sesuai, serta mendaftarkan metadata set pencadangan.

  # -*- coding: utf-8 -*-
  # File ini dihasilkan secara otomatis, jangan edit. Terima kasih.
  import os
  import argparse
  import re
  import time
  import json
  from datetime import datetime
  
  import oss2
  from alibabacloud_openapi_util.client import Client as OpenApiUtilClient
  from alibabacloud_tea_openapi import models as open_api_models
  from alibabacloud_tea_openapi.client import Client as OpenApiClient
  from alibabacloud_tea_util import models as util_models
  
  import xtrabackup_info_parser
  import xtrabackup_log_parser
  
  def init_command_args():
      parser = argparse.ArgumentParser(description="A sample command-line parser.")
      parser.add_argument("--access_key_id", help="Aliyun AccessKeyId.")
      parser.add_argument("--access_key_secret", help="Aliyun AccessKeySecret.")
      parser.add_argument("--endpoint", help="Aliyun API Endpoint.")
      parser.add_argument("--region_code", help="Aliyun DataSource RegionCode.")
      parser.add_argument("--datasource_id", help="Aliyun DataSourceId.")
      parser.add_argument("--data_type", help="BackupSet DataType: FullBackup | LogBackup.")
      parser.add_argument("--file_path", help="BackupSet File Path.")
      parser.add_argument("--xtrabackup_info_path", help="Xtrabackup Info Path.")
      parser.add_argument("--xtrabackup_log_path", help="Xtrabackup Log Path.")
      parser.add_argument("--begin_time", help="Binlog Begin Time.")
      parser.add_argument("--end_time", help="Binlog End Time.")
  
      args = parser.parse_args()
      if args.access_key_id:
          print(f"access_key_id: ************")
      if args.access_key_secret:
          print(f"access_key_secret: ************")
      if args.endpoint:
          print(f"endpoint: {args.endpoint}")
      if args.region_code:
          print(f"region_code: {args.region_code}")
      if args.datasource_id:
          print(f"datasource_id: {args.datasource_id}")
      if args.data_type:
          print(f"data_type: {args.data_type}")
      if args.file_path:
          print(f"file_path: {args.file_path}")
      if args.xtrabackup_info_path:
          print(f"xtrabackup_info_path: {args.xtrabackup_info_path}")
      if args.xtrabackup_log_path:
          print(f"xtrabackup_log_path: {args.xtrabackup_log_path}")
      if args.begin_time:
          print(f"begin_time: {args.begin_time}")
      if args.end_time:
          print(f"end_time: {args.end_time}")
  
      print('\n')
      return args
  
  
  def date_to_unix_timestamp(date_str):
      dt_obj = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")
      # Gunakan metode .time() untuk mendapatkan tuple waktu, dan gunakan time.mktime untuk mengonversi tuple waktu menjadi timestamp hingga detik.
      timestamp_seconds = time.mktime(dt_obj.timetuple())
      return int(timestamp_seconds) * 1000
  
  
  def create_oss_client(params):
      # Informasi autentikasi Layanan Penyimpanan Objek Alibaba Cloud (OSS)
      access_key_id = params['AccessKeyId']
      access_key_secret = params['AccessKeySecret']
      security_token = params['SecurityToken']
      bucket_name = params['BucketName']
      endpoint = params['OssEndpoint']
  
      # Inisialisasi klien OSS.
      auth = oss2.StsAuth(access_key_id, access_key_secret, security_token)
      return oss2.Bucket(auth, endpoint, bucket_name)
  
  
  def upload_oss_file(oss_client, file_path, object_key):
      """
      Jalankan tugas unggah multipart untuk mengunggah objek besar ke OSS.
      :param oss_client:
      : param file_path: jalur file lokal.
      : param object_key: kunci objek di OSS, yaitu nama file.
      """
      # Atur ukuran bagian dalam byte. Nilai defaultnya adalah 1 MB.
      part_size = 1024 * 1024 * 5
      # Mulai tugas unggah multipart.
      upload_id = oss_client.init_multipart_upload(object_key).upload_id
  
      # Buka file dan baca kontennya.
      with open(file_path, 'rb') as file_obj:
          parts = []
          while True:
              data = file_obj.read(part_size)
              if not data:
                  break
              # Unggah bagian.
              result = oss_client.upload_part(object_key, upload_id, len(parts) + 1, data)
              parts.append(oss2.models.PartInfo(len(parts) + 1, result.etag))
  
          # Selesaikan unggah multipart.
          oss_client.complete_multipart_upload(object_key, upload_id, parts)
  
  
  class OssUploader:
  
      def __init__(self, access_key_id, access_key_secret, endpoint, region_code, datasource_id):
          self.access_key_id = access_key_id
          self.access_key_secret = access_key_secret
          self.endpoint = endpoint
          self.region_code = region_code
          self.datasource_id = datasource_id
  
          config = open_api_models.Config(access_key_id, access_key_secret)
          # Untuk informasi lebih lanjut tentang titik akhir, kunjungi https://api.aliyun.com/product/Rds.
          config.endpoint = endpoint
          self.client = OpenApiClient(config)
  
      """
      Daftarkan informasi set pencadangan.
      """
      def configure_backup_set_info(self, req_param):
          params = open_api_models.Params(
              # Operasi yang ingin Anda panggil.
              action='ConfigureBackupSetInfo',
              # Nomor versi operasi.
              version='2021-01-01',
              # Protokol operasi.
              protocol='HTTPS',
              # Metode HTTP operasi.
              method='POST',
              auth_type='AK',
              style='RPC',
              # URL operasi.
              pathname='/',
              # Format badan permintaan.
              req_body_type='json',
              # Format badan respons.
              body_type='json'
          )
          # opsi runtime
          runtime = util_models.RuntimeOptions()
          request = open_api_models.OpenApiRequest(
              query=OpenApiUtilClient.query(req_param)
          )
          # Respons bertipe MAP, yang berisi badan respons, header respons, dan kode status HTTP. 
          print(f"ConfigureBackupSetInfo request: {req_param}")
          data = self.client.call_api(params, request, runtime)
  
          print(f"ConfigureBackupSetInfo response: {data}")
          return data['body']['Data']
  
      """
      Dapatkan informasi unggah OSS.
      """
      def describe_bak_datasource_storage_access_info(self, req_param):
          params = open_api_models.Params(
              # Operasi yang ingin Anda panggil.
              action='DescribeBakDataSourceStorageAccessInfo',
              # Nomor versi operasi.
              version='2021-01-01',
              # Protokol operasi.
              protocol='HTTPS',
              # Metode HTTP operasi.
              method='POST',
              auth_type='AK',
              style='RPC',
              # URL operasi.
              pathname='/',
              # Format badan permintaan.
              req_body_type='json',
              # Format badan respons.
              body_type='json'
          )
          # opsi runtime
          runtime = util_models.RuntimeOptions()
          request = open_api_models.OpenApiRequest(
              query=OpenApiUtilClient.query(req_param)
          )
          # Respons bertipe MAP, yang berisi badan respons, header respons, dan kode status HTTP. 
          print(f"DescribeBakDataSourceStorageAccessInfo request: {req_param}")
          data = self.client.call_api(params, request, runtime)
  
          print(f"DescribeBakDataSourceStorageAccessInfo response: {data}")
          return data['body']['Data']
  
      def _fetch_oss_access_info(self, params):
          info = self.describe_bak_datasource_storage_access_info({
              'RegionId': params['RegionId'],
              'DataSourceId': params['DataSourceId'],
              'RegionCode': params['RegionCode'],
              'BackupType': params['BackupType'],
              'BackupSetId': params['BackupSetId']
          })
          return info['OssAccessInfo']
  
      def upload_and_register_backup_set(self, file_path, data_type, extra_meta):
          filename = os.path.basename(file_path)
          params = {'BackupMode': 'Automated', 'BackupMethod': 'Physical', 'RegionId': self.region_code,
                    'RegionCode': self.region_code, 'DataSourceId': self.datasource_id, 'BackupSetName': filename,
                    'ExtraMeta': extra_meta, 'BackupType': data_type, 'UploadStatus': 'WaitingUpload'}
  
          # Daftarkan set pencadangan untuk pertama kali. ID set pencadangan dikembalikan.
          data = self.configure_backup_set_info(params)
          params['BackupSetId'] = data['BackupSetId']
          print(f"------ configure_backup_set_info success: {file_path}, {data_type}, {params['BackupSetId']}, WaitingUpload\n")
  
          # Unggah data ke OSS.
          oss_info = self._fetch_oss_access_info(params)
          oss_client = create_oss_client(oss_info)
          upload_oss_file(oss_client, file_path, oss_info['ObjectKey'])
          print(f"------ upload_oss_file success: {file_path}, {data_type}, {params['BackupSetId']}\n")
  
          # Tandai bahwa set pencadangan telah diunggah.
          params['UploadStatus'] = 'UploadSuccess'
          self.configure_backup_set_info(params)
          print(f"------ configure_backup_set_info success: {file_path}, {data_type}, {params['BackupSetId']}, UploadSuccess\n")
  
  
  if __name__ == '__main__':
      args = init_command_args()
      uploader = OssUploader(args.access_key_id, args.access_key_secret,
                             args.endpoint, args.region_code, args.datasource_id)
  
      # Konstruksi extraMeta dengan metode berbeda untuk pencadangan penuh dan pencadangan log.
      extra_meta = '{}'
      if args.data_type == 'FullBackup':
          obj = {}
          if args.xtrabackup_log_path is not None:
              obj = xtrabackup_log_parser.analyze_slave_status(logpath=args.xtrabackup_log_path)
          elif args.xtrabackup_info_path is not None:
              parser = xtrabackup_info_parser.ExtraMetaParser(file_path=args.xtrabackup_info_path)
              obj = parser.get_extra_meta()
          extra_meta = {'BINLOG_FILE': obj.get('BINLOG_FILE'),
                        'version': obj.get("SERVER_VERSION"),
                        'dataBegin': date_to_unix_timestamp(obj.get("START_TIME")),
                        'dataEnd': date_to_unix_timestamp(obj.get("END_TIME")),
                        'consistentTime': int(date_to_unix_timestamp(obj.get("END_TIME")) / 1000)}
          extra_meta = json.dumps(extra_meta)
  
      elif args.data_type == 'LogBackup':
          obj = {'dataBegin': date_to_unix_timestamp(args.begin_time),
                 'dataEnd': date_to_unix_timestamp(args.end_time)}
          extra_meta = json.dumps(obj)
      print(f"get extra meta json string: {extra_meta}")
  
      # Unggah data dan daftarkan metadata set pencadangan.
      uploader.upload_and_register_backup_set(file_path=args.file_path, data_type=args.data_type, extra_meta=extra_meta)

• xtrabackup_info_parser.py: Mengurai metadata menggunakan file xtrabackup_info.

  # -*- coding: utf-8 -*-
  # File ini dihasilkan secara otomatis, jangan edit. Terima kasih.
  import re
  import json
  
  
  class ExtraMetaParser:
      def __init__(self, file_path):
          self.file_path = file_path
          pass
  
      def _parse_xtrabackup_info(self):
          config_data = {}
          with open(self.file_path, 'r') as file:
              for line in file:
                  line = line.strip()
                  if line and not line.startswith('#'):
                      key, value = line.split('=', 1)
                      config_data[key.strip()] = value.strip()
          return config_data
  
      def get_extra_meta(self):
          config_data = self._parse_xtrabackup_info()
          print(f"xtrabackup_info: {config_data}")
  
          binlog_pos = config_data.get("binlog_pos")
          pattern = f"filename '(.*)', position (.*)"
          match = re.search(pattern, binlog_pos)
  
          return {'BINLOG_FILE': match.group(1),
                  'SERVER_VERSION': config_data.get("server_version"),
                  'START_TIME': config_data.get("start_time"),
                  'END_TIME': config_data.get("end_time")}

• xtrabackup_log_parser.py: Mengurai metadata menggunakan file xtrabackup.log.

  #!/usr/bin/python
  # coding:utf8
  
  import io
  import re
  import sys
  from datetime import datetime
  
  from six import binary_type, text_type
  
  
  def parse_date_part(date_part, time_part):
      """
      Parse dan kembalikan string datetime lengkap berdasarkan bagian tanggal dan bagian waktu yang diberikan. 
      """
      # Dapatkan dua digit pertama tahun saat ini.
      current_century = datetime.now().strftime("%Y")[:2]
  
      year_short = date_part[:2]
      # Dapatkan nilai tahun lengkap.
      year_full = current_century + year_short
      date_full = year_full + date_part[2:]
      datetime_str = date_full + " " + time_part
  
      dt = datetime.strptime(datetime_str, "%Y%m%d %H:%M:%S")
      formatted_datetime = dt.strftime("%Y-%m-%d %H:%M:%S")
      return text_type(formatted_datetime)
  
  
  def analyze_slave_status(logpath=None):
      slave_status = {}
      start_time_pattern = (
          r"(\d{6}) (\d{2}:\d{2}:\d{2}) .*Connecting to MySQL server host:"
      )
      """
          240925 17:46:58 completed OK!
          240925 02:22:58 innobackupex: completed OK!
          240925 02:22:58 xtrabackup: completed OK!
      """
      end_time_pattern = r"(\d{6}) (\d{2}:\d{2}:\d{2}) .*completed OK!"
      with io.open(logpath, "rb") as fp:
          lines = fp.read().splitlines()
          for i in reversed(range(len(lines))):
              line = lines[i]
              if isinstance(line, binary_type):
                  line = line.decode("utf-8")
  
              m = re.search(start_time_pattern, line)
              if m:
                  # Ekstrak bagian tanggal dan bagian waktu.
                  date_part = m.group(1)
                  time_part = m.group(2)
                  slave_status["START_TIME"] = parse_date_part(date_part, time_part)
                  continue
  
              m = re.search(r"Using server version (\S*)", line)
              if m:
                  slave_status["SERVER_VERSION"] = text_type(m.group(1))
                  continue
  
              m = re.search("MySQL binlog position:", line)
              if m:
                  binlog_line = line
                  m = re.search(r"filename '(\S*)'", binlog_line)
                  if m:
                      slave_status["BINLOG_FILE"] = text_type(m.group(1))
                  m = re.search(r"position (\d+)", binlog_line)
                  m2 = re.search(r"position '(\d+)'", binlog_line)
                  if m:
                      try:
                          slave_status["BINLOG_POS"] = int(m.group(1))
                      except ValueError:
                          pass
                  elif m2:
                      try:
                          slave_status["BINLOG_POS"] = int(m2.group(1))
                      except ValueError:
                          pass
                  continue
  
              m = re.search("consistent_time (\d+)", line)
              if m:
                  try:
                      slave_status["CONSISTENT_TIME"] = int(m.group(1))
                  except ValueError:
                      pass
                  continue
  
              m = re.search(end_time_pattern, line)
              if m:
                  date_part = m.group(1)
                  time_part = m.group(2)
                  slave_status["END_TIME"] = parse_date_part(date_part, time_part)
                  continue
  
      return slave_status
  
  if __name__ == "__main__":
      logpath = sys.argv[1]
      slave_status = analyze_slave_status(logpath)
      print(slave_status)