Data Management:Cadangan database on-premises dan cloud pihak ketiga

Ini adalah titik masuk untuk eksekusi skrip. Ganti parameter placeholder dengan nilai aktual Anda sesuai kebutuhan.

boot_backup.sh: Skrip ini mensimulasikan pembuatan cadangan penuh lokal dengan menggunakan Xtrabackup.

#!/bin/bash
## Pasangan AccessKey Akun Alibaba Cloud memiliki izin atas semua operasi API. Menggunakan kredensial ini untuk melakukan operasi merupakan praktik berisiko tinggi. Kami menyarankan Anda membuat dan menggunakan Pengguna RAM untuk akses API atau O&M rutin.
## Untuk alasan keamanan, jangan hardcode ID AccessKey dan Rahasia AccessKey Anda dalam kode. Hal ini dapat menyebabkan kebocoran kredensial. Sebagai gantinya, simpan dalam file konfigurasi atau gunakan variabel lingkungan.
AK=<YOUR_ACCESS_KEY_ID>
SK=<YOUR_ACCESS_KEY_SECRET>
DBS_API=dbs-api.<your-region-id>.aliyuncs.com
DATASOURCE_ID=<your-data-source-id>
## Dapatkan path 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 Xtrabackup dan arahkan log error ke file xtrabackup.log.
~/innobackupex --defaults-file=/etc/my.cnf --user='<your-db-account>' --password='<your-db-password>' --host='<your-db-host>' --port=<your-db-port> --socket='/var/lib/mysql/mysql.sock' --parallel=4 $XTRA_DATA_DIR/ 2>$XTRA_DATA_DIR/xtrabackup.log
## Dapatkan direktori cadangan 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 direktori menjadi 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
## Panggil skrip Python untuk mengunggah file cadangan dan mendaftarkan metadata-nya.
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: Skrip ini mengunggah data cadangan penuh dan cadangan log, mengurai metadata yang sesuai, dan mendaftarkan metadata set cadangan ke Layanan Cadangan Basis Data (DBS).

  # -*- coding: utf-8 -*-
  # File ini di-generate secara otomatis, jangan diedit. 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="Parser command-line contoh.")
      parser.add_argument("--access_key_id", help="ID AccessKey Alibaba Cloud Anda.")
      parser.add_argument("--access_key_secret", help="Rahasia AccessKey Alibaba Cloud Anda.")
      parser.add_argument("--endpoint", help="Titik akhir API untuk layanan Anda.")
      parser.add_argument("--region_code", help="Kode wilayah sumber data Anda.")
      parser.add_argument("--datasource_id", help="ID sumber data Anda.")
      parser.add_argument("--data_type", help="Jenis set cadangan: FullBackup atau LogBackup.")
      parser.add_argument("--file_path", help="Path file set cadangan.")
      parser.add_argument("--xtrabackup_info_path", help="Path ke file xtrabackup_info.")
      parser.add_argument("--xtrabackup_log_path", help="Path ke file xtrabackup.log.")
      parser.add_argument("--begin_time", help="Waktu mulai binlog.")
      parser.add_argument("--end_time", help="Waktu akhir binlog.")
      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):
      # Mengonversi string tanggal ke timestamp Unix dalam milidetik.
      dt_obj = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")
      timestamp_seconds = time.mktime(dt_obj.timetuple())
      return int(timestamp_seconds) * 1000
  def create_oss_client(params):
      # Detail autentikasi untuk OSS Alibaba Cloud.
      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):
      """
      Mengunggah file besar ke OSS menggunakan unggah multi-bagian.
      :param oss_client: Instance klien OSS.
      :param file_path: Path ke file lokal.
      :param object_key: Kunci objek di OSS, yang berfungsi sebagai nama file.
      """
      # Atur ukuran bagian dalam byte. Default adalah 5 MB.
      part_size = 1024 * 1024 * 5
      # Mulai unggah multi-bagian.
      upload_id = oss_client.init_multipart_upload(object_key).upload_id
      # Buka file dan baca isinya.
      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 multi-bagian.
          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, lihat https://api.alibabacloud.com/product/Dbs.
          config.endpoint = endpoint
          self.client = OpenApiClient(config)
      """
      Mendaftarkan metadata set cadangan.
      """
      def configure_backup_set_info(self, req_param):
          params = open_api_models.Params(
              # Nama operasi API.
              action='ConfigureBackupSetInfo',
              # Versi operasi API.
              version='2021-01-01',
              # Protokol.
              protocol='HTTPS',
              # Metode HTTP.
              method='POST',
              auth_type='AK',
              style='RPC',
              # Path operasi API.
              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 adalah map yang berisi badan respons, header, dan kode status HTTP.
          print(f"Permintaan ConfigureBackupSetInfo: {req_param}")
          data = self.client.call_api(params, request, runtime)
          print(f"Respons ConfigureBackupSetInfo: {data}")
          return data['body']['Data']
      """
      Mendapatkan informasi unggah OSS.
      """
      def describe_bak_datasource_storage_access_info(self, req_param):
          params = open_api_models.Params(
              # Nama operasi API.
              action='DescribeBakDataSourceStorageAccessInfo',
              # Versi operasi API.
              version='2021-01-01',
              # Protokol.
              protocol='HTTPS',
              # Metode HTTP.
              method='POST',
              auth_type='AK',
              style='RPC',
              # Path operasi API.
              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 adalah map yang berisi badan respons, header, dan kode status HTTP.
          print(f"Permintaan DescribeBakDataSourceStorageAccessInfo: {req_param}")
          data = self.client.call_api(params, request, runtime)
          print(f"Respons DescribeBakDataSourceStorageAccessInfo: {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 cadangan untuk pertama kali guna memperoleh ID set cadangan.
          data = self.configure_backup_set_info(params)
          params['BackupSetId'] = data['BackupSetId']
          print(f"------ configure_backup_set_info berhasil: {file_path}, {data_type}, {params['BackupSetId']}, WaitingUpload\n")
          # Unggah file 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 berhasil: {file_path}, {data_type}, {params['BackupSetId']}\n")
          # Tandai set cadangan sebagai telah diunggah.
          params['UploadStatus'] = 'UploadSuccess'
          self.configure_backup_set_info(params)
          print(f"------ configure_backup_set_info berhasil: {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)
      # Bangun extra_meta secara berbeda untuk cadangan penuh dan cadangan 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"Dapatkan string JSON extra meta: {extra_meta}")
      # Unggah data dan daftarkan metadata set cadangan.
      uploader.upload_and_register_backup_set(file_path=args.file_path, data_type=args.data_type, extra_meta=extra_meta)

• xtrabackup_info_parser.py: Skrip ini mengurai metadata dari file xtrabackup_info.

  # -*- coding: utf-8 -*-
  # File ini di-generate secara otomatis, jangan diedit. 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: Skrip ini mengurai metadata dari file xtrabackup.log yang dihasilkan selama proses pencadangan.

  #!/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):
      """
      Mengurai dan mengembalikan string datetime lengkap dari bagian tanggal dan waktu yang diberikan.
      """
      # Dapatkan dua digit pertama tahun saat ini (abad).
      current_century = datetime.now().strftime("%Y")[:2]
      year_short = date_part[:2]
      # Bangun tahun penuh.
      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 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)