More than 3 years have passed since last update.

簡単Webサイトホスティング④CloudFrontのリアルタイムログをKibanaで可視化する

Last updated at 2021-01-26Posted at 2021-01-23

はじめに

Amazon Web Services（AWS）が提供する、Amazon CloudFront や Amazon S3 と呼ばれるサービスを組み合わせることで、 HTMLやJavaScript、画像、ビデオなどで構成される静的Webサイトの配信基盤 を安価に構築することができます。本記事では、リソースのセットアップを自動で行うことのできる、AWS CloudFormation を用いることで、これらの配信基盤を ミスなく迅速に構築 する手順をご説明します。なお、今回使用する CloudFormation テンプレートは、以下の GitHub リポジトリで公開しています。

AWSCloudFormationTemplates/static-website-hosting-with-ssl

TL;DR

以下の CloudFormation テンプレートを実行することで、 静的Webサイトのホスティング基盤 を 迅速かつお手軽に実現 します。下にあるボタンをクリックすると、自身のAWSアカウント（Asia Pacific Tokyo - ap-northeast-1）で、この CloudFormation テンプレートを実行することが可能となります。

作成されるAWSリソース全体のアーキテクチャ図は、過去の記事をご覧ください。このうち本記事では、以下のリソースに焦点を当ててご説明します。

Kibana を用いた CloudFront リアルタイムログの可視化

ログが生成されてから利用できるまでに数分を要していた従来の 標準ログ に加えて、生成されてからわずか数秒でアクセス可能 となる CloudFront のリアルタイムログ が提供開始となり、これを用いることで より詳細かつ迅速なモニタリング と 発生した事象に合わせた迅速なリソース設定の変更 を行えるようになりました。このリアルタイムログは、

取得するログのサンプリング率
取得するログのフィールド
どのCloudFront Behavior に適用するか

を指定することが可能で、指定した 任意のログを Kinesis Data Streams に送信する ことができます。また、Kinesis ストリームに到達したログは、 Kinesis Data Firehose を経由して Elasticsearch Service に蓄積することが可能で、さらに Kibanaを用いることで、ログの中身を簡単に可視化 することができます。

これらの手順は、 Amazon CloudFront ログを使用したリアルタイムダッシュボードの作成 というタイトルで、 Amazon Web Services ブログに公開されており、ここに掲載されている手順に沿ってリソースを作成することで、以下のアーキテクチャを構成することができます。

本記事では、 上記のアーキテクチャ を CloudFormationテンプレートを用いてワンクリックで作成 するとともに、それぞれのAWSサービスに掛かる負荷に対して どの程度のリソースをプロビジョニングしておけば良いか についても合わせてご説明します。

Amazon Kinesis Data Streams

まず最初に、 CloudFrontから送られた リアルタイムログを受信する Kinesis ストリームを作成 します。

Parameters:
  KinesisShardCount:
    Type: Number
    Default: 1
    MinValue: 1 
    Description: The shard count of Kinesis [required]

Resources:
  Kinesis:
    Type: 'AWS::Kinesis::Stream'
    Properties:
      Name: !Ref AWS::StackName
      RetentionPeriodHours: 24
      StreamEncryption:
        EncryptionType: KMS
        KeyId: alias/aws/kinesis
      ShardCount: !Ref KinesisShardCount

ここで重要となるのは、 Kinesisストリームを何シャードで構えておくか についてです。この値をどう算出するのかについては、公式ドキュメントの Kinesis データストリームのシャード数を推定するには という項に推定方法の記述があり、全てのフィールドを含んだリアルタイムログを出力する場合には、

1,000 Byte x 秒間リクエスト数 / 1,000,000 x 1.25 = シャード数

として算出できます。

例えば最大秒間5,000リクエストのアクセスが発生する可能性がある場合は、

5,000 x 1,000 / 1,000,000 x 1.25 = 7

となり、バッファも含めて約7シャード必要になることが分かります。

ただし、 バーストトラフィックが発生するようなサイト の場合は、 特定の1秒にログの出力が集中することで受信できるデータ量を超過 してしまい、 ProvisionedThroughputExceededExceptionが発生 するケースがあり得ることから、ドキュメントに記載のある値より大きなバッファ値、例えば 想定するデータ量を処理できるシャード数の倍のシャードをプロビジョニング しておくなどしておいた方が安全です。

トラフィックに合わせてシャード数を変更する

CloudFrontへのトラフィックは、時間帯やイベントの有無によって変動します。 Kinesis Data Streams が、シャード数単位（シャード時間）で課金されること、キャパシティを超えたリアルタイムログを受信できないことなどから、 CloudFrontへのトラフィック量に応じて 、 Kinesis ストリームの シャード数を変更する 必要があります。

ただし、シャード数を変更する際には、 いくつかの制約事項が存在 します。まず、シャード数を変更する際にコールされる UpdteShardCount APIは、 現在のシャード数を2倍にするか、もしくは1/2にするかの操作しかできません 。したがって、3シャードのKinesisストリームを7シャードに変更するといったことはできません。そこで シャード数は、2の階乗（1, 2, 4, 8, 16, 32, 64, 128...）に設定しておく ことをオススメします。また、 1リージョンあたりのシャード数 の初期値は、北部バージニア（us-east-1）リージョンで500シャード、それ以外のリージョンで 200シャード です。これに加えて、シャード数の変更を行う UpdteShardCount APIの実行回数にも上限 があります。これらの上限値を超えた利用を希望する場合には、 クオータ制限の緩和申請 が必要となります。

Kinesis Data Streams に設定したキャパシティが負荷に対して適切であるかどうかを確認するためには、以下のCloudWatchメトリクスを監視してください。こちらのリンク から、これらのメトリクスを基にしたCloudWatchアラームを一括で有効化することができます。

ネームスペース	メトリクス	閾値
AWS/Kinesis	GetRecords.IteratorAgeMilliseconds	テンプレートで指定した値
AWS/Kinesis	PutRecord.Success	テンプレートで指定した値
AWS/Kinesis	WriteProvisionedThroughputExceeded	1分間に1回以上

Amazon CloudFront

次に、CloudFront から出力する リアルタイムログの設定 を行います。

まずは、 CloudFront が Kinesis Data Streams に対してログを出力 できるように、IAM Role を用いて権限の付与を行います。 CloudFront に与える権限は、 Kinesisへの書き込み権限 と、データを暗号化する際に使用する KMSキーを作成する権限 です。

Resources:
  IAMRoleForCloudFrontRealtimeLog:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: cloudfront.amazonaws.com
            Action: 'sts:AssumeRole'
      Policies:
        - PolicyName: !Sub '${AWS::StackName}-KinesisPutPolicy-${AWS::Region}'
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - 'kinesis:DescribeStreamSummary'
                  - 'kinesis:DescribeStream'
                  - 'kinesis:PutRecord'
                  - 'kinesis:PutRecords'
                Resource:
                  - !GetAtt Kinesis.Arn
              - Effect: Allow
                Action:
                  - 'kms:GenerateDataKey'
                Resource:
                  - !GetAtt KMSKey.Arn
      RoleName: !Sub '${aws::StackName}-CloudFrontRealtimeLog-${AWS::Region}'

次に、先ほど作成した Kinesis Data Streams と IAM Role のARNを指定して、リアルタイムログの設定を行います。 Fields では ログに出力するフィールド を選択することができますが、本テンプレートでは以下の 全てのフィールドを出力 する設定としています。

フィールド名	内容
timestamp	エッジサーバーがリクエストへの応答を終了した日時
c-ip	リクエスト元のビューワーの IP アドレス
time-to-first-byte	サーバー上で測定される、要求を受信してから応答の最初のバイトを書き込むまでの秒数
sc-status	サーバーのレスポンスの HTTP ステータスコード
sc-bytes	サーバーがリクエストに応じてビューワーに送信したデータのバイトの合計数
cs-method	ビューワーから受信した HTTP リクエストメソッド
cs-protocol	ビューワーリクエストのプロトコル
cs-host	CloudFront ディストリビューションのドメイン名
cs-uri-stem	パスとオブジェクトを識別するリクエスト URL の部分
cs-bytes	ビューワーがリクエストに含めたデータのバイトの合計数
x-edge-location	リクエストを処理したエッジロケーション
x-edge-request-id	リクエストを一意に識別する不透明な文字列
x-host-header	ビューワーが、このリクエストの Host ヘッダーに追加した値
time-taken	サーバーが、ビューワーのリクエストを受信してからレスポンスの最後のバイトを出力キューに書き込むまでの秒数
cs-protocol-version	ビューワーがリクエストで指定した HTTP バージョン
c-ip-version	リクエストの IP バージョン
cs-user-agent	リクエスト内の `User-Agent` ヘッダーの値
cs-referer	リクエスト内の `Referer` ヘッダーの値
cs-cookie	リクエスト内の Cookie ヘッダー
cs-uri-query	リクエスト URL のクエリ文字列の部分
x-edge-response-result-type	ビューワーにレスポンスを返す直前にサーバーがレスポンスを分類した方法
x-forwarded-for	リクエスト元のビューワーの IP アドレス
ssl-protocol	リクエストとレスポンスを送信するためにビューワーとサーバーがネゴシエートした SSL/TLS プロトコル
ssl-cipher	リクエストとレスポンスを暗号化するためにビューワーとサーバーがネゴシエートした SSL/TLS 暗号
x-edge-result-type	サーバーが、最後のバイトを渡した後で、レスポンスを分類した方法
fle-encrypted-fields	サーバーが暗号化してオリジンに転送したフィールドレベル暗号化フィールドの数
fle-status	リクエストボディが正常に処理されたかどうかを示すコード
sc-content-type	レスポンスの HTTP Content-Type ヘッダーの値
sc-content-len	レスポンスの HTTP Content-Length ヘッダーの値
sc-range-start	範囲の開始値
sc-range-end	範囲の終了値
c-port	閲覧者からのリクエストのポート番号
x-edge-detailed-result-type	x-edge-result-type と同じ値
c-country	ビューワーの IP アドレスによって決定される、ビューワーの地理的位置を表す国コード
cs-accept-encoding	ビューワーリクエスト内の Accept-Encoding ヘッダーの値
cs-accept	ビューワーリクエスト内の Accept ヘッダーの値
cache-behavior-path-pattern	ビューワーリクエストに一致したキャッシュ動作を識別するパスパターン
cs-headers	ビューワーリクエスト内の HTTP ヘッダー
cs-header-names	ビューワーリクエスト内の HTTP ヘッダーの名前
cs-headers-count	ビューワーリクエスト内の HTTP ヘッダーの数

また、 SamplingRate の値を変更することで、CloudFront が Kinesis Data Streams に送信する ログのサンプリングレートを、1%から100%の間で指定 することができます。

Parameters:
  SamplingRate:
    Type: Number
    Default: 100
    MinValue: 1
    MaxValue: 100
    Description: The sampling rate of logs sent by CloudFront [required]

Resources:
  CloudFrontRealtimeLogConfig:
    Type: 'AWS::CloudFront::RealtimeLogConfig'
    Properties: 
      EndPoints: 
        - KinesisStreamConfig: 
            RoleArn: !GetAtt IAMRoleForCloudFrontRealtimeLog.Arn
            StreamArn: !GetAtt Kinesis.Arn
          StreamType: Kinesis
      Fields: 
        - timestamp
        - c-ip
        - time-to-first-byte
        - sc-status
        - sc-bytes
        - cs-method
        - cs-protocol
        - cs-host
        - cs-uri-stem
        - cs-bytes
        - x-edge-location
        - x-edge-request-id
        - x-host-header
        - time-taken
        - cs-protocol-version
        - c-ip-version
        - cs-user-agent
        - cs-referer
        - cs-cookie
        - cs-uri-query
        - x-edge-response-result-type
        - x-forwarded-for
        - ssl-protocol
        - ssl-cipher
        - x-edge-result-type
        - fle-encrypted-fields
        - fle-status
        - sc-content-type
        - sc-content-len
        - sc-range-start
        - sc-range-end
        - c-port
        - x-edge-detailed-result-type
        - c-country
        - cs-accept-encoding
        - cs-accept
        - cache-behavior-path-pattern
        - cs-headers
        - cs-header-names
        - cs-headers-count
      Name: RealtimeLogConfig
      SamplingRate: !Ref SamplingRate

なお、プロビジョニングした Kinesis ストリームのキャパシティを超えるリアルタイムログが生成された場合は、 キャパシティを超えた分のリアルタイムログが欠落 します。しかし、それによって、 CloudFront ディストリビューションの挙動に異常が発生することはありません 。

また、 CloudFront はグローバルに提供されるAWSサービスですが、リアルタイムログを受信する Kinesis ストリームには全てのエッジロケーションのアクセスログが出力 されます。どのエッジロケーションでリクエストを処理したかについては、 x-edge-location フィールドで確認することができます。

そして、過去に作成したCloudFrontディストリビューションに上記の リアルタイムログの設定をアタッチ すると、今設定したばかりの リアルタイムログの出力がCloudFront上で有効 となります。

  CloudFront:
    Type: 'AWS::CloudFront::Distribution'
    Properties:
      DistributionConfig:
        DefaultCacheBehavior:
          RealtimeLogConfigArn: !GetAtt CloudFrontRealtimeLogConfig.Arn

Amazon Elasticsearch Service

Elasticsearchは、 分散型の分析エンジン で、構造化、非構造化、地理情報、メトリックなどの 様々なタイプの検索 を行なったり、 大規模なデータに対して分析 を行うことができます。この Elasticsearch を簡単かつ大規模にデプロイ、保護、実行を可能とするマネージドサービスが、 Amazon Elasticsearch Service です。

この Elasticsearch Service を使用することで、 アプリケーションやインフラストラクチャのログを保存、分析して問題を迅速に発見 したり、 アプリケーションに検索機能を追加 したりすることができます。そこで今回は、 Elasticsearch を用いて CloudFront のリアルタイムログを分析し、 Elasticsearch で使えるグラフツールとして知られる Kibanaを用いてこれを可視化 します。

Parameters:
  ElasticSearchVolumeSize:
    Type: Number
    Default: 10
    MinValue: 10
    Description: The volume size (GB) of ElasticSearch Service [required]
  ElasticSearchDomainName:
    Type: String
    Default: cloudfront-realtime-logs
    AllowedPattern: .+
    Description: The domain name of ElasticSearch Service [required]
  ElasticSearchInstanceType:
    Type: String
    Default: r5.large.elasticsearch
    AllowedValues:
      - t3.small.elasticsearch
      - t3.medium.elasticsearch
      - t2.micro.elasticsearch
      - t2.small.elasticsearch
      - t2.medium.elasticsearch
      - m5.large.elasticsearch
      - m5.xlarge.elasticsearch
      - m5.2xlarge.elasticsearch
      - m5.4xlarge.elasticsearch
      - m5.12xlarge.elasticsearch
      - m4.large.elasticsearch
      - m4.xlarge.elasticsearch
      - m4.2xlarge.elasticsearch
      - m4.4xlarge.elasticsearch
      - m4.10xlarge.elasticsearch
      - c5.large.elasticsearch
      - c5.xlarge.elasticsearch
      - c5.2xlarge.elasticsearch
      - c5.4xlarge.elasticsearch
      - c5.9xlarge.elasticsearch
      - c5.18xlarge.elasticsearch
      - c4.large.elasticsearch
      - c4.xlarge.elasticsearch
      - c4.2xlarge.elasticsearch
      - c4.4xlarge.elasticsearch
      - c4.8xlarge.elasticsearch
      - r5.large.elasticsearch
      - r5.xlarge.elasticsearch
      - r5.2xlarge.elasticsearch
      - r5.4xlarge.elasticsearch
      - r5.12xlarge.elasticsearch
      - r4.large.elasticsearch
      - r4.xlarge.elasticsearch
      - r4.2xlarge.elasticsearch
      - r4.4xlarge.elasticsearch
      - r4.8xlarge.elasticsearch
      - r4.16xlarge.elasticsearch
      - r3.large.elasticsearch
      - r3.xlarge.elasticsearch
      - r3.2xlarge.elasticsearch
      - r3.4xlarge.elasticsearch
      - r3.8xlarge.elasticsearch
      - i3.large.elasticsearch
      - i3.xlarge.elasticsearch
      - i3.2xlarge.elasticsearch
      - i3.4xlarge.elasticsearch
      - i3.8xlarge.elasticsearch
      - i3.16xlarge.elasticsearch
    Description: The instance type of ElasticSearch Service [required]
  ElasticSearchMasterUserName:
    Type: String
    AllowedPattern: .+
    Description: The user name of ElasticSearch Service [required]
  ElasticSearchMasterUserPassword:
    Type: String
    AllowedPattern: .+
    NoEcho: true
    Description: The password of ElasticSearch Service [required]

Resources:
  KMSKey:
    Type: AWS::KMS::Key
    Properties: 
      Description: Encrypt CloudTrail Logs
      Enabled: true
      EnableKeyRotation: true
      KeyPolicy: 
        Version: 2012-10-17
        Id: DefaultKeyPolicy
        Statement:
          - Sid: Enable IAM User Permissions
            Effect: Allow
            Principal:
              AWS: !Sub arn:aws:iam::${AWS::AccountId}:root
            Action: 'kms:*'
            Resource: '*'
          - Effect: Allow
            Principal:
              Service: cloudtrail.amazonaws.com
            Action:
              - 'kms:GenerateDataKey*'
            Resource:
              - '*'
            Condition:
              StringLike:
                kms:EncryptionContext:aws:cloudtrail:arn:
                  - !Sub arn:aws:cloudtrail:*:${AWS::AccountId}:trail/*
          - Effect: Allow
            Principal:
              Service: cloudtrail.amazonaws.com
            Action:
              - 'kms:DescribeKey'
            Resource:
              - '*'
      KeyUsage: ENCRYPT_DECRYPT
      PendingWindowInDays: 30
  ElasticSearchDomain:
    Type: 'AWS::Elasticsearch::Domain'
    Properties: 
      AccessPolicies:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              AWS:
                - '*'
            Action:
              - es:*
            Resource: !Sub arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/*
      AdvancedSecurityOptions: 
        Enabled: true
        InternalUserDatabaseEnabled: true
        MasterUserOptions:
          MasterUserName: !Ref ElasticSearchMasterUserName
          MasterUserPassword: !Ref ElasticSearchMasterUserPassword
      DomainEndpointOptions: 
        EnforceHTTPS: true
      DomainName: !Ref ElasticSearchDomainName
      EBSOptions: 
        EBSEnabled: true
        VolumeSize: !Ref ElasticSearchVolumeSize
        VolumeType: gp2
      ElasticsearchClusterConfig: 
        DedicatedMasterCount: 3
        DedicatedMasterEnabled: true
        DedicatedMasterType: c5.large.elasticsearch
        InstanceCount: 3
        InstanceType: !Ref ElasticSearchInstanceType
        ZoneAwarenessConfig: 
          AvailabilityZoneCount: 3
        ZoneAwarenessEnabled: true
      ElasticsearchVersion: 7.8
      EncryptionAtRestOptions:
        Enabled: true
        KmsKeyId: !GetAtt KMSKey.Arn
      NodeToNodeEncryptionOptions: 
        Enabled: true
      SnapshotOptions: 
        AutomatedSnapshotStartHour: 0

本テンプレートの構成と注意点は、以下の通りです。

DomainName には、「アカウントおよびリージョンに固有」「先頭が小文字」「3～28文字」「小文字のアルファベット、数字、ハイフンのみ使用可能」という制約が課せられています。
EBSOptions にて、アタッチするEBSボリュームのタイプとサイズを指定しています。
ElasticsearchClusterConfig にて、本番稼働用として奨励されている、マルチAZ + 専用データノード構成を規定しており、 データノード3台 + マスターノード3台 の構成としています。 AvailabilityZoneCount を3に設定しているため、 インスタンスは3つのAZに分散配置 されます。
EncryptionAtRestOptions にて、 保管時のデータ暗号化 を指定しています。これと同時に 暗号化の際に使用するAWS KMSのカスタマーマスターキー（CMK）も作成 しています。
NodeToNodeEncryptionOptions にて、 ノード間の暗号化 を指定しています。
AutomatedSnapshotStartHour にて、 UTC時刻の午前0時にスナップショットが作成 されます。

アクセスコントロール

本テンプレートは きめ細かなアクセスコントロール （= FGAC）を有効化しており、以下の設定としています。

パブリックアクセスを許可 します。
DomainEndpointOptions にて、 HTTPSによるアクセスを強制 します。
InternalUserDatabaseEnabled にて 内部ユーザデータベースを有効化 した上で、 MasterUserOptions にて、 マスターユーザのユーザ名およびパスワードを規定 します。
AccessPolicies にて、 Elasticsearch Service の全ての操作を許可 します。

上記の設定によりこのドメインおよびKibanaへは、 事前に設定したユーザ名とパスワードを用いて外部からアクセスする ことが可能となります。

ドメインのサイジング

Elasticsearch Service を使用するに当たって注意すべき点は、 どのインスタンスタイプにすべきか、そして EBSボリュームはどの程度の大きさを用意しておけばよいか 、についてです。これについては、公式ドキュメントの Amazon ES ドメインのサイジングの項に記載があります。

例えば、ストレージサイズについては、

ソースデータ x (1+ レプリカの数) x 1.45 = 最小ストレージ要件

という式が掲載されています。秒間5,000リクエストのトラフィックが発生するCloudFrontディストリビューションのリアルタイムログを24時間保存する場合は、

5,000（件） x 3,600（秒）x 24（時間） x 1,000(byte) = 432（GB）

ソースデータは432GBとなり、上記の式を適用すると、

432（GB） x (1 + 1) x 1.45 = 1252 (GB）

1252GBのストレージが必要となります。

なお、上記の例では 1時間あたり18GBの割合でソースデータが増加する 計算となり、これは Elasticsearch Service にとって大きな負荷となります。。公式ドキュメントに、「高負荷の集計処理、頻繁なドキュメント更新、または大量のクエリ処理が発生している場合 、それらのリソースではニーズを満たせない可能性があります。クラスターがこのようなカテゴリに分類される場合はまず、 ストレージ要件の 100 GiB ごとに vCPU x 2 コア、メモリの 8 GiB に近い構成」を勧める旨の記載があり、上記例にこれを当てはめると、

1252（GB）/ 100（GB）x 2 = 25（コア）
1252（GB）/ 100（GB）x 8 = 100（GBメモリ）

が必要になると考えられます。上述のように、本テンプレートでは データノードを3台用意 しているため、1インスタンスあたりで必要とされるコア数およびメモリサイズは、

25（コア）/ 3 （台） = 8.3（コア）
100（GBメモリ）/ 3 （台） = 33.3（GBメモリ）

となり、1インスタンスあたりに必要なEBSボリュームは、

1252（GB） / 3（台） = 417（GB）

となります。これを満たす インスタンスタイプ は、

m5.2xlarge.elasticsearch 以上のインスタンスタイプ
c5.4xlarge.elasticsearch 以上のインスタンスタイプ
r5.2xlarge.elasticsearch 以上のインスタンスタイプ
i3.2xlarge.elasticsearch 以上のインスタンスタイプ

となりますが、経験上 データノードはヒープ領域が枯渇することが多い ため、上記例の場合は、 メモリ最適化 インスタンスである R5インスタンスを選択 するのが良いかもしれません。なお、マスターノードに関しては、 専用マスターノードの項に、

Instance Count	推奨される最小専用マスターインスタンスタイプ
1–10	c5.large.elasticsearch

との記述があるため、本テンプレートが構築する構成では c5.large.elasticsearch で問題なさそうです。なお、これらの値はあくまで計算上の値であることから、データノードおよびマスターノードのインスタンスタイプを決定する際には、 事前に想定と同程度の負荷を掛けて挙動を検証する必要 があります。

ドメイン作成後にこれらの設定を変更する場合、Blue/Greenデプロイメントプロセスが実行されて新たな環境が作成されます。この 設定変更には時間が掛かる上にマスターノードに大きな負荷が掛かる ため、このデプロイプロセスに関連した オーバヘッドを処理するための十分なリソース が必要となります。十分なリソースが無い状態で設定変更した場合、 設定変更（In Progress）に数時間掛かる こともあります。

Elasticsearch Service に設定したキャパシティが負荷に対して適切であるかどうかを確認するためには、以下のCloudWatchメトリクスを監視してください。こちらのリンク から、これらのメトリクスを基にしたCloudWatchアラームを一括で有効化することができます。

ネームスペース	メトリクス	閾値
AWS/ES	ClusterStatus.green	0だった場合
AWS/ES	ClusterIndexWritesBlocked	1分間に1回以上
AWS/ES	MasterReachableFromNode	0だった場合
AWS/ES	AutomatedSnapshotFailure	1分間に1回以上
AWS/ES	KibanaHealthyNodes	0だった場合
AWS/ES	FreeStorageSpace	テンプレートで指定した値
AWS/ES	MasterCPUUtilization	50%を超えた場合
AWS/ES	MasterJVMMemoryPressure	80%を超えた場合
AWS/ES	CPUUtilization	50%を超えた場合
AWS/ES	JVMMemoryPressure	80%を超えた場合
AWS/ES	SysMemoryUtilization	80%を超えた場合

Kinesis Data Firehose

最後に Kinesis Data Firehose の設定を行います。Firehoseは、ストリーミングデータを取り込んで変換し、 Amazon S3、Amazon Redshift、Amazon Elasticsearch Service、汎用 HTTP エンドポイントなどに配信できるサービスで、今回は Kinesisストリームに配信されたリアルライムログをElasticsearch Service に配信 します。

なお、Kinesisストリームでは、レコードを格納する際に Base64エンコードする必要があるため、 リアルタイムログもBase64エンコードされた状態で格納 されています。そこで、 Firehoseが持つデータ変換機能を用いて、対象のカラムをBase64デコード します。Firehoseのデータ変換機能は、変換処理を記述したAWS Lambdaを紐づけることで実現します。

このLambda関数にアタッチするIAM Roleは、以下の通りです。Lambda関数に与える権限は、CloudWatch Logs への書き込み権限 です。

Resources:
  IAMRoleForLambda:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: 'sts:AssumeRole'
      Description: A role required for Lambda to execute.
      Policies:
        - PolicyName: !Sub '${AWS::StackName}-AWSLambdaCloudWatchLogsPolicy-${AWS::Region}'
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'logs:CreateLogStream'
                Resource: '*'
              - Effect: Allow
                Action:
                  - 'logs:PutLogEvents'
                Resource: '*' 
      RoleName: !Sub '${AWS::StackName}-Lambda-${AWS::Region}'

また、データ変換処理を行う Lambda 関数は、以下の通りです。

Resources:
  Lambda:
    Type: 'AWS::Lambda::Function'
    Properties:
      Code:
        ZipFile: |
          import logging
          import base64
          import json

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

          logger.info("Loading function")

          def lambda_handler(event, context):
              output = []

              # Based on the fields chosen during the creation of the Real-time log configuration.
              # The order is important and please adjust the function if you have removed certain default fields from the configuration.
              realtimelog_fields_dict = {
                  "timestamp": "float",
                  "c-ip": "str",
                  "time-to-first-byte": "float",
                  "sc-status": "int",
                  "sc-bytes": "int",
                  "cs-method": "str",
                  "cs-protocol": "str",
                  "cs-host": "str",
                  "cs-uri-stem": "str",
                  "cs-bytes": "int",
                  "x-edge-location": "str",
                  "x-edge-request-id": "str",
                  "x-host-header": "str",
                  "time-taken": "float",
                  "cs-protocol-version": "str",
                  "c-ip-version": "str",
                  "cs-user-agent": "str",
                  "cs-referer": "str",
                  "cs-cookie": "str",
                  "cs-uri-query": "str",
                  "x-edge-response-result-type": "str",
                  "x-forwarded-for": "str",
                  "ssl-protocol": "str",
                  "ssl-cipher": "str",
                  "x-edge-result-type": "str",
                  "fle-encrypted-fields": "str",
                  "fle-status": "str",
                  "sc-content-type": "str",
                  "sc-content-len": "int",
                  "sc-range-start": "int",
                  "sc-range-end": "int",
                  "c-port": "int",
                  "x-edge-detailed-result-type": "str",
                  "c-country": "str",
                  "cs-accept-encoding": "str",
                  "cs-accept": "str",
                  "cache-behavior-path-pattern": "str",
                  "cs-headers": "str",
                  "cs-header-names": "str",
                  "cs-headers-count": "int",
              }

              for record in event["records"]:

                  # Extracting the record data in bytes and base64 decoding it
                  payload_in_bytes = base64.b64decode(record["data"])

                  # Converting the bytes payload to string
                  payload = "".join(map(chr, payload_in_bytes))

                  # dictionary where all the field and record value pairing will end up
                  payload_dict = {}

                  # counter to iterate over the record fields
                  counter = 0

                  # generate list from the tab-delimited log entry
                  payload_list = payload.strip().split("\t")

                  # perform the field, value pairing and any necessary type casting.
                  # possible types are: int, float and str (default)
                  for field, field_type in realtimelog_fields_dict.items():
                      # overwrite field_type if absent or '-'
                      if payload_list[counter].strip() == "-":
                          field_type = "str"
                      if field_type == "int":
                          payload_dict[field] = int(payload_list[counter].strip())
                      elif field_type == "float":
                          payload_dict[field] = float(payload_list[counter].strip())
                      else:
                          payload_dict[field] = payload_list[counter].strip()
                      counter = counter + 1

                  # JSON version of the dictionary type
                  payload_json = json.dumps(payload_dict)

                  # Preparing JSON payload to push back to Firehose
                  payload_json_ascii = payload_json.encode("ascii")
                  output_record = {
                      "recordId": record["recordId"],
                      "result": "Ok",
                      "data": base64.b64encode(payload_json_ascii).decode("utf-8"),
                  }
                  output.append(output_record)

              logger.info("Successfully processed {} records.".format(len(event["records"])))

              return {"records": output}
      Description: CloudFrontログを変換します
      FunctionName: realtimeLogsTransformer
      Handler: index.lambda_handler
      MemorySize: 512
      Role: !GetAtt IAMRoleForLambda.Arn
      Runtime: python3.8
      Timeout: 60
      TracingConfig:
        Mode: Active
  LambdaLogGroup:
    Type: 'AWS::Logs::LogGroup'
    Properties: 
      LogGroupName: !Sub /aws/lambda/${Lambda}
      RetentionInDays: 60

次に、Firehose から Elasticsearch Service への配信が失敗した場合に、 代わりにリアルタイムログを格納するS3バケットも事前に作成 しておきます。

Resources:
  S3ForKinesisFirehose:
    Type: 'AWS::S3::Bucket'
    UpdateReplacePolicy: Retain
    DeletionPolicy: Retain
    Properties:
      BucketEncryption:
        ServerSideEncryptionConfiguration: 
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256
      BucketName: !Sub ${ElasticSearchDomainName}-${AWS::Region}-${AWS::AccountId}
      LifecycleConfiguration:
        Rules:
          - Id: ExpirationInDays
            ExpirationInDays: 60
            Status: Enabled
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true

さらに、Firehoseに付与する権限をIAM Roleで規定します。

  IAMRoleForKinesisFirehose:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: firehose.amazonaws.com
            Action: 'sts:AssumeRole'
      Description: A role required for KinesisFirehose to access Glue, S3, Lambda, CloudWatch Logs, Kinesis and KMS.
      Policies:
        - PolicyName: !Sub '${AWS::StackName}-FirehoseDelivery-${AWS::Region}'
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - 's3:AbortMultipartUpload'
                  - 's3:GetBucketLocation'
                  - 's3:GetObject'
                  - 's3:ListBucket'
                  - 's3:ListBucketMultipartUploads'
                  - 's3:PutObject'
                Resource:
                  - !Sub 'arn:aws:s3:::${S3ForKinesisFirehose}'
                  - !Sub 'arn:aws:s3:::${S3ForKinesisFirehose}/*'
              - Effect: Allow
                Action:
                  - 'kms:Decrypt'
                  - 'kms:GenerateDataKey'
                Resource:
                  - !GetAtt KMSKey.Arn
                Condition:
                  StringEquals:
                    'kms:ViaService': s3.region.amazonaws.com
                  StringLike:
                    'kms:EncryptionContext:aws:s3:arn': !Sub 'arn:aws:s3:::${S3ForKinesisFirehose}/*'
              - Effect: Allow
                Action:
                  - 'es:DescribeElasticsearchDomain'
                  - 'es:DescribeElasticsearchDomains'
                  - 'es:DescribeElasticsearchDomainConfig'
                  - 'es:ESHttpPost'
                  - 'es:ESHttpPut'
                Resource:
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/*'
              - Effect: Allow
                Action:
                  - 'es:ESHttpGet'
                Resource:
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/_all/_settings'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/_cluster/stats'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/realtime*/_mapping/*'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/_nodes'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/_nodes/stats'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/_nodes/*/stats'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/_stats'
                  - !Sub 'arn:aws:es:${AWS::Region}:${AWS::AccountId}:domain/${ElasticSearchDomainName}/realtime*/_stats'
              - Effect: Allow
                Action:
                  - 'kinesis:DescribeStream'
                  - 'kinesis:GetShardIterator'
                  - 'kinesis:GetRecords'
                  - 'kinesis:ListShards'
                Resource: !GetAtt Kinesis.Arn
              - Effect: Allow
                Action:
                  - 'logs:PutLogEvents'
                Resource:
                  - !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:${CloudWatchLogsGroupForFirehose}:log-stream:${CloudWatchLogsStreamForFirehoseElasticSearch}'
                  - !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:${CloudWatchLogsGroupForFirehose}:log-stream:${CloudWatchLogsStreamForFirehoseS3}'
              - Effect: Allow
                Action:
                  - 'lambda:InvokeFunction'
                  - 'lambda:GetFunctionConfiguration'
                Resource:
                  - !GetAtt Lambda.Arn
              - Effect: Allow
                Action:
                  - 'kms:Decrypt'
                Resource:
                  - !GetAtt KMSKey.Arn
                Condition:
                  StringEquals:
                    'kms:ViaService': kinesis.ap-northeast-1.amazonaws.com
                  StringLike:
                    'kms:EncryptionContext:aws:kinesis:arn': !GetAtt Kinesis.Arn
      RoleName: !Sub '${AWS::StackName}-Firehose-${AWS::Region}'

最後に、リアルタイムログの配信先となる Elasticsearch Service と S3、データ変換機能を提供するLambda関数、それぞれを指定して Firehoseを作成 します。なお、Elasticsearch Serviceへの データ配信に係る遅延量をできるだけ少なくするため、 BufferingHints を設定可能な最小の値 にしています。また、 S3BackupMode の値を FailedDocumentsOnly とすることで、 Elasticsearch Serviceへの配信が失敗した場合のみ、S3にリアルタイムログを保存 する挙動にしています。AWS::KinesisFirehose::DeliveryStream の一部属性は更新時の動作が Replacement 、つまり 更新時にはFirehoseリソースが再作成され物理IDも新規で作成 する必要があります。このため、当該属性の値を更新する場合は DeliveryStreamName の値も同時に更新してください。

Resources:
  KinesisFirehose:
    Type: 'AWS::KinesisFirehose::DeliveryStream'
    Properties:
      DeliveryStreamName: !Sub ${AWS::StackName}-${KinesisFirehoseStreamNameSuffix}
      DeliveryStreamType: KinesisStreamAsSource
      KinesisStreamSourceConfiguration:
        KinesisStreamARN: !GetAtt Kinesis.Arn
        RoleARN: !GetAtt IAMRoleForKinesisFirehose.Arn
      ElasticsearchDestinationConfiguration:
        BufferingHints:
          IntervalInSeconds: 60
          SizeInMBs: 1
        CloudWatchLoggingOptions: 
          Enabled: true
          LogGroupName: !Ref CloudWatchLogsGroupForFirehose
          LogStreamName: !Ref CloudWatchLogsStreamForFirehoseS3
        DomainARN: !GetAtt ElasticSearchDomain.DomainArn
        IndexName: realtime
        IndexRotationPeriod: NoRotation
        ProcessingConfiguration: 
          Enabled: true
          Processors: 
            - Parameters: 
                - ParameterName: LambdaArn
                  ParameterValue: !GetAtt Lambda.Arn
              Type: Lambda
        RetryOptions: 
          DurationInSeconds: 300
        RoleARN: !GetAtt IAMRoleForKinesisFirehose.Arn
        S3BackupMode: FailedDocumentsOnly
        S3Configuration: 
          BucketARN: !GetAtt S3ForKinesisFirehose.Arn
          CloudWatchLoggingOptions:
            Enabled: true
            LogGroupName: !Ref CloudWatchLogsGroupForFirehose
            LogStreamName: !Ref CloudWatchLogsStreamForFirehoseElasticSearch
          RoleARN: !GetAtt IAMRoleForKinesisFirehose.Arn
        TypeName: ''
  CloudWatchLogsGroupForFirehose:
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: !Sub '/aws/kinesisfirehose/${AWS:StackName}'
      RetentionInDays: 60
  CloudWatchLogsStreamForFirehoseS3:
    Type: 'AWS::Logs::LogStream'
    Properties:
      LogGroupName: !Ref CloudWatchLogsGroupForFirehose
      LogStreamName: S3
  CloudWatchLogsStreamForFirehoseElasticSearch:
    Type: 'AWS::Logs::LogStream'
    Properties:
      LogGroupName: !Ref CloudWatchLogsGroupForFirehose
      LogStreamName: ElasticSearch

以上で、CloudFront のリアルタイムログを可視化するために必要な全てのリソースの設定が完了しました。この CloudFormation テンプレートを実行することで、それぞれのリソースがデプロイされます。

なお、Firehose が正常に動作しているかどうかを確認するためには、以下のCloudWatchメトリクスを監視してください。こちらのリンク から、これらのメトリクスを基にしたCloudWatchアラームを一括で有効化することができます。

ネームスペース	メトリクス	閾値
AWS/Firehose	DeliveryToElasticsearch.DataFreshness	テンプレートで指定した値
AWS/Firehose	ThrottledGetShardIterator	1分間に1回以上
AWS/Firehose	ThrottledGetRecords	1分間に1回以上
AWS/Firehose	DeliveryToElasticsearch.Success	1より小さい場合

Kibana の設定

上記のリソースのデプロイが完了したあとは、 Elasticsearch に取り込んだデータにインデックスを指定し、Kibana を用いてこれを可視化する ための設定を行います。この手順の詳細については、 こちら にも記載がありますので、本記事と合わせてご一読ください。下記の作業を行うことで、 Firehose が Elasticsearch に対してデータを投入できるようになる とともに、 可視化に必要なグラフおよびダッシュボードが自動で作成 されます。

Security の Roles を選択します。

+ アイコンをクリックして新しいロールを追加します。
作成したロールに firehose という名前をつけます。

Cluster Permissions タブの Cluster-wide permissions で cluster_composite_ops cluster_monitor グループを追加します。

Index Permissions タブの Add index permissions から Index Patterns を選んで realtime* を入力します。Permissions: Action Groups で crud create_index manage アクショングループを追加します。

Save Role Definition をクリックします。
Security の Role Mappings を選択します。

Add Backend Role をクリックします。
先ほど作成した firehoseを選択します。
Backend roles に Kinesis Data Firehose が Amazon ES および S3 に書き込むために使用する IAM ロールの ARN を入力します。

Submit をクリックします。
Dev Tools を選択します。
timestamp フィールドを date タイプと認識させるために、以下のコマンドを入力して実行します。

PUT _template/custom_template
{
    "template": "realtime*",
    "mappings": {
        "properties": {
            "timestamp": {
                "type": "date",
                "format": "epoch_second"
            }
        }
    }
}