AWS Lambdaの書き方まとめ

AWS Lambda使って○○がしたいときってありますよね。

そんなときLambdaやプログラミング言語のスキルが低い私はけっこう苦戦したので、攻略方法を自分のためにも世のためにもまとめます。

なおこの記事で扱う言語はPythonのみです。

基本

何をしたいのかを整理する

まずLambdaで何をどうしたいのかを整理します。
（他のサービスで実現できないかを考えておくことが前提）

以下、例を挙げていきます。

• RDSのスナップショットを2時間おきに取りたい

自動スナップショットは1日1回しかバックアップを取得してくれないため作成する。必要なことは、RDSインスタンスの一覧を取得して、タグのついた対象のRDSインスタンス名を取得、スナップショットを実行する。もしくはRDS名を指定してスナップショットを実行する。

• EBSをスナップショットしたら別リージョンにコピーしたい

EBSのスナップショットはAmazon Data Lifecycle Manager (DLM)を用いれば自動化できるが、リージョンコピーまではできないため作成する。必要なことは、CloudWatchイベントでEBSのスナップショット完了を選択し、ターゲットをLambdaに設定すること、それをトリガーに対象のEBS名を取得しDRコピーを実行する。

このような形でやりたいことを整理していきましょう。Lambdaを利用してやりたいことの多くは自動化やサービス間連携が目的かと思われます。自動化の場合はコンソール上の操作のどこの部分を自動化したいのか、サービス間連携の場合は始まりと終わりを考えてみてください。AWSのコンソール上のすべての操作はAPIを利用して操作できますし、サービス間連携はCloudWatchやSNSなどを利用すればうまく繋げることができるはずですので、わりと整理しやすいですよ。

やりたい動作をboto3のドキュメントで探す

https://boto3.amazonaws.com/v1/documentation/api/latest/index.html

Lambdaでやりたいことが決まったら、boto3のドキュメントを見てやりたいことを探します。

例えば、SNSで通知を行いたい場合は、以下のURLを参考にします。

Request Syntax が構文で、Returns が結果です。

探し方は簡単、boto3のトップページから「API reference」の中で対象のサービスを選んで、やりたい動作を探すだけ。それでは構文にならってPythonを記述しましょう。

import boto3

client = boto3.client('sns')

response = client.publish(
    TopicArn='xxxxxx',
    Message='Hello',
    Subject='This is test'
)

各パラメータの説明もドキュメントに書いてあります。上記の場合、「TopicArn」がSNSトピックのARNで、「Message」が本文、「Subject」が件名です。

また、boto3を利用する際は必ずインスタンスオブジェクトの生成が必要となりますので、お忘れなく。ドキュメントを見ると目次があると思いますが、それを選択すると一番始めに記載されているかと思います。そのあとに各動作の一覧がありますので、使いたいものに必要なインスタンスオブジェクトを生成することを忘れないでください（なんでもかんでも boto3.client じゃないんだぜ）。ちなみに、上記でも利用しているSNSの「 publish() 」の場合は以下を参考にしています。

環境変数を使う

環境変数を使うことで、汎用的な関数にしましょう。なるべくコード内に情報は記述しない方向でいくと管理が楽です。

先ほどのSNSへの通知するLambdaの場合は以下のようにできます。

import boto3
import os

client = boto3.client('sns')
arn_sns = os.environ['ARN']

response = client.publish(
    TopicArn=arn_sns,
    Message='Hello',
    Subject='This is test'
)

こうすることで対象のSNSトピックを変えたいときは、環境変数を修正するだけで済みますので、コードに触れることはありません。よって人的ミスを減らすことにつながります。ぜひ活用しましょう。

よく使う技

CloudWatchイベントの値を受け取る

例えば、CloudWatchをトリガーにしているとき、CloudWatchから渡される値を使いたい場合は、event変数に値が入っております。以下は、EBSのスナップショット完了をトリガーにしたものになります。

import boto3

def lambda_handler(event,context):
    snaphost_create_time = event['time']
    print(snaphost_create_time)

「cloudwatach event ec2 snapshot」とかでググると、AWSドキュメントがヒットします。渡される内容が確認できますので、自分の取得したいものを確認しましょう。ほかのCloudWatchイベントも基本「cloudwatch event xxx」で検索すればAWSドキュメントが検索結果にでますので、どんな値が渡されるか事前に見ておくといいですね。

ちなみに、CloudWatchイベントのターゲットの設定で、「定数(JSONテキスト)」を選択してLambdaの動作をコントロールすることにより無駄な関数を作成しなくてもよいケースがあります。以下は、EC2の起動停止を行うLambdaです。

定数
{"Action": "start", "Instances": ["************"]}

import boto3

ec2 = boto3.client('ec2')

def lambda_handler(event, context):
    instances = event['Instances']
    if event['Action'] == 'start':
        ec2.start_instances(InstanceIds=instances)
        print ('started your instances: ' + ", ".join(instances))
    elif event['Action'] == 'stop':
        ec2.stop_instances(InstanceIds=instances)
        print ('stopped your instances: ' + ", ".join(instances))

こちらの方法は以下の記事を参考にしております。

SNSをトリガーにする

トリガーをSNSトピックに選択することでいろんなサービスと連携できます。

例えば、RDSイベントサブスクリプションのスナップショット完了通知をトリガーにDRコピーをしたい場合、RDSイベントサブスクリプションをLambdaのトリガーに設定することはできませんが、一度SNSトピックに通知し、そのSNSトピックをトリガーにLambdaを実行することはできます。

つまりSNSトピックを間に置いておくことで、各サービスとLambdaを連携することができるわけです。SNSはいろんなサービスと連携ができるので、Lambdaと連携できないなと思ったらSNSと連携できるかどうかを確認してみるといいかもです。

S3からファイルを読み込む

S3からファイルを読み込みたいときってたまにあります。

そんなときは以下のようにやってみてください。

import boto3

s3 = boto3.client('s3')
bucket_name = os.environ['BUCKET']

def lambda_handler(event, context):
    file_name = 'syukujitsu.csv' 

    response = s3.get_object(Bucket=bucket_name, Key=file_name)
    body = response['Body'].read()

    bodystr = body.decode('utf-8')
    lines = bodystr.split('\r\n')

    print(lines)

上記は日本の祝日がリスト化されているCSVファイルを読み込んでいます。

こちらの方法は以下の記事を参考にしております。

一時的なファイルを利用する

Lambda関数内でファイルを生成してS3などに保存したいとき、一時的なファイルを利用すると便利です。

def lambda_handler(event, context):
    # ready to tmpfile
    filepath = '/tmp/tmp.csv'

    savepath = test.csv
    header = ['name', 'age']
    data = [['judy','18'],['mary','20']]

    # create tmpfile
    with open(filepath, 'w', newline='', encoding='utf-16', errors='ignore') as f:
        writer = csv.writer(f, dialect='excel-tab', quoting=csv.QUOTE_ALL)
        writer.writerow(header)
        writer.writerows(data)

    # upload
    s3 = boto3.resource('s3')
    s3.meta.client.upload_file(filepath, bucket, savepath)

    # delete tmpfile
    os.remove('/tmp/data.csv')

上記のように一時ファイルにデータを書き込むことで、S3にLambda関数内で生成したファイルをアップロードできます。

こちらの方法は以下の記事を参考にしております。

Lambda関数の失敗検知

Lambdaのコンソールにて、「送信先を追加」を選択肢すると、失敗したときの通知設定が可能です。SNSトピックを指定すれば失敗したとき、登録しているメールアドレスに通知がいくようになります。

正常時の場合でも送信が可能ですので、このLambdaが成功したのをトリガーに○○をしたいと思ったときに使ってみるといいかもです。送信先は、SNSトピック、SQSキュー、Lambda関数、EventBridgeイベントパス、を指定することが可能です。

注意点

• グローバル変数の再利用

Lambdaは一定の時間を過ぎると初期化されますが、そうでない場合は（頻繁にLambdaが実行される、例えば30秒以内に2回とか）、グローバル変数が再利用されます。

例えば以下のようにコードを書いている場合、

from datetime import datetime

now = datetime.now()

def lambda_handler(event, context):
    # code.....

現在時刻が正しく取得できず、同じ時刻が繰り返し利用されてしまいます。対処方法は簡単で、内容が変化する変数は関数内に記述すると影響を受けません。

こちらの方法は以下の記事を参考にしております。

個人的にやっていること

• インスタンスオブジェクトの書き方、リージョン名は基本いれない

region_name = "ap-northeast-1"
ec2 = boto3.client('ec2', region_name=region)

上記のようにリージョン名書いたりするのをよく見かけますが、別リージョン使わないならいらないと思いますので、私は記述しません。

• 環境変数は関数の外に置いておく

import boto3

s3 = boto3.client('s3')
bucket_name = os.environ['BUCKET']

def lambda_handler(event, context):
    response = s3.get_object(Bucket=bucket_name, Key=file_name)
    print(response['Body'])

定数となるものは関数の外に置いておきます。
これは単純に、間違えて代入しないようにです。

• 外部ライブラリをなるべく使わない

外部ライブラリを使うとzipでまとめてアップロードしなきゃいけなくなるので、なるべく標準ライブラリで実装します。Lambdaのテストを行うときに、いちいちアップロードするのがめんどくさいだけです。

---

以上、ご参考になれば幸いです。

Thanks.