NATインスタンス起動・停止のSlackスラッシュコマンドを作る

Posted at 2022-04-20

はじめに

2,3年前に作成したスラッシュコマンドの詳細を、やっとこさですがまとめようと思います。
※ですので、一部情報が古い可能性があります。

できること

/xxxnat status →NATインスタンスの状態を確認
/xxxnat start →NATインスタンスを起動
/xxxnat stop →NATインスタンスを停止

※実行結果が設定したSlackチャンネル全員に見える形で投稿されます

作ったもの（大きく分けて２つ）

NATインスタンスがstopped→runnninng/runnninng→stoppedに変わった時に通知するSlackbotを作る
「/xxxnat start」「/xxxnat stop」と設定したチャンネルで発言すると、NATが起動/停止するWebhookを作る（「/xxxnat status」で今の状況も返してくれる）

フロー

1. インスタンスステータス変化時に通知するSlackbot

Amazon EventBridge (CloudWatch Events) で検知
Lambdaで通知内容作成
SlackのIncoming WebHooksに渡る
対象のSlackチャンネルに通知

2. Slackスラッシュコマンドを打つとNATが起動/停止するWebhook

１． SlackのSlash Commands を実行
２． APIGateway 作動
３． Lambda(APIGatewayから呼ばれて処理内容をSQSに渡す)
４． SQSで処理実施
５． Lambda（SQSから呼ばれて結果をSlackに返す）

＜ポイント＞
Slackのスラッシュコマンドが3秒でタイムアウトとなる問題があったため、Lambdaを2つに分け、1つはSQSに処理をお願いする内容に、もう一つはSQSにお願いした内容を受け取ってSlackに返す内容にしました。

作成手順

1. インスタンスステータス変化時に通知するSlackbot

Incoming WebHooksを作成する

Slackの管理画面で、Custom Integrations > Incoming WebHooks からWebhook を作成し、Webhook URLを払い出します。
Integration Settingsに通知を飛ばしたいチャンネルを設定します。
Customize Name・Customize Iconにbotの発言時の名前・アイコンを登録します。

NATインスタンスの状態変化が起こったときにSlackに投稿するLambdaを作成する

Lambda「check_nat_instance_state」を作成する。
ランタイム: Python 3.7
パッケージタイプ: Zip
コード：lambda_function.pyの中身

import json
import urllib.request

def lambda_handler(event, context):

    ec2_status = event['ec2_status']
    post_slack(ec2_status)
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

def post_slack(ec2_status):

    if ec2_status == "running":
        message = "NATインスタンスが起動しました。"
    elif ec2_status == "stopped":
        message = "NATインスタンスが停止しました。"
    elif ec2_status == "stopping":
        message = "NATインスタンス停止中です。"

    send_data = {
        "username": "NAT",
        "icon_emoji": ":vertical_traffic_light:",
        "text": message,
    }
    send_text = "payload=" + json.dumps(send_data)
    request = urllib.request.Request(
        "https://hooks.slack.com/services/XXX/XXXXX", 
        data=send_text.encode("utf-8"), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode("utf-8")

※「https://hooks.slack.com/services/XXX/XXXXX」はSlackの管理画面で発行したWebhook URLを記載。

設定：トリガーには Amazon EventBridge (CloudWatch Events) で作成したルールを３つ紐づける
- Change_NAT_Instance_State_running
- Change_NAT_Instance_State_stopping
- Change_NAT_Instance_State_stopped

3つのルールのイベントパターン内容は以下の通り（以下、runnningの場合。stopping、stoppedも同様に作成）

{
  "source": ["aws.ec2"],
  "detail-type": ["EC2 Instance State-change Notification"],
  "detail": {
    "state": ["running"],
    "instance-id": ["i-XXXXXXXXX"]
  }
}

※「i-XXXXXXXXX」はインスタンスのID

2. Slackスラッシュコマンドを打つとNATが起動/停止するWebhook

SlackのSlash Commandsを作成する

Slackの管理画面で、Custom Integrations > Slash Commands から Slash Commands を作成します。
Integration SettingsのCommand入力欄で設定したいコマンド「/xxxnat」を入力。
URL欄は、後ほどAPIGatewayを作成したあとにAPIエンドポイントのURLを記入。
MethodはPOST
Token欄に表示される内容をメモっておきます。
Customize Nameは「slash-command」としました。
Customize Iconを適当なものにします。
Descriptive Labelにはこのスラッシュコマンドの用途を入れておきます。

Lambda(APIGatewayから呼ばれてSQSに渡す)を作成

Lambda「startstop_nat_instance_from_apigw」を作成する。
ランタイム: Python 3.7
パッケージタイプ: Zip
コード：lambda_function.pyの中身

import boto3
import json
import logging
import os
import urllib

region = 'ap-northeast-1'
instances = ['i-XXXXXXXXX']
expected_token = 'XXXSLASH-CMD-TOKENXXX'
sqs_que_name = 'XXXSQS-QUE-NAMEXXX'

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def respond(err, res=None):
    return {
        'statusCode': '400' if err else '200',
        'body': err.message if err else json.dumps(res),
        'headers': {
            'Content-Type': 'application/json',
        },
    }

def lambda_handler(event, context):
    params = urllib.parse.parse_qs(event['body'])
    token = params['token'][0]
    if token != expected_token:
        logger.error("Request token (%s) does not match expected", token)
        return respond(Exception('Invalid request token'))

    user = params['user_name'][0]
    command = params['command'][0]
    channel = params['channel_name'][0]

    if 'text' in params:
        command_text = params['text'][0]
    else:
        command_text = ''

    sqs = boto3.resource('sqs')
    queue = sqs.get_queue_by_name(QueueName=sqs_que_name)
    message = {
        'user': user,
        'command': command,
        'channel': channel,
        'command_text': command_text
    }

    response = queue.send_message(MessageBody='SlackMessage', MessageAttributes={
        'Message': {
            'StringValue': str(message),
            'DataType': 'String'
        },
    })

    return respond(None, { 'text': "Please wait..."} )

以下は任意内容を入れて下さい。（できればコードの中に直接書かずに外出しするのがベストです）

インスタンス名、i-XXXXXXXXX
SlackスラッシュコマンドToken、XXXSLASH-CMD-TOKENXXX
SQS名、XXXSQS-QUE-NAMEXXX（後から作成するsqs名を入れる）
設定：トリガーには APIGatewayを作成して紐づける

APIGatewayの内容
API名：startstopNatInstance-API
API タイプ: REST
ステージ: default
メソッド: ANY
リソースパス: /startstop_nat_instance_from_apigw
認証: NONE

APIGatewayのAPIエンドポイントは、SlackのSlash CommandsのURL欄に記入する。

Lambda（SQSから呼ばれてSlackに返す）を作成

Lambda「startstop_nat_instance_from_sqs」を作成する。
ランタイム: Python 3.7
パッケージタイプ: Zip
コード：lambda_function.pyの中身

import boto3
import json
import logging
import os
import urllib
import ast

region = 'ap-northeast-1'
instances = ['i-XXXXXXXXX']

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):

    # SQSからのメッセージ受け取り
    que_message = event['Records'][0]['messageAttributes']['Message']['stringValue']
    print(que_message)

    # 文字列を辞書型に変換
    que_message_json = ast.literal_eval(que_message)

    user = que_message_json['user']
    command = que_message_json['command']
    channel = que_message_json['channel']
    if 'command_text' in que_message_json:
        command_text = que_message_json['command_text']
    else:
        command_text = ''

    # EC2操作
    ec2 = boto3.client('ec2', region_name=region)
    addtext = ''

    if command_text == 'stop':
        ec2.stop_instances(InstanceIds=instances)
        msg = 'stopped your instances: ' + ", ".join(instances)
        print(msg)
    elif command_text == 'start':
        ec2.start_instances(InstanceIds=instances)
        msg = 'started your instances: ' + ", ".join(instances)
        print(msg)
    elif command_text == 'status':
        mystatus= ec2.describe_instances(Filters=[{'Name':'instance-id', 'Values':instances}])["Reservations"][0]["Instances"][0]['State']['Name'] 
        msg = 'your instances status is: ' + ", ".join(mystatus)
        print(msg)
        addtext = str(mystatus)
    else:
        addtext = " bad option."

    # Slack投稿する
    post_slack(user, command, command_text, addtext)
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

def post_slack(user, command, command_text, addtext):

    message = user + "が " + command + " " + command_text + "を実行しました。"
    if len(addtext) != 0:
        message += " 結果は: `" + addtext + "`"

    send_data = {
        "username": "NAT",
        "icon_emoji": ":vertical_traffic_light:",
        "text": message,
    }
    send_text = "payload=" + json.dumps(send_data)
    request = urllib.request.Request(
        "https://hooks.slack.com/services/XXXWEBHOOKS-URLXXX", 
        data=send_text.encode("utf-8"), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode("utf-8")

以下は任意内容を入れて下さい。（できればコードの中に直接書かずに外出しするのがベストです）

インスタンス名、i-XXXXXXXXX
Slack Incoming WebHooks URL、https://hooks.slack.com/services/XXXWEBHOOKS-URLXXX
設定：トリガーには SQSを作成して紐づける
名前：任意名称
バッチウインドウ：なし
バッチサイズ：10

動作例

kosugeが /xxxnat startを実行しました。
NATインスタンスが起動しました。

kosugeが /xxxnat stopを実行しました。
NATインスタンス停止中です。
NATインスタンスが停止しました。

kosugeが /xxxnat statusを実行しました。 結果は: running

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up