LoginSignup
8
9

More than 5 years have passed since last update.

@kikeyama

Kinesis Firehoseを使ってSplunkでAWSユーザーアクティビティを可視化してみた

Last updated at Posted at 2018-02-23

AWS re:invent 2017でKinesis Firehose経由でのSplunkへのデータ投入ができるようになったと発表されました。

Amazon Kinesis Data Firehose、配信先として Splunk が一般利用可能に

CloudTrailやVPC Flow、CloudWatch Event、IoTをFirehose経由でストリームできるらしい。

ということで、試してみた。

はじめに

Splunk: バージョン7.0.2
OS: Amazon Linux on EC2

AWSユーザーアクティビティをモニタリングするには CloudTrail から証跡情報をとる必要があります。

データの流れは↓
CloudTrail --> CloudWatch Logs --> Kinesis Firehose --> Splunk

（参考）Power data ingestion into Splunk using Amazon Kinesis Data Firehose

Splunk設定

最初にデータの受け口となるSplunkをセットアップ

Appインストール

Splunk Add-on for Amazon Kinesis Firehose をSplunkにインストール

HTTP Event Collector設定

Firehose用のHECトークンを作成
[設定] > [データ入力] から [HTTPイベントコレクタ] を選択

[新規トークン] をクリック
Screen Shot 2018-02-23 8.19.25 PM.png

名前を入力して インデクサー確認を有効化 にチェック
HEC入力設定ごとにソースタイプを決めることになるので、CloudTrailからのデータとわかるよう aws_kinesis_firehose_cloudtrail と命名
Screen Shot 2018-02-23 8.27.21 PM.png

ソースタイプ を aws:cloudtrail に、 インデックス を main に設定
（データへのアクセス権を制御したい場合は個別にインデックスを作成）
Screen Shot 2018-02-23 8.23.58 PM.png

そのまま次の画面で内容確認して問題なければ [実行]

作成したHEC入力のトークン値を控えておく
Screen Shot 2018-02-23 8.29.52 PM_mosaic.png

SplunkのHTTPS設定

Amazon Kinesis Firehoseはデータ送信先のエンドポイントが商用のSSL証明書で暗号化されていることが必須要件となるらしい。
オレオレはダメみたい。

前回、Let's Encryptの証明書をSplunkにインストールしてみた ところ、FirehoseはLet's Encryptを認証してくれませんでした（2018年1月時点）。

今回のSplunkインスタンスはEC2で稼働しているので、前段にELBを入れて、ACM (Amazon Certificate Manager) からAmazon署名の証明書を取得して対処しました。

ターゲットのEC2には、プロトコル HTTPS でポート 8088 に転送するようにしました。
Screen Shot 2018-02-23 8.41.25 PM.png

ELBが受ける通信は HTTPS ポート 443 で設定
Screen Shot 2018-02-23 8.42.22 PM.png

これでSplunk側の設定は完了！

AWS設定

続いてAWS側の設定
Lambda, Firehose, CloudWatch Logs, CloudTrailの順に設定します。

Lambda設定

まずはLambdaから

なぜLambdaが必要かって？
なぜなら、CloudWatch Logsから送られるデータは圧縮されているので、それをFirehose経由で送ってもSplunkが受け取れないから、解凍する処理を経由させる必要があるんです。

ちなみに、解凍の処理を入れずにFirehoseからSplunkに流そうとすると、以下のような InvalidEncodingException エラーが出ます。


{"attemptsMade":0,"arrivalTimestamp":1517204317073,"errorCode":"InvalidEncodingException","errorMessage":"The data could not be decoded as UTF-8","attemptEndingTimestamp":25314598570688,"rawData":"...","EventId":null}

ということで、Lambdaで新しい関数を作成

設計図（Blueprint）から kinesis-firehose-cloudwatch-logs-processor を選択して新規作成
検索フィルタに firehose splunk と入力すると出てきます。

ソースコード


/*
For processing data sent to Firehose by Cloudwatch Logs subscription filters.

Cloudwatch Logs sends to Firehose records that look like this:

{
  "messageType": "DATA_MESSAGE",
  "owner": "123456789012",
  "logGroup": "log_group_name",
  "logStream": "log_stream_name",
  "subscriptionFilters": [
    "subscription_filter_name"
  ],
  "logEvents": [
    {
      "id": "01234567890123456789012345678901234567890123456789012345",
      "timestamp": 1510109208016,
      "message": "log message 1"
    },
    {
      "id": "01234567890123456789012345678901234567890123456789012345",
      "timestamp": 1510109208017,
      "message": "log message 2"
    }
    ...
  ]
}

The data is additionally compressed with GZIP.

The code below will:

1) Gunzip the data
2) Parse the json
3) Set the result to ProcessingFailed for any record whose messageType is not DATA_MESSAGE, thus redirecting them to the
   processing error output. Such records do not contain any log events. You can modify the code to set the result to
   Dropped instead to get rid of these records completely.
4) For records whose messageType is DATA_MESSAGE, extract the individual log events from the logEvents field, and pass
   each one to the transformLogEvent method. You can modify the transformLogEvent method to perform custom
   transformations on the log events.
5) Concatenate the result from (4) together and set the result as the data of the record returned to Firehose. Note that
   this step will not add any delimiters. Delimiters should be appended by the logic within the transformLogEvent
   method.
*/

'use strict';

const zlib = require('zlib');

/**
 * logEvent has this format:
 *
 * {
 *   "id": "01234567890123456789012345678901234567890123456789012345",
 *   "timestamp": 1510109208016,
 *   "message": "log message 1"
 * }
 *
 * The default implementation below just extracts the message and appends a newline to it.
 *
 * The result must be returned in a Promise.
 */
function transformLogEvent(logEvent) {
    return Promise.resolve(`${logEvent.message}\n`);
}

exports.handler = (event, context, callback) => {
    Promise.all(event.records.map(r => {
        const buffer = new Buffer(r.data, 'base64');
        const decompressed = zlib.gunzipSync(buffer);
        const data = JSON.parse(decompressed);
        if (data.messageType !== 'DATA_MESSAGE') {
            return Promise.resolve({
                recordId: r.recordId,
                result: 'ProcessingFailed',
            });
        } else {
            const promises = data.logEvents.map(transformLogEvent);
            return Promise.all(promises).then(transformed => {
                const payload = transformed.reduce((a, v) => a + v, '');
                const encoded = new Buffer(payload).toString('base64');
                return {
                    recordId: r.recordId,
                    result: 'Ok',
                    data: encoded,
                };
            });
        }
    })).then(recs => callback(null, { records: recs }));
};

Kinesis Firehose設定

続いてFirehose
Kinesisの設定画面の左側メニューから Data Firehose を選択
Screen Shot 2018-02-23 10.16.30 PM.png

Create Delivery Stream をクリックして新しいストリームを作成
ストリーム名を入力して
firehose__01_01.png

Sourceは Direct PUT or other sources を選択
firehose_01_02.png

Transform records with AWS Lambda で作成した関数を指定
firehose_02_01.png

IAMロールのポリシーはデフォルトでおk

続いてデータ送信先を指定
firehose_03_01.png

S3、Redshift、Elasticsearch とAWSのサービスと同列にサードパーティーの Splunkが並んでるのすごい！

で、　Splunkのエンドポイントとトークンを指定
firehose_03_02_mosaic.png

エンドポイントはELBで設定したFQDNを（ELB介さない場合はポート :8088 の指定が必要）、トークンは先のステップでSplunkで作成したトークンを指定
Splunk endpoint type は Raw endpoint を指定

送信失敗した場合の転送先S3バケットは適当に

より早く送信してほしいので、バッファサイズと時間はなるべく小さくしてみた
firehose_04_01.png

内容確認して作成！

CloudWatch Logs設定

（参考） 例 3: Amazon Kinesis Data Firehose のサブスクリプションフィルタ

まずはCWLロググループをひとつ kikeyama/CloudTrail 作成

その後、SubscriptionでFirehoseに流す

aws logs put-subscription-filter \
    --log-group-name "kikeyama/CloudTrail" \
    --filter-name "Destination" \
    --filter-pattern "" \
    --destination-arn "arn:aws:firehose:us-west-2:123456789012:deliverystream/kikeyama-kinesis-firehose-cloudtrail" \
    --role-arn "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"

--filter-pattern は必須のパラメータなので、何かしらのフィルタリングが必要。 "" 指定で全条件になるっぽい。
フィルタの指定方法 → CloudWatch Logs サブスクリプションフィルタの使用

フィルタでの絞り込みなしで全データ取り込みたい場合は、 --filter-pattern "" でいけるはず。

ちなみにIAMロールのポリシーは↓

CWLtoKinesisFirehoseRole
{
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "firehose:*"
            ],
            "Resource": [
                "arn:aws:firehose:us-west-2:123456789012:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "iam:PassRole"
            ],
            "Resource": [
                "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"
            ]
        }
    ]
}

-- 2018年2月27日追記 ｺｺｶﾗ --
ロールの信頼関係を編集して権限付与

Screen Shot 2018-02-26 7.32.43 PM.png

{
  "Statement": {
    "Effect": "Allow",
    "Principal": { "Service": "logs.amazonaws.com" },
    "Action": "sts:AssumeRole"
  }
}

-- 2018年2月27日追記 ｺｺﾏﾃﾞ --

以上！

CloudTrail設定

最後に、CloudTrailを作ってオシマイ

CWLの設定に先ほど作ったロググループを指定
Screen Shot 2018-02-23 10.54.54 PM.png

IAMは↓

CloudTrail_CloudWatchLogs_Role
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AWSCloudTrailCreateLogStream20141101",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogStream"
      ],
      "Resource": [
        "arn:aws:logs:us-west-2:123456789012:log-group:kikeyama/CloudTrail:log-stream:123456789012_CloudTrail_us-west-2*"
      ]
    },
    {
      "Sid": "AWSCloudTrailPutLogEvents20141101",
      "Effect": "Allow",
      "Action": [
        "logs:PutLogEvents"
      ],
      "Resource": [
        "arn:aws:logs:us-west-2:123456789012:log-group:kikeyama/CloudTrail:log-stream:123456789012_CloudTrail_us-west-2*"
      ]
    }
  ]
}

Splunkでサーチ

Screen Shot 2018-02-23 11.02.12 PM_mosaic.png

いいねー、データ入ってきてるねー

ユーザーアクティビティを可視化

可視化には Splunk App for AWS を使います。
SplunkにこのAppをインストールして、Appのメニューから Security > User Activity に進むと、、、
Screen Shot 2018-02-23 11.36.12 PM.png

↓↓↓

おっ！
Screen Shot 2018-02-23 11.38.25 PM.png

エラーや非認証のアクティビティが集計して可視化されてる！
非認証多いな、なんでだろう...

CloudTrailで 証跡情報を全てのリージョンに適用 にチェック入れてると、全リージョン情報が出てくるし、ダッシュボードでは自分が見たいリージョンだけをフィルタリングして表示することもできるみたい。

おわりに

この設定、AWS素人の自分にはなかなか難しかったです。

つまずいたポイントは、、、

  • FirehoseからSplunkに送信する際に "errorCode":"InvalidEncodingException","errorMessage":"The data could not be decoded as UTF-8" がS3に吐き出されていたものの、対処法がわからず右往左往
  • 各サービスでIAMのロール設定が必要になるけど、どんな権限付与すればいいのかわからない（というより、ポリシーの設定方法がよくわかってなかった）
  • FirehoseがLet's Encryptを認証してくれない

ってところですかね。

VPC Flow Logも同じようにCWLからFirehose経由でSplunkに送ることができます。
その際、Splunkに取り込む際のソースタイプがCloudTrailとは異なるので、HECトークンもCloudTrailとは別々に作る必要があります。

うまくいってなかなかの達成感でした。

8
9
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
8
9