API GatewayとKinesisを用いたリアルタイムログ転送

# はじめに

　今回はAPI GatewayとKinesis firehoseを用いて、イベントログのリアルタイム収集、処理、分析を行います。
APIGatewayでは、イベントログを受け取りKinesisへストリーミングデータとして転送します。
またその際にLambdaは使用せず、**サービス統合プロクシ（ReguestTemplate）**を使用することで、lambdaを経由せずノーコードでFirehoseへのリクエストを行うことが可能です。

またKinesis firehoseとGlueと連携することで、収集と同時に**Parquet変換や圧縮**などもノーコード行うことが可能です。

以下は今回構築する構成図になります。

![Callbackシステムフロー-Kinesis(ServiceProxy)構成.drawio (1).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63356/22d66460-f13b-5cca-0b82-2d0273dbe2d9.png)

# API Gateway

　API GatewayはServerless FlameworkやSAMを用いて定義することが多いかもしれませんが、
今回は全体構成を全てCloudformationで一括管理するため使用しておりません。
この辺はSeverless FrameworkやSAMに慣れていると、Cloudformationによる定義は少し冗長に感じるかもしれません。

## RestApi

　以下がAPIGatewayに関するCfn定義です。
`Statement`ではイベントログの送信元IPが固定のため、許可するIPを制限しています。
また`execute-api:/${ApiGatewayStage}/GET/*/mobile_attribution`はエンドポイントを定義しています。
こちらはもちろん複数の定義も可能です。


```yml
  ## ------------- ApiGateway -------------
  ApiGatewayRestApi:
    Type: "AWS::ApiGateway::RestApi"
    Properties:
      Name: !Sub "${ProjectName}-${Env}-api"
      EndpointConfiguration:
        Types:
          - EDGE
      Policy:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal: "*"
            Action:
              - "execute-api:Invoke"
            Resource:
              - !Sub "execute-api:/${ApiGatewayStage}/GET/*/mobile_attribution"
            Condition:
              IpAddress:
                aws:SourceIp: !Split
                  - ","
                  - !Join
                    - ","
                    - - !Join [ ",", !Ref "OfficeIpAddressList" ]
                      - !Join [ ",", !Ref "VPNIpAddressList" ]

```

## Resource

　APIパスが`/{app_token}/mobile_attribution`なので、`/{app_token}`と`/mobile_attribution`を以下のように階層関係で定義します。


```yml
  # ApiGateway Resource
  ApiGatewayResourceAppTokenVar:
    Type: "AWS::ApiGateway::Resource"
    Properties:
      ParentId: !Sub "${ApiGatewayRestApi.RootResourceId}"
      PathPart: "{app_token}"
      RestApiId: !Ref "ApiGatewayRestApi" #前述のAPIGateway定義
  # ApiGateway Resource        
  ApiGatewayResourceMobileAttributionVar:
    Type: "AWS::ApiGateway::Resource"
    Properties:
      ParentId: !Ref "ApiGatewayResourceAppTokenVar"
      PathPart: "mobile_attribution"
      RestApiId: !Ref "ApiGatewayRestApi" #前述のAPIGateway定義
    DependsOn:
      - ApiGatewayResourceAppTokenVar

```

## Method

![CallbackシステムAPIGateway.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63356/d3b5a586-edc5-3b16-b809-8007228552f8.png)

　前述定義したAPI Gateway Resourceを用いてAPI Methodの定義を行います。
こちらではAWSサービスプロクシを用いてKinesis firehoseへのリクエストを定義します。
Lambdaを用いることでも実現可能ですが、メンテナンス工数を削減するためにノーコードでリクエストが可能なAWSサービスプロクシを選択しています。
　そして`RequestTemplate`ではどのようなデータをリクエストするか定義しています。
こちらでは、URLパラメータやパス、パスに定義されたapp_tokenなど全てをリクエスト情報に追加しています。

こちらの構文はVelocityTemplateLanguage(VTL)で定義されており、詳細は[API Gateway mapping template](https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html)で確認することができます。

```yml
  # ApiGateway Method    
  ApiGatewayMethodMobileAttributionCheckVarGet:
    Type: "AWS::ApiGateway::Method"
    Properties:
      HttpMethod: GET
      RequestParameters: { }
      ResourceId: !Ref "ApiGatewayResourceMobileAttributionVar"
      RestApiId: !Ref "ApiGatewayRestApi"
      ApiKeyRequired: true
      AuthorizationType: NONE
      Integration:
        Type: AWS
        IntegrationHttpMethod: POST
        Uri: !Sub "arn:aws:apigateway:${AWS::Region}:firehose:action/PutRecord"
        Credentials: !GetAtt "ApiGatewayIntegrationRequestRole.Arn"
        PassthroughBehavior: WHEN_NO_TEMPLATES
        RequestTemplates:
          application/json:
            !Sub |
            #set($params = $input.params().get('querystring'))
            #set($paths = $input.params().get('path'))
            #set($data = "{
              #foreach($paramName in $params.keySet())
                ""$paramName"": ""$util.escapeJavaScript($params.get($paramName))"",
              #end
              ""app_token"": ""$util.escapeJavaScript($method.request.path.app_token)""
            }")
            {
                "DeliveryStreamName": "${FirehoseMobileAttribution}",
                "Record": {
                    "Data": "$util.base64Encode($data)"
                }
            }
        IntegrationResponses:
          - StatusCode: 200
      MethodResponses:
        - StatusCode: 200
          ResponseModels: { application/json: "Empty" }
    DependsOn:
      - ApiGatewayResourceMobileAttributionVar

```

## Deployment

今までに定義したAPI Gatewayの`deployment`定義を行います。
コンソール画面で設定変更後デプロイするかと思いますが、その際の定義となります。
アクセスログが必要な場合は、こちらでログフォーマットの設定なども行えます。

```yml
  # ApiGateway deployment
  ApiGatewayDeployment:
    Type: "AWS::ApiGateway::Deployment"
    Properties:
      RestApiId: !Ref "ApiGatewayRestApi"
      StageName: !Ref "ApiGatewayStage"
      StageDescription:
        Description: !Sub "Stage - ${Env}"
        LoggingLevel: INFO
        MetricsEnabled: False
        AccessLogSetting:
          DestinationArn: !GetAtt "ApiGatewayAccessLogGroup.Arn"
          Format: "{ \"requestId\":\"$context.requestId\", \"ip\": \"$context.identity.sourceIp\", \"caller\":\"$context.identity.caller\", \"user\":\"$context.identity.user\",\"requestTime\":\"$context.requestTime\", \"httpMethod\":\"$context.httpMethod\",\"resourcePath\":\"$context.resourcePath\", \"status\":\"$context.status\",\"protocol\":\"$context.protocol\", \"responseLength\":\"$context.responseLength\" }"
    DependsOn:
      - ApiGatewayResourceMobileAttributionVar
      - ApiGatewayMethodMobileAttributionCheckVarGet
      - ApiGatewayAccessLogGroup

```

## ApiKey

前述の`Method`で `ApiKeyRequired: true`と定義していたと思いますが、そちらのApiKey定義となります。
以下の`Name`と`Value`でヘッダーに渡すApiKeyのキー名と値を定義しています。


```yml
  ## ------------- Api key --------------
  ApiGatewayRestApiBaseKey:
    Type: AWS::ApiGateway::ApiKey
    Properties:
      Name: !Sub "${ProjectName}-${Env}-api-base-key" # Replacement
      Enabled: true
      Value: !FindInMap [ !Ref Env, ApiGateway, BaseApiKey ] # Replacement
      StageKeys:
        - RestApiId: !Ref "ApiGatewayRestApi"
          StageName: !Ref "ApiGatewayStage"
    DependsOn:
      - ApiGatewayDeployment
```

## UsagePlan

![スクリーンショット 2021-09-28 12.46.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63356/ae115107-7110-37cd-342b-814fef058f05.png)

　こちらはApiKeyを定義しているため必要な設定で、上記コンソール画面の使用量プランに相当します。
`UsagePlan` を対象のApiIdとStateを指定して紐付け、それを`UsagePlanKey`のUsagePlanIdにセットします。

```yml
  # ApiGateway usage plan
  ApiGatewayRestApiBaseUsagePlan:
    Type: AWS::ApiGateway::UsagePlan
    Properties:
      ApiStages:
        - ApiId: !Ref "ApiGatewayRestApi"
          Stage: !Ref "ApiGatewayStage"
      UsagePlanName: !Sub "${ProjectName}-${Env}-usage-plan"
    DependsOn:
      - ApiGatewayRestApi
      - ApiGatewayDeployment
  # ApiGateway usage plan key
  ApiGatewayRestApiBaseUsagePlanKey:
    Type: 'AWS::ApiGateway::UsagePlanKey'
    Properties:
      KeyId: !Ref "ApiGatewayRestApiBaseKey"              # Replacement
      KeyType: "API_KEY"                                  # Replacement
      UsagePlanId: !Ref "ApiGatewayRestApiBaseUsagePlan"  # Replacement
    DependsOn:
      - ApiGatewayRestApiBaseUsagePlan

```

# S3

こちらはFirehoseで処理したデータの保存先となるS3定義です。

```yml
  ## ------------- S3 -------------
  S3Firehose:
    Type: "AWS::S3::Bucket"
    DeletionPolicy: Retain # cfn delete時にS3Bucketを残す.
    Properties:
      BucketName: !FindInMap [ !Ref Env, S3, Bucket ] 
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
```

# Glue

![Callbackシステムフロー-Kinesis(ServiceProxy)構成.drawio (1)のコピー.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63356/620f9a02-9370-89a4-02d1-9d3bdd8a77b3.png)


　Glueとはデータの分類、クリーニング、加工をデータストア間、データストリーム間で行う完全マネージドETLです。
以下はストリーミングETLを行うための定義となっており、
リアルタイムParque変換に必要で、テーブル定義やパーティション定義も行っております。
テーブル定義やパーティション定義は必要な項目に応じて定義してください。

```yml

  # Glue database
  GlueDatabase:
    Type: "AWS::Glue::Database"
    Properties:
      CatalogId: !Ref "AWS::AccountId" 
      DatabaseInput:
        Name: !FindInMap [ !Ref Env, Glue, Database ] 

  # Glue table
  GlueTable:
    Type: "AWS::Glue::Table"
    Properties:
      CatalogId: !Ref "AWS::AccountId"  
      DatabaseName: !Ref "GlueDatabase" 
      TableInput: 
        Name: !Ref "GlueTableNameMobileAttribution"
        TableType: EXTERNAL_TABLE
        Parameters:
          has_encrypted_data: true
          EXTERNAL: true
        PartitionKeys:
          - Name: year
            Type: string
          - Name: month
            Type: string
          - Name: day
            Type: string
        StorageDescriptor:
          Columns:
            - Name: event_token
              Type: string
            - Name: event_name
              Type: string
            - Name: app_name
              Type: string
            - Name: activity_kind
              Type: string
            - Name: created_at
              Type: timestamp
          Location: !Sub
            - "s3://${S3Bucket}/${S3Dir}"
            - S3Bucket: !Sub "{{resolve:ssm:${ProjectName}-${Env}-S3FirehoseBucketName}}"
              S3Dir: !Ref "S3DirMobileAttribution"
          InputFormat: org.apache.hadoop.mapred.TextInputFormat
          OutputFormat: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
          SerdeInfo:
            SerializationLibrary: org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe
            Parameters:
              serialization.format: '1'
          BucketColumns: [ ]
          SortColumns: [ ]

```

# Kinesis Firehose

　以下はFirehose定義となります。
`ExtendedS3DestinationConfiguration`でS3に関する定義を行います。
`BucketArn` で先ほど作成したS3バケットのArnを、`Prefix` ではGlueで定義したパーティションを定義します。ここが一致していないとエラーになるので注意してください。
Prefixの定義方法については[Amazon S3 オブジェクトのカスタムプレフィックス](https://docs.aws.amazon.com/ja_jp/firehose/latest/dev/s3-prefixes.html)で詳しく記載されているので参考にしていただければと思います。
　また`CompressionFormat`はRecordFormatを使用している場合自動でSnappy圧縮となるのでUNCOMPRESSEDを定義します。
　またバックアップが必要な場合は、`S3BackupMode: Enabled`としてバックアップ先を指定する必要があります。

```yml
  ## ------------- Firehose -------------
  FirehoseMobileAttribution:
    Type: "AWS::KinesisFirehose::DeliveryStream"
    Properties:
      DeliveryStreamEncryptionConfigurationInput:
        KeyType: AWS_OWNED_CMK
      DeliveryStreamName: !FindInMap [ !Ref Env, Firehose, DeliveryStreamMobileAttribution ] # Replacement
      DeliveryStreamType: DirectPut                                                          # Replacement
      ExtendedS3DestinationConfiguration:
        RoleARN: !GetAtt FirehoseRole.Arn
        BucketARN: !Sub "{{resolve:ssm:${ProjectName}-${Env}-S3FirehoseArn}}"
        Prefix: !Sub "${S3DirMobileAttribution}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
        ErrorOutputPrefix: !Sub "${S3DirMobileAttribution}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/!{firehose:error-output-type}"
        BufferingHints:
          SizeInMBs: 64
          IntervalInSeconds: 60
        CompressionFormat: UNCOMPRESSED # Record Format変換を使用する場合, 自動的にSnappy圧縮が選択されるため指定しない
        EncryptionConfiguration:
          KMSEncryptionConfig:
            AWSKMSKeyARN: !Ref "S3KeyArn"
        CloudWatchLoggingOptions:
          Enabled: true
          LogGroupName: !Ref "FirehoseLogGroup"
          LogStreamName: !Ref "FirehoseLogStream"
        S3BackupMode: Enabled
        S3BackupConfiguration:
          BucketARN: !Sub "{{resolve:ssm:${ProjectName}-${Env}-S3FirehoseArn}}"
          RoleARN: !GetAtt "FirehoseRole.Arn"
          Prefix: !Sub "backup/${S3DirMobileAttribution}/"
        DataFormatConversionConfiguration:
          SchemaConfiguration:
            CatalogId: !Ref "AWS::AccountId"
            RoleARN: !GetAtt "FirehoseRole.Arn"
            DatabaseName: !Sub "{{resolve:ssm:${ProjectName}-${Env}-GlueDatabaseName}}"
            TableName: !Sub "{{resolve:ssm:${ProjectName}-${Env}-GlueTableName}}"
            Region: !Ref "AWS::Region"
            VersionId: LATEST
          InputFormatConfiguration:
            Deserializer:
              OpenXJsonSerDe: { }
          OutputFormatConfiguration:
            Serializer:
              ParquetSerDe: { }
          Enabled: true
    DependsOn:
      - FirehoseLogStream

```

# 最後に

いかがでしたでしょうか？
今回はイベントログのリアルタイム処理をユースケースとしていましたが、少し構成を変えると他のユースケースにも対応できるかと思います。
例えばストレージへの永続化が必要な場合、間にDynamoDBをはさみ、DynamoDB Streamsの変更データキャプチャをストリームデータとして転送するなどで、その他にも様々な活用方法があると思いますので、ぜひ活用してみてください。


# 参考文献

* [AWS CloudFormation Documentation](https://docs.aws.amazon.com/cloudformation/index.html)

* [x-amazon-apigateway-integration.requestTemplates オブジェクト](https://docs.aws.amazon.com/ja_jp/apigateway/latest/developerguide/api-gateway-swagger-extensions-integration-requestTemplates.html)

* [AWS Glue](https://docs.aws.amazon.com/ja_jp/glue/latest/dg/what-is-glue.html)

* [Firehose でS3プリフィックスのカスタマイズが可能になりました](https://dev.classmethod.jp/articles/firehose-custom-s3-prefixes/)

* [Amazon S3 オブジェクトのカスタムプレフィックス](https://docs.aws.amazon.com/ja_jp/firehose/latest/dev/s3-prefixes.html)