「コース中の質問をサーバーレスなアーキテクチャで受けて共有」で掲載しました質問フォームのアーキテクチャを一部変更しましたので、今回は設計を書いておきたいと思います。
動作などは上記リンクをご参照ください。
Amazon Route 53
個人で所有しているドメインでアクセスできるようにしていますので、Route 53のAレコードのエイリアスで、CloudFrontのドメインを指定しています。
AWS Certificate Manager
Certificate Managerでドメインの証明書を発行して、Route 53のCNAMEでドメイン認証をしています。
Amazon CloudFront
CloudFrontにはCertificate Managerで発行した証明書を設定しています。
オリジンにはS3バケットを指定して、OAIにてバケットへのCloudFrontを介さない直接のアクセスをできないようにしています。
Amazon S3
S3のバケットポリシー
{
"Version": "2008-10-17",
"Id": "PolicyForCloudFrontPrivateContent",
"Statement": [
{
"Sid": "1",
"Effect": "Allow",
"Principal": {
"AWS": "arn:aws:iam::cloudfront:user/CloudFront Origin Access Identity XXXXXXXXX"
},
"Action": "s3:GetObject",
"Resource": "arn:aws:s3:::qa.trainocate/*"
}
]
}
S3バケットにはHTML, CSS, JavaScriptを配置して入力フォームをCloudFront経由で配信しています。
Amazon API Gateway
[質問]ボタンが押されると質問内容がURLパラメータとしてAPI GatewayでデプロイしたAPIにPOSTされます。
API GatewayではURlパラメータをJSONに変換してLambdaにEventデータとして渡しています。
最初のAWS Lambda
最初のLambdaはメッセージを受け取って、SNSトピックへ発行します。
import logging
import boto3
import traceback
import json
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info(event)
try:
sns = boto3.client('sns')
qa_topic = 'Qa'
snsTopicArn = [t['TopicArn'] for t in sns.list_topics()['Topics'] if t['TopicArn'].endswith(':' + qa_topic)][0]
sns.publish(
TopicArn=snsTopicArn,
Message=json.dumps(event),
Subject='question'
)
except:
logger.error(traceback.format_exc())
Amazon SNS, Amazon SQS
SNSトピックは3つのSQSキューへメッセージをファンアウトします。
Teamsに投稿するAWS Lambda
1つ目のSQSキューからメッセージを受信したLambdaはTeamsに通知します。
import logging, boto3, json, traceback, os
from teams import Teams
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
teams = Teams()
try:
for record in event['Records']:
message_body = json.loads(record['body'])
message = json.loads(message_body['Message'])
project_id = message['project_id']
description = message['contact_detail'].replace(
'newline',
'\n'
)
teams.send_message(
'受講者からの質問',
'{description} {project_id}'.format(
description=description,
project_id=project_id
)
)
except:
logger.error(traceback.format_exc())
raise Exception(traceback.format_exc())
Backlogにチケット登録するAWS Lambda
2つ目のSQSキューからメッセージを受信したLambdaはBacklog APIにPOSTしてチケット登録を行います。
import logging, boto3, json, traceback, os, requests
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
try:
backlog_key = os.environ.get('BACKLOG_KEY', '')
project_id = os.environ.get('PROJECT_ID', '')
issue_type_id = os.environ.get('ISSUE_TYPE_ID', '')
priority_id = os.environ.get('PRIORITY_ID', '')
assignee_id = os.environ.get('ASSIGNEE_ID', '')
backlog_url =os.environ.get('BACKLOG_URL', '')
mile_stone_id = os.environ.get('MILE_STONE_ID', '')
for record in event['Records']:
message_body = json.loads(record['body'])
logger.info(message_body)
logger.info(message_body['Message'])
message = json.loads(message_body['Message'])
summary = message['contact_detail'].replace(
'newline',
''
)[:50]
description = message['contact_detail'].replace(
'newline',
'\n'
)
payload_dic = {
'projectId': project_id,
'issueTypeId': issue_type_id,
'priorityId': priority_id,
'assigneeId': assignee_id,
'milestoneId': [mile_stone_id],
'summary': summary,
'description': description
}
headers = {
'Content-Type': 'application/json'
}
response = requests.post(
backlog_url + backlog_key,
data=json.dumps(
payload_dic
),
headers = headers
)
if response.text:
logger.info(response.text)
else:
logger.info(response)
except:
logger.error(traceback.format_exc())
raise Exception(traceback.format_exc())
DynamoDBにデータを入れるAWS Lambda
3つ目のSQSキューからメッセージを受信したLamndaはリクエスト内容と時間、送信元IPアドレスなどの情報とともにDynamoDBへ書きます。
import traceback, json, os, logging, boto3
from datetime import datetime
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info(event)
try:
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('qa')
with table.batch_writer() as batch:
for record in event['Records']:
message_body = json.loads(record['body'])
message = json.loads(message_body['Message'])
message['submit_datetime'] = datetime.now().strftime('%Y%m%d%H%M%S%f')
batch.put_item(Item=message)
except:
logger.error(traceback.format_exc())
トレノケートのAWS研修（AWS認定トレーニング）
トレノケートのAWS認定トレーニングでは、AWS社の厳格なテクニカルスキル及びティーチングスキルチェックに合格した認定トレーナーがコースを担当します。AWS初心者向けの研修や、AWS認定資格を目指す人向けの研修をご提供し、皆様のAWS知識修得のサポートをいたします。
・トレノケートのAWS研修（AWS認定トレーニング）はこちら
▼AWS初心者の方は、AWS Cloud Practitioner Essentialsから！
座学中心の研修で、AWSを初めて学ぶ方や、営業などで提案に関わる方におすすめです。
「AWS Certified Cloud Practitioner」資格取得を目指す方の基礎知識修得にも最適です。
→ AWS Cloud Practitioner Essentials 詳細・日程はこちらから