はじめに

本投稿は、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に取り込んでみた