AWS SQS + Lambda を利用した Slack 通知アプリケーションを Scala で実装して Serverless Framework でデプロイする

社内ブログを書いたのでその転載です。

[AWS SQS + Lambda を利用した Slack 通知アプリケーションを Scala で実装して Serverless Framework でデプロイする](http://labs.septeni.co.jp/entry/2018/08/20/210538)

表題の通り AWS のサービスを活用した Slack 通知アプリケーションを
開発する機会がありましたのでアウトプットです。

尚、今回は[イベントソース](https://dev.classmethod.jp/etc/aws-lambda-support-sqs-event-source/)について詳細には触れませんのでご了承ください。

こちらについてはまたの機会に。

# アプリケーション概要

1. SQS に溜めたメッセージを Lambda がポーリング
2. 読み込んだメッセージを Slack に POST
3. 通知が成功したメッセージを SQS から削除

![Image from iOS.jpg](https://qiita-image-store.s3.amazonaws.com/0/142263/604b92e2-bc5e-ed97-84af-8b5e54724683.jpeg)


本ブログにおいてはコアの機能となっている、
メッセージの読み込み、通知が成功した場合にメッセージを削除する機能について解説していきます。

# 開発の準備

### Serverless Framework インストール

タイトルにある通り、Lambdaのデプロイ及び動作確認には、
[Serverless Framework](https://serverless.com/) を利用しました。

インストールは [Quick Start#Pre-requisites](https://serverless.com/framework/docs/providers/aws/guide/quick-start#pre-requisites) から

YAMLファイルをちょこっと書くだけで、
アプリケーション構成を定義できるので大変便利でした。

今回はAWS Lambdaの構成管理に利用しましたが、
他にもGCP, Azureなど様々なプラットフォームに対応しているところも魅力的です。

### 環境変数 AWS_PROFILE
SQS, Lambda, S3, CloudFormation への更新が必要なので、
対象の credential に適切なポリシーを設定してください。

### サービスの作成

インストールした Serverless を利用してコマンドラインからサービスを作成していきます。

今回は Lambda を Scala で実装するので、
テンプレートには `aws-scala-sbt` を指定します。


```bash
$ sls create --template aws-scala-sbt
```

結果は下記の通り。

```bash
├── build.sbt
├── project
│   ├── assembly.sbt
│   ├── build.properties
│   └── plugins.sbt
├── serverless.yml
└── src
    └── main
        └── scala
            └── hello
                ├── ApiGatewayResponse.scala
                ├── Handler.scala
                ├── Request.scala
                └── Response.scala
```

### build.sbt 修正

libraryDependencies に必要なライブラリを追加していきます。

```scala
libraryDependencies ++= Seq(
  "com.amazonaws"  % "aws-lambda-java-events" % "1.3.0",
  "com.amazonaws"  % "aws-lambda-java-core"   % "1.1.0",

  /* 下記を追加 */

  "com.eed3si9n" %% "gigahorse-okhttp" % "0.3.0",
  "io.circe" %% "circe-core" % "0.9.3",
  "io.circe" %% "circe-generic" % "0.9.3",
  "io.circe" %% "circe-parser" % "0.9.3",
  "com.typesafe" % "config" % "1.3.2",
  "com.amazonaws" % "aws-java-sdk-sqs" % "1.11.386",
  "com.amazonaws" % "aws-java-sdk-core" % "1.11.386",
  "com.amazonaws" % "jmespath-java" % "1.11.386"
)
```

用途としては下記の通りです。

- aws-java-sdk-sqs
   - SQSからのメッセージ読み込みと削除処理の実装
- aws-java-sdk-core, jmespath-java
   - aws-java-sdk-sqs のコンパイル依存
- gigahorse-okhttp
   - Slack Web Api の Http Client に利用
- circe-core, circe-generic, circe-parser
   - メッセージ の Decode 処理
- config
   - application.conf 読み込み

# Lambda関数を開発していく

Lambda関数を実装していきます。
まずは、`src / main / scala` ディレクトリの配下に適当なディレクトリを用意して実装していきます。

今回は `sample` ディレクトリでいきます。

### Amazon SQS メッセージの受信・削除を行うクラス

`aws-java-sdk-sqs` を利用して
メッセージの受信・削除を行うクラスを作成します。

基本的にはキューのURLを指定して、
行いたいオペレーションのメソッドを定義するだけなので実装はシンプルです。

```scala
package sample

import com.amazonaws.services.sqs.model.{Message, ReceiveMessageRequest}
import com.amazonaws.services.sqs.{AmazonSQS, AmazonSQSClientBuilder}
import com.typesafe.config.Config

import scala.collection.JavaConverters._
import scala.concurrent.{ExecutionContext, Future}

class SQSClient(config: Config) {

  private val queueUrl = config.getString("aws.sqs.queue.url")
  private val sqsClient: AmazonSQS = AmazonSQSClientBuilder.defaultClient()
  private val receiveRequest: ReceiveMessageRequest = {
    val request = new ReceiveMessageRequest
    request.setMaxNumberOfMessages(10)
    request.setQueueUrl(queueUrl)
    request
  }

  def receive(implicit ec: ExecutionContext): Future[Seq[Message]] =
    Future {
      sqsClient.receiveMessage(receiveRequest).getMessages.asScala
    }

  def delete(msg: Message)(implicit ec: ExecutionContext): Future[Unit] =
    Future{
      sqsClient.deleteMessage(queueUrl, msg.getReceiptHandle)
    }

}

```

各オペレーションのリクエストに対し細かいオプションを設定することができるので、
今回は受信時のリクエストに最大受信数を設定することにしました。

詳しくは [ReceiveMessageRequest (AWS SDK for Java - 1.11.391)](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/sqs/model/ReceiveMessageRequest.html) 参照。

### Slack通知を行うクラス

Slack API の [chat.postMessage](https://api.slack.com/methods/chat.postMessage) に対してPOSTリクエストするクラスです。
Httpクライアントには [Gigahorse](http://eed3si9n.com/gigahorse/) を利用しました。


```scala
package sample

import com.amazonaws.services.sqs.model.Message
import com.typesafe.config.Config
import gigahorse.support.okhttp.Gigahorse
import gigahorse.{HeaderNames, MimeTypes, Request}

import scala.concurrent.{ExecutionContext, Future}

class SlackClient(config: Config) {

  private val Url = config.getString("slack.api.post.message.url")
  private val ApiToken = config.getString("slack.api.token")
  private val requestWithBody: String => Request = { body: String =>
    Gigahorse
      .url(Url)
      .post(body)
      .addHeaders(
        HeaderNames.AUTHORIZATION -> s"Bearer $ApiToken",
        HeaderNames.CONTENT_TYPE -> MimeTypes.JSON
      )
  }

  def post(message: Message)(implicit ec: ExecutionContext): Future[Unit] =
    Gigahorse.withHttp(Gigahorse.config) { http =>
      http
        .run(requestWithBody(message.getBody), Gigahorse.asString)
        .flatMap { s =>
          SlackAPIResponse
            .fromJsString(s)
            .fold(e => Future.failed(e), _ => Future.successful(()))
        }
    }
}

```

今回は割愛してますが、 [Gigahorse — Extending Gigahorse](http://eed3si9n.com/gigahorse/plugin.html) で紹介されている
OAuth認証処理をラップしたクラスの実装例など非常に参考になりました。

### Slack APIレスポンスの処理

Slack APIのレスポンスは通知が失敗した場合でもステータスコード200で結果を返してくるため、
下記のようなモデルにパースして、失敗していた場合は独自例外を返すようしました。

```scala
package sample

import io.circe.Decoder
import io.circe.parser.decode

case class SlackAPIResponse(ok: Boolean, error: Option[String])

object SlackAPIResponse {

  class SlackPostMessageError(
      message: String,
      cause: Option[Throwable] = None
  ) extends Exception(message, cause.orNull) {
    def this(message: String, cause: Throwable) = this(message, Some(cause))
  }

  object SlackPostMessageError {
    def apply(stringOpt: Option[String]): SlackPostMessageError =
      stringOpt
        .map(msg => new SlackPostMessageError(msg))
        .getOrElse(new SlackPostMessageError("There is no errors"))
  }

  implicit val decoder: Decoder[SlackAPIResponse] =
    Decoder.forProduct2("ok", "error")(SlackAPIResponse.apply)

  def fromJsString(
      jsonString: String): Either[SlackPostMessageError, SlackAPIResponse] =
    decode[SlackAPIResponse](jsonString) match {
      case Right(response @ SlackAPIResponse(true, _)) =>
        Right(response)
      case Right(_ @SlackAPIResponse(false, msgOpt)) =>
        Left(SlackPostMessageError(msgOpt))
      case Left(e) =>
        Left(new SlackPostMessageError(e.getMessage, e))
    }
}

```

[circe](https://circe.github.io/circe/) が非常に便利です。


### Handlerの実装

AWS Lambda から呼び出される関数を定義します。
先に解説した SQSClient, SlackClient を駆動してコアとなる機能を実装していきます。


```scala
package sample

import com.amazonaws.services.lambda.runtime.{Context, RequestHandler}
import com.typesafe.config.ConfigFactory
import sample.SampleHandler.{SampleRequest, SampleResponse}

import scala.beans.BeanProperty
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.util.control.NonFatal

object SampleHandler {

  class SampleRequest(@BeanProperty var value: String) {
    def this() = this("")
  }
  case class SampleResponse(
      @BeanProperty value: String = "slack message post done.")
}

class SampleHandler extends RequestHandler[SampleRequest, SampleResponse] {

  import scala.concurrent.ExecutionContext.Implicits.global

  private val config = ConfigFactory.load()
  private val sqsClient = new SQSClient(config)
  private val slackClient = new SlackClient(config)

  def handleRequest(input: SampleRequest, context: Context): SampleResponse = {
    val eventualUnit = for {
      messages <- sqsClient.receive
      eventualMessages = messages.map(m => slackClient.post(m).map(_ => m))
      eventualUnits = eventualMessages.map(_.flatMap(sqsClient.delete))
      result = eventualUnits.map(_.recover(printStackTraceInCaseOfFailure()))
      _ <- Future.sequence(result)
    } yield SampleResponse()

    Await.result(eventualUnit, 300.second)
  }

  private def printStackTraceInCaseOfFailure(): PartialFunction[Throwable, Unit] = {
    case NonFatal(e) => e.printStackTrace()
  }
}
```

`Await.result` の第二引数( `atMost` ) に設定しているのは Lambda の最大実行時間です。

Slack通知が失敗したメッセージの削除はスキップしてひとまずスタックトレースを吐くことにしました。
キューに残ったメッセージの後始末は最大受信数を設定して [デッドレターキュー](https://docs.aws.amazon.com/ja_jp/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-dead-letter-queues.html) に投げてしまうのが良さそうですね。


### application.conf

`src / main / resources` に `application.conf` を作成します。
キューのURLとSlackのAPIトークンは環境変数から読み込むことにました。

```conf
aws.sqs.queue.url=${QUEUE_URL}
slack.api.token=${SLACK_API_TOKEN}
slack.api.post.message.url="https://slack.com/api/chat.postMessage"
```

### serverless.yml

Lambda関数やSQSの構成を定義していきます。

```yaml
service: sample-slack-post

frameworkVersion: "=1.30.0"

provider:
  name: aws
  runtime: java8
  stage: ${opt:stage, 'dev'}
  region: ap-northeast-1
  iamRoleStatements:
    -  Effect: "Allow"
       Action:
       - "sqs:*"
       Resource:
         Fn::ImportValue: ${self:service}:${self:provider.stage}:SampleQueueArn

package:
  artifact: target/scala-2.12/hello.jar

functions:
  sample:
    handler: sample.SampleHandler
    name: sample-slack-post-function-${self:provider.stage}
    timeout: 300
    memorySize: 256
    environment:
      QUEUE_URL:
        Ref: sampleQueue
      SLACK_API_TOKEN: "slack api token required"

resources:
  Resources:
    sampleQueue:
      Type: AWS::SQS::Queue
      Properties:
        QueueName: sample-slack-post-${self:provider.stage}
        ReceiveMessageWaitTimeSeconds: 5
        VisibilityTimeout: 10
  Outputs:
    SampleQueueArn:
      Description: The ARN for the Sample Queue
      Value:
        "Fn::GetAtt": [ sampleQueue, Arn ]
      Export:
        Name: ${self:service}:${self:provider.stage}:SampleQueueArn
```

SQSに関して、実際の業務では CloudFormationでは構築しましたが、
せっかくなので `resources` で構築してみました。

Scalaのコードが参照するキューのURLとSlackのAPIトークンは、
`environment` から環境変数に登録しています。

実際の開発では、キューのURLなど環境毎に切り替えが必要な情報は、
開発用、ステージング用にファイルを切り出して対応しました。

[Serverless Variables - reference-variables-in-other-files](https://serverless.com/framework/docs/providers/aws/guide/variables/#reference-variables-in-other-files)


# デプロイ

カレントディレクトリ上で下記のコマンドを実行していきます。

##### パッケージング

```
$ sbt assembly
```

##### デプロイ

```bash
$ sls deploy -v
```

AWSコンソール上からも serverless.yml に定義した Lambda と SQS がクラウド上に、
作成されていることが確認できます。

<b>Lambda</b>
![スクリーンショット 2018-08-20 1.30.16.png](https://qiita-image-store.s3.amazonaws.com/0/142263/4d02501b-68c9-0651-2afc-ef2cc5d06169.png)

<b>SQS</b>
![スクリーンショット 2018-08-20 1.34.10.png](https://qiita-image-store.s3.amazonaws.com/0/142263/16d03bab-608c-296d-a2cc-b1a31018b848.png)

デプロイすると Serverless が CloudFormation に Stack を作成します。

失敗することがありますが、
環境変数 AWS_PROFILE に登録に入力したプロファイルの権限が問題なことが多いのでポリシーの見直しが必要でしょう。

また、正しい権限に修正しても一度作成した CloudFormation の Stack をうまく削除することができずに再実行が失敗することがあるので、
そんなときは迷わず下記のコマンドを実行すれば大体のことは解決できそうでした。

```
$ sls remove
```

# 動かしてみる

### メッセージの送信

テストデータとして aws-cli からキューにメッセージを送信します。
メッセージの内容は Slack API のリクエストボディに設定する Json です。

今回は `channel`, `text`, `username` を指定したシンプルな Json を送信します。

```bash
$ aws sqs send-message --queue-url <キューのURL> --message-body '{"channel":"@t_horikoshi","text":"test","username":"LambdaBot"}'
```

### Lambdaの実行

serverless.yml の `functions.name` に指定した名前を指定して、
Lambdaを実行します。

```bash
$ sls invoke -f sample -l
```

🎉🎉🎉<b>ヒャッホウ!!!</b>🎉🎉🎉

通知を確認できました

![スクリーンショット 2018-08-20 1.52.32.png](https://qiita-image-store.s3.amazonaws.com/0/142263/407f30c4-aa14-3dc0-a03e-a6635570794d.png)


# 開発してみて

### Serverless Framework

Lambda の設定に関して、
実際の開発では `timeout` や `memorySize` に加えて `reservedConcurrency`, `events.schedule` を `serverless.yml` に定義しました。

+ [Serverless Framework - AWS Lambda Guide - Functions](https://serverless.com/framework/docs/providers/aws/guide/functions)
+ [Serverless Framework - AWS Lambda Guide - Events](https://serverless.com/framework/docs/providers/aws/guide/events/:embed:cite)


これらの構成管理やクラウドへの反映が大変容易に感じました。
また依存するSQSのリソース定義が一元管理できる点も魅力的です。

CloudFormation や terraform の知見がある人にとっては、
キャッチアップが容易な気がします。

便利。

### AWS SQS

可視性タイムアウトやデッドレターキューの仕組みなど、
よく考えられているなーっと感心しました。

デッドレターキューに送信した不正メッセージの処理については、
AWS SNSなどで分析できるという話でした。

+ [Amazon Simple Notification Service とは - Amazon Simple Notification Service](https://docs.aws.amazon.com/ja_jp/sns/latest/dg/welcome.html)

素敵。

### AWS Lambda

動きっぱなしのEC2なんかで動く簡素な日次バッチ処理なんかは、
全部Lambdaに移行してしまえばコストカットにつながるのかなという印象でした。

同時実行起動数やイベントソースのサポートが大変充実しているので、
割と色んなユースケースに柔軟に対応できる気がします。

実はSQSのイベントソースがサポートされているので、
開発でも試してみたかったのですが進捗の都合で今回は断念しました...、無念。

機会があれば触ってみたいと思います。

無念。

# 辛かったところ

Slack API の [chat.postMessage](https://api.slack.com/methods/chat.postMessage) ですが
 Channel 毎に秒間1リクエストしか受け付けないという [RateLimit](https://docs.aws.amazon.com/ja_jp/sns/latest/dg/welcome.html) がありまして、
このあたりの制御がちょっと厄介でした。

Slackに通知する際、リクエストが1回で済むようにチャンネル単位でメッセージを集計したり、
Throttle制御入れたり、Lambdaの同時実行数を絞ったりなどなど...

解説では割愛してますがいろいろやってます。

このあたりはちょっとしんどかったです。

何でもかんでもSlack通知するっていうのはもうやめようぜ。
そんな気持ちになりました。

# 終わりに

今回紹介したコードは GitHub に公開しておきました。

+ https://github.com/takat0-h0rikosh1/SlackPostLambdaFunction

どなたかの参考になればと。

総じてAWSのクラウドサービスは便利ですね。
いろいろできるようになりたいので今後も仲良くしていきたいです。

本当はイベントソースやデッドレターキューの
お話などをもう少し詳しく解説に交えたかったのですがそれはまたの機会に。

では、また。