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

Skrip ini merupakan titik masuk.

boot_backup.sh: Mengotomatiskan proses pembuatan cadangan penuh Xtrabackup lokal.

#!/bin/bash
## Pasangan AccessKey Akun Alibaba Cloud memberikan akses penuh ke semua API dan menimbulkan risiko keamanan tinggi. Kami sangat menyarankan agar Anda membuat dan menggunakan Pengguna RAM untuk panggilan API dan operasi harian.
## Untuk mencegah risiko keamanan, jangan hard-code ID AccessKey dan AccessKey Secret Anda dalam kode. Kami menyarankan agar Anda menyimpannya dalam file konfigurasi sesuai kebutuhan bisnis Anda.

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 skrip saat ini.
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 innobackupex untuk membuat cadangan dan arahkan log error 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 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 ke 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

## Panggil skrip Python untuk mengunggah data cadangan dan mendaftarkan metadata set cadangan.
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 cadangan penuh dan cadangan log, mengurai metadata yang sesuai, dan mendaftarkan metadata set cadangan.

  # -*- 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="AccessKey Secret Alibaba Cloud Anda.")
      parser.add_argument("--endpoint", help="Titik akhir API Alibaba Cloud.")
      parser.add_argument("--region_code", help="ID wilayah sumber data.")
      parser.add_argument("--datasource_id", help="ID sumber data.")
      parser.add_argument("--data_type", help="Jenis set cadangan: FullBackup atau LogBackup.")
      parser.add_argument("--file_path", help="Path ke 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):
      dt_obj = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")
      # Gunakan `time.mktime` untuk mengonversi tuple waktu dari `dt_obj.timetuple()` ke stempel waktu UNIX dalam detik.
      timestamp_seconds = time.mktime(dt_obj.timetuple())
      return int(timestamp_seconds) * 1000
  
  
  def create_oss_client(params):
      # Informasi autentikasi 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, yaitu nama file.
      """
      # Atur ukuran bagian dalam byte. Default adalah 5 MB.
      part_size = 1024 * 1024 * 5
      # Inisialisasi 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 satu 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.aliyun.com/product/Rds.
          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 API.
              version='2021-01-01',
              # Protokol permintaan API.
              protocol='HTTPS',
              # Metode HTTP permintaan API.
              method='POST',
              auth_type='AK',
              style='RPC',
              # Path permintaan 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)
          )
          # Mengembalikan map yang berisi badan respons, header respons, dan kode status.
          print(f"ConfigureBackupSetInfo request: {req_param}")
          data = self.client.call_api(params, request, runtime)
  
          print(f"ConfigureBackupSetInfo response: {data}")
          return data['body']['Data']
  
      """
      Mengambil informasi unggah OSS.
      """
      def describe_bak_datasource_storage_access_info(self, req_param):
          params = open_api_models.Params(
              # Nama operasi API.
              action='DescribeBakDataSourceStorageAccessInfo',
              # Versi API.
              version='2021-01-01',
              # Protokol permintaan API.
              protocol='HTTPS',
              # Metode HTTP permintaan API.
              method='POST',
              auth_type='AK',
              style='RPC',
              # Path permintaan 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)
          )
          # Mengembalikan map yang berisi badan respons, header respons, dan kode status.
          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 cadangan untuk pertama kali. ID set cadangan 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 set cadangan sebagai berhasil 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)
  
      # Bangun extraMeta 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"get extra meta json string: {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: 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: Mengurai metadata dari file xtrabackup.log yang dihasilkan selama 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 bagian tanggal dan waktu yang diberikan serta mengembalikan string tanggal dan waktu lengkap.
      """
      # Dapatkan dua digit pertama tahun saat ini.
      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)