おぎろぐはてブロ

なんだかんだエンジニアになって10年以上

AWS SAM + Slack Bolt for PythonでSlack botをつくる

書いたこと

  • Lazy Listenerを利用し、ackが必要なイベントかつ3秒以上時間がかかる処理を実現する
  • Lambda Function URLを利用してHTTP Endpointをつくる

Slack Bolt for PythonでSlack botをつくる

このチュートリアルを参照しつつ、つくります。割愛。

slack.dev

ある程度、チュートリアルに沿って、ローカルで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_somethingstep_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 ALLLATEST 以外のバージョンも取得する
  • 見つかったら、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 の中の KeyName である要素をフィルタしてきて、これは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アドレス順に並んでくれる。

fishでAWS Organizations配下のアカウントに同じコマンドを投げるスクリプト

環境調査のためにOrganization配下のAWSアカウントすべてにコマンド実行したいということがあったりする。fishを使っているので、関数を作って使っているのでメモ。

~/.config/fish/functions/ 配下に maws.fish としてつくる

function maws
  set accounts (aws organizations list-accounts --query 'Accounts[].Id' --output text | string match -ra '\d+')
  echo "call aws command in these accounts: $accounts"

  for account in $accounts
    echo "[$account]"

    set assumerole (aws sts assume-role --role-arn arn:aws:iam::$account:role/OrganizationAccountAccessRole --role-session-name (whoami))

    if test $status -eq 0
      set -x AWS_ACCESS_KEY_ID (echo $assumerole | jq -r '.Credentials.AccessKeyId')
      set -x AWS_SECRET_ACCESS_KEY (echo $assumerole | jq -r '.Credentials.SecretAccessKey')
      set -x AWS_SESSION_TOKEN (echo $assumerole | jq -r '.Credentials.SessionToken')

      aws $argv
    end

    echo ""

    set -e AWS_ACCESS_KEY_ID
    set -e AWS_SECRET_ACCESS_KEY
    set -e AWS_SESSION_TOKEN
  end

end

一旦、 ふつうのaws cli

aws ec2 describe-vpcs --query Vpcs[].[OwnerId,VpcId,CidrBlock] --output text

といった感じで単一アカウントで確認してOKだったら

maws ec2 describe-vpcs --query Vpcs[].[OwnerId,VpcId,CidrBlock] --output text

とするだけ。

動作の補足

環境変数に認証情報が存在すると、優先順位として強いので、Organizationの親アカウントの認証情報をもって実行し、Assume Roleして、環境変数をセット、削除するという流れ

jq 使ったら負けだよなと思いつつ、手抜き。