AWS AppSync のリゾルバから AWS Step Functions ステートマシンを呼び出す

こんにちは、広野です。

React アプリから実行できる Amazon DynamoDB へのデータインポート画面を作成したときの話です。

インプットするデータ量は少ないのですが、データを加工したりデータから AI サービスを使用してさらにデータを作成したりしなければならなかったので、裏で AWS Step Functions ステートマシンを使用します。AWS AppSync から Mutation でインポートデータをステートマシンに渡し、開始させる仕組みを作成しました。そこで AWS AppSync の HTTP リゾルバという設定を使いましたが、少し工夫が必要な面があったので紹介します。

つくったもの

主に、以下の図の 緑色の線 の部分を説明します。

AWS AppSync への命令は Mutation にしましたが、Query でも動くと思います。ただ、データを取得するのが目的ではなかったので Mutation にしました。

HTTP リゾルバから実行するアクションは states:StartExecution を使用します。AWS AppSync はステートマシンを開始させたら一旦役目は終了です。以降、結果は非同期に通知される仕様とします。※本記事では言及しません。

AWS AppSync には主にデータベースと接続するための専用リゾルバが用意されていますが、それ以外の AWS サービスを呼び出すには HTTP リゾルバを使用します。呼び出すのに必要な Sigv4 と呼ばれる署名プロセスも自動で実施してくれます。

AWS AppSync での HTTP リゾルバーの使用 - AWS AppSync

AWS AppSync の HTTP リゾルバーのチュートリアル。

docs.aws.amazon.com

以降、必要な設定を紹介します。

AWS AppSync スキーマ

例として、startImportJob という Mutation を作成し、data という string 型のデータをやり取りする簡単な設定にしています。

type ImportData {
	data: String
}

type Mutation {
	startImportJob(data: String!): ImportData
}

schema {
	query: Query
	mutation: Mutation
	subscription: Subscription
}

AWS AppSync データソース

AWS Step Functions ステートマシンをデータソースと見立てて、設定します。タイプは HTTP にします。このデータソースに関連付ける IAM ロールを別途作成する必要があります。今回は、AWS AppSync が states:StartExecution を実行できる権限さえあれば良いです。

AWS AppSync リゾルバ

今回は JavaScript リゾルバを使用しています。

import { util } from '@aws-appsync/utils';
export function request(ctx) {
  return {
    method: 'POST',
    params: {
      headers: {
        'Content-Type': 'application/x-amz-json-1.0',
        'x-amz-target': 'AWSStepFunctions.StartExecution'
      },
      body: {
        'stateMachineArn': 'arn:aws:states:ap-northeast-1:231376435333:stateMachine:xxxxxxxxxxx',
        'input': JSON.stringify({
          "data": ctx.args.data, //アプリから渡された引数
          "username": ctx.identity.claims.email //emailアドレスをIDトークンから取得している
        })
      }
    },
    resourcePath: '/'
  };
}
export function response(ctx) {
  if (ctx.error) {
    util.error(ctx.error.message, ctx.error.type);
  }
  return {
    "data": JSON.stringify({
      "statusCode": ctx.result.statusCode, //ステートマシンの実行結果
      "result": JSON.parse(ctx.result.body) //結果メッセージ
     })
  }
}

ステートマシンに渡すパラメータを説明します。

headers の部分はこのままにします。

body.stateMachineArn には呼び出したいステートマシンの ARN を入れます。

body.input に入れたデータが、そのままステートマシンへの input になります。ただし、格納するデータは JSON.stringify で文字列化した JSON データに変換する必要があります。(VTL の場合、$util.toJson を使用します)

レスポンスについて説明します。

あらかじめ定義していた data (string 型) 変数を使用してアプリに結果を戻します。そのため、戻りのデータを JSON.stringify で文字列化しています。AWS Step Functions から戻ってきた結果は、ctx.result に入ります。エラーがあると、ctx.result.statusCode が 200 以外になります。エラーの内容は ctx.result.body に入っているのですが、これが JSON オブジェクトではなく文字列化された JSON で格納されているので、ctx.result.body を JSON.parse をかけて JSON オブジェクトとして読み取れる状態に変換しています。そうして statusCode とマージさせた JSON を stringify させて返す、という処理にしています。当初ここで結果を読み取ることができず、苦労しました。

この呼び出し方は、説明は雑でしたが以下の AWS ブログに書いてありました。

Invoking AWS Step Functions short- and long-running workflows from AWS AppSync | Amazon Web Services

AWS AppSync is a fully managed GraphQL service allowing developers to easily build GraphQL APIs. AWS AppSync lets develo...

aws.amazon.com

React 

React アプリ側の GraphQL コードも書いておきます。

import { generateClient } from 'aws-amplify/api';
import gql from 'graphql-tag';

//create AppSync client
const client = generateClient();
//インポートジョブ開始関数
const startImportJob = async (data) => {
  const mutateImportJob = gql`
    mutation startImportJob($data: String!) {
      startImportJob(data: $data) {
        data
      }
    }
  `;
  const res = await client.graphql({
    query: mutateImportJob,
    variables: {
      data: data
    }
  });
  return JSON.parse(res.data.startImportJob.data); //文字列化されたJSONが返されるのでparseしている
};

export { startImportJob };

AWS CloudFormation テンプレート

抜粋ですが、関連する部分のテンプレートを貼っておきます。詳細な設定はこちらをご覧ください。AWS Step Functions は省略しています。検証した AWS AppSync が Amazon Cognito ユーザープールで認証するタイプになっているので、その記述が紛れ込んでいます。

AWSTemplateFormatVersion: 2010-09-09
Description: The CloudFormation template that creates an AppSync API to start a Step Functions state machine and relevant IAM roles.

# ------------------------------------------------------------#
# Input Parameters
# ------------------------------------------------------------#
Parameters:
  SubName:
    Type: String
    Description: System sub name that is used for all deployed resources. (e.g. prod or dev)
    Default: dev
    MaxLength: 10
    MinLength: 1

Resources:
# ------------------------------------------------------------#
# AppSync
# ------------------------------------------------------------#
  AppSyncApi:
    Type: AWS::AppSync::GraphQLApi
    Description: AppSync API for example
    Properties:
      Name: !Sub example-${SubName}
      AuthenticationType: AMAZON_COGNITO_USER_POOLS
      AdditionalAuthenticationProviders:
        - AuthenticationType: AWS_IAM
      UserPoolConfig:
        UserPoolId: xxxxx //CognitoのユーザープールID を入れる
        AwsRegion: !Sub ${AWS::Region}
        DefaultAction: "ALLOW"
      IntrospectionConfig: DISABLED
      LogConfig:
        CloudWatchLogsRoleArn: !GetAtt AppSyncCloudWatchLogsPushRole.Arn
        ExcludeVerboseContent: true
        FieldLogLevel: ALL
      Visibility: GLOBAL
      XrayEnabled: true
      Tags:
        - Key: Cost
          Value: !Sub example-${SubName}
    DependsOn:
      - AppSyncCloudWatchLogsPushRole

  AppSyncSchema:
    Type: AWS::AppSync::GraphQLSchema
    Properties:
      ApiId: !GetAtt AppSyncApi.ApiId
      Definition: |
        schema {
          query: Query
          mutation: Mutation
          subscription: Subscription
        }
        type ImportData {
          data: String
        }
        type Mutation {
          startImportJob(data: String!): ImportData
        }
    DependsOn:
      - AppSyncApi

  AppSyncDataSourceStateMachines:
    Type: AWS::AppSync::DataSource
    Properties:
      ApiId: !GetAtt AppSyncApi.ApiId
      Name: !Sub example${SubName}StateMachines
      Description: AppSync DataSource to call State Machines.
      Type: HTTP
      ServiceRoleArn: !GetAtt AppSyncStepFunctionsRole.Arn
      HttpConfig:
        Endpoint: !Sub https://states.${AWS::Region}.amazonaws.com/
        AuthorizationConfig:
          AuthorizationType: AWS_IAM
          AwsIamConfig:
            SigningRegion: !Ref AWS::Region
            SigningServiceName: states
    DependsOn:
      - AppSyncApi
      - AppSyncStepFunctionsRole

  AppSyncResolverStartImportJob:
    Type: AWS::AppSync::Resolver
    Properties:
      ApiId: !GetAtt AppSyncApi.ApiId
      TypeName: Mutation
      FieldName: startImportJob
      DataSourceName: !GetAtt AppSyncDataSourceStateMachines.Name
      Kind: UNIT
      Runtime:
        Name: APPSYNC_JS
        RuntimeVersion: 1.0.0
      Code: |
        import { util } from '@aws-appsync/utils';
        export function request(ctx) {
          return {
            method: 'POST',
            params: {
              headers: {
                'Content-Type': 'application/x-amz-json-1.0',
                'x-amz-target': 'AWSStepFunctions.StartExecution'
              },
              body: {
                'stateMachineArn': 'arn:aws:states:ap-northeast-1:231376435333:stateMachine:example-xxxxx',
                'input': JSON.stringify({
                  "data": ctx.args.data,
                  "username": ctx.identity.claims.email
                })
              }
            },
            resourcePath: '/'
          };
        }
        export function response(ctx) {
          if (ctx.error) {
            util.error(ctx.error.message, ctx.error.type);
          }
          return {
            "data": JSON.stringify({
              "statusCode": ctx.result.statusCode,
              "result": JSON.parse(ctx.result.body)
             })
          }
        }
    DependsOn:
      - AppSyncDataSourceStateMachines

# ------------------------------------------------------------#
# AppSync CloudWatch Invocation Role (IAM)
# ------------------------------------------------------------#
  AppSyncCloudWatchLogsPushRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub example-AppSyncCloudWatchLogsPushRole-${SubName}
      Description: This role allows AppSync to push logs to CloudWatch Logs.
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
              - appsync.amazonaws.com
            Action:
              - sts:AssumeRole
      Path: /
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSAppSyncPushToCloudWatchLogs

# ------------------------------------------------------------#
# AppSync Step Functions Invocation Role (IAM)
# ------------------------------------------------------------#
  AppSyncStepFunctionsRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub example-AppSyncStepFunctionsRole-${SubName}
      Description: This role allows AppSync to invoke Step Functions.
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
              - appsync.amazonaws.com
            Action:
              - sts:AssumeRole
      Path: /
      Policies:
        - PolicyName: !Sub example-AppSyncStepFunctionsPolicy-${SubName}
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - "states:StartExecution"
                Resource:
                  - "arn:aws:states:ap-northeast-1:231376435333:stateMachine:example-xxxxx"

まとめ

いかがでしたでしょうか？

時間のかかる処理を AWS AppSync の裏側で AWS Step Functions を非同期実行させるのに活用できると思います。もちろん、AWS AppSync のタイムアウト範囲内で終わることが確実な処理であれば同期処理にしてしまってもよいでしょう。

本記事が皆様のお役に立てれば幸いです。