こんにちは、デザイナー部テクニカルアーティストチームスペシャリストの岸川貴紀と Cyllista Game Engine ゲームエンジニアの足立和弥です。Cygames Tech Conference で「AAA タイトル開発と在宅勤務を支えるゲームエンジンエンジニアとテクニカルアーティストの取り組み」と題して講演を行いました。
ご視聴いただいた方々、コメントをくださった方々本当にありがとうございました。
講演の資料と動画はこちらとなります。
このフォローアップ記事では講演内ではお伝えしきれなかったことを補足していきます。
ShotGrid -> Slack 統合の AWS SAM 実装
今回の Cygames Tech Conference では、岸川からは、「ネットワークの環境の変化に対するツールリリースに関する対応」と「コミュニケーションパイプライン」についてお話しさせていただきました。
このフォローアップ記事では、その中でもテクニカルな部分だった「コミュニケーションパイプライン」の ShotGrid から Slack への自動通知の部分を AWS SAM を使用して実装した点について、基本的な部分の詳細を解説していきます。
当セッションにおいて、ShotGrid から Slack への通知のしくみは次の様な AWS アーキテクチャーになっていると解説しました。
この全体を構築する際には、AWS SAM (Serverless Application Model) での構築と AWS コンソール / 別途 CloudFormation 上での構築を併用しています。
主な使い分けは次のようになっています。
【AWS SAM】
- 影響範囲
- プロジェクト単位でそれぞれ実装するもの。
- サービス
- AWS Lambda (以下 Lambda)
- Amazon API Gateway (以下、API Gatway)
- Amazon SQS (以下 SQS)
- Amazon CloudWatch (以下 CloudWatch)
【AWS コンソール / 別途 CloudFormation】
- 影響範囲
- TA チームとしてサービスを共有するもの。
- サービス
- AWS ChatBot (以下、 ChatBot)
- Amazon SNS (以下、 SNS)
- Amazon DynamoDB (以下、 DynamoDB)
- 加えて、AWS IAM などのセキュリティ・認証機構も共有で管理。
これらの構成設計と運用は基本的に岸川が行なっていますが、他 PJ でも同様に自動化を構築したいという要望もあり、AWS SAM で書いたものをプロジェクト用にカスタマイズするテンプレートとして用意し、TA で統括管理する部分はリソースを共有しています。
それでは次に、特に核になる、AWS SAM 側の実装をどう行なっているかを解説していきます。
【全体の処理順】
全体は次のような順で処理されます。
- ShotGrid Webhook から更新イベントが実行される。
- API Gateway のエンドポイントにリクエストが送信される。
- API Gateway から SQS にペイロード本体が送信され、SQS 内でポーリングされる。
- SQS から Lambda への実行イベントが発行され、Lambda が実行される。
- Lambda から、DynamoDB に問い合わせて実行済みイベントかそうでないかを判断する。未実行のものであれば DyanamoDB に書き込み、実行済みであれば Lambda の処理自体をスキップする。 (履歴管理と冪等性の確保)
- Lambda から ShotGrid に問い合わせてペイロード内の情報から、必要情報を問い合わせる。
- Lambda から Slack の Slack API を元に問い合わせて、通知の処理を完了がされる。
加えて、監視・デバッグ用として、次の機能が常時動作しています。
- Lambda でエラーが発生した場合、CloudWatch Alarm で閾値をベースにアラームが発行される。
- こちらのみ SAM 内で構築。
- SNS 、ChatBot という順で経由して、Slack のデバッグ用チャンネルに送信される。
次に、上記の処理系の中から、特に核になっている ShtoGrid から Lambda までの処理を解説します。
【ShotGrid → Lambda】
ShotGrid から Lambda までの出発点として、ShotGrid Webhook を使用しています。ShotGrid Webhook は 2021年11月 に正式リリースされたサービス（プレゼン資料の執筆当時はオープンベータ）となっており、ShotGrid 上の権限があるユーザーであれば誰でもアクセスできる様になっています。
ShotGrid Webhook には次の設定項目があります。
- Name / 名前
- Description / 説明 (オプション)
- URL
- Secret Token / シークレットトークン (オプション)
- Entity Filters / エンティティ フィルタ
- Validate SSL Certificate / SSL 証明書を検証 (オプション)
- Deliver in Batched Format / バッチ形式で配信 (オプション)
- Notify when unstable / 不安定な場合に通知 (オプション)
- Projects / プロジェクト (オプション)
必ず必要になってくる、この URL 部分に入力するエンドポイントが、ここで作成する API Gateway で発行される URL です。
AWS SAM では、基本的には AWS::Serverless::Function のプロパティに記載し、Output にてエンドポイントのテンプレートを記載すればいいだけなのですが、今回は SQS を挟む都合上、直接 AWS::ApiGatewayV2::Api 、AWS::ApiGatewayV2::Integration 、AWS::ApiGatewayV2::Route、AWS::ApiGatewayV2::Stage を記載していきます。
- AWS::ApiGatewayV2::Api
- URLエンドポイントを構築する設定
- ここでは、ApiGatewayV2 を使用しており、HTTP/Websocket タイプを前提に作成しています。これは、SQS との接続のために設定しているものです。(もう一つの ApiGateway の方は、REST API を前提に設計されており、ApiGateway”V2″ となっているものの別物であることに注意して下さい。)
- AWS::ApiGatewayV2::Integration
- URL エンドポイントにリクエストが投げられた際に実行される SQS との統合機能の設定。
- AWS::ApiGatewayV2::Route
- URL エンドポイントの末尾パス。Ex. /enqueue_sg_event
- AWS::ApiGatewayV2::Stage
- URL エンドポイントのリリースステージで、本番用のエンドポイントなのか、開発用なのか、バージョン違いなのかをここで指定する。また、ここの AutoDeploy を有効にする事によって、通常はマニュアルで Api をデプロイする必要があるところを、自動でデプロイしてくれます。つまり、SAM 全体をデプロイした時点で使用できるようになっています。
SgEventSQS:
Type: AWS::SQS::Queue
Properties:
FifoQueue: True
ContentBasedDeduplication: True
SgEventApi:
Type: AWS::ApiGatewayV2::Api
Properties:
Name: <プロジェクト名>_sg_event_api
ProtocolType: HTTP
SgEventApiSQSIntegration:
Type: AWS::ApiGatewayV2::Integration
Properties:
ApiId: !Ref SgEventApi
CredentialsArn:
IntegrationType: AWS_PROXY
IntegrationSubtype: SQS-SendMessage
PayloadFormatVersion: 1.0
RequestParameters:
QueueUrl: !Ref SgEventSQS
MessageBody: $request.body.data
MessageGroupId: <プロジェクト名>_sg
SgEventApiRoutes:
Type: AWS::ApiGatewayV2::Route
Properties:
ApiId: !Ref SgEventApi
RouteKey: 'POST /enqueue_sg_event'
Target: !Join
- /
- - integrations
- !Ref SgEventApiSQSIntegration
SgEventApiStageLogs:
Type: AWS::Logs::LogGroup
roperties:
LogGroupName: /<組織名>/ShotGrid/SgEventApiStageLogs/<プロジェクト名>
RetentionInDays: 1
SgEventApiStage:
Type: AWS::ApiGatewayV2::Stage
Properties:
AccessLogSettings:
DestinationArn: !GetAtt SgEventApiStageLogs.Arn
Format: $context.requestId
ApiId: !Ref SgEventApi
AutoDeploy: True
DefaultRouteSettings:
DetailedMetricsEnabled: true
StageName: prod
あとは、実際に実行する部分である AWS::Serverless::Function とAWS::CloudWatch::Alarm 部分を作成していくだけです。
SgEventFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: sg_event/
Events:
SQSEvent:
Type: SQS
Properties:
Queue: !GetAtt SgEventSQS.Arn
BatchSize: 10
Role: <Lambda用IAMロール>
SgEventFunctionAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
AlarmActions:
- <SNS用IAMロール>
ComparisonOperator: GreaterThanOrEqualToThreshold
Dimensions:
- Name: FunctionName
Value: !Ref SgEventFunction
EvaluationPeriods: 1
MetricName: Errors
Namespace: AWS/Lambda
Period: 60
Statistic: Sum
Threshold: '1'
このテンプレートを踏まえたディレクトリ構成は次の様になります。
(* エンティティイベントごとに処理をモジュール化しているため、py ファイルを個別にする構成を取っています。)
- sam_app_root
- sg_event
- app.py
- <entity.py>
- template.yaml
- sg_event
これらをまとめて SAM テンプレートに起こすと次のようになります。
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: >
<プロジェクト名>-shotgrid-application
Sample SAM Template for <プロジェクト名>-shotgrid-application
# グローバル設定 / Lambda に記載しても良いが、
# Lambda が増えた場合などのためにグローバルに埋め込む
Globals:
Function:
# 執筆時点での不具合だろうが、初回デプロイ時は最大 30 秒
# 次回以降は最大 900 秒で設定できる
Timeout: 30
Runtime: python3.8
Handler: app.lambda_handler
Environment:
Variables:
SG_SERVER: <ShotGrid のサーバーURL>
SG_SCRIPT_NAME: <ShotGrid のスクリプト名>
SG_SCRIPT_APPKEY: <ShotGrid のスクリプトのアプリケーションキー>
PROJECT_NAME: <ShotGrid のプロジェクト名>
SLACK_WEBHOOK_URL: <Slack の Webhook URL>
# リソース定義
Resources:
SgEventSQS:
Type: AWS::SQS::Queue
Properties:
FifoQueue: True
ContentBasedDeduplication: True
SgEventFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: sg_event/
Events:
SQSEvent:
Type: SQS
Properties:
Queue: !GetAtt SgEventSQS.Arn
BatchSize: 10
Role: <Lambda用IAMロール>
SgEventFunctionAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
AlarmActions:
- <SNS用IAMロール>
ComparisonOperator: GreaterThanOrEqualToThreshold
Dimensions:
- Name: FunctionName
Value: !Ref SgEventFunction
EvaluationPeriods: 1
MetricName: Errors
Namespace: AWS/Lambda
Period: 60
Statistic: Sum
Threshold: '1'
SgEventApi:
Type: AWS::ApiGatewayV2::Api
Properties:
Name: <プロジェクト名>_sg_event_api
ProtocolType: HTTP
SgEventApiSQSIntegration:
Type: AWS::ApiGatewayV2::Integration
Properties:
ApiId: !Ref SgEventApi
CredentialsArn:
IntegrationType: AWS_PROXY
IntegrationSubtype: SQS-SendMessage
PayloadFormatVersion: 1.0
RequestParameters:
QueueUrl: !Ref SgEventSQS
MessageBody: $request.body.data
MessageGroupId: <プロジェクト名>_sg
SgEventApiRoutes:
Type: AWS::ApiGatewayV2::Route
Properties:
ApiId: !Ref SgEventApi
RouteKey: 'POST /enqueue_sg_event'
Target: !Join
- /
- - integrations
- !Ref SgEventApiSQSIntegration
SgEventApiStageLogs:
Type: AWS::Logs::LogGroup
Properties:
LogGroupName: /<組織名>/ShotGrid/SgEventApiStageLogs/<プロジェクト名>
RetentionInDays: 1
SgEventApiStage:
Type: AWS::ApiGatewayV2::Stage
Properties:
AccessLogSettings:
DestinationArn: !GetAtt SgEventApiStageLogs.Arn
Format: $context.requestId
ApiId: !Ref SgEventApi
AutoDeploy: True
DefaultRouteSettings:
DetailedMetricsEnabled: true
StageName: prod
Outputs:
SgEventFunction:
Description: ""
Value: !GetAtt SgEventFunction.Arn
SgEventSQS:
Description: "ShotGrid SQS integrated with SgEventFunction Function"
Value: !GetAtt SgEventSQS.Arn
【SQS】
SQS から Lambda に渡ってくるレコードは、ただ API Gateway から渡ってくる通常のペイロードとは形態が異なります。これは sam local invoke する際などの障害になる事や、実際の処理を記載していく際に障害になります。
これらを簡単にするため、ヘルパ関数を作成します。
def get_datas(event):
"""SQS からの Records や、ローカル・コンソールテスト・Proxy/Non-Proxy
のペイロードに対応。
"""
logger.debug(sys._getframe().f_code.co_name)
datas = []
if event.get('Records'):
# SQS で渡ってきた event には Records としてまとまって格納されてくるので、
# こちらを読み取る。
# 更に、中身はヘッダ等を含むリクエスト全体が渡されるので、
# 本来使うペイロードだけを抜き出す。
datas = [json.loads(record["body"]) for record in event['Records']]
elif event.get('data'):
# ローカルや AWS コンソール上からテストする際や、
# Proxy を使わないで REST API で Lambda とつなぐ際には、
# そのままのペイロードを渡します。
datas = [event['data']]
elif event.get('body') and type(event['body']) is str:
# API Gateway の統合設定で Proxy を設定した場合には、
# ヘッダ等を含むリクエスト全体が渡されるので、
本来使うペイロードだけを抜き出す。
body = json.loads(event['body'])
datas = [body["data"]]
return datas
帰ってくる値は、ペイロードのリストとなるので、それをループで回します。
場合によっては、この部分を AWS Step Functions などにして別 Lambda 関数に渡しても良いでしょう。
def lambda_handler(event, context):
datas = get_datas(event)
for data in datas:
# 各ペイロード毎に処理
exec_(data)
【DynamoDB】
サーバレス環境の冪等性に対応するためを最初の目的として、DynamoDB を設定しています。
そこに記載するコードが必要になるのですが、これは boto3 を使う事によってとても簡潔に書けます。
import boto3
import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
def check_idempotent_event(data, debug=False):
# デバッグ時はチェックを無視
if debug:
return True
table_name = ''
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)
request = table.get_item(Key={'delivery_id': data['delivery_id'],
'user_id': data['user']['id']})
# すでにレコードが見つかった場合は Item として取得できるので、
# 検知してスキップさせる処理に。
if request.get('Item'):
logger.debug('A record is found in DynamoDB.')
return False
logger.debug('Write a record to DynamoDB.')
table.put_item(
Item={
'user_id': data['user']['id'],
'user_type': data['user']['type'],
'session_uuid': data['session_uuid'],
'entity_id': data['entity']['id'],
'entity_type': data['entity']['type'],
'operation': data['operation'],
'created_at': data['created_at'],
'project_id': data['project']['id'],
'event_type': data['event_type'],
'delivery_id': data['delivery_id'],
'id': data['id'],
'meta_type': data['meta']['type']
}
)
return True
そして、こちらの関数を先ほどのペイロードをループ処理している部分に統合させて完了です。
def lambda_handler(event, context):
datas = get_datas(event)
for data in datas:
if not check_idempotent_event(data, debug=False):
continue
# 各ペイロード毎に処理
exec_(data)
【まとめ】
ShotGrid Webhook が公開されたことにより、これまで、ShotGrid Events で起こっていた、一つのプラグインが停止するだけで全てのプラグイン処理が停止してしまうといった問題が回避できるようになり、また、物理サーバーを管理する必要性もなくなりました。その上で、AWS SAM として、実装・インフラコードをテンプレート化することにより、復旧や他 PJ との共有も容易になりました。実際、この構成は他の開発・運用 PJ の両方でも運用が開始されています。
昨今のリモートワークでの開発・運用環境においては、こうしたクラウドリソースをうまく活用しながら業務効率化を行っていくことは当たり前になっていくのではないでしょうか。
皆さんも、是非試してみてください。
Slack の運用ルールについて
講演内では運用ルールを参考程度に紹介しただけでしたので少し補足します。
このルールは在宅勤務が始まった当初に設定してプロジェクトメンバー間で共有されたのですが、在宅勤務時に限らずプロジェクト内のコミュニケーションを円滑にするうえで大切なものだと思っています。
自身が含まれるメンションに対しての返信やリアクションを付けるということは投稿した側にその投稿が読んだことを伝えるための一つのコミュニケーションで、実際の会話であれば返答や相槌にあたるものです。投稿した側は相手がちゃんとそれを読んでくれたのかどうか知れるだけでも安心できます。
通話のかけ方については業務時間内であればいつでも通話をかけるということをよしとしています。これはオフィス勤務時に直接声をかけるのと同じ感覚で話しかけられるように設定したルールです。これを全員が共通の認識として持っていることで通話をかける際の心理的な障壁が多少なりとも軽くなるはずです。
このようなルールを明文化しておくことによって、コミュニケーションに対する認識のずれを減らし人によって情報伝達の精度にバラつきが出ることを軽減する効果を期待しています。
今回紹介したルールが正解というのではなく、その組織に合ったルールがあると思います。大事なのはその組織内でそのルールが共通認識として存在していることです。もし、組織内のコミュニケーションの認識が揃わずコミュニケーションが円滑にならないとお悩みの方がいらっしゃれば我々のルールをベースに共通認識を作ってみてはいかがでしょうか。
最後に
今回の講演では大規模プロジェクトの開発環境を在宅勤務への対応を軸に紹介しました。在宅勤務で抱えている問題の解決の一助となれば幸いです。また、コミュニケーションパイプラインについてはオフィス勤務でも活かせる内容になっていますので参考にしていただければと思います。
Cygames では Cyllista Game Engine と Cyllista Game Engine を採用したゲーム「Project Awakening」を共に製作する仲間を募集しています。詳細はこちらをご覧ください。