【Elasticsearch系列廿】Logstash 学习

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。

推荐:kwan 的首页,持续学习,不断总结,共同进步,活到老学到老
导航
檀越剑指大厂系列:全面总结 java 核心技术,jvm,并发编程 redis,kafka,Spring,微服务等
常用开发工具系列:常用的开发工具,IDEA,Mac,Alfred,Git,typora 等
数据库系列:详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
新空间代码工作室:提供各种软件服务,承接各种毕业设计,毕业论文等
懒人运维系列:总结好用的命令,解放双手不香吗?能用一个命令完成绝不用两个操作
数据结构与算法系列:总结数据结构和算法,不同类型针对性训练,提升编程思维,剑指大厂

非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

博客目录

- - 1.什么是 Logstash
  - 2.配置文件
  - 3.启动操作
  - 4.input 原理
  - 5.Grok 正则捕获
  - 6.Mutate 插件
  - 7.output
  - 8.综合案例

1.什么是 Logstash

下载地址

输入插件地址

过滤器插件

输出插件

logstash 是一个数据抽取工具，将数据从一个地方转移到另一个地方。如 hadoop 生态圈的 sqoop 等。

logstash 之所以功能强大和流行，还与其丰富的过滤器插件是分不开的，过滤器提供的并不单单是过滤的功能，还可以对进入过滤器的原始数据进行复杂的逻辑处理，甚至添加独特的事件到后续流程中。
Logstash 配置文件有如下三部分组成，其中 input、output 部分是必须配置，filter 部分是可选配置，而 filter 就是过滤器插件，可以在这部分实现各种日志过滤功能。

2.配置文件

input {
    #输入插件
}
filter {
    #过滤匹配插件
}
output {
    #输出插件
}

标准输入stdin:

input{
    stdin{

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

读取 TCP 网络数据:

input {
  tcp {
    port => "1234"
  }
}

filter {
  grok {
    match => { "message" => "%{SYSLOGLINE}" }
  }
}

output {
    stdout{
        codec=>rubydebug
    }
}

3.启动操作

#windows启动
logstash.bat -e 'input{stdin{}} output{stdout{}}'

#脚本启动
logstash.bat -f ../config/test1.conf

4.input 原理

logstash 使用一个名为 filewatch 的 ruby gem 库来监听文件变化,并通过一个叫.sincedb 的数据库文件来记录被监听的日志文件的读取进度（时间戳），这个 sincedb 数据文件的默认路径在 <path.data>/plugins/inputs/file 下面，文件名类似于.sincedb_123456，而<path.data>表示 logstash 插件存储目录，默认是 LOGSTASH_HOME/data。

input {
    file {
        path => ["/var/*/*"]
        start_position => "beginning"
    }
}
output {
    stdout{
        codec=>rubydebug
    }
}

默认情况下，logstash 会从文件的结束位置开始读取数据，也就是说 logstash 进程会以类似 tail -f 命令的形式逐行获取数据。

5.Grok 正则捕获

grok 是一个十分强大的 logstash filter 插件，他可以通过正则解析任意文本，将非结构化日志数据弄成结构化和方便查询的结构。他是目前 logstash 中解析非结构化日志数据最好的方式。

Grok 的语法规则是：

%{语法: 语义}

例如输入的内容为：

172.16.213.132 [07/Feb/2019:16:24:19 +0800] "GET / HTTP/1.1" 403 5039

%{IP:clientip}匹配模式将获得的结果为：clientip: 172.16.213.132
%{HTTPDATE:timestamp}匹配模式将获得的结果为：timestamp: 07/Feb/2018:16:24:19 +0800
%{QS:referrer}匹配模式将获得的结果为：referrer: “GET / HTTP/1.1”

下面是一个组合匹配模式，它可以获取上面输入的所有内容：

%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}

通过上面这个组合匹配模式，我们将输入的内容分成了五个部分，即五个字段，将输入内容分割为不同的数据字段，这对于日后解析和查询日志数据非常有用，这正是使用 grok 的目的。

例子：

input{
    stdin{}
}
filter{
    grok{
        match => ["message","%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}"]
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}

输入内容：

172.16.213.132 [07/Feb/2019:16:24:19 +0800] "GET / HTTP/1.1" 403 5039

时间处理(Date):

date 插件是对于排序事件和回填旧数据尤其重要，它可以用来转换日志记录中的时间字段，变成 LogStash::Timestamp 对象，然后转存到@timestamp 字段里，下面是 date 插件的一个配置示例（这里仅仅列出 filter 部分）：

filter {
    grok {
        match => ["message", "%{HTTPDATE:timestamp}"]
    }
    date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
}

GeoIP 地址查询归类:

filter {
    geoip {
        source => "ip_field"
    }
}

kibana测试Grok:

6.Mutate 插件

数据修改(Mutate):

正则表达式替换匹配字段

gsub 可以通过正则表达式替换字段中匹配到的值，只对字符串字段有效，下面是一个关于 mutate 插件中 gsub 的示例（仅列出 filter 部分）：

filter {
    mutate {
        gsub => ["filed_name_1", "/" , "_"]
    }
}

这个示例表示将 filedname_1 字段中所有"/“字符替换为”"。

分隔符分割字符串为数组:

split 可以通过指定的分隔符分割字段中的字符串为数组，下面是一个关于 mutate 插件中 split 的示例（仅列出 filter 部分）：

filter {
    mutate {
        split => ["filed_name_2", "|"]
    }
}

这个示例表示将 filed_name_2 字段以"|"为区间分隔为数组。

重命名字段:

rename 可以实现重命名某个字段的功能，下面是一个关于 mutate 插件中 rename 的示例（仅列出 filter 部分）：

filter {
    mutate {
        rename => { "old_field" => "new_field" }
    }
}

这个示例表示将字段 old_field 重命名为 new_field。

删除字段:

remove_field 可以实现删除某个字段的功能，下面是一个关于 mutate 插件中 remove_field 的示例（仅列出 filter 部分）：

filter {
    mutate {
        remove_field  =>  ["timestamp"]
    }
}

这个示例表示将字段 timestamp 删除。

综合例子:

input {
    stdin {}
}
filter {
    grok {
        match => { "message" => "%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}" }
        remove_field => [ "message" ]
   }
date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
mutate {
          convert => [ "response","float" ]
           rename => { "response" => "response_new" }
           gsub => ["referrer","\"",""]
           split => ["clientip", "."]
        }
}
output {
    stdout {
        codec => "rubydebug"
    }

7.output

output 是 Logstash 的最后阶段，一个事件可以经过多个输出，而一旦所有输出处理完成，整个事件就执行完成。一些常用的输出包括：

file：表示将日志数据写入磁盘上的文件。
elasticsearch：表示将日志数据发送给 Elasticsearch。Elasticsearch 可以高效方便和易于查询的保存数据。

输出到标准输出(stdout):

output {
    stdout {
        codec => rubydebug
    }
}

保存为文件file:

output {
    file {
        path => "/data/log/%{+yyyy-MM-dd}/%{host}_%{+HH}.log"
    }
}

输出到 elasticsearch:

output {
    elasticsearch {
        host => ["192.168.1.1:9200","172.16.213.77:9200"]
        index => "logstash-%{+YYYY.MM.dd}"
    }
}

host：是一个数组类型的值，后面跟的值是 elasticsearch 节点的地址与端口，默认端口是 9200。可添加多个地址。
index：写入 elasticsearch 的索引的名称，这里可以使用变量。Logstash 提供了%{+YYYY.MM.dd}这种写法。在语法解析的时候，看到以+ 号开头的，就会自动认为后面是时间格式，尝试用时间格式来解析后续字符串。这种以天为单位分割的写法，可以很容易的删除老的数据或者搜索指定时间范围内的数据。此外，注意索引名中不能有大写字母。
manage_template:用来设置是否开启 logstash 自动管理模板功能，如果设置为 false 将关闭自动管理模板功能。如果我们自定义了模板，那么应该设置为 false。
template_name:这个配置项用来设置在 Elasticsearch 中模板的名称。

8.综合案例

input {
    file {
        path => ["D:/ES/logstash-7.3.0/nginx.log"]
        start_position => "beginning"
    }
}

filter {
    grok {
        match => { "message" => "%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}" }
        remove_field => [ "message" ]
   }
	date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
	mutate {
           rename => { "response" => "response_new" }
           convert => [ "response","float" ]
           gsub => ["referrer","\"",""]
           remove_field => ["timestamp"]
           split => ["clientip", "."]
        }
}

output {
    stdout {
        codec => "rubydebug"
    }

elasticsearch {
    host => ["localhost:9200"]
    index => "logstash-%{+YYYY.MM.dd}"
}

}