AWS SAM + Slack Bolt for PythonでSlack botをつくる
書いたこと
- Lazy Listenerを利用し、ackが必要なイベントかつ3秒以上時間がかかる処理を実現する
- Lambda Function URLを利用してHTTP Endpointをつくる
Slack Bolt for PythonでSlack botをつくる
このチュートリアルを参照しつつ、つくります。割愛。
ある程度、チュートリアルに沿って、ローカルでSocket Mode有効でつくってから、Lambdaで動かすように次のLazy Listernerを有効にするように自分はしています。
Lazy Listenerの利用
Boltで受けられるイベントは各種ありますが、アクション(action)、コマンド(command)、ショートカット(shortcut)、オプション(options)、およびモーダルからのデータ送信(view_submission)を処理する場合は、Slack側からのリクエストに対して3秒以内に ack()
を返す必要があります。
ただ、Lambdaで動かす場合、HTTPレスポンスを返却したタイミングでプロセスが終了されるため、ackを返したあとに処理を継続させることができません。この対応として、Slack Bolt for Pythonでは、Lazy Listenerという機能が提供されています。
これは、HTTPでSlackからリクエストを受け取ったLambdaプロセスが、時間のかかる処理を実行するLambdaを非同期で呼び出して、ackレスポンスを返却することで実現されています。
例にあるように、ack
と、 lazy
の引数にそれぞれack処理と、時間のかかる処理を分けて書けます。lazyには複数の関数を渡せ、それぞれ並列に実行されます。
lazyに1つ処理を指定した場合、1つのイベントに対してack処理とlazy処理でLambdaが2度invokeされることになります。上に書いてあるackが必要なイベント以外をハンドルする場合はackが不要なので、その場合Lazy Listenerを使うと無駄にinvokeの回数が増えるので、この場合はLazy Listenerを有効にしないほうがよいです。
以下は、行頭にHelloもしくはhelloという文字列があったときにHello!を発言する場合です。 (messageイベントはackが不要なのでlazyを使う必要はないです)
import logging import re from slack_bolt import App, Say from slack_bolt.adapter.aws_lambda import SlackRequestHandler SlackRequestHandler.clear_all_log_handlers() logging.basicConfig(level=logging.INFO) app = App(process_before_response=True) def handle_message(message, say: Say): print(message) say("Hello!") rule = re.compile("^[Hh]ello") app.message(rule)( ack=lambda ack: ack(), lazy=[handle_message] ) def handler(event, context): print("invoked", event) slack_handler = SlackRequestHandler(app=app) return slack_handler.handle(event, context)
とにかくackだけ返したければ
ack=lambda ack: ack(),
と書けばOK
呼び出しをCloudWatch Metricsにputする
Slack botを作った場合、Slackからの呼び出しは同じLambda Functionが呼び出されるだけなので、複数のイベントをハンドリングする場合、どのイベントがどのくらい呼び出されているのかを調べるのが困難です。このときCloudWatch metricsにメトリックを記録すると便利です。
以下のように定義して、handle_shortcut
ではackしつつmodalを開いたりし、do_something
では時間のかかる処理を行い、step_handle_shortcut
ではCloudWatch metricsにメトリックを記録します。
do_something
と step_handle_shortcut
は並列にinvokeされます。
app.shortcut("shortcut")(
ack=handle_shortcut,
lazy=[do_something, metric.step_handle_shortcut]
)
metric.py
では
import boto3 import os import datetime def put(step: str): client = boto3.client("cloudwatch") env = os.getenv("ENV") # 別途定義する response = client.put_metric_data( Namespace=f"FooBarSlackApp/{env}", MetricData=[ { "MetricName": "Request", "Dimensions": [ { "Name": "Step", "Value": step }, ], "Timestamp": datetime.datetime.utcnow(), "Value": 1, }, ] ) print(response) def step_handle_shortcut(): put("handle_shortcut")
SAMでデプロイ
最低限必要なのは Lambda Function と、HTTP EndpointとしてのLambda Function URLです。 HTTP EndpointとしてAPI Gateway (v1, v2) を利用している例があったりしますが、特に高度な定義も必要なければ、Lambda Function URLを使用するほうが、定義が簡単で、かつ、利用料金もLambda側に含まれるので安価になります。(大抵はまぁ誤差みたいな金額だと思いますが)
src/
配下にコードを配置した場合、以下のようなコードで実現できます。
AWSTemplateFormatVersion: "2010-09-09" Transform: AWS::Serverless-2016-10-31 Description: Slack bot sample Resources: HandlerFunction: Type: AWS::Serverless::Function Properties: CodeUri: src/ Handler: app.handler Runtime: python3.11 Timeout: 30 Policies: - AWSLambdaRole FunctionUrlConfig: AuthType: NONE Outputs: HandlerFunctionUrl: Value: !GetAtt HandlerFunctionUrl.FunctionUrl
FunctionUrlConfig
を定義してやると、Lambda Function URLが有効になります。SlackからのリクエストにIAM認証を利用することはできないので、AuthType: NONE
になります。(bolt内でSlackの特定のアプリから来た正当なリクエストか signing_secret
を利用して検証しています)
Slack appの Interactivity eventsのURL、Slash commandのURL、Event SubscriptionsのURL (それぞれ有効にするかは使うイベント次第) にこのLambda Function URLのURLを指定してやる必要があるので、Outputs
で出力をしています。
SAMでリソースが勝手に作られて、その名前が Lambda Function名に Url
を足したものになるので、Lambda Functionが HandlerFunction
の場合、 HandlerFunctionUrl
になります。
AWS CLIで存在するNode.js 10.xのLambda関数を一覧する
[要対応] AWS Lambda における Node.js 10 のサポート終了 | [Action Required] AWS Lambda end of support for Node.js 10 のメール来てるけど、どこにあるんだとおもったら
tl;dr
以下を実行
REGIONS=`aws ec2 describe-regions --query 'Regions[*].RegionName' --output text` for region in $REGIONS; do aws lambda list-functions \ --function-version ALL \ --region $region \ --query "Functions[?Runtime=='nodejs10.x']" --no-cli-pager done
- region一覧とって回そう
--function-version ALL
でLATEST
以外のバージョンも取得する- 見つかったら、CloudWatch MetricでInvocationされているかを確認して、使われてなかったら削除しよう
AWS Configでリソースに紐付かないセキュリティグループを抽出
tl;dr
- 使っていないセキュリティグループを列挙したい
- セキュリティグループはENIに対して割り当て、ENIがEC2などのリソースにアタッチされている。使っていないセキュリティグループ = どのENIにも割り当てられていないセキュリティグループ、なのだが、取得するのはセキュリティグループを列挙し、ENIを列挙して、という作業が必要でEC2 API叩いて処理するの地味に面倒
- AWS Configを有効にしているなら、クエリでいい感じにとれるし、リージョンもまたげるし、Aggregatorを使っているならマルチアカウントも一発
コード
import json import boto3 client = boto3.client('config') query = "SELECT resourceId, awsRegion, resourceName, " \ "configuration.description, configuration.ipPermissions, " \ "configuration.ipPermissionsEgress, relationships " \ "WHERE resourceType = 'AWS::EC2::SecurityGroup'" results = [] response = client.select_resource_config(Expression=query, Limit=100) results.extend(response['Results']) while True: if 'NextToken' in response: response = client.select_resource_config(Expression=query, NextToken=response['NextToken'], Limit=100) results.extend(response['Results']) else: break for result in results: result = json.loads(result) for relationship in result['relationships']: if relationship['resourceType'] == 'AWS::EC2::NetworkInterface': #eni_id = relationship['resourceId'] break else: # 関連するENIが無い = リソースに関連付けられていないセキュリティグループ print('\t'.join( [result['awsRegion'], result['resourceId'], result['resourceName'], result['configuration']['description']]))
マルチアカウント版コード
select_aggregate_resource_config()
に差し替えるだけで、AWS Config Aggregatorに対してクエリを実行できます。便利。
import json import boto3 client = boto3.client('config') query = "SELECT accountId, resourceId, awsRegion, resourceName, " \ "configuration.description, configuration.ipPermissions, " \ "configuration.ipPermissionsEgress, relationships " \ "WHERE resourceType = 'AWS::EC2::SecurityGroup'" results = [] response = client.select_aggregate_resource_config(Expression=query, Limit=100, ConfigurationAggregatorName='<aggregator name>') results.extend(response['Results']) while True: if 'NextToken' in response: response = client.select_aggregate_resource_config(Expression=query, NextToken=response['NextToken'], Limit=100, ConfigurationAggregatorName='<aggregator name>') results.extend(response['Results']) else: break for result in results: result = json.loads(result) for relationship in result['relationships']: if relationship['resourceType'] == 'AWS::EC2::NetworkInterface': #eni_id = relationship['resourceId'] break else: # 関連するENIが無い = リソースに関連付けられていないセキュリティグループ print('\t'.join( ['"' + result['accountId'] + '"', result['awsRegion'], result['resourceId'], result['resourceName'], result['configuration']['description']]))
VPCのサブネットを抽出して、きれいに並べたい
tl;dr
aws ec2 describe-subnets --filter Name=vpc-id,Values=vpc-123456 \ --query "Subnets[*].[CidrBlock,(Tags[?Key=='Name'].Value)[0]]" \ --output text | sort -V
10.0.0.0/22 public-subnet-ap-northeast-1a 10.0.4.0/22 public-subnet-ap-northeast-1c 10.0.8.0/22 public-subnet-ap-northeast-1d 10.0.12.0/22 private-subnet-ap-northeast-1a 10.0.16.0/22 private-subnet-ap-northeast-1c 10.0.20.0/22 private-subnet-ap-northeast-1d
aws ec2 describe-subnets
タグを取り出すのがちょっとだるい
{ "Subnets": [ { "AvailabilityZone": "ap-northeast-1d", "AvailabilityZoneId": "apne1-az2", "AvailableIpAddressCount": 1019, "CidrBlock": "10.0.20.0/22", "State": "available", "SubnetId": "subnet-...", "VpcId": "vpc-....", "Tags": [ { "Key": "Name", "Value": "private-subnet-ap-northeast-1d" } ], }, ...
今回は指定したVPC内のサブネットと、その名前を取り出したかったので、CidrBlockと、Tags
の中の Key
が Name
である要素をフィルタしてきて、これは1つなので (正確には1つかゼロ)、リストの先頭をとってくる。
--query "Subnets[*].[CidrBlock,(Tags[?Key=='Name'].Value)[0]]"
sort -V
queryの中で sort_by を使ってソートも (きっと) できるけれど、さすがに辛いので、sort
を使う。
-V
は natural sort of (version) numbers within text ということで、ドットで区切られたバージョン番号をソートするオプションで、バージョン番号ではないけれど、ドットの後の数字で並べてくれるので、これでIPアドレス順に並んでくれる。