Data Management:Back up an on-premises database or a cloud database from a third-party provider

Replace with the actual data source parameters as needed.

boot_backup.sh: simulates the process of a full backup by using the xtrabackup tool.

#!/bin/bash
# The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. Using these credentials to perform operations is a high-risk operation. We recommend that you create and use a RAM user for API access or routine O&M.
## Do not save the AccessKey pair in the code. This may lead to key leakage. You can save the Access pair in a configuration file based on your business requirements.

AK=<ALIBABA_CLOUD_ACCESS_KEY_ID>
SK=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>
DBS_API=dbs-api.<ID of the region where your data source resides, for example, cn-hangzhou>.aliyuncs.com
DATASOURCE_ID=<Your data source ID>

## Obtain the current path of the script.
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

## Run the xtrabackup backup command to output the error log to the xtrabackup.log file.
~/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

## Obtain the directory of the current xtrabackup backup.
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

## Package the data in a tar.gz file.
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

## Use the Python script to upload the backup data and register the backup set metadata.
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: uploads data of the full backup and log backup, parse the corresponding metadata, and register the backup set metadata.

  # -*- coding: utf-8 -*-
  # This file is auto-generated, don't edit it. Thanks.
  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")
      # Use the .time() method to obtain the time tuple, and use time.mktime to convert the time tuple to a timestamp accurate to second.
      timestamp_seconds = time.mktime(dt_obj.timetuple())
      return int(timestamp_seconds) * 1000
  
  
  def create_oss_client(params):
      # Alibaba Cloud Object Storage Service (OSS) authentication information
      access_key_id = params['AccessKeyId']
      access_key_secret = params['AccessKeySecret']
      security_token = params['SecurityToken']
      bucket_name = params['BucketName']
      endpoint = params['OssEndpoint']
  
      # Initialize the OSS client.
      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):
      """
      Run a multipart upload task to upload a large object to OSS.
      :param oss_client:
      : param file_path: the local file path.
      : param object_key: the object key in OSS, which is a file name.
      """
      # Set the part size in bytes. The default value is 1 MB.
      part_size = 1024 * 1024 * 5
      # Initiate the multipart upload task.
      upload_id = oss_client.init_multipart_upload(object_key).upload_id
  
      # Open the file and read the content.
      with open(file_path, 'rb') as file_obj:
          parts = []
          while True:
              data = file_obj.read(part_size)
              if not data:
                  break
              # Upload parts.
              result = oss_client.upload_part(object_key, upload_id, len(parts) + 1, data)
              parts.append(oss2.models.PartInfo(len(parts) + 1, result.etag))
  
          # Complete multipart upload.
          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)
          # For more information about endpoints, visit https://api.aliyun.com/product/Rds.
          config.endpoint = endpoint
          self.client = OpenApiClient(config)
  
      """
      Register backup set metadata.
      """
      def configure_backup_set_info(self, req_param):
          params = open_api_models.Params(
              # The operation that you want to call.
              action='ConfigureBackupSetInfo',
              # The version number of the operation.
              version='2021-01-01',
              # The protocol of the operation.
              protocol='HTTPS',
              # The HTTP method of the operation.
              method='POST',
              auth_type='AK',
              style='RPC',
              # The URL of the operation.
              pathname='/',
              # The format of the request body.
              req_body_type='json',
              # The format of the response body.
              body_type='json'
          )
          # runtime options
          runtime = util_models.RuntimeOptions()
          request = open_api_models.OpenApiRequest(
              query=OpenApiUtilClient.query(req_param)
          )
          # The response is of the MAP type, which contains the response body, response headers, and HTTP status code. 
          print(f"ConfigureBackupSetInfo request: {req_param}")
          data = self.client.call_api(params, request, runtime)
  
          print(f"ConfigureBackupSetInfo response: {data}")
          return data['body']['Data']
  
      """
      Obtain the OSS upload information.
      """
      def describe_bak_datasource_storage_access_info(self, req_param):
          params = open_api_models.Params(
              # The operation that you want to call.
              action='DescribeBakDataSourceStorageAccessInfo',
              # The version number of the operation.
              version='2021-01-01',
              # The protocol of the operation.
              protocol='HTTPS',
              # The HTTP method of the operation.
              method='POST',
              auth_type='AK',
              style='RPC',
              # The URL of the operation.
              pathname='/',
              # The format of the request body.
              req_body_type='json',
              # The format of the response body.
              body_type='json'
          )
          # runtime options
          runtime = util_models.RuntimeOptions()
          request = open_api_models.OpenApiRequest(
              query=OpenApiUtilClient.query(req_param)
          )
          # The response is of the MAP type, which contains the response body, response headers, and HTTP status code. 
          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'}
  
          # Register a backup set for the first time. The ID of the backup set is returned.
          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")
  
          # Upload data to 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")
  
          # Mark that the backup set is uploaded.
          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)
  
      # Construct extraMeta in different methods for full backup and log backup.
      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}")
  
      # Upload data and register backup set metadata.
      uploader.upload_and_register_backup_set(file_path=args.file_path, data_type=args.data_type, extra_meta=extra_meta)

• xtrabackup_info_parser.py: parses metadata by using the xtrabackup_info file.

  # -*- coding: utf-8 -*-
  # This file is auto-generated, don't edit it. Thanks.
  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: parses metadata by using the xtrabackup.log file.

  #!/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 and return the complete datetime string based on the given date part and time part. 
      """
      # Obtain the first two digits of the current year.
      current_century = datetime.now().strftime("%Y")[:2]
  
      year_short = date_part[:2]
      # Obtain the complete year value.
      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:
                  # Extract the date part and time part.
                  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)