Data Management:オンプレミスおよびサードパーティクラウドのデータベースバックアップ

このスクリプトはエントリポイントです。

boot_backup.sh: ローカルの Xtrabackup 完全バックアップを作成するプロセスを自動化します。

#!/bin/bash
## Alibaba Cloud アカウントの AccessKey ペアは、すべての API へのフルアクセスを許可するため、高いセキュリティリスクをもたらします。API 呼び出しや日常の操作には、RAM ユーザーを作成して使用することを強く推奨します。
## セキュリティリスクを防ぐため、コードに AccessKey ID と AccessKey Secret をハードコーディングしないでください。ビジネス要件に基づいて、これらを構成ファイルに保存することをお勧めします。

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>

## スクリプトの現在のパスを取得します。
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

## innobackupex コマンドを実行してバックアップを作成し、エラーログを 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

## 現在の Xtrabackup バックアップのディレクトリを取得します。
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

## ディレクトリを .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

## Python スクリプトを呼び出してバックアップデータをアップロードし、バックアップセットのメタデータを登録します。
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: 完全バックアップとログバックアップデータをアップロードし、対応するメタデータを解析して、バックアップセットのメタデータを登録します。

  # -*- coding: utf-8 -*-
  # このファイルは自動生成されたものです。編集しないでください。
  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="Your Alibaba Cloud AccessKey ID.")
      parser.add_argument("--access_key_secret", help="Your Alibaba Cloud AccessKey Secret.")
      parser.add_argument("--endpoint", help="The API endpoint for Alibaba Cloud.")
      parser.add_argument("--region_code", help="The region ID of the data source.")
      parser.add_argument("--datasource_id", help="The ID of the data source.")
      parser.add_argument("--data_type", help="The type of the backup set: FullBackup or LogBackup.")
      parser.add_argument("--file_path", help="The path to the backup set file.")
      parser.add_argument("--xtrabackup_info_path", help="The path to the xtrabackup_info file.")
      parser.add_argument("--xtrabackup_log_path", help="The path to the xtrabackup.log file.")
      parser.add_argument("--begin_time", help="The start time of the binlog.")
      parser.add_argument("--end_time", help="The end time of the 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")
      # `time.mktime` を使用して、`dt_obj.timetuple()` からの time タプルを秒単位の UNIX タイムスタンプに変換します。
      timestamp_seconds = time.mktime(dt_obj.timetuple())
      return int(timestamp_seconds) * 1000
  
  
  def create_oss_client(params):
      # Alibaba Cloud OSS 認証情報。
      access_key_id = params['AccessKeyId']
      access_key_secret = params['AccessKeySecret']
      security_token = params['SecurityToken']
      bucket_name = params['BucketName']
      endpoint = params['OssEndpoint']
  
      # 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):
      """
      マルチパートアップロードを使用して大きなファイルを OSS にアップロードします。
      :param oss_client: OSS クライアントインスタンス。
      :param file_path: ローカルファイルのパス。
      :param object_key: OSS のオブジェクトキー (ファイル名)。
      """
      # パートサイズをバイト単位で設定します。デフォルトは 5 MB です。
      part_size = 1024 * 1024 * 5
      # マルチパートアップロードを初期化します。
      upload_id = oss_client.init_multipart_upload(object_key).upload_id
  
      # ファイルを開き、その内容を読み取ります。
      with open(file_path, 'rb') as file_obj:
          parts = []
          while True:
              data = file_obj.read(part_size)
              if not data:
                  break
              # パートをアップロードします。
              result = oss_client.upload_part(object_key, upload_id, len(parts) + 1, data)
              parts.append(oss2.models.PartInfo(len(parts) + 1, result.etag))
  
          # マルチパートアップロードを完了します。
          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)
          # エンドポイントの詳細については、https://api.aliyun.com/product/Rds をご参照ください。
          config.endpoint = endpoint
          self.client = OpenApiClient(config)
  
      """
      バックアップセットのメタデータを登録します。
      """
      def configure_backup_set_info(self, req_param):
          params = open_api_models.Params(
              # API オペレーションの名前。
              action='ConfigureBackupSetInfo',
              # API のバージョン。
              version='2021-01-01',
              # API リクエストのプロトコル。
              protocol='HTTPS',
              # API リクエストの HTTP メソッド。
              method='POST',
              auth_type='AK',
              style='RPC',
              # API リクエストのパス。
              pathname='/',
              # リクエストボディのフォーマット。
              req_body_type='json',
              # レスポンスボディのフォーマット。
              body_type='json'
          )
          # ランタイムオプション。
          runtime = util_models.RuntimeOptions()
          request = open_api_models.OpenApiRequest(
              query=OpenApiUtilClient.query(req_param)
          )
          # レスポンスボディ、レスポンスヘッダー、およびステータスコードを含むマップを返します。
          print(f"ConfigureBackupSetInfo request: {req_param}")
          data = self.client.call_api(params, request, runtime)
  
          print(f"ConfigureBackupSetInfo response: {data}")
          return data['body']['Data']
  
      """
      OSS アップロード情報を取得します。
      """
      def describe_bak_datasource_storage_access_info(self, req_param):
          params = open_api_models.Params(
              # API オペレーションの名前。
              action='DescribeBakDataSourceStorageAccessInfo',
              # API のバージョン。
              version='2021-01-01',
              # API リクエストのプロトコル。
              protocol='HTTPS',
              # API リクエストの HTTP メソッド。
              method='POST',
              auth_type='AK',
              style='RPC',
              # API リクエストのパス。
              pathname='/',
              # リクエストボディのフォーマット。
              req_body_type='json',
              # レスポンスボディのフォーマット。
              body_type='json'
          )
          # ランタイムオプション。
          runtime = util_models.RuntimeOptions()
          request = open_api_models.OpenApiRequest(
              query=OpenApiUtilClient.query(req_param)
          )
          # レスポンスボディ、レスポンスヘッダー、およびステータスコードを含むマップを返します。
          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'}
  
          # バックアップセットを初めて登録します。バックアップセット ID が返されます。
          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")
  
          # データを 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")
  
          # バックアップセットを正常にアップロード済みとしてマークします。
          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)
  
      # 完全バックアップとログバックアップで extraMeta を異なる方法で構築します。
      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}")
  
      # データをアップロードし、バックアップセットのメタデータを登録します。
      uploader.upload_and_register_backup_set(file_path=args.file_path, data_type=args.data_type, extra_meta=extra_meta)

• xtrabackup_info_parser.py: xtrabackup_info ファイルからメタデータを解析します。

  # -*- coding: utf-8 -*-
  # このファイルは自動生成されたものです。編集しないでください。
  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: バックアップ中に生成された 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):
      """
      指定された日付と時刻の部分を解析し、完全な日付と時刻の文字列を返します。
      """
      # 現在の年の最初の 2 桁を取得します。
      current_century = datetime.now().strftime("%Y")[:2]
  
      year_short = date_part[:2]
      # 完全な年を構築します。
      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:
                  # 日付と時刻の部分を抽出します。
                  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)