如何快速采集分析平台日志，并进行展示监控？( 二 ) _日志

Dissect的使用规则：https://www.elastic.co/guide/en/logstash/current/plugins-filters-dissect.html
Logstash配置如下：
input {
beats {
port => "5045"
}
}
filter {
if "/user/if_ia_pro/output/test" in [message] {
dissect {
mApping => { "message" => "%{logd} %{drop} %{level} %{log-type}: %{?allowed}=%{&allowed}%{?ugi}=%{&ugi} (%{?authtype})%{?ip}=/%{&ip}%{?cmd}=%{&cmd}%{}=/user/if_ia_pro/output/test/%{src2}/%{src3}/%{}%{?dst}=%{&dst}%{?perm}=%{&perm}%{?proto}=%{&proto}" }
add_field => {
"srctable" => "/user/if_ia_pro/output/test/%{src2}/%{src3}"
"logdate" => "%{logd} %{drop}"
}
remove_field => ['message','src2','src3','logd','drop']
}
}
else if "/user/lf_zh_pro/lf_safedata_pro/output/" in [message] {
dissect {
mapping => { "message" => "%{logd} %{drop} %{level} %{log-type}: %{?allowed}=%{&allowed}%{?ugi}=%{&ugi} (%{?authtype})%{?ip}=/%{&ip}%{?cmd}=%{&cmd}%{}=/user/lf_zh_pro/lf_safedata_pro/output/%{src2}/%{}%{?dst}=%{&dst}%{?perm}=%{&perm}%{?proto}=%{&proto}" }
add_field => {
"srctable" => "/user/lf_zh_pro/lf_safedata_pro/output/%{src2}"
"logdate" => "%{logd} %{drop}"
}
remove_field => ['message','src2','drop']
}
}
else if "/files/" in [message] {
dissect {
mapping => { "message" => "%{logd} %{drop} %{level} %{log-type}: %{?allowed}=%{&allowed}%{?ugi}=%{&ugi} (%{?authtype})%{?ip}=/%{&ip}%{?cmd}=%{&cmd}%{}=/files/%{src2}/%{}%{?dst}=%{&dst}%{?perm}=%{&perm}%{?proto}=%{&proto}" }
add_field => {
"srctable" => "/files/%{src2}"
"logdate" => "%{logd} %{drop}"
}
remove_field => ['message','src2','drop']
}
}
else {
dissect {
mapping => { "message" => "%{logd} %{drop} %{level} %{log-type}: %{?allowed}=%{&allowed}%{?ugi}=%{&ugi} (%{?authtype})%{?ip}=/%{&ip}%{?cmd}=%{&cmd}%{}=/%{src}/%{src1}/%{src2}/%{src3}/%{}%{?dst}=%{&dst}%{?perm}=%{&perm}%{?proto}=%{&proto}" }
add_field => {
"srctable" => "/%{src}/%{src1}/%{src2}/%{src3}"
"logdate" => "%{logd} %{drop}"
}
remove_field => ['message','src','src1','src2','src3','logd','drop']
}
}
date {
match => [ "logdate","ISO8601" ]
target => "@times"
remove_field => ['logdate']
}
}
output {
elasticsearch {
hosts => ["es-host:9200"]
index => "logstash-hdfs-auit-%{+YYYY.MM.dd}"
user => "elastic"
password => "password"
}
stdout { }
}
3、ES上观察数据
Filebeat和Logstash配置好采集分析hdfs-audit.log之后启动进程，到ES上观察会发现创建有logstash-hdfs-auit- YYYY.MM.dd的index 。

文章插图
具体查看数据，可以看到已经具备多个需要使用到的字段。

文章插图
Grafana配置NameNode RPC操作
最后一步就需要在Grafana上配置连接ES数据库。

文章插图
然后创建Dashboard依次配置以下几种查询展示：
1）集群整体RPC每分钟连接次数

文章插图

文章插图
2）HDFS路径All下All类型每分钟操作计数

文章插图

文章插图
3）All类型操作计数最多的hdfs路径

文章插图

文章插图
4）路径All下操作计数排行前五的类型和All操作类型下操作计数前五的路径

文章插图
总结
那么现在对于企业来说，不管是在物理机上还是云上，玩自己的大数据平台跑生产任务，就不可避免会有不够合理不够优化的任务，比如最简单的集群对拷任务出现异常中断时，我们通常会挂定时任务并对hadoop distcp添加-update参数，进行对比更新覆盖，这时当定时吊起的过于频繁，就会发现当对拷目录下文件数越来越多，NameNode对该目录的listStatus类型的RPC连接会激增，这时我们就需要优化对拷任务。