AWS DMS タスク ID を使用して DMS タスクログをダウンロードするための Python コード

AWS Database Migration Service (AWS DMS) を使用すると、データベースを迅速かつ安全に AWS に移行できます。この記事では、AWS DMS タスク ID を使用して AWS DMS タスクログをローカルコンピュータにダウンロードするために必要なサンプル Python コードについて説明します。

概要

DMS タスクログには、移行プロセス中に記録されたタスク情報が含まれています。記事「AWS DMS 移行のデバッグ: 問題が発生した場合の対処方法 (パート1)」で示唆されているように、AWS DMS を使用してデータを移行する際、これらのログを使用して問題を解決することができます。

この記事では、DMS は Amazon CloudWatch Logs のログストリームにタスクログを送信するので、DMS 移行タスクに関する情報を取得できます。けれども、トラブルシューティングをスピードアップするためにローカルにログをダウンロードしたい場合はどうすればよいでしょうか?

Download DM Task ログの GitHub レポジトリで、DMS タスク ID を使用して AWS DMS タスクログをダウンロードするための完璧な Python コードを見つけてください。この記事の残りの部分では、Python コードを分解してそのソリューションの使用方法を次の順序で説明します。

必要なライブラリをインポートします。
DMS タスク ID の引数とログの時間範囲を読み取ります。
エポックからの経過時間をミリ秒に変換します。
AWS DMS タスク ID に基づいてレプリケーションタスクを取得します。
レプリケーションインスタンス ARN を取得します。
レプリケーションインスタンス情報を取得します。
レプリケーションインスタンス ID を取得します。
ロググループ名を作成します。
DMS タスクの CloudWatch ログイベントを取得します。
CloudWatch のログを印刷します。
ソリューションを使用します。

前提条件

このソリューションを使用するための前提条件は次のとおりです。

Python をダウンロードしてインストールします。
次の pip コマンドを使用して、boto3 および maya ライブラリをダウンロードします。
pip install boto3
pip install maya
AWS CLI をインストールします。
AWS CLI を設定します。
Python 用のコードエディタをダウンロードします。この記事では、Visual Studio Code を使用します。

必要なライブラリをインポートする

boto3、sys、および maya ライブラリをインポートすることから始めます。

import boto3

import sys

import maya

DMS タスク ID の読み取り引数とログの時間範囲

AWS DMS タスク ID の引数、およびログの開始時刻と終了時刻の範囲を読み取ります。

replication_task_id = sys.argv[1]

time_string = sys.argv[2]

end_time_string = sys.argv[3]

エポックからの経過時間をミリ秒に変換する

次の方法を使用して、開始時刻と終了時刻をエポックからのミリ秒に変換します。変換には maya ライブラリを使用します。

def start_time_milliseconds_since_epoch(time_string):
    datetime = maya.when(time_string)
    seconds = datetime.epoch
    return seconds * 1000

start_time = start_time_milliseconds_since_epoch(time_string)

def end_time_milliseconds_since_epoch(end_time_string):
    datetime = maya.when(end_time_string)
    seconds = datetime.epoch
    return seconds * 1000

end_time = end_time_milliseconds_since_epoch(end_time_string)

AWS DMS タスク ID に基づいてレプリケーションタスクを取得する

引数として指定した AWS DMS タスク ID に基づいてレプリケーションタスクを取得します。boto3 ライブラリの describe_replication_tasks メソッドを使用します。このメソッドは、現在のリージョンのアカウントのレプリケーションタスクに関する情報を返します。次に、メソッドによって返されたレプリケーションタスクを返します。

def get_replication_tasks():

    client = boto3.client('dms')

    response = client.describe_replication_tasks(Filters=[
        {
            'Name': 'replication-task-id',
            'Values': [
                replication_task_id,
            ]
        },
    ],
    MaxRecords=100,
    Marker='')

    return response['ReplicationTasks']

レプリケーションインスタンス ARN を取得する

前のメソッドを使用して取得したレプリケーションタスクからレプリケーションインスタンス ARN を取得します。レプリケーションインスタンス ARN を返し、将来使用するためにそれを変数 rep_instance_arn に割り当てます。

def get_replication_instance_arn():
    for ReplicationTasks in get_replication_tasks():
        ReplicationInstanceArn = ReplicationTasks['ReplicationInstanceArn']

        return ReplicationInstanceArn

rep_instance_arn = get_replication_instance_arn()

レプリケーションインスタンス情報を取得する

次に、以下のコード例に示すように、describe_replication_instances メソッドにレプリケーションインスタンス ARN を指定して、レプリケーションインスタンス情報を取得します。

def get_replication_instances():

    client = boto3.client('dms')
    
    response = client.describe_replication_instances(Filters=[
        {
            'Name': 'rep-instance-arn',
            'Values': [
                rep_instance_arn,
            ]
        },
    ],
    MaxRecords=100,
    Marker='')
    
    
    return response['ReplicationInstances']

レプリケーションインスタンス ID を取得する

前のメソッドで取得したレプリケーションインスタンスの結果から、AWS DMS タスクのロググループ名を構成するために使用するレプリケーションインスタンス識別子を取得します。

def get_replication_instance_id():
    for ReplicationInstances in get_replication_instances():
     ReplicationInstanceIdentifier = ReplicationInstances['ReplicationInstanceIdentifier']

     return ReplicationInstanceIdentifier

ロググループ名を作成する

レプリケーションインスタンス ID を取得したら、レプリケーションインスタンス ID の先頭に dms-tasks- を付けてロググループ名を構成します。

log_group = "dms-tasks-" + get_replication_instance_id()

DMS タスクの CloudWatch ログイベントを取得する

上記で作成したロググループを get_cloudwatch_log_events メソッドに渡します。このメソッドは、filter_log_events メソッドを呼び出して DMS タスクの CloudWatch ログイベントを取得します。このメソッドにより、指定されたロググループのログイベントが一覧表示されます。

def get_cloudwatch_log_events(log_group):
    
    client = boto3.client('logs')
    kwargs = {
        'logGroupName': log_group,
        'limit': 10000,
        'startTime': start_time,
        'endTime': end_time
    }
    while True:
        response = client.filter_log_events(**kwargs)
        yield from response['events']
        try:
            kwargs['nextToken'] = response['nextToken']
        except KeyError:
            break

CloudWatch ログを印刷する

最後に、メインメソッドで、DMS タスクの CloudWatch ログイベントを印刷します。

def main():
    for event in get_cloudwatch_log_events(log_group):
        sys.stdout.write(event['message'].rstrip() + '\n')

if __name__ == '__main__':
    main()

ソリューションを使用する

このソリューションを使用するには、aws-database-migration-tools GitHub リポジトリからコードファイルをコンピュータ上のフォルダにダウンロードします。Python コードを実行するには、端末またはコマンドプロンプトから次のコマンドを実行します。

python GetCWLogData.py <Your DMS Task ID> <Start Time Filter> <End Time Filter> > dmslogs.log

コマンド内の <Your DMS task ID> を DMS タスク ID に置き換えます。タスク ID は describe-replication-tasks AWS CLI コマンドを使用して見つけることができます。

次のスクリーンショットに示すように、AWS マネジメントコンソールから DMS タスク ID を取得することもできます。

<Start Time Filter> を、ログエントリに含める開始日時に置き換えます。ログファイルに含める必要があるログエントリのカットオフ日時で <End Time Filter> を置き換えます。このコマンドを実行すると、ログエントリが dmslogs.log というファイルにダウンロードされます。

次のコマンド例には、1 月 21 日から 1 月 22 日までのログエントリが含まれています。

python GetCWLogData.py [DMS task ID] 2019-01-21T00:00 2019-01-22T00:00 > dmslogs.log

Python コードを実行したコマンドを実行したフォルダに、dmslogs.log という名前のファイルがあります。ファイルを開くと、次のスクリーンショットのようなログエントリがあります。

Linux および UNIX ベースのシステムでは、tail コマンドを使用して dmslogs.log ファイルの終了を確認することもできます。次のスクリーンショットでは、dmslogs.log ファイルの最後の 50 行が表示されています。

結論

AWS DMS を使用して異なるデータストア間でデータを移行している間に問題が発生する可能性があります。これは、クラウド内、オンプレミスインスタンス間 (AWS Cloud の設定により) か、またはクラウドとオンプレミス間を問わず発生する可能性があります。この記事で提供されている Python コードを使用すると、AWS マネジメントコンソールにログオンしなくても DMS タスクログをダウンロードして問題を解決できます。

この記事で概説したソリューションの実装についてのコメントまたは質問がございましたら、以下の「コメント」欄に送信してください。

著者について

Zafar Kapadia はアマゾンウェブサービスのシニアソリューションアーキテクトです。 彼はアプリケーション開発と最適化を専門としています。また、熱心なクリケット選手であり、さまざまなローカルリーグでプレーしています。

David Rader はアマゾンウェブサービスのシニアプラクティスマネージャーです。彼はアプリケーションモダナイゼーションを専門としています。ジョギング、キャンプ、それにマシュマロを炙って食べることが大好きです。

Amazon Web Services ブログ