Redshift Spectrum을 사용하여 감사 로그를 분석하려면 어떻게 해야 하나요?

간략한 설명

Redshift Spectrum을 사용하기 전에 다음 작업을 완료하세요.

1.    감사 로그를 켭니다.

참고: 감사 로그가 Amazon Simple Storage Service(S3) 버킷에 표시되는 데 시간이 걸릴 수 있습니다.

2.    AWS 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, 리전을 사용자의 버킷 이름, 계정 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;

반환된 파일은 숨겨진 $path 열에 의해 connectionlog 항목과 일치하도록 제한됩니다.

다음 예에서는 숨겨진 $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