一、简介与安装

logstash的作用：日志收集

如何工作：

• 输入（input）:必须，如stdin、file、http、exec…
• 过滤器（filter）：可选，如grok、mutate…
• 输出（output）：必须，如stdout、elasticsearch、file…

编解码器

• 作为输入或输出的一部分，分离传输与序列化过程

运行Logstash的两种方式

logstash -e CONFIG_STRING //使用给定的配置字符串，其格式同配置文件
logstash -f CONFIG_PATH //读取配置文件

 

安装环境：java8 Centos7

安装直接解压压缩包就可以，移动并重命名到你想到的文件中去，就酱🥳

二、简单体验

输入输出（配置字符串）

首先需要进入你安装的logstash的bin目录下，执行以下语句，如下图我们可以进行简单的输入输出
./logstash -e 'input { stdin{} } output { stdout {} }'

修改输出格式为json格式

  如果想修改输出的格式，我们可以将语句进行简单的修改,即可得到如json格式的输出类型
./logstash -e 'input { stdin{} } output { stdout {codec => rubydebug} }'

三、读取配置文件

输入输出

  1.读取配置文件，首先我们要自定义配置文件，新建一个目录，在目录下创建一个自定义的配置文件，开始编辑

注意格式规范

input {stdin {}
}
output {stdout {codec => rubydebug}
}

2.加载配置文件，简单执行检查
./logstash -f /opt/config/mylog.conf

读取文件中的内容

（1）读取普通文本

   若要读取文件中的内容，仅需修改配置文件即可

   1.首先你要有需要读取的文件，这里以新建的opt目录下先建的a.txt为例

2.修改自定义配置文件
格式入下，注意格式规范

input {file {path => "/opt/a.txt"start_position => "beginning"sincedb_path => "/dev/null"}
}
output {stdout {codec => rubydebug}
}

3.测试是否能够成功读取

（2）读取json格式内容

    若要读取为json格式的内容，修改配置文件如下

input {file {path => "/opt/a.txt"start_position => "beginning"sincedb_path => "/dev/null"codec => json}
}
output {stdout {codec => rubydebug}
}

以下内容作为测试内容

{"browse":"chrome:true version:81.1","custid":"10000","jzy":{"eventCategory":"btn_click","position":"128 65","time":"2000-01-01 12:00:00","pageName":"index.html","msg":"body"}}
{"browse":"chrome:true version:81.1","custid":"29070","jzy":{"eventCategory":"btn_click","position":"128 65","time":"2000-01-01 12:01:00","pageName":"index.html","msg":"every"}}
{"browse":"chrome:true version:81.2","custid":"26775","jzy":{"eventCategory":"btn_click","position":"128 65","time":"2000-01-01 12:02:00","pageName":"index.html","msg":"body"}}
{"browse":"chrome:true version:81.1","custid":"43694","jzy":{"eventCategory":"href_click","position":"128 65","time":"2000-01-01 12:03:00","pageName":"index.html","msg":"hello"}}

执行测试结果如下：

（3）多层嵌套json平级格式

   以上测试出现大json套小json的内容为多层json，输出格式存在不平级的现象

需要使用filter过滤，修改配置如下

input {file {path => "/opt/a.txt"start_position => "beginning"sincedb_path => "/dev/null"codec => json}
}
filter {mutate {add_field => { "@adv" => "%{jzy}" }}json {source => "@adv" remove_field => [ "@adv","jzy" ]}
}
output {stdout {codec => rubydebug}
}

   思路：用@adv接受jzy的json格式的内容，最后删除不需要的内容，是的jzy中的内容与browse的内容平级

   测试结果展示

四、grok语法与正则的使用

   现要对以下非json格式进行分割，并将其以json格式输出

68738|event_login|947700000426|192.168.56.1
50553|event_login|948042000426|192.168.56.2
76286|event_login|951224460426|192.168.56.3

   以上内容输入在/opt/b.txt中

方法1：

grok语法

input {file {path => "/opt/b.txt"start_position => "beginning"sincedb_path => "/dev/null"}
}
filter {grok {match => { "message" => "%{NUMBER:usid}\|%{WORD:uname}\|%{NUMBER:times}\|%{IP:client_ip}" }remove_field => [ "message" ]}
}
output {stdout {codec => rubydebug}
}

方法2：

正则匹配：

input {file {path => "/opt/b.txt"start_position => "beginning"sincedb_path => "/dev/null"}
}
filter {grok {        match => { "message" => "(?<userid>[0-9]+)\|(?<event_name>[a-zA-Z_]+)\|(?<times>[0-9]+)\|(?<clientip>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})" }remove_field => [ "message" ]}
}
output {stdout {codec => rubydebug}
}

展示结果

五、双线程处理

   同时读取两个文件，这里以a.txt json格式，与b.txt普通格式为例

   修改配置如下

input {file {path => "/opt/b.txt"start_position => "beginning"sincedb_path => "/dev/null"type => "system"}file {path => "/opt/a.txt"start_position => "beginning"sincedb_path => "/dev/null"codec => jsontype => "action"}
}
filter {if [type] == "system" {grok {            match => { "message" => "(?<userid>[0-9]+)\|(?<event_name>[a-zA-Z_]+)\|(?<times>[0-9]+)\|(?<clientip>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})" }remove_field => [ "message" ]}}else {mutate {add_field => { "@adv" => "%{jzy}" }}json {source => "@adv"remove_field => [ "@adv","jzy" ]}}
}
output {stdout {codec => rubydebug}
}

结果展示: