集約データを転送、分析、可視化するためのシステムを作るために使われるソフトウェアについて、アーキテクチャーを調べたり触ってみたりした上で分かったことのメモ書き。

Fluentd

一言でいえば、ログ集約用のソフトウェア、ログのフォーマットもJSONやCSV等幅広く選択可能だったり、ログ転送先もS3やその他対象が広く、例えば各種ログを集約してデータレイクにするなどの使い方が可能なのでは？と思う。ちなみにFluentd自体はrubyのgemだが、それをLinuxサーバで使いやすくするためにrpm化したものがtd-agentである。簡単に、構築の仕方を記載しておく（我流）。

###①td-agentのマスターとなるconf。基本的なログレベルや、ログ種別ごとのconfの在りかを指定しておく。`

/etc/td-agent/td-agent.conf

<system>
  log_level error
</system>

#外部ファイル読み込み
@include config.d/*.conf

###②3パターン程度各ログ種別の特性にあったconfの書き方をガイド

標準的なログの場合(Nginxログ)

#Nginx
#Accesslog Source
<source>
  @type tail
  path /var/log/nginx/access.log
  pos_file /var/log/td-agent/pos/nginx.accesslog.pos
  tag nginx.accesslog
  <parse>
    @type json
    keep_time_key true
    time_format %d/%b/%Y:%H:%M:%S %z
  </parse>
</source>

#Errorlog Source
<source>
  @type tail
  path /var/log/nginx/error.log
  pos_file /var/log/td-agent/pos/nginx.errorlog.pos
  tag nginx.errorlog
  <parse>
    @type regexp
    expression /^(?<time>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) \[(?<log_level>\w+)\] (?<pid>\d+).(?<tid>\d+): (?<message>.*)$/
    keep_time_key true
    time_format %Y/%m/%d %H:%M:%S
  </parse>
</source>

#Add Key:Value
<filter nginx.*>
  @type record_transformer
  <record>
    hostname "#{Socket.gethostname}"
  </record>
</filter>

#Accesslog Destination
<match nginx.accesslog>
  @type s3
  s3_bucket test-log-env
  s3_region ap-northeast-1
  s3_object_key_format "%{path}/year=%Y/month=%m/day=%d/#{Socket.gethostname}/${tag}.#{Socket.gethostname}.%{time_slice}.%{index}.%{file_extension}"
  path fluentd/${tag[0]}/${tag[1]}
  store_as gzip
  time_slice_format %Y%m%d%H%M%S
  output_tag false
  output_time false
 <buffer tag,time>
    @type file
    path /var/log/td-agent/buffer/nginx.accesslog/s3
    timekey 60
    timekey_wait 60
  </buffer>
</match>

#Errorlog Destination
<match nginx.errorlog>
  @type s3
  s3_bucket test-log-env
  s3_region ap-northeast-1
  s3_object_key_format "%{path}/year=%Y/month=%m/day=%d/#{Socket.gethostname}/${tag}.#{Socket.gethostname}.%{time_slice}.%{index}.%{file_extension}"
  path fluentd/${tag[0]}/${tag[1]}
  store_as gzip
  time_slice_format %Y%m%d%H%M%S
  output_tag false
  output_time false
  <buffer tag,time>
    @type file
    path /var/log/td-agent/buffer/nginx.errorlog/s3
    timekey 60
    timekey_wait 60
  </buffer>
</match>

同一行が複数行に渡るログの場合(log4jフォーマット)

#Java

#TEST Application Log
<source>
  @type tail
  path /opt/java/logs/application*.log
  pos_file /var/log/td-agent/pos/java-test.applicationlog.pos
  tag java-test.applicationlog
  <parse>
    @type multiline
    format_firstline /^\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}.\d{3}/
    format1 /^(?<time>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}.\d{3})\s+(?<thread>.+)\s+(?<level>.)\s+(?<logger>.+)#(?<method>\S+)\s+:(?<line>\d{1}*)\s+(?<msg>.+)$/
    keep_time_key true
    time_format %Y/%m/%d %H:%M:%S.%L
  </parse>
</source>

#Add Key:Value
<filter java-test.*>
  @type record_transformer
  <record>
    hostname "#{Socket.gethostname}"
  </record>
</filter>

#TEST Application Log
<match java-test.applicationlog>
  @type s3
  s3_bucket test-log-env
  s3_region ap-northeast-1
  s3_object_key_format "%{path}/year=%Y/month=%m/day=%d/#{Socket.gethostname}/${tag}.#{Socket.gethostname}.%{time_slice}.%{index}.%{file_extension}"
  path fluentd/${tag[0]}/${tag[1]}
  store_as gzip
  time_slice_format %Y%m%d%H%M%S
  output_tag false
  output_time false
  <buffer tag,time>
    @type file
    path /var/log/td-agent/buffer/java-test.applicationlog/s3
    timekey 60
    timekey_wait 60
  </buffer>
</match>

RDSのログを抽出してローカルとS3の複数の宛先に出力させる場合

#Aurora(PostgreSQL)
#PostgreSQL errorログ


#########RDS Master

<source>
  @type rds_pgsql_log
  region ap-northeast-1
  db_instance_identifier test-env-rds01
  tag aurora.postgresqllog.test-env-rds01
  pos_file /var/log/td-agent/pos/aurora.postgresqllog.rds01.pos
</source>

<filter aurora.postgresqllog.test-env-rds01>
  @type record_transformer
  <record>
    db_identifer test-env-rds01
  </record>
</filter>

<match aurora.postgresqllog.test-env-rds01>
 @type copy

 <store>
  @type exec
  command cat $1 | logger -p local5.debug
  format json
  <buffer>
    @type file
    path /var/log/td-agent/buffer/aurora.postgresqllog.rds01/file
    flush_interval 60s
  </buffer>
 </store>

 <store>
  @type s3
  s3_bucket test-log-env
  s3_region ap-northeast-1
  s3_object_key_format %{path}/year=%Y/month=%m/day=%d/${tag[2]}/${tag}.%{time_slice}.%{index}.%{file_extension}
  path fluentd/${tag[0]}/${tag[1]}
  store_as gzip
  time_slice_format %Y%m%d%H%M%S
  output_tag false
  output_time false
  <buffer tag,time>
    @type file
    path /var/log/td-agent/buffer/aurora.postgresqllog.rds01/s3
    timekey 60
    timekey_wait 60
  </buffer>
 </store>

</match>

###パラメータに関して詳細

`sourceディレクティブ``
ログの入力方法を決める。
デフォルトで使えるものは、標準入力、ファイル、ポート指定のHTTP通信等あり。

in_tailプラグイン(@tail)
in_tailはログを末尾から読み取るプラグイン。syslogやapache access.logのようにファイルに吐き出させるログをfluentdに読み込ませる時に使用するプラグイン。

tag
ログを指定するID。matchディレクティブで指定する「debug.test」とかがタグの一例。

path
fluentdに送り込みたいログファイル名

pos_file
どこまで読み込んだかを記録するファイル

<parse>~</parse>
パースの方式を指定する。syslogやnginx等で事前定義されているものもあるので便利

matchディレクティブ
ここでマッチしたログに、指定した処理を適用する。
デフォルトで使えるものには、標準出力、ファイル、他のfluentサーバーへの転送、
これまでに挙げた方法を組み合わせる、等がある。

@type s3
転送先を定義。この例ではAWSのS3。S３転送できる権限を付けておくことが必要

s3_bucket
AWS S3バケット名を指定。

s3_region
AWS S3の存在するリージョンを指定。

time_slice_format
出力されるファイル名の一部になる。下記は出力時に実際の時刻に変換される

%Y: 年(4桁の数字)
%m: 月 (01..12)
%d: 日付 (01..31)
%H: 時間(00..23)
%M: 分(00..59)
%S: 秒(00..60)

timekey_zone
タイムゾーン指定

format
S3にアップロードするファイルの形式指定。デフォルトはout_fileで、他にjson, ltsv, single_valueがある

<buffer>~</buffer>
buffer_type
fluentdが受け取ったデータをどう保存するか、メモリかファイルが選べる

path
buffer_timeがfileの場合のファイル名

time_key

time_key_wait

includeディレクティブ
複数の設定ファイルを設定することができる。/etc/log/httpd/conf.d/*.confのようなイメージ。

tag(タグ)
ログを指定するID。matchディレクティブで指定する「debug.test」とかがタグの一例。

Fluented
http://fluentular.herokuapp.com/parse?regexp=%5E%28%3F%3Cremote%3E%5B%5E+%5D*%29+%28%3F%3Chost%3E%5B%5E+%5D*%29+%28%3F%3Cuser%3E%5B%5E+%5D*%29+%5C%5B%28%3F%3Ctime%3E%5B%5E%5C%5D%5D*%29%5C%5D+%22%28%3F%3Cmethod%3E%5CS%2B%29%28%3F%3A+%2B%28%3F%3Cpath%3E%5B%5E+%5D*%29+%2B%5CS*%29%3F%22+%28%3F%3Ccode%3E%5B%5E+%5D*%29+%28%3F%3Csize%3E%5B%5E+%5D*%29%28%3F%3A+%22%28%3F%3Creferer%3E%5B%5E%5C%22%5D*%29%22+%22%28%3F%3Cagent%3E%5B%5E%5C%22%5D*%29%22+%22%28%3F%3Cforwarder%3E%5B%5E%5C%22%5D*%29%22%29%3F&input=60.236.159.133+-+-+%5B24%2FAug%2F2019%3A22%3A04%3A05+%2B0900%5D+%22GET+%2Fpoweredby.png+HTTP%2F1.1%22+200+2811+%22http%3A%2F%2Fxxxxx.xyz%3A81%2F%22+%22Mozilla%2F5.0+%28Windows+NT+10.0%3B+Win64%3B+x64%29+AppleWebKit%2F537.36+%28KHTML%2C+like+Gecko%29+Chrome%2F76.0.3809.100+Safari%2F537.36%22+%22-%22&time_format=%25d%2F%25b%2F%25Y%3A%25H%3A%25M%3A%25S+%25z

AWS Athenaでログ分析する。

Athenaへテーブル作成

#nginxログ用
CREATE EXTERNAL TABLE IF NOT EXISTS fluentd.nginx (
         `remote` string,
         `host` string,
         `user` string,
         `time` string,
         `method` string,
         `path` string,
         `code` string,
         `size` string,
         `referer` string,
         `agent` string,
         `forwarder` string,
         `hostname` string
) PARTITIONED BY (
         year int,
         month int,
         day int      
) 
ROW FORMAT serde 'org.apache.hive.hcatalog.data.JsonSerDe'
WITH serdeproperties ( 'paths'='remote,host,user,time,method,path,code,size,referer, agent,forwarder,hostname') 
LOCATION 's3://test12349999/fluentd/nginx/accesslog/';

#Auroraログ用
CREATE EXTERNAL TABLE IF NOT EXISTS fluentd.aurora (
         `time` string,
         `remote_host` string,
         `user_name` string,
         `database_name` string,
         `process_id` string,
         `severity` string,
         `message` string,
         `hostname` string
) PARTITIONED BY (
         year int,
         month int,
         day int      
) 
ROW FORMAT serde 'org.apache.hive.hcatalog.data.JsonSerDe'
WITH serdeproperties ( 'paths'='time,remote_host,user_name,database_name,process_id,severity,message,hostname') 
LOCATION 's3://test12349999/fluentd/aurora/postgresqllog/';

パーティションロード処理

MSCK REPAIR TABLE access_database.access_table;

手動パーティション追加・削除

ALTER TABLE nginx ADD PARTITION (year=2019,month=09,day=01);
ALTER TABLE nginx DROP PARTITION (year=2019,month=09,day=01);