Amazon Kinesis Firehoseを使ってApacheのログをステータスごとに出し分ける方法を考える
こんにちは、せーのです。今日は先日ご紹介した新機能のアレンジ編です。
イベント毎にファイルを分けたい
先日Firehoseから出力先にデータを送る際にLambdaをはさんでApacheのログを簡単にJSON化する機能をご紹介しました。
【新機能】Amazon Kinesis FirehoseにAmazon Lambdaを使ったデータ変換機能が追加。Apacheログが簡単にJSON形式に変換可能に！｜クラスメソッドブログ
Firehoseからデータを流す時にLambdaを間にはさんでETL処理することが出来る、しかもApacheやSyslogのデータ加工に関しては既にblueprinｔが用意されていて何もコードを書かなくても変換してくれる、という分析系の方には涙が溢れて止まらない機能でした。
こちらのブログはSNSで色んなコメントをいただき、概ね「便利そう」というものだったのですが、中にこのようなコメントがありました。
うーん。これだけではイベント毎にファイルを分けて書き出せないなあ…惜しい。イベント毎にストリームを分けるのは大仰だからな－。 / “【新機能】Amazon Kinesis FirehoseにAmazon Lambdaを使ったデー…” https://t.co/cVnXjBgoG6
— kazutomo (@kazutomo) 2016年12月22日
なるほど。確かにステータスやIP、パスなどでファイルを分けるような機能はついていませんでした。どうにかして出来ないものでしょうか。考えてみましょう。
考えてみた
Athenaで取り出す
一番わかりやすいのはこの方法ではないでしょうか。S3には全てのログを入れておいてAthenaで欲しいステータスのものを拾い出す、という方法です。
こんな感じでApacheのJSONログをAthenaで拾ってるブログとかないかなー、とググってみると、なんとDevelopers.IOが引っかかりましたｗ
ApacheのアクセスログをAmazon Athenaを使って分析する｜クラスメソッドブログ
もうこのまんまいけそうです。
Lambdaで分岐する
次にLambdaを使ってETLしているのだからそのLambdaを使って分岐してみよう、というものです。
Lambdaですから基本コードでなんとでもなりそうな感じがしますね。やってみましょう。
やってみた
まずは前回の記事と同じようにFirehose, S3, Lambdaを準備します。詳しい構築方法は前回の記事を参考にして下さい。
今回は加工前の生データもS3に残してみることにします。こんなことも設定一つでできる辺りはありがたいですね。
次にデータを流すためのAWS IoTを用意します。
軽くテストしてみます。
ファイルが入っていますね。中身を見てみます。
{"host":"127.0.0.1","ident":"-","authuser":"frank","datetime":"10/Oct/2000:13:55:36 -0700","request":"GET /apache_pb.gif HTTP/1.0","response":"200"}
JSON化されていますね。成功です。
次にETLをしているLambdaの中身を見てみましょう。
'use strict';
console.log('Loading function');
/* Apache Log format parser */
const parser = /^([\d.]+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(.+?)" (\d{3})/;
exports.handler = (event, context, callback) => {
let success = 0; // Number of valid entries found
let failure = 0; // Number of invalid entries found
/* Process the list of records and transform them */
const output = event.records.map((record) => {
const entry = (new Buffer(record.data, 'base64')).toString('utf8');
const match = parser.exec(entry);
if (match) {
/* Prepare JSON version from Apache log data */
const result = {
host: match[1],
ident: match[2],
authuser: match[3],
datetime: match[4],
request: match[5],
response: match[6],
};
const payload = (new Buffer(JSON.stringify(result), 'utf8')).toString('base64');
success++;
return {
recordId: record.recordId,
result: 'Ok',
data: payload,
};
} else {
/* Failed event, notify the error and leave the record intact */
failure++;
return {
recordId: record.recordId,
result: 'ProcessingFailed',
data: record.data,
};
}
});
console.log(`Processing completed. Successful records ${success}, Failed records ${failure}.`);
callback(null, { records: output });
};
Common Log Formatの形式でマッチングをして、マッチしたらそれぞれの値をオブジェクトに入れ直してJSON.stringify()でJSONとして加工、Kinesis用にbase64に戻してCallbackしているんですね。
ということはこのresult.responseの値で分岐して、欲しいステータスのものが来たら別のファイルに吐き出せば良さそうです。実装してみます。
'use strict';
console.log('Loading function');
/* Apache Log format parser */
const parser = /^([\d.]+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(.+?)" (\d{3})/;
var AWS = require('aws-sdk');
var s3 = new AWS.S3();
var params = {
Bucket: 'separate-status',
Key: "error/500.log",
Body: ""
};
exports.handler = (event, context, callback) => {
let success = 0; // Number of valid entries found
let failure = 0; // Number of invalid entries found
/* Process the list of records and transform them */
const output = event.records.map((record) => {
const entry = (new Buffer(record.data, 'base64')).toString('utf8');
const match = parser.exec(entry);
if (match) {
/* Prepare JSON version from Apache log data */
const result = {
host: match[1],
ident: match[2],
authuser: match[3],
datetime: match[4],
request: match[5],
response: match[6],
};
const payload = (new Buffer(JSON.stringify(result), 'utf8')).toString('base64');
if (Number(result.response) >= 500){
params.Body = JSON.stringify(result);
s3.upload(params, function(err, data) {
if (err) {
console.log("Error uploading data: ", err);
} else {
console.log("Successfully uploaded data to error.");
}
return {
recordId: record.recordId,
result: 'Ok',
data: payload,
};
});
} else {
success++;
return {
recordId: record.recordId,
result: 'Ok',
data: payload,
};
}
} else {
/* Failed event, notify the error and leave the record intact */
failure++;
return {
recordId: record.recordId,
result: 'ProcessingFailed',
data: record.data,
};
}
});
console.log(`Processing completed. Successful records ${success}, Failed records ${failure}.`);
callback(null, { records: output });
};
LambdaのIAM RoleにS3へのアクセス許可を加えて、もう一度テストしてみます。今度はステータスを501にしてみましょう。
バケットを見てみます。
指定したフォルダができていますね。
新しいファイルが出来上がりました。中身を見てみます。
{"host":"127.0.0.1","ident":"-","authuser":"frank","datetime":"10/Oct/2000:13:55:36 -0700","request":"GET /apache_pb.gif HTTP/1.0","response":"501"}
501のレスポンスのものだけデータがきていますね。成功です。今回は固定のファイル名にしましたが、本番で使う際にはUnixtimeをつけるなどして上書きされないようにしてみて下さい。
ちなみに元データのファイルものぞいてみます。
127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 200 2326 127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 501 2326
元データがきちんと無加工で入っています。
まとめ
いかがでしたでしょうか。前回のブログのちょい足しレシピをご紹介しました。基本はAthenaで取り出す、どうしてもファイルを分けたい方は条件をLambdaに書き込めば可能です。是非お試しください。
![[Tips] Amazon Athena 6億レコードのサンプルデータ「SSB」を準備する](https://images.ctfassets.net/ct0aopd36mqt/wp-thumbnail-764c44f07bcd41184fe62a432ae120ab/512a4cdcd05a0ed3ddea950fdf619fa0/amazon-athena)
