Redshift Spectrum を使用して監査ログを分析する方法を教えてください。

簡単な説明

Redshift Spectrum を使用する前に、以下のタスクを完了してください。

1. 監査ログをオンにします。

注: 監査ログが Amazon Simple Storage Service (Amazon S3) バケットに表示されるまでに時間がかかる場合があります。

2. AWS ID およびアクセス管理 (IAM) ロールを作成します。

3. IAM ロールを Amazon Redshift クラスターに関連付けます。

Redshift Spectrum で監査ログをクエリするには、外部テーブルを作成し、(ファイルで使用される) 共通フォルダーを指すように設定します。次に、非表示の $path 列と 正規表現関数を使用して、分析用の行を生成するビューを作成します。

解決策

Redshift Spectrum で監査ログをクエリするには、以下の手順に従ってください。

1. 外部スキーマを作成します。

create external schema s_audit_logs
from data catalog
database 'audit_logs'
iam_role 'arn:aws:iam::your_account_number:role/role_name' create external database if not exists

your_account_number を実際のアカウント番号と一致するように置き換えてください。role_name には、Amazon Redshift クラスターにアタッチされている IAM ロールを指定します。

2. 外部テーブルを作成します。

注: 次の例では、bucket_name、your_account_id、および region をバケット名、アカウント ID、および AWS リージョンに置き換えてください。

ユーザーアクティビティログテーブルを作成します。

create external table s_audit_logs.user_activity_log(
	logrecord varchar(max)
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '|'
STORED AS INPUTFORMAT
	'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
	'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
	's3://bucket_name/logs/AWSLogs/your_account_id/redshift/region'

接続ログテーブルを作成します。

CREATE EXTERNAL TABLE s_audit_logs.connections_log(
	event varchar(60), recordtime varchar(60),
	remotehost varchar(60), remoteport varchar(60),
	pid int, dbname varchar(60),
	username varchar(60), authmethod varchar(60),
	duration int, sslversion varchar(60),
	sslcipher varchar(150), mtu int,
	sslcompression varchar(70), sslexpansion varchar(70),
	iamauthguid varchar(50), application_name varchar(300))
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '|'
STORED AS INPUTFORMAT
	'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
	'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://bucket_name/logs/AWSLogs/your_account_id/redshift/region';

ユーザーログテーブルを作成します。

create external table s_audit_log.user_log(
	userid varchar(255),
	username varchar(500),
	oldusername varchar(500),
	usecreatedb varchar(50),
	usesuper varchar(50),
	usecatupd varchar(50),
	valuntil varchar(50),
	pid varchar(50),
	xid varchar(50),
	recordtime varchar(50))
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '|'
STORED AS INPUTFORMAT
	'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
	'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://bucket_name/logs/AWSLogs/your_account_id/redshift/region’;

3. 監査ログを表示するローカルスキーマを作成します。

create schema audit_logs_views

4. 外部テーブルにアクセスするには、WITH NO SCHEMA BINDING オプションを使用してデータベースにビューを作成します。

CREATE VIEW audit_logs_views.v_connections_log AS
select *
FROM s_audit_logs.connections_log
WHERE "$path" like '%connectionlog%'
with no schema binding;

返されるファイルは、connectionlog エントリと一致するように非表示の $path 列によって制限されます。

次の例では、非表示の $path 列と正規表現関数で、v_connections_log に対して返されるファイルが制限されています。

CREATE or REPLACE VIEW audit_logs_views.v_useractivitylog AS
SELECT    logrecord,
          substring(logrecord,2,24) as recordtime,
          replace(regexp_substr(logrecord,'db=[^" "]*'),'db=','') as db,
          replace(regexp_substr(logrecord,'user=[^" "]*'),'user=','') AS user,
          replace(regexp_substr(logrecord, 'pid=[^" "]*'),'pid=','') AS pid,
          replace(regexp_substr(logrecord, 'userid=[^" "]*'),'userid=','') AS userid,
          replace(regexp_substr(logrecord, 'xid=[^" "]*'),'xid=','') AS xid,
          replace(regexp_substr(logrecord, '][^*]*'),']','') AS query
   FROM s_audit_logs.user_activity_log
   WHERE "$path" like '%useractivitylog%'
   with no schema binding;

返されるファイルは、useractivitylog のエントリと一致します。

注: ユーザーアクティビティログの複数行クエリに関連する制限があります。列のログレコードを直接クエリするのがベストプラクティスです。

関連情報

Amazon Redshift Spectrum を使用してデータベース監査ログを分析し、セキュリティとコンプライアンスを確認する

STL_CONNECTION_LOG