概要

https://qiita.com/okadarhi/items/5ff734c6472b83d51597 の続き。
AWS側で転送元のデータを作成するが、作成処理が完全に完了してから転送ジョブを実行して欲しいという要望があり、スケジュール実行ではなく、AWS側から手動で転送ジョブを実行する方法を検討した。
転送ジョブを実行するAPIがあるのだが、それは実行しかしてくれず、転送ジョブの成否は別のAPIで確認する必要があった。
実行するAPI: https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs/run
確認するAPI: https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferOperations/get
なので、転送ジョブの実行と確認を管理するジョブを作成しようと思い、Step Functionsを採用することにした。
その時行った検証作業を備忘録としてまとめる。

手順

Lambda関数の作成

レイヤーの作成

Google APIを実行するためのpythonクライアントライブラリーをLambdaで実行するためのレイヤーを作成する。
まずは、Cloud9上でLambdaレイヤーに使用するライブラリーファイルを作成する。

admin:~/environment $ mkdir python
admin:~/environment $ pip install -t ./python google-api-python-client boto3
admin:~/environment $ zip -r google-api-python-client.zip python

次に、作成したライブラリーファイルをLambdaレイヤーのアップロードする。

Workload Identityの認証情報ファイルの配置

Workload Identityを使用してLambda関数でGoogle APIを実行するため、参照用の認証情報ファイルをS3バケットに格納する。
まずは、バケットを作成する。

次に、認証情報ファイルをアップロードする。

Storage Transfer Serviceの転送ジョブを実行するLambda関数の作成

Storage Transfer Serviceの転送ジョブを実行するLambda関数を作成する。

import os
import boto3
from pprint import pprint
from googleapiclient import discovery
import google.auth

s3 = boto3.resource('s3')

def lambda_handler(event, context):

    bucket = s3.Bucket(os.environ['config_bucket_name'])
    bucket.download_file(os.environ['client_library_config'], '/tmp/' + os.environ['client_library_config'])

    os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '/tmp/' + os.environ['client_library_config']
    scopes=['https://www.googleapis.com/auth/cloud-platform']
    credentials, project = google.auth.default(scopes=scopes)
    service = discovery.build('storagetransfer', 'v1', credentials=credentials)

    job_name = os.environ['job_name']
    job_body = {
        "projectId": os.environ['project_id']
    }

    request = service.transferJobs().run(jobName=job_name, body=job_body)
    response = request.execute()

    pprint(response)

    return response

ランタイムはpython3.9（x86_64）を指定する。

作成したレイヤーを指定する。

デフォルトのタイムアウト（３秒）は短いので、1分に変更する。

コードで指定した環境変数を設定する。
※ここでは https://qiita.com/okadarhi/items/5ff734c6472b83d51597 で作成した転送ジョブを指定している。

Storage Transfer Serviceの転送ジョブのステータスを確認するLambda関数の作成

Storage Transfer Serviceの転送ジョブのステータスを確認するLambda関数を作成する。

import os
import boto3
from pprint import pprint
from googleapiclient import discovery
import google.auth

s3 = boto3.resource('s3')

def lambda_handler(event, context):

    bucket = s3.Bucket(os.environ['config_bucket_name'])
    bucket.download_file(os.environ['client_library_config'], '/tmp/' + os.environ['client_library_config'])

    os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '/tmp/' + os.environ['client_library_config']
    scopes=['https://www.googleapis.com/auth/cloud-platform']
    credentials, project = google.auth.default(scopes=scopes)
    service = discovery.build('storagetransfer', 'v1', credentials=credentials)

    name = event['Payload']['name']

    request = service.transferOperations().get(name=name)
    response = request.execute()

    pprint(response)

    return response

ランタイムはpython3.9（x86_64）を指定する。

作成したレイヤーを指定する。

デフォルトのタイムアウト（３秒）は短いので、1分に変更する。

コードで指定した環境変数を設定する。

Step Functionsの作成

Storage Transfer Serviceの転送ジョブを実行して、ステータスを確認するStep Functionsを作成する

Run: Storage Transfer Serviceの転送ジョブを実行するLambda関数を呼び出している。
Get: Storage Transfer Serviceの転送ジョブのステータスを確認するLambda関数を呼び出している。
　　　　　　　　　　Runで実行した転送ジョブを確認する。
Check: Getで取得した転送ジョブのステータスを確認する。
　　　　　　　　　　　　　　・ステータスがIN_PROGRESS→Waitステートに遷移し、一定時間経過後に再度Getを実行する。
　　　　　　　　　　　　　　・ステータスがSUCCESS→Successステートに遷移する。
　　　　　　　　　　　　　　・ステータスが上記以外→Failステートに遷移する。

実際の定義内容は以下の通り。

{
  "Comment": "This is your state machine",
  "StartAt": "Run",
  "States": {
    "Run": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "<Storage Transfer Serviceの転送ジョブを実行するLambda関数>"
      },
      "Retry": [<デフォルト設定>],
      "Next": "Get"
    },
    "Get": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "Payload.$": "$",
        "FunctionName": "<Storage Transfer Serviceの転送ジョブのステータスを確認するLambda関数>"
      },
      "Retry": [<デフォルト設定>],
      "Next": "Check"
    },
    "Check": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.Payload.metadata.status",
          "StringMatches": "IN_PROGRESS",
          "Next": "Wait"
        },
        {
          "Variable": "$.Payload.metadata.status",
          "StringMatches": "SUCCESS",
          "Next": "Success"
        }
      ],
      "Default": "Fail"
    },
    "Wait": {
      "Type": "Wait",
      "Seconds": 10,
      "Next": "Get"
    },
    "Fail": {
      "Type": "Fail"
    },
    "Success": {
      "Type": "Succeed"
    }
  }
}

Step Functionsの実行

作成したStep Functionsを実行する。

→正常終了したことが確認できる。

→Google Cloud側でもStorage Transfer Serviceの転送ジョブが成功していることが確認できる。

Step Functionsを使ってStorage Transfer Serviceの転送ジョブを管理する

概要

手順