🤔

Serverless Framework で作成した Lambda の関数名を直書きしないためにちょっとしたツールを作った

2022/12/29に公開

AWS

個人的にServerless Framework を運用する際にちょっとした面倒ポイントになっていた「Lambdaの関数名を直書きしなくてはいけない」問題を解決するために、ちょっとしたツールを開発した話をまとめました。もし共感する人がいたら、ちょっとずつでも育てていこうかと思っており、Contribution もらえると嬉しいです！（歓迎します）

どのような問題を解決したかったのか

サーバーレスアプリケーションを構築する際にはさまざまな選択肢があります。例えば、Serverless Framework^[1] や AWS Serverless Application Model^[2]、Chalice^[3] などなど。どれも便利なのですが、アプリケーションが少しずつ大きくなってくると「困ったな」と感じる点が出てくるのも事実です。

この困ったポイントが明らかに「やばい」レベルであれば、アーキテクチャを設計し直したりすれば良いのですが、「まぁ、努力と注意でなんとかなるか」と思えるレベルだと、放置しがちで運用時に気を遣う点が増えていく問題をかねてから感じていました。そのうち、今回はたった１つの問題だけを解決するツールを作りました。

LambdaからLambda関数を呼び出す時に関数名を直書きしてしまう問題

LambdaをAppSyncやAPIGateway の後段に配置するアーキテクチャを構成していくと、「非同期で別のロジックを動かしたい」と思うシーンが増えてきます。

例えば、APIGatewayの場合はタイムアウトが30秒、AppSyncのGraphQLエンドポイントでもタイムアウトは60秒です。タイムアウト制約を満たすために、リクエストを受け付けた後にレスポンスを即座に返しつつ、裏側で非同期のワーカーに当たるものを動かす場合があります。シーケンス図にすると、以下のような流れになります。

この時、Workerサービスを作る方法はさまざまありますが、今回、Lambdaで別のLambdaを非同期で呼び出す際に抱えていた「Lambda関数の名前を直書きする」問題を解消してみることにしました。具体的にはこんな問題です。

import boto3 

def handler(event: dict, context: dict) -> dict: 
    process_something(event) 
   
    client = boto3.client("lambda")
    response = client.invoke(
      FunctionName='sample-service-function-name',  # ここが直書きになる
      InvocationType='Event',
    )
   
   if response['StatusCode'] > 400: 
       return {"statusCode": 500}
   return {"statusCode": 200}

「そんなの気をつければいいだけだし、設計上手くやればもっと楽になるでしょ」と思った人。間違っていません。例えば、SQSをサーバーレスアプリケーション外で作成しておいて環境変数にARNを注入しておくとか、やりようは本当にたくさんあります。しかしまあ、さまざまな歴史的経緯で直接読んでいる場合にそんな正論を言っても仕方なく、上手くミスしないような運用を考える方が建設的です。

Enum でとりあえず直書きを避ける

とはいえ、関数の数が増えてくると困ったことにFaaS特有の「関数が増えすぎて関数の把握が難しい」問題が表出してきます。単純に関数名を覚えていられなくなります。しかも、こういう類のLambda関数は、ある特定のシーンでしか使わないことが多く、一層忘却の彼方に行ってしまいます。

そんなこんなで、関数名を直接書かない方が安全な場合があるのです。今回作成したツールでは、Enumに関数名を詰め込むことにして、直接関数名を扱わないようにしてみました。

作成されるEnumを用いると、以下のようになります。
（*lambda_handlers.py に Enum クラスが入っているとします）

import boto3 
from lambda_handlers import LambdaHandlers

def handler(event: dict, context: dict) -> dict: 
    process_something(event) 
   
    client = boto3.client("lambda")
    response = client.invoke(
      FunctionName=LambdaHandlers.sample_service_function_name.name,  # ここが直書きになる
      InvocationType='Event',
    )
   
   if response['StatusCode'] > 400: 
       return {"statusCode": 500}
   return {"statusCode": 200}

作成したツールの紹介

さて、ツールの紹介です。

今回、Python製のCLIツール、ptmpdyを作成しました。（読み方は決まってません）

https://pypi.org/project/ptmpdy/

PyPI^[4] にも公開してあるため、お手持ちのパッケージマネージャー(pip, pipenv, poetry など）で、 ptmpdy を指定していただくとインストールできる状態になっています。基本的には、CLIツールとしての用途のみを考えていますため、よほどのことがない限りは開発用ライブラリとしてインストールすれば事足ります。例えば、Poetry でインストールする場合は以下のようにコマンドを打てばOKです。

poetry add -D ptmpdy

使い方

インストールした後に、

ptmpdy <serverless.ymlのパス> <生成するファイルのファイルパス>

とすれば、serverless.yml に記載されている functions の情報を用いて Lambda の関数名を参照できる Enum クラスが作成されるようになっています。と言っても、大したことはしておらず、ちょっとしたテンプレートに従うように、 serverless.yml のデータを解析し、ファイルに書き出すだけです。以下に例を出しておきます。

from enum import Enum


class LambdaHandlers(str, Enum):
    sample_function1 = "sample-service-sample-function1"
    sample_function2 = "sample-service-sample-function2"
    sample_function3 = "sample-service-sample-function3"

今後

現在、このツールができることはとても限定的であまり多くのシーンを満たせているとは思いません。例えば、目下できていないこととして以下のようなことが挙げられます。

stage などで環境ごとに関数名を変えている場合の対応
Enum の反映漏れを検知するための diff メソッド
TypeScript など Python以外の言語への対応
Serverless Framework の Plugin としての提供

ま、時間がある時に困っていることベースに機能を追加していきつつ、育てていく予定です。

おわりに

今回、CLIツールをTyperベースに作成しました。作成したツールの検証から配布まで一連を丁寧に解説したページがあり、とてもスムーズに進めることができました。（一連の開発の公開からこの記事完成までで、1日もかかっていません）

脚注