はじめに
本投稿は、Aurora MySQLのスロークエリログと監査ログを
Amazon Elasticsearch Serviceに取り込んでみたという内容になります^^
【参考】
・Auroraの特徴:MySQL互換エディション
・Amazon Auroraで監査ログに対応しました
利用環境
| product | version |
|---|---|
| logstash (OSS版) | 7.7.1 |
| Java (Corretto) | 11.0.8 |
| OS(EC2) | Amazon Linux2 (t3.small) |
| AMI ID | ami-03657b56516ab7912 |
| Elasticsearch | 7.7 (latest) |
| MySQL | 5.7 |
| Region | us-east-2 |
※投稿時点における最新版を採用しています。
【構成図】
- CloudWatch Logsに出力したAurora PostgreSQLのログをLogstashがInputしています。
前提条件
- Elasticsearchへのログ出力にはElastic社のETLツールであるLogstashを利用しています。
- Logstashのバージョンは、Amazon Elasticsearch Serviceの最新バージョンに合わせています。
- Amazon Elasticsearch ServiceはOSS版のため、LogstashもOSS版としています。
- Amazon Elasticsearch Service(以下、Amazon ES)は、パブリックアクセスとしています。
(IPアドレスによるホワイトリスト制御でセキュリティを確保しています)
【参考】
・ Logstashとは
実施内容
- パラメータグループ作成(ログ出力設定)
- データベース作成
- 各種ログの確認
- Amazon ESのドメイン作成
- IAM Role作成
- Logstashの構築
- Kibanaでの各設定
1. パラメータグループ作成(ログ出力設定)
- MySQLのログ出力設定のためのパラメータグループを以下の内容で作成します。
- MySQLのログ記録をオンにした状態では、エラーログのみが記録されます。
※ 全般、スロークエリ、監査のログは追加設定が必要です。
【パラメータグループの設定】
| 項目 | 値 |
|---|---|
| パラメータグループファミリー | aurora-mysql5.7 |
| タイプ | DB Cluster Parameter Group |
| グループ名 | testmysql57-paramatergroup (任意) |
| 説明 | testmysql57 (任意) |
上記で作成したパラメータグループにてパラメータを編集します。
以下のパラメータを変更して保存します。
【パラメータの説明】
| 項目 | 値 | 説明 |
|---|---|---|
| slow_query_log | 1 | スロークエリログの出力を有効にする |
| long_query_time | 0 | スロークエリログの閾値を0ミリ秒にする (今回) |
| server_audit_logging | 1 | 監査ログの出力を有効にする |
| server_audit_events | QUERY | 監査対象のイベントをクエリのみとする (今回) |
| server_audit_excl_users | <無し> | 監査対象外のユーザは指定しない (今回) |
| server_audit_incl_users | <無し> | 監査対象のユーザは指定しない (今回) |
| log_output | FILE | CloudWatch Logsへの出力の場合はデフォルトのままとする |
【参考】
・Auroraでスロークエリや一般ログがCloudWatch Logsへ出力可能に
・Amazon CloudWatch LogsへのAmazon Aurora MySQLログの発行
・Amazon Aurora MySQL DB クラスターでの高度な監査の使用
2. データベース作成
- 下記内容でAurora MySQLのDBクラスタを作成します。
※ 作成したDBクラスタのパラメータグループを割り当てます。
| 項目 | 値 |
|---|---|
| データベース作成方法 | 標準作成 |
| エンジンのタイプ | Amazon Aurora |
| エディション | MySQLとの互換性を持つAmazon Aurora |
| キャパシティタイプ | プロビジョニング済み |
| レプリケーション機能 | シングルマスター |
| バージョン | Aurora (MySQL 5.7) 2.07.2 |
| テンプレート | 開発/テスト |
| DBクラスター識別子 | database-2 |
| マスターユーザー名 | admin |
| DBインスタンスクラス | バースト可能クラス (db.t3.small) |
| マルチAZ配置 | Auroraレプリカを作成しない |
| VPC | Default VPC (今回は) |
| サブネットグループ | default-vpc-xxxxxxxxx (今回は) |
| パブリックアクセス可能 | なし |
| VPCセキュリティグループ | 新規作成 |
| 新しいVPCセキュリティグループ名 | ForMySQL (今回は) |
| アベイラビリティゾーン | us-east-2a |
| データベースポート | 3306 |
| データベースの選択肢 | testdatabase |
| DBクラスターのパラメータグループ | testmysql57-paramatergroup |
| DBパラメータグループ | default.aurora-mysql5.7 |
| フェールオーバー優先順位 | 指定なし |
| バックアップ保持期間 | 1日間 |
| スナップショットにタグをコピー | ☑︎ |
| 暗号を有効化 | ☑︎ |
| マスターキー | (default) aws/rds |
| バックトラックを有効にする | □ |
| 拡張モニタリングの有効化 | □ |
| ログのエクスポート | ☑︎監査ログ、☑︎エラーログ、☑︎全般ログ、☑︎スロークエリログ |
| マイナーバージョン自動アップグレードの有効化 | □ |
| メンテナンスウィンドウ | 設定なし |
| 削除保護の有効化 | □ |
3. 各種ログの確認
- [Systems Manager] > [Session Manager]からLogstash用のEC2に接続します。
- 設定したパラメータが反映されていることを確認します。
※ EC2からMySQLSQLにTCP3306で通信許可されている状態としています。
※ MySQLクライアントがインストールされていることを前提としています。
$ mysql -h [Auroraのエンドポイント] -D testdatabase -P 3306 -u [マスターユーザ名] -p
Enter password:
Welcome to the MySQL monitor. Commands end with ; or \g.
Your MySQL connection id is 15
Server version: 5.7.12 MySQL Community Server (GPL)
Copyright (c) 2000, 2020, Oracle and/or its affiliates. All rights reserved.
Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql> SHOW VARIABLES LIKE '%slow_query_log%';
+---------------------+----------------------------------------------+
| Variable_name | Value |
+---------------------+----------------------------------------------+
| slow_query_log | ON |
| slow_query_log_file | /rdsdbdata/log/slowquery/mysql-slowquery.log |
+---------------------+----------------------------------------------+
2 rows in set (0.00 sec)
mysql> SHOW VARIABLES LIKE '%long_query_time%';
+-----------------+----------+
| Variable_name | Value |
+-----------------+----------+
| long_query_time | 0.000000 |
+-----------------+----------+
1 row in set (0.00 sec)
mysql> SHOW VARIABLES LIKE '%server_audit%';
+------------------------------+-------+
| Variable_name | Value |
+------------------------------+-------+
| server_audit_cw_upload | OFF |
| server_audit_events | QUERY |
| server_audit_excl_users | |
| server_audit_incl_users | |
| server_audit_logging | ON |
| server_audit_mode | 0 |
| server_audit_query_log_limit | 1024 |
+------------------------------+-------+
7 rows in set (0.00 sec)
- testtableという名前のテーブルを作成し、適当なデータ(name: guest01、text: test01)をINSERTします。
- その後、SELECTコマンドでクエリします。
mysql> create table testtable (name varchar(255), text varchar(255));
Query OK, 0 rows affected (0.08 sec)
mysql> insert into testtable values ('guest01', 'test01');
Query OK, 1 row affected (0.00 sec)
mysql> select * from testtable;
+---------+--------+
| name | text |
+---------+--------+
| guest01 | test01 |
+---------+--------+
1 row in set (0.00 sec)
[CloudWatch Logs]のロググループに以下のようなログが出力されます。
スロークエリログは、以下のような形式になります。
slowquery
# Time: 2020-10-31T17:05:58.746864Z
# User@Host: admin[admin] @ [172.31.41.200] Id: 10
# Query_time: 0.000067 Lock_time: 0.000000 Rows_sent: 0 Rows_examined: 0
SET timestamp=1604163958;
select * from testtable;
- 監査ログは、以下のような形式になります。
audit
1604163668366363,database-2-instance-1,admin,172.31.41.200,10,1703,QUERY,testdatabase,'select * from testtable',0
【参考】
・CentOSにmysqlコマンドをインストールする方法
4. Amazon ESのドメイン作成
- 下記内容でAmazon ESのドメインを作成します。
| 項目 | 値 |
|---|---|
| リージョン | us-east-2 |
| デプロイタイプ | 開発およびテスト |
| Elasticsearchのバージョン | 7.7 (latest) |
| Elasticsearchのドメイン | test-es |
| インスタンスタイプ | t3.small.elasticsearch |
| ノードの数 | 1 |
| データノードのストレージタイプ | EBS |
| EBS ボリュームタイプ | 汎用(SSD) |
| EBS ボリュームサイズ | 10 GiB |
| 自動スナップショットの開始時間 | 00:00 UTC (デフォルト) |
| ネットワークアクセス | パブリックアクセス |
| 細かいアクセスコントロールを有効化 | 無効 |
| Amazon Cognito認証を有効化 | 無効 |
| ドメインアクセスポリシー | カスタムアクセスポリシー (IPv4アドレス) |
| ドメインへのすべてのトラフィックにHTTPSを要求 | 有効 |
| ノード間の暗号化 | 無効 |
| 保管時のデータの暗号化の有効化 | 無効 |
※アクセスポリシーに追加するIPは、Amazon ESにアクセスする自宅IPとLogstashのIPになります。
ドメインアクセスポリシー
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"AWS": "*"
},
"Action": "es:*",
"Resource": "arn:aws:es:us-east-2:<AWSアカウント>:domain/test-es/*",
"Condition": {
"IpAddress": {
"aws:SourceIp": [
"<自宅IP>/32",
"<LogstasのEIP>/32"
]
}
}
}
]
}
5. IAM Role作成
- EC2として構築するLogstashに割り当てるIAM Roleを作成します。
- logatashのEC2に割り当てるIAM Roleとして
RoleForEC2というロールを作成し、
CloudWatchLogsReadOnlyAccessというIAM Policyを割り当てます。
CloudWatchLogsReadOnlyAccess
{
"Version": "2012-10-17",
"Statement": [
{
"Action": [
"logs:Describe*",
"logs:Get*",
"logs:List*",
"logs:StartQuery",
"logs:StopQuery",
"logs:TestMetricFilter",
"logs:FilterLogEvents"
],
"Effect": "Allow",
"Resource": "*"
}
]
}
6. Logstashの構築
Logstashの構築手順は、Amazon Corretto 11で構築するLogstashになります。
スロークエリログおよび監査ログは、正規表現パターンを使って以下のようなフィールドと値にパースします。
【スロークエリログ】
| フィールド | 値(サンプル) | データ型 |
|---|---|---|
| タイムスタンプ | 1604163958 | 日付型 (UNIX) |
| ユーザ名 | admin | 文字列型 (keyword) |
| IPアドレス | 172.31.41.200 | IP型 |
| コネクションID | 10 | 文字列型 (keyword) |
| クエリ応答時間 | 0.000067 | 数値型 (float) |
| ロック取得時間 | 0.000000 | 数値型 (float) |
| クライアントに送信された行数 | 0 | 数値型 (integer) |
| 検査された行数 | 0 | 数値型 (integer) |
| クエリ文 | select * from testtable; | 文字列型 (keyword) |
【監査ログ】
| フィールド | 値(サンプル) | データ型 |
|---|---|---|
| タイムスタンプ | 1604163668366363 | 日付型 (UNIX) |
| DBインスタンスID | database-2-instance-1 | 文字列型 (keyword) |
| ユーザ名 | admin | 文字列型 (keyword) |
| IPアドレス | 172.31.41.200 | IP型 |
| コネクションID | 10 | 文字列型 (keyword) |
| クエリID | 1703 | 文字列型 (keyword) |
| オペレーション | QUERY | 文字列型 (keyword) |
| データベース名 | testdatabase | 文字列型 (keyword) |
| クエリ文 | select * from testtable | 文字列型 (text) |
| リターンコード | 0 | 文字列型 (keyword) |
※リターンコードは、クエリ文が正常に実行されている場合に「0」となります。
- スロークエリログの形式は以下の3パターンあります。いずれの場合にもマッチ出来るようにGrokパターンファイルを作成する必要があります。
slowquery_log
# Time: 2020-10-31T21:11:06.305379Z\n# User@Host: rdsadmin[rdsadmin] @ localhost [] Id: 7\n# Query_time: 0.000090 Lock_time: 0.000000 Rows_sent: 1 Rows_examined: 0\nSET timestamp=1604178666;\nselect @@session.tx_read_only;"
# Time: 2020-10-31T19:52:54.112559Z\n# User@Host: admin[admin] @ [172.31.41.200] Id: 14\n# Query_time: 0.000090 Lock_time: 0.000000 Rows_sent: 0 Rows_examined: 0\nSET timestamp=1604173974;\nselect * from testtable;
# Time: 2020-10-31T21:01:35.031575Z\n# User@Host: admin[admin] @ [172.31.41.200] Id: 20\n# Query_time: 0.000009 Lock_time: 0.000000 Rows_sent: 0 Rows_examined: 0\nuse testdatabase;\nSET timestamp=1604178095;\n# administrator command: Quit;
- Grokパターン(
/etc/logstash/grok_patterns/aurora-mysql-log)を保存するディレクトリを作成します。
$ sudo mkdir /etc/logstash/grok_patterns
$ sudo vi /etc/logstash/grok_patterns/aurora-mysql-log
- 下記の内容を上記のGrokパターンファイルに記述します。
aurora-mysql-log
### MySQL slowLog
MYSQL_SLOWLOG ^#\sTime:\s%{TIMESTAMP_ISO8601:timestamp1}\n#\sUser@Host:\s%{WORD:username1}\[%{WORD:username2}\]\s@\s+(?:%{HOSTNAME:hostname})?\s+\[(?:%{IP:ipaddress})?\]\s*Id:\s+%{NUMBER:connectionid}\n#\sQuery_time:\s%{NUMBER:query_time}\s+Lock_time:\s%{NUMBER:lock_time}\s+Rows_sent:\s%{NUMBER:rows_sent}\s+Rows_examined:\s%{NUMBER:rows_examined}(?:\nuse\s%{WORD:db_name};)?\nSET timestamp=%{NUMBER:timestamp2};\n%{GREEDYDATA:statement}
### MySQL auditLog
UNIX_TS1 ([0-9]{13})
UNIX_TS2 ([0-9]{3})
MYSQL_AUDITLOG %{UNIX_TS1:timestamp1}%{UNIX_TS2:timestamp2}\,%{HOSTNAME:db_instance_id}\,%{USER:username}\,%{HOSTNAME:ipaddress}\,%{NUMBER:connectionid}\,%{NUMBER:queryid}\,%{WORD:operation}\,(?:|%{WORD:database})\,(?:|'%{DATA:object}')\,%{NUMBER:retcode}
※ localhostというホスト名が出るログと出ないログがあるため、(?:%{HOSTNAME:hostname})?と記述しています。(?:)?で括ると存在しない場合は無視されます。
※ (?:\nuse\s%{WORD:db_name};)?も上記と同様です。user テーブル名;が表示されるのは、DBからquitで抜けた場合になります。
※ スペースは\s、改行は\nとしています。
- 下記のパイプライン構成ファイル(
/etc/logstash/conf.d/logstash.conf)を作成します。
(このタイミングではLogstashのプロセスは起動しません)
$ sudo vi /etc/logstash/conf.d/logstash.conf
- 下記の内容を上記のパイプライン構成ファイルに記述します。
logstash.conf
input {
cloudwatch_logs {
log_group => [ "/aws/rds/cluster/database-2/slowquery" ]
region => "us-east-2"
interval => 5
}
cloudwatch_logs {
log_group => [ "/aws/rds/cluster/database-2/audit" ]
region => "us-east-2"
interval => 5
}
}
filter {
if "slowquery" in [cloudwatch_logs][log_group] {
### 読み込むGrok Patternファイルを"patterns_dir"で指定
grok {
patterns_dir => [ "/etc/logstash/grok_patterns" ]
match => { "message" => "%{MYSQL_SLOWLOG}" }
}
### dateフィールドから@timestampを抽出
date {
match => [ "timestamp2", "UNIX" ]
timezone => "Asia/Tokyo"
target => "@timestamp"
}
### @timstampから日本時間を抽出
ruby {
code => "event.set('[@metadata][local_time]',event.get('[@timestamp]').time.localtime.strftime('%Y-%m-%d'))"
}
### document_idに利用する一意のIDを作成
fingerprint {
source => "message"
target => "[@metadata][fingerprint]"
method => "MURMUR3"
}
### デフォルトの型がstringのため、フィールド定義で定義した型に変換
mutate {
### typeフィールドを追加
add_field => { "type" => "aurora-mysql-slow-cwl" }
### 不要なフィールドを削除
remove_field => [ "timestamp1", "timestamp2" ]
}
}
if "audit" in [cloudwatch_logs][log_group] {
### 読み込むGrok Patternファイルを"patterns_dir"で指定
grok {
patterns_dir => [ "/etc/logstash/grok_patterns" ]
match => { "message" => "%{MYSQL_AUDITLOG}" }
}
### dateフィールドから@timestampを抽出
date {
match => [ "timestamp1", "UNIX_MS" ]
timezone => "Asia/Tokyo"
target => "@timestamp"
}
### @timstampから日本時間を抽出
ruby {
code => "event.set('[@metadata][local_time]',event.get('[@timestamp]').time.localtime.strftime('%Y-%m-%d'))"
}
### document_idに利用する一意のIDを作成
fingerprint {
source => "message"
target => "[@metadata][fingerprint]"
method => "MURMUR3"
}
### デフォルトの型がstringのため、フィールド定義で定義した型に変換
mutate {
### typeフィールドを追加
add_field => { "type" => "aurora-mysql-audit-cwl" }
### 不要なフィールドを削除
remove_field => [ "timestamp1", "timestamp2" ]
}
}
}
output {
if [type] == "aurora-mysql-slow-cwl" {
### 出力先のAmazonESのIndexを指定
elasticsearch {
hosts => [ "https://search-test-es-xxxxxxxxxxxxxxxxxxxxxxxxxx.us-east-2.es.amazonaws.com:443" ]
index => "%{type}-%{[@metadata][local_time]}"
document_id => "%{[@metadata][fingerprint]}"
ilm_enabled => false
}
}
if [type] == "aurora-mysql-audit-cwl" {
### 出力先のAmazonESのIndexを指定
elasticsearch {
hosts => [ "https://search-test-es-xxxxxxxxxxxxxxxxxxxxxxxxxx.us-east-2.es.amazonaws.com:443" ]
index => "%{type}-%{[@metadata][local_time]}"
document_id => "%{[@metadata][fingerprint]}"
ilm_enabled => false
}
}
}
【参考】
・ Logstashの実践的な説明
・ cloudwatch-logs input
・ grok filter
・ date filter
・ ruby filter
・ fingerprint filter
・ mutate filter
・ elasticsearch output
7. Kibanaでの各設定
Amazon ESの
KibanaのURLをクリックします。
[Dev Tools]のConsoleからAurora MySQLのスロークエリログと監査ログのIndex Templateを追加します。
上記で張り付けたIndex Templateは以下の通りです。
index_templete_aurora-mysql-slow-cwl
PUT _template/aurora-mysql-slow-cwl
{
"index_patterns": ["aurora-mysql-slow-cwl-*"],
"settings": {
"number_of_shards": 1,
"number_of_replicas" : 1
},
"mappings": {
"properties": {
"@timestamp": {
"type": "date"
},
"@version" : {
"type" : "keyword"
},
"username1" : {
"type" : "keyword"
},
"username2" : {
"type" : "keyword"
},
"hostname" : {
"type" : "keyword"
},
"ipaddress" : {
"type" : "ip"
},
"connectionid" : {
"type" : "keyword"
},
"query_time" : {
"type" : "float"
},
"lock_time" : {
"type" : "float"
},
"rows_sent" : {
"type" : "integer"
},
"rows_examined" : {
"type" : "integer"
},
"db_name" : {
"type" : "keyword"
},
"statement" : {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above" : 8191
}
}
},
"type" : {
"type" : "keyword"
},
"tags" : {
"type" : "keyword"
},
"message" : {
"type" : "text"
},
"cloudwatch_logs" : {
"properties" : {
"event_id" : {
"type" : "keyword"
},
"ingestion_time" : {
"type" : "date"
},
"log_group" : {
"type" : "keyword"
},
"log_stream" : {
"type" : "keyword"
}
}
}
}
}
}
index_templete_aurora-mysql-audit-cwl
PUT _template/aurora-mysql-audit-cwl
{
"index_patterns": ["aurora-mysql-audit-cwl-*"],
"settings": {
"number_of_shards": 1,
"number_of_replicas" : 1
},
"mappings": {
"properties": {
"@timestamp": {
"type": "date"
},
"@version" : {
"type" : "keyword"
},
"db_instance_id" : {
"type" : "keyword"
},
"username" : {
"type" : "keyword"
},
"ipaddress" : {
"type" : "ip"
},
"connectionid" : {
"type" : "keyword"
},
"queryid" : {
"type" : "keyword"
},
"operation" : {
"type" : "text"
},
"database" : {
"type" : "text"
},
"object" : {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above" : 8191
}
}
},
"retcode" : {
"type" : "keyword"
},
"type" : {
"type" : "keyword"
},
"tags" : {
"type" : "keyword"
},
"message" : {
"type" : "text"
},
"cloudwatch_logs" : {
"properties" : {
"event_id" : {
"type" : "keyword"
},
"ingestion_time" : {
"type" : "date"
},
"log_group" : {
"type" : "keyword"
},
"log_stream" : {
"type" : "keyword"
}
}
}
}
}
}
※クエリ文が含まれるフィールド(statementとobject)は、全文検索ができるようtext型で格納しつつ、グラフ化できるようkeyword型でも格納しています。
※Luceneのトークンの上限値である32,766バイトを超えて破棄されないよう、推奨値8,191で制限しています。
- Logstashを起動します。
logstash_start
$ sudo systemctl start logstash
$ sudo systemctl status logstash
● logstash.service - logstash
Loaded: loaded (/etc/systemd/system/logstash.service; disabled; vendor preset: disabled)
Active: active (running) since Sun 2020-10-30 17:11:21 UTC; 4min 42s ago
Main PID: 32168 (java)
CGroup: /system.slice/logstash.service
└─32168 /bin/java -Xms1g -Xmx1g -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -XX:+UseCMSInitiatingOccupancyOnly -Djava.awt.h...
- Kibanaの [Management] > [Index Patterns]で
Create index patternをクリックします。 aurora-mysql-*という名前でIndex Patternを作成します。
@timestampを指定します。
[Discover]を開き、Index Patternに
aurora-mysql-*を指定します。デフォルトでは直近15分間のログが表示されます。表示されていればOKです。
取り込んだログは以下のような感じです。
aurora-mysql_slowquery_sample
{
"_index": "aurora-mysql-slow-cwl-2020-11-01",
"_type": "_doc",
"_id": "2522881698",
"_version": 1,
"_score": null,
"_source": {
"@timestamp": "2020-11-01T11:22:24.000Z",
"statement": "select * from testtable;",
"username2": "admin",
"@version": "1",
"lock_time": "0.000000",
"connectionid": "24",
"ipaddress": "172.31.41.200",
"rows_sent": "0",
"username1": "admin",
"cloudwatch_logs": {
"ingestion_time": "2020-11-01T11:22:25.225Z",
"log_stream": "database-2-instance-1",
"event_id": "35775518769277692425490291398483945867814577320597454853",
"log_group": "/aws/rds/cluster/database-2/slowquery"
},
"type": "aurora-mysql-slow-cwl",
"rows_examined": "0",
"query_time": "0.000065",
"message": "# Time: 2020-11-01T11:22:24.378067Z\n# User@Host: admin[admin] @ [172.31.41.200] Id: 24\n# Query_time: 0.000065 Lock_time: 0.000000 Rows_sent: 0 Rows_examined: 0\nSET timestamp=1604229744;\nselect * from testtable;"
},
"fields": {
"cloudwatch_logs.ingestion_time": [
"2020-11-01T11:22:25.225Z"
],
"@timestamp": [
"2020-11-01T11:22:24.000Z"
]
},
"sort": [
1604229744000
]
}
aurora-mysql_audit_sample
{
"_index": "aurora-mysql-audit-cwl-2020-10-31",
"_type": "_doc",
"_id": "2976207611",
"_version": 1,
"_score": null,
"_source": {
"operation": "QUERY",
"@version": "1",
"db_instance_id": "database-2-instance-1",
"database": "testdatabase",
"queryid": "53944",
"retcode": "0",
"ipaddress": "172.31.41.200",
"username": "admin",
"connectionid": "14",
"object": "select * from testtable",
"type": "aurora-mysql-audit-cwl",
"cloudwatch_logs": {
"log_stream": "database-2-instance-1.audit.log.0.2020-10-31-19-30.0.2",
"event_id": "35774275050785972149828474772771999174825091470427422722",
"ingestion_time": "2020-10-31T19:52:54.181Z",
"log_group": "/aws/rds/cluster/database-2/audit"
},
"@timestamp": "2020-10-31T19:52:54.112Z",
"message": "1604173974112542,database-2-instance-1,admin,172.31.41.200,14,53944,QUERY,testdatabase,'select * from testtable',0"
},
"fields": {
"cloudwatch_logs.ingestion_time": [
"2020-10-31T19:52:54.181Z"
],
"@timestamp": [
"2020-10-31T19:52:54.112Z"
]
},
"sort": [
1604173974112
]
}
まとめ
いかがでしたでしょうか？
MySQLの監査ログはシングルラインで出力されるため、取り込みのための処理は比較的簡単ですが
スロークエリログは、マルチラインで出力されます。しかも、クエリ内容に応じて、出力フォーマットが
多少異なります。そのため、綺麗に漏れなく取り込むにはそれなりの工夫が必要になります。
それを今回は、LogstashのGrok Filterで吸収しています。
PostgreSQLでも同じようにAmazon ESに取り込む方法を記事にしています。
もし興味がありましたら、以下の記事も合わせて参照頂けると良いと思います^^
【参考】
・Aurora PostgreSQLのスロークエリログをAmazon ESに取り込んでみた
・Aurora PostgreSQLの監査ログをAmazon ESに取り込んでみた