安装logstash及logstash的初步使用-处理DNS日志
安装logstash及logstash的初步使用-处理DNS日志
安装logstash
需要高版本的java
使用1.4版本的java会有报错
# Can't start up: not enough memory查询java信息
rpm -qa | grep java
查询yum源内含有的所有java的安装包
yum -y list java*
选择需要的版本
yum install java-xxx-openjdk.x86_64
安装完成后，验证版本
java -verison
下载logstash
cd /opt/soft
wget https://artifacts.elastic.co/downloads/logstash/logstash-6.3.2.tar.gz
解压
tar zxvf logstash-6.3.2.tar.gz
cd logstash-6.3.2
若需要安装插件，可以通过logstash-plugin 来安装
查看可以安装的插件
./bin/logstash-plugin list
./bin/logstash-plugin install pkg_name
安装插件
https://github.com/logstash-plugins/logstash-patterns-core
需要安装gem
yum install -y rubygem #centos6.5
yum install -y gem #centos7
安装完成后，解压github上的插件然后进入目录执行
gem build *.gemspec
会产生一个新的*.gem文件
./bin/logstash-plugin install *.gem
然后就安装成功了
logstash的使用方式
在任意文件夹下创建一个conf文件(这里演示就直接创建在程序目录下了)
touch /opt/soft/logstash/bin/std.conf
vi /opt/soft/logstash/bin/std.conf
以下是一个实例
需要处理的日志格式:17-Jul-2018 10:54:06.822 client 127.0.0.1 18524: view default: kugou.xdwscache.ourwebpic.com IN A NOERROR + NS NE NT ND NC H 25 Response: kugou.xdwscache.ourwebpic.com 50 IN A 127.0.0.1;kugou.xdwscache.ourwebpic.com 50 IN A 127.0.0.1;std.conf内容如下
#input 日志文件来源，还可以使用redis等等具体可查询官网的input选项
input {
file {
#指定路径
path => ["/home/dnslog/*.log"]
}
}
#格式处理
filter{
#判断message是否含有字符“Respose”
if "Response" in [message] {
#通过正则的方式来切割日志，并且赋予需要的数据变量。本文导入了模块logstash-patterns-core
#logstash-patterns-core此模块来自于github：https://github.com/logstash-plugins/logstash-patterns-core
grok{
match =>{
"message" => "%{BIND9_TIMESTAMP:timestamp} client %{IP:clientip} %{POSINT:clientport}: view %{DATA:view}\:\ %{DATA:query} IN %{DATA:aaa} %{DATA:status} \+ %{GREEDYDATA:res}"
}
}
#在此通过正则过滤
grok {
match => {
#这里的res由上面grok过滤获得
"res" => ".*?(%{HOSTNAME:query1} %{NONNEGINT:clientport1} IN A %{IP:resolution};)$.*?"
}
}
#if判断失败的就直接丢弃
}else {
drop {}
}
#geoip插件通过plugin install 安装，这个插件对移动的地址识别有异常。
geoip{
#由以上第二个正则grok过滤得到的resolution获取IP
source => "resolution"
target => "geoip"
#可以注释掉fields，那就是输出完成的geoip的数据。这边只取了需要的
fields => ["city_name","country_name","region_name"]
}
#日期插件会替换原本自身的@timestamp项
date {
#将日期格式化为标准输出
match => ["timestamp", "dd-MMM-yyyy HH:mm:ss.SSS"]
locale => "en"
timezone => "+00:00"
# target=>"@timestamp"
# timezone=>"Asia/Shanghai"
}
#格式化日志将服务器名称发送过去
mutate {
rename => { "[host][name]" => "host" }
}
#去除不需要的项
mutate {
remove_field => ["_type","_source","path","message", "@version","clientport","clientport1","res"]
}
}
#输出
output {
#输出到elasticsearch,并且定义index名字
elasticsearch{
index => "logstash_dns%{[host]}_%{+YYYY.MM.dd}"
hosts => "127.0.0.1:8200"
}
#输出到界面，以rubydebug的格式，还有json等等格式
stdout {codec => rubydebug}
}
这样一个DNS的日志格式就处理完成了