[AWS] Cloudwatch_LogsからLambda経由で軽いETLしつつOpenSearchでApacheログを可視化する方法

# やりたい事

2022/01/13開催の [SB Tech Festival](https://www.softbank.jp/biz/events/techfestival-deeptech-2022/) で登壇した内容、「ビッグデータ活用の第一歩AWS環境での大容量ログ可視化」の詳細手順(後半)ページです。

### このエントリで解説するのは以下のイメージ
![2022-01-11_09h20_031.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/23980234-e06e-96dd-3d8b-0bba4493cd57.png)

前半部分のFargate+Firelens+CloudwatchLogsについては[こちら](https://qiita.com/Kedamari/items/fba00598c0bd9e08fdb9)の記事を参照してください。

# 手順
2022/01/11時点でECSコンソールは新しいエクスペリエンスが提供されていますが、本エントリでは旧UIをベースに解説しています。

## 1.OpenSearchクラスターを作成する

AWSコンソールからOpenSearchダッシュボードを開き、新しいドメインを作成します。
![2022-01-11_15h46_39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/d1d2ac8b-03db-9871-1b92-cc99e256b57d.png)

|項目名|設定内容|
|---|---|
|ドメイン名|<任意の名前>|
|カスタムエンドポイント|チェック無し|
|デプロイタイプ|開発及びテスト|
|バージョン|1.0 <br> ※より新しいバージョンがリリースされていたとしても、Lambdaコードの後方互換が切られる可能性があります。自身でコード改修出来ない場合は、1.0選択をお願いします|
|自動調整|無効化|
|データノード|t3.small.search<br> ※インスタンスサイズを大きくするのは任意ですが、課金に直結するのでご注意ください。t3.smallは無料利用枠有り|
|ノードの数|1|
|ストレージタイプ|EBS|
|EBS ボリュームタイプ|汎用(SSD)|
|ノード当たりのEBSストレージサイズ|20GiB|
|きめ細かなアクセスコントロール|チェック無し|
|アクセスポリシー|ドメインレベルのアクセスポリシーの設定(ビジュアルエディタでダッシュボードへのアクセスを許可するIPv4/v6アドレスを指定します)|
|暗号化|全てチェック|
|AWS KMSキーを選択する|AWS 所有キーを使用する|

作成後、10分ほど時間をおいてデプロイが完了する事を確認してください。
(クラスターのヘルスは黄色or緑のいずれかであれば、次の手順へ進んで問題ありません。)
![2022-01-11_15h57_31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/be1f8e2f-3861-d9a7-0a75-617d2243abc5.png)

## 2.CloudWatch Logs用IAMロールを作成する

IAM管理ポリシーを新規作成し、
![2022-01-11_09h42_56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/12015437-6ff2-e444-b1fa-f2ef04bc71f6.png)

JSONタブを開き
![2022-01-11_09h43_27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/ddab8dd6-10dd-76cf-4124-20ecd1462662.png)

以下のJSONをコピペする。これはLambdaからCloudwatchLogs及びElasticSearchの必要な操作が出来るように権限を付与する内容。
より権限を絞ったり、リソースを指定するのがより良いですが、今回はその他リソースが無い前提で広く権限を持たせます。

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "logs:*",
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": "es:*",
            "Resource": "*"
        }
    ]
}
```

続いてIAMロールを新規作成し、
![2022-01-11_09h45_45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/9a9fcd6e-5e5e-8a22-5056-a513b47a2771.png)

以下の通りにロールの利用方法を指定。
![2022-01-11_19h31_32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/bae9fb48-0bac-1de6-7a62-6380d0ba03cc.png)

次の画面で、作成済みのIAM管理ポリシーを指定しLambda向けロール作成完了。

:::note info
IAM管理ポリシー、IAMロールのタグや名前は任意の内容でOK。
:::

## 3.CloudWatch Logsサブスクリプションフィルターを作成する

前半で作成したCloudWatch Logグループ **/apache-dummy-log**を選択し、図示の通りOpenSearch向けのサブスクリプションフィルターを作成
![2022-01-11_16h10_53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/20c14337-5950-fd64-57fb-0f43d4ef926c.png)

|項目名|設定内容|
|---|---|
|アカウントを選択|This account|
|Amazon OpenSearch Service cluster|<手順1で作成したOpenSearchドメインを指定>|
|ログの形式|JSON|
|サブスクリプションフィルターのパターン|<未記入のまま>|
|サブスクリプションフィルター名|<任意の名前を記載>|

以下のようにサブスクリプションフィルターが作成された事を確認してください。
![2022-01-11_16h15_52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/f50a5837-4eff-62c1-6202-b62d09edb866.png)

## 4.Lambdaで軽いETL処理を実装する

今回はApache2のダミーログを発生させていますが、ご存じの通りApache2のログはスペース区切りでフィールド名がありません。
OpenSearchへ投入する前に、Lambda上でパースしてフィールド名を割り当てましょう。

Lambdaコンソールを開き、関数一覧の中に**LogsToElasticsearch_apachedummylog-opensearch**がある事を確認し、クリックします。
![2022-01-11_16h18_47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/113a13cc-2140-b8e5-f725-0ec283c448b2.png)

設定タブを開き、実行ロールの編集をクリックします。
![2022-01-12_19h54_05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/74b85b5c-1aff-592d-c7a3-bce35ca500ba.png)

手順#2で作成したIAMロールを指定し、保存をクリックします。
![2022-01-12_19h54_00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/9a4d1961-3570-8160-82db-ee19e4a93e11.png)

続いてコードタブを開き、関数の76行目以降に以下を追加します。

```Node.js
        var messageConvJson = logEvent.message.replace(/"/g,'\"');
        messageConvJson = messageConvJson.replace(/{\\"/,'{"');
        messageConvJson = messageConvJson.replace(/\\"\s*:\s*\\"/g,'":"');
        messageConvJson = messageConvJson.replace(/\\"\s*,\s*\\"/g,'","');
        messageConvJson = messageConvJson.replace(/\\"}/,'"}');
        var messageJson = JSON.parse(messageConvJson);

        let [ remote_host, remote_logname, remote_user, ts, tz, method, resource, protocol, statusCode, size, referer, ...userAgent ] = messageJson.log.split(' ');
        
        statusCode = +statusCode;
        size = +size;
        ts = ts.substring(1);
        tz = tz.substring(0, tz.length - 1);
        const accesstime = ts + tz;
        userAgent = userAgent.join('');
        
        source['@remote_host'] = remote_host;
        source['@accesstime'] = accesstime;
        source['@method'] = method;
        source['@resource'] = resource;
        source['@protocol'] = protocol;
        source['@statusCode'] = statusCode;
        source['@size'] = size;
        source['@referer'] = referer;
        source['@userAgent'] = userAgent;
```

CloudWatch Logsから自動デプロイされるコードに、上記を含めたコード全体も載せておきます。
こちらをコピペしていただくのが簡単です。
<details><summary>コード全体</summary><div>

```Node.js
// v1.1.2
var https = require('https');
var zlib = require('zlib');
var crypto = require('crypto');

var endpoint = 'search-apachedummylog-opensearch-amepwsh24gynwv63reszss5u4e.ap-northeast-1.es.amazonaws.com';

// Set this to true if you want to debug why data isn't making it to
// your Elasticsearch cluster. This will enable logging of failed items
// to CloudWatch Logs.
var logFailedResponses = false;

exports.handler = function(input, context) {
    // decode input from base64
    var zippedInput = new Buffer.from(input.awslogs.data, 'base64');

    // decompress the input
    zlib.gunzip(zippedInput, function(error, buffer) {
        if (error) { context.fail(error); return; }

        // parse the input from JSON
        var awslogsData = JSON.parse(buffer.toString('utf8'));

        // transform the input to Elasticsearch documents
        var elasticsearchBulkData = transform(awslogsData);

        // skip control messages
        if (!elasticsearchBulkData) {
            console.log('Received a control message');
            context.succeed('Control message handled successfully');
            return;
        }

        // post documents to the Amazon Elasticsearch Service
        post(elasticsearchBulkData, function(error, success, statusCode, failedItems) {
            console.log('Response: ' + JSON.stringify({
                "statusCode": statusCode
            }));

            if (error) {
                logFailure(error, failedItems);
                context.fail(JSON.stringify(error));
            } else {
                console.log('Success: ' + JSON.stringify(success));
                context.succeed('Success');
            }
        });
    });
};

function transform(payload) {
    if (payload.messageType === 'CONTROL_MESSAGE') {
        return null;
    }

    var bulkRequestBody = '';

    payload.logEvents.forEach(function(logEvent) {
        var timestamp = new Date(1 * logEvent.timestamp);

        // index name format: cwl-YYYY.MM.DD
        var indexName = [
            'cwl-' + timestamp.getUTCFullYear(),              // year
            ('0' + (timestamp.getUTCMonth() + 1)).slice(-2),  // month
            ('0' + timestamp.getUTCDate()).slice(-2)          // day
        ].join('.');

        var source = buildSource(logEvent.message, logEvent.extractedFields);
        source['@id'] = logEvent.id;
        source['@timestamp'] = new Date(1 * logEvent.timestamp).toISOString();
        source['@message'] = logEvent.message;
        source['@owner'] = payload.owner;
        source['@log_group'] = payload.logGroup;
        source['@log_stream'] = payload.logStream;

        //console.log("logEvent.message = " + logEvent.message);
        var messageConvJson = logEvent.message.replace(/"/g,'\"');
        messageConvJson = messageConvJson.replace(/{\\"/,'{"');
        messageConvJson = messageConvJson.replace(/\\"\s*:\s*\\"/g,'":"');
        messageConvJson = messageConvJson.replace(/\\"\s*,\s*\\"/g,'","');
        messageConvJson = messageConvJson.replace(/\\"}/,'"}');
        //console.log("messageConvJson = " + messageConvJson);
        var messageJson = JSON.parse(messageConvJson);
        //console.log("messageJson.log = " + messageJson.log);

        let [ remote_host, remote_logname, remote_user, ts, tz, method, resource, protocol, statusCode, size, referer, ...userAgent ] = messageJson.log.split(' ');
        
        statusCode = +statusCode;
        size = +size;
        ts = ts.substring(1);
        tz = tz.substring(0, tz.length - 1);
        const accesstime = ts + tz;
        userAgent = userAgent.join('');
        
        source['@remote_host'] = remote_host;
        source['@accesstime'] = accesstime;
        source['@method'] = method;
        source['@resource'] = resource;
        source['@protocol'] = protocol;
        source['@statusCode'] = statusCode;
        source['@size'] = size;
        source['@referer'] = referer;
        source['@userAgent'] = userAgent;
        console.log(source)

        var action = { "index": {} };
        action.index._index = indexName;
        action.index._type = payload.logGroup;
        action.index._id = logEvent.id;

        bulkRequestBody += [
            JSON.stringify(action),
            JSON.stringify(source),
        ].join('\n') + '\n';
    });
    return bulkRequestBody;
}

function buildSource(message, extractedFields) {
    if (extractedFields) {
        var source = {};

        for (var key in extractedFields) {
            if (extractedFields.hasOwnProperty(key) && extractedFields[key]) {
                var value = extractedFields[key];

                if (isNumeric(value)) {
                    source[key] = 1 * value;
                    continue;
                }

                var jsonSubString = extractJson(value);
                if (jsonSubString !== null) {
                    source['$' + key] = JSON.parse(jsonSubString);
                }

                source[key] = value;
            }
        }
        return source;
    }

    var jsonSubString = extractJson(message);
    if (jsonSubString !== null) {
        return JSON.parse(jsonSubString);
    }

    return {};
}

function extractJson(message) {
    var jsonStart = message.indexOf('{');
    if (jsonStart < 0) return null;
    var jsonSubString = message.substring(jsonStart);
    return isValidJson(jsonSubString) ? jsonSubString : null;
}

function isValidJson(message) {
    try {
        JSON.parse(message);
    } catch (e) { return false; }
    return true;
}

function isNumeric(n) {
    return !isNaN(parseFloat(n)) && isFinite(n);
}

function post(body, callback) {
    var requestParams = buildRequest(endpoint, body);

    var request = https.request(requestParams, function(response) {
        var responseBody = '';
        response.on('data', function(chunk) {
            responseBody += chunk;
        });

        response.on('end', function() {
            var info = JSON.parse(responseBody);
            var failedItems;
            var success;
            var error;

            if (response.statusCode >= 200 && response.statusCode < 299) {
                failedItems = info.items.filter(function(x) {
                    return x.index.status >= 300;
                });

                success = {
                    "attemptedItems": info.items.length,
                    "successfulItems": info.items.length - failedItems.length,
                    "failedItems": failedItems.length
                };
            }

            if (response.statusCode !== 200 || info.errors === true) {
                // prevents logging of failed entries, but allows logging
                // of other errors such as access restrictions
                delete info.items;
                error = {
                    statusCode: response.statusCode,
                    responseBody: info
                };
            }

            callback(error, success, response.statusCode, failedItems);
        });
    }).on('error', function(e) {
        callback(e);
    });
    request.end(requestParams.body);
}

function buildRequest(endpoint, body) {
    var endpointParts = endpoint.match(/^([^\.]+)\.?([^\.]*)\.?([^\.]*)\.amazonaws\.com$/);
    var region = endpointParts[2];
    var service = endpointParts[3];
    var datetime = (new Date()).toISOString().replace(/[:\-]|\.\d{3}/g, '');
    var date = datetime.substr(0, 8);
    var kDate = hmac('AWS4' + process.env.AWS_SECRET_ACCESS_KEY, date);
    var kRegion = hmac(kDate, region);
    var kService = hmac(kRegion, service);
    var kSigning = hmac(kService, 'aws4_request');

    var request = {
        host: endpoint,
        method: 'POST',
        path: '/_bulk',
        body: body,
        headers: {
            'Content-Type': 'application/json',
            'Host': endpoint,
            'Content-Length': Buffer.byteLength(body),
            'X-Amz-Security-Token': process.env.AWS_SESSION_TOKEN,
            'X-Amz-Date': datetime
        }
    };

    var canonicalHeaders = Object.keys(request.headers)
        .sort(function(a, b) { return a.toLowerCase() < b.toLowerCase() ? -1 : 1; })
        .map(function(k) { return k.toLowerCase() + ':' + request.headers[k]; })
        .join('\n');

    var signedHeaders = Object.keys(request.headers)
        .map(function(k) { return k.toLowerCase(); })
        .sort()
        .join(';');

    var canonicalString = [
        request.method,
        request.path, '',
        canonicalHeaders, '',
        signedHeaders,
        hash(request.body, 'hex'),
    ].join('\n');

    var credentialString = [ date, region, service, 'aws4_request' ].join('/');

    var stringToSign = [
        'AWS4-HMAC-SHA256',
        datetime,
        credentialString,
        hash(canonicalString, 'hex')
    ] .join('\n');

    request.headers.Authorization = [
        'AWS4-HMAC-SHA256 Credential=' + process.env.AWS_ACCESS_KEY_ID + '/' + credentialString,
        'SignedHeaders=' + signedHeaders,
        'Signature=' + hmac(kSigning, stringToSign, 'hex')
    ].join(', ');

    return request;
}

function hmac(key, str, encoding) {
    return crypto.createHmac('sha256', key).update(str, 'utf8').digest(encoding);
}

function hash(str, encoding) {
    return crypto.createHash('sha256').update(str, 'utf8').digest(encoding);
}

function logFailure(error, failedItems) {
    if (logFailedResponses) {
        console.log('Error: ' + JSON.stringify(error, null, 2));

        if (failedItems && failedItems.length > 0) {
            console.log("Failed Items: " +
                JSON.stringify(failedItems, null, 2));
        }
    }
}
```

</div></details>

コピペが完了したら、デプロイボタンをクリックします。
![2022-01-11_16h22_02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/9c684f33-71a1-3908-5a00-cf89412aba96.png)

最後にCloudWatch Logsコンソール上に、本Lambda関数のログストリームが作成され、エラー発生が無い事を確認します。
![2022-01-11_16h16_03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/40d26bbe-e12a-e11b-04ef-689715cbb95d.png)

## 5.OpenSearchの初期設定を実行する

OpenSearchコンソールから、OpenSearch Dashboards の URLを確認し、アクセスします。
![2022-01-11_16h25_42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/dc2dabdf-71d1-196b-fb58-7e87a034f4c4.png)

※正常にアクセスできない場合、アクセス元IPアドレスの指定が不完全な可能性が大きいです。
　ドメインの「セキュリティ設定>編集」 からアクセスポリシーを開き、ビジュアルエディタから許可したいIPアドレスを設定してください。

OpenSearchにアクセスできたら、左上のハンバーガーメニューから、[Index Management>Indices]と遷移し、「cwl-yyyy.mm.dd」という名前のインデックスが作成されている事を確認します。
![2022-01-11_16h28_22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/34b04741-5406-79e9-1361-55951122f294.png)

確認が出来たら、Index Patternsを登録します。
左上のハンバーガーメニューから、[Stack Management>Index patterns]と遷移し、「Create index pattern」をクリックします。
![2022-01-11_16h48_51.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/783d69e1-47eb-6e6c-8f90-63126e89e932.png)

「Index pattern name」に「cwl-*」を指定し、Next Step。
![2022-01-11_16h28_51.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/efd4f5f2-43a9-4626-c2e3-57ca318733b1.png)

「Time Field」に「@timestamp」を指定し、「Create index pattern」をクリック。
![2022-01-11_16h28_59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/806a6200-19f3-bff0-4ccc-93c932fd754d.png)

46のFieldを持つインデックスパターンが作成された事を確認します。
![2022-01-11_16h51_55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/2a0e89e9-88d6-8d00-3ae1-d69679b4d229.png)

## 6.OpenSearchでダッシュボードを作る

OpenSearch左上のハンバーガーメニューから、[Discover]へ遷移し、CHANGE INDEX PATTERNメニューから手順5で作成した、[cwl-*]を選択し、正しく転送されたログが表示される事を確認してください。
![2022-01-11_16h29_39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/ce303662-4727-8913-1922-f79c0bc219fd.png)

続いて左上のハンバーガーメニューから、[Visualize]へ遷移し、「Create Visualize」をクリック。
以下3パターンのグラフを作成します。
![2022-01-11_16h31_19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/f2ac6b11-4be8-efba-8060-39d92dc42194.png)

■Line(単位時間あたりの応答サイズ数合計)　→　Name:Apache_SizeChart
![2022-01-11_16h41_59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/7a51abea-1e22-19bf-9213-37b899ee4afc.png)

■Vertical Bar(UA TOP30)　→　Name:	Apache_StatusCode
![2022-01-11_16h36_31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/5f7d02d7-7ad6-d834-66d7-f3565a08d606.png)

■Pie(UA TOP30)　→　Name:Apache_UA
![2022-01-11_16h37_39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/3fa63d79-e65a-76a9-5991-46becb1fb784.png)

最後に左上のハンバーガーメニューから、[DashBoard]へ遷移し、今作成したVizualizeを並べます。
![2022-01-11_16h44_31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/94374b07-a018-0c75-31f2-c8e1412a8c2d.png)

簡単ですが、Apacheログの可視化が完了しました。

# 番外編
## OpenSearchで用意されているダミーデータで遊んでみる

OpenSearchダッシュボードへ最初に接続した段階でお気づきだと思いますが、OpenSearchには予め用意されているダミーデータがあります。
![2022-01-11_17h04_46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/9fd07752-25bc-7d53-f605-b44a2db711ea.png)

ECサイトのオーダーデータ、フライトデータ、WEB Logの3点セットになっていて、インデックスファイルのみではなくVizualizeやDashBoardも含まれています。
![2022-01-11_18h18_14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/def19467-3678-8442-a935-d742ef9a5861.png)

OpenSearchでどんなことが出来るのか？や、ダッシュボード作成のコツをつかむ事が出来ると思います。
是非活用してみてください。
セッションの中でお話した通り、OpenSearchだけを試してみるアプローチならば、後編手順の #1とこの番外編のみでも触り始める事が出来ます。

# 全行程終了

:::note warn
もし、個人アカウントで試されている方おられましたら、Cloudwatch Logsのログ課金とOpenSearchクラスタの課金は高額になる可能性がありますので、その点だけ留意ください。
:::


今回はApache2のダミーログを生成してテストをしてみましたが、ここまでの道のりをマスターすれば、あとはデータを本番に切り替えるだけです。
本番データと似た構造のダミーデータを使うなどして検証をさらに1歩2歩進める事で、どんなアウトプットが必要なのか、そのためにどういったデータが必要か、どんなダッシュボードが必要か等、どんどん明確になってくるはずです。
また一旦このような環境を作ってしまえば、例えば以下のように他のAWSサービスを利用して、検証を進めるのもずいぶんとハードルが低くなった事と思います。
![2022-01-11_15h36_48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99499/4c90aa7b-95fd-470c-c61e-5320914d57b9.png)

データ活用とは、初めから「おむつとビール」が得られるものではなく、そのような相関に気づく為の土台を作る事こそが本質です。
このエントリが、皆様のデータ活用の推進や、ログ分析基盤を構築する一助になれば幸いです。