Slack と Lambda で ChatOps してみる #Python

Slack と AWS Lambda を連携させて、権限の異なるチーム間での作業依頼とその実行を簡略化する ChatOps を考えてみる。

お題

以下のような事前条件と制約があるとする。

AWS 上の RDS で、開発用、ステージング用、本番用のデータベースが運用されている。
一般開発メンバーは開発用とステージング用のみにアクセスできる。
admin チームは本番用を含むすべてのデータベースにアクセスできる。

現状では、運用中の障害対応等で一般開発メンバーが本番DBのデータを調べたい場合に、以下のフローでやり取りしているとする。

指定の Slack チャンネルで、メンション @admin-group で select 文を書き込む
admin チームのメンバーは、本番DBに接続し書き込まれたクエリをコピペして実行する
admin チームのメンバーは、クエリの結果をコピペして「スレッドに返信」する

このフローを以下のように改善して、一手間減らしたい。

指定の Slack チャンネルで、メンション @admin-group で select 文を書き込む
admin チームのメンバーは、クエリに問題がなければ所定の絵文字でリアクションをする
Slack のイベントを購読している Slack App が、リアクションがつけられた書き込みからクエリを抽出する
Slack App はクエリを実行し、スレッドへの返信として結果を書き込む

これを、Python + Lambda + API Gateway + Slack API + RDS で実装してみる。

ステップ１: Lambda 関数作成

Lambda 関数の作成から始める

作成

まず IAM Role を作っておく。

IAM コンソールで Role 画面に遷移して〔 Create Role 〕
Choose the service that will use this role で Lambda を選択し、〔Next: Permission〕
Attach permissions policies で AWSLambdaVPCAccessExecutionRole にチェックを入れて〔Next: Review〕
Role name に slack-ops-test2-role を指定して〔Create Role〕

次に Lambda 関数を作る

Lambda コンソールで〔Create function〕
Author from scratch を選択して、下記仕様で〔Create function〕
- Name: slack-ops-test2-func
- Runtime: Python 3.6
- Role: slack-ops-test2-role
画面上部の Congratulations! を確認

テスト

画面上部の〔select a test event〕で〔Confiture test event〕を選択。適当にテンプレートや名前を入れて〔Create〕
〔Test〕を押下して、正常動作したことを Console 上で確認。
画面上部に表示された〔logs〕リンクで CloudWatch コンソールに遷移し、出力を確認。（もしロールに正しく権限が付与されていないと CloudWatch にログが出力されない。）

ステップ２: API Gateway 作成

API Gateway を作って Lambda 関数を外部から呼べるようにする。

作成

Amazon API Gateway コンソールで、〔Create API〕押下
以下の仕様で Create API 押下
- New API: 選択
- API Name: slack-ops-test2-api
- その他: そのまま
/ が選択された状態で〔 Actions 〕プルダウンから〔 Create Method 〕
/ 直下に出現したプルダウンで〔 Any 〕を選択しを押下
下記の設定で〔 Save 〕押下
- Integration type: Lambda Function Proxy
- Use Lambda Proxy integration: チェックなし
- Lambda Function: slack-ops-test2-func
表示される Add Permission to Lambda Function メッセージボックスで〔 OK 〕押下

テスト

/ 下の ANY が選択された状態で、Client エリアの〔 TEST 〕をクリック
Method プルダウンで GET を選択し〔 Test 〕押下
右側のペインで正常実行を確認

デプロイ

Actions プルダウンで Deploy API を選択し、下記を指定して〔 Deploy 〕押下
- Deployment stage: [New Stage]
- Stage Name: dev
上部に表示される Invoke URL を踏んで、ブラウザ上に結果が正しく表示されることを確認。この URL は、次の Slack 連携で使う。

ステップ３: Slack 連携

Lambda 側の準備

Slack と連携する Web API は、下記の形式のリクエストを受け取った場合、中の challenge を取り出してレスポンスとして返すようにしておく必要がある。

{
  "token": "Jhj5dZrVaK7ZwHHjRyZWjbDl",
  "challenge": "3eZbrw1aBm2rZgRNFdxV2595E9CY3gmdALWMmHkvFXO7tYXAYM8P",
  "type": "url_verification"
}

下のようにLambda関数を書き換える（ソース）。あとで使うので、イベントを出力するログコードも追加しておいた。

import json
import logging

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

def lambda_handler(event, context):
  logger.info(json.dumps(event))
  if ('challenge' in event):
    return {
      "statusCode": 200,
      "body": json.dumps({'challenge': event['challenge']})
    }
  else: 
    return { "statusCode": 200 }

上記の JSON で test event を作成し実行。うまく動くことを確認。

Slack App 作成

Slack API の Your Apps ページに行って〔 Create New App 〕
遷移先で下記を指定し〔 Create App 〕
- App Name: slack-ops-test2-app
- Workspace: { インストール先のワークスペース }
Add features and functionality から〔 Event Subscriptions 〕押下
Enable Events を On にして、Request URL に dev ステージの URLを指定
Verified が表示されるのを確認（もしchallengeに対応できていなかったら、ここでエラーとなる）。
Subscribe to Workspace Events 欄に下がって、reaction_added を追加し、〔 Save Changes 〕押下
画面左の Features 下のリンクから、OAuth & Permissions ページに遷移し、Scopes 欄で reactions:read を追加

ここまでで Slack と Lambda がつながったはず。以下のように確認。

slack の対象ワークスペースに戻って何か投稿する
投稿したものに適当な絵文字でリアクションする。
CloudWatch のロググループ /aws/lambda/slack-ops-test2-func の出力を見て、上手く動作していることを確認。

reaction_added イベントでは、こんな仕様のJSONデータが渡される。event/item/channel、event/item/ts、event/reactionあたりをあとで使うので、これを取り出してみる。

以下のようなメソッドを追加して、呼び出しコードを追加する。

def process_event(event):
  item     = event['item']
  channel  = item['channel']
  ts       = item['ts']
  reaction = event['reaction']
  logger.info('{}, {}, {}'.format(channel, ts, reaction))
  # TODO

もう一度、なんかの emoji でリアクションしてみてログを確認すると、リアクションが付いた書き込みのチャンネル、タイムスタンプ、絵文字の種類が出力されている。

ここまでのコード

ステップ４: Slack の内容を取得する

Slack API の channels.history メソッドを使って、リアクションがつけられた書き込みを取得する。

手順

Slack API の OAuth & Permissions で、Sopes に channels:history を追加
画面上部に表示される〔 click here 〕リンクをクリックし、遷移先で許可
OAuth Access Token を〔 Copy 〕する → (1)
Lambda コンソールに戻り、Environment variable にキー=TOKEN、値=(1) の組を追加
下記のようなメソッドと、これにchannel と ts を渡すコードを追加

def retrieve_item_text(channel, ts):
  url    = 'https://slack.com/api/channels.history'
  params = {
    'token':     token,
    'channel':   channel,
    'count':     1,
    'inclusive': True,
    'latest':    ts
  }
  requestUrl = '{}?{}'.format(url, parse.urlencode(params))
  req        = request.Request(requestUrl)

  with request.urlopen(req) as res:
    body = res.read().decode('utf-8')

  return json.loads(body)['messages'][0]['text']

Slack でリアクションすると、CloudWatch に本文が出力されているのを確認できる。

ここまでのコード

ステップ５: Lambda から「スレッドに返信」する

このステップではリアクションがつけられたメッセージに返信してみる。

手順

Slack API の OAuth & Permissions に戻り、Sopes に chat:write:bot を追加。reinstall が求められるので従う。
下記メソッドと呼び出しコードを追加する( ソース)

def post_result(channel, thread_ts, text):
  url    = 'https://slack.com/api/chat.postMessage'
  params = {
    'token':     token,
    'channel':   channel,
    'text':      text,
    'thread_ts': thread_ts
  }
  requestUrl = '{}?{}'.format(url, parse.urlencode(params))
  req = request.Request(requestUrl)

  with request.urlopen(req) as response:
    response_body = response.read().decode("utf-8")
  logger.info(str(response_body))

Slack でリアクションすると、今度はログだけじゃなく「スレッドに返信」される。ここでは単にオウム返ししているだけだが、あとで SQLを抽出し、RDS で実行して結果を返すコードに変更する。

ステップ６: RDS に接続する

このステップでは RDSへの接続だけやってみる。若干手順が多い。

手順

ローカルに適当に作業用フォルダを作って、現在のソースで lambda_function.py を作成。
右のようにして pymysql をインストール: pip3 install pymysql -t ./
右のようにして zip を作る: zip -r app.zip lambda_function.py pymysql cryptography
Lambda コンソールに戻り、Code entry type で Upload a .ZIP file を選択し、app.zip をアップロード。〔 Save 〕して、lambda_function.py とともに pymysql フォルダと cryptography フォルダが上がっているのを確認。
Environment variables に、接続用のパラメータ、RDS_HOST、NAME、PASSWORD、DB_NAME を追加する。
コード冒頭あたりに以下を追加する(ソース)。

import pymysql
...
rds_host = os.environ['RDS_HOST']
name     = os.environ['NAME']
password = os.environ['PASSWORD']
db_name  = os.environ['DB_NAME']
...
try:
    logger.info("connecting: {}, {}, {}, {}".format(rds_host, name, password, db_name))
    conn = pymysql.connect(rds_host, user=name, passwd=password, db=db_name, connect_timeout=20)
    logger.info("connected")
except:
    logger.error("ERROR: Unexpected error")
    sys.exit()

VPC にデータベースと同じ VPC を指定する。
Subnet にデータベースの Subnet Group を構成してるものを（複数）指定する
VPC を指定すると、そのままでは外部にある Slack Webhook にアクセスできなくなるため、Security Group に outbound コネクションが許可されているものを指定する。生成する場合は VPC 以外デフォルトのままで可。

Slack から試行して CloudWatch で RDS 接続まわりのログと、ステップ６までの動きが変わらない事を確認。
※ VPC は NAT を持っている必要がある。もし、 Security Group、Route Table、NAT Gateway が上手く関連付けられていなければ、リアクションがつけられたメッセージを取得しに行くところでタイムアウトになる。

ステップ７: クエリを実行する

上で接続設定した DB に下のようなテーブルがあるとして、Lambda からクエリを投げてみる。

create table Employee3 (
  EmpID  int NOT NULL,
  Name varchar(255) NOT NULL,
  PRIMARY KEY (EmpID)
)

手順

Slack の書き込みからクエリを抽出するメソッドを追加する

def extract_query(text):
  r = re.compile(".*```(.*)```.*", re.MULTILINE | re.DOTALL)
  return r.search(text).group(1).strip().replace('\n', ' ')

抽出したクエリを実行するメソッドを追加する

def exec_query(query):
  lines = []
  with conn.cursor() as cur:
      cur.execute(query)
      for row in cur:
          lines.append(str(row))
  return "\n".join(lines)

これらが連携するようにコードを書き換える（ソース）

確認

Slack に例えば以下のように書き込んで、

@admin-group おねがいしますー
```
select empid, name from Employee3
```

適当な絵文字でリアクションすると、結果がスレッドに返信される。

TODO

Slack、Lambda、RDS をつなげるのが主旨なので、単純な Python コーディングだけの要素や、使用感の追求などはいろいろ省略した。以下のような TODOが残っている。

クエリ実行を許可できるユーザの制限と絵文字の種類の設定。これは Lambda の環境変数で指定し、関数内でバリデートするように実装できる。
エラー処理。書き込みに select 文がない場合や、select 文が不正でクエリ実行が失敗した場合などのハンドリング。
クエリ実行結果の書式や、件数が多すぎる場合の制限やページングなど。
bot ベースの対話的な UX の検討や、ボタンを使った UI の提供など。
RDS のアクセス権の設定などで、更新系のクエリが誤って実行されないようにする工夫。