1 资源

资源信息	版本号	备注
服务器	Centos7	IP: 192.168.51.4
logstash	7.4.2	IP: 192.168.51.4

2 介绍

Logstash 是一个数据流引擎：

• 它是用于数据物流的开源流式 ETL（Extract-Transform-Load）引擎
• 在几分钟内建立数据流管道
• 具有水平可扩展及韧性且具有自适应缓冲
• 不可知的数据源
• 具有200多个集成和处理器的插件生态系统
• 使用 Elastic Stack 监视和管理部署

Logstash 几乎可以摄入各种类别的数据：

2020050716093388.png

它可以摄入日志，文件，指标或者网路真实数据。经过 Logstash 的处理，变为可以使用的 Web Apps 可以消耗的数据，也可以存储于数据中心，或变为其它的流式数据。

20200507161418149.png

• Logstash 可以很方便地和 Beats一起合作，这也是被推荐的方法
  Logstash 也可以和那些著名的云厂商的服务一起合作处理它们的数据
• 它也可以和最为同样的信息消息队列，比如 redis 或 kafka 一起协作
• Logstash 也可以使用 JDBC 来访问 RDMS 数据
• 它也可以和 IoT 设备一起处理它们的数据
• Logstash 不仅仅可以把数据传送到 Elasticsearch，而且它还可以把数据发送至很多其它的目的地，并作为它们的输入源做进一步的处理

附上官网地址：https://www.elastic.co/cn/logstash

3 安装与配置

3.1 安装

logstash单机版的安装，可参考博文：

ES Logstash安装以及数据同步配置：https://www.jianshu.com/p/657bb28caf8e

3.2 配置

3.2.1 配置文件说明

## conf下配置文件说明：
# logstash配置文件：/config/logstash.yml
# JVM参数文件：/config/jvm.options
#  日志格式配置文件：log4j2.properties
#  制作Linux服务参数：/config/startup.options

3.2.2 自定义线程数

[root@localhost software]# vim /usr/local/logstash-7.4.2/config/logstash.yml
## 增加workers工作线程数 可以有效的提升logstash性能
pipeline.workers: 16

3.3.3 自定义配置

[root@localhost software]# cd /usr/local/logstash-7.4.2/
[root@localhost logstash-7.4.2]# mkdir script
[root@localhost logstash-7.4.2]# cd script
[root@localhost script]# vim logstash-script.conf

## multiline 插件也可以用于其他类似的堆栈式信息，比如 linux 的内核日志。
input {
  kafka {
    ## app-log-服务名称
    topics_pattern => "app-log-.*"
    bootstrap_servers => "192.168.51.4:9092"
    codec => json
    consumer_threads => 1   ## 增加consumer的并行消费线程数
    decorate_events => true
    #auto_offset_rest => "latest"
    group_id => "app-log-group"
   }

   kafka {
    ## error-log-服务名称
    topics_pattern => "error-log-.*"
    bootstrap_servers => "192.168.51.4:9092"
    codec => json
    consumer_threads => 1
    decorate_events => true
    #auto_offset_rest => "latest"
    group_id => "error-log-group"
   }

}

filter {

  ## 时区转换
  ruby {
    code => "event.set('index_time',event.timestamp.time.localtime.strftime('%Y.%m.%d'))"
  }

  if "app-log" in [fields][logtopic]{
    grok {
        ## 表达式,这里对应的是Springboot输出的日志格式
        match => ["message", "\[%{NOTSPACE:currentDateTime}\] \[%{NOTSPACE:level}\] \[%{NOTSPACE:thread-id}\] \[%{NOTSPACE:class}\] \[%{DATA:hostName}\] \[%{DATA:ip}\] \[%{DATA:applicationName}\] \[%{DATA:location}\] \[%{DATA:messageInfo}\] ## (\'\'|%{QUOTEDSTRING:throwable})"]
    }
  }

  if "error-log" in [fields][logtopic]{
    grok {
        ## 表达式
        match => ["message", "\[%{NOTSPACE:currentDateTime}\] \[%{NOTSPACE:level}\] \[%{NOTSPACE:thread-id}\] \[%{NOTSPACE:class}\] \[%{DATA:hostName}\] \[%{DATA:ip}\] \[%{DATA:applicationName}\] \[%{DATA:location}\] \[%{DATA:messageInfo}\] ## (\'\'|%{QUOTEDSTRING:throwable})"]
    }
  }

}

## 测试输出到控制台：
output {
  stdout { codec => rubydebug }
}


## elasticsearch：
output {

  if "app-log" in [fields][logtopic]{
    ## es插件
    elasticsearch {
        # es服务地址
        hosts => ["192.168.51.4:9200"]
        # 用户名密码
        user => "elastic"
        password => "123456"
        ## 索引名，+ 号开头的，就会自动认为后面是时间格式：
        ## javalog-app-service-2019.01.23
        index => "app-log-%{[fields][logbiz]}-%{index_time}"
        # 是否嗅探集群ip：一般设置true；http://192.168.11.35:9200/_nodes/http?pretty
        # 通过嗅探机制进行es集群负载均衡发日志消息
        sniffing => true
        # logstash默认自带一个mapping模板，进行模板覆盖
        template_overwrite => true
    }
  }

  if "error-log" in [fields][logtopic]{
    elasticsearch {
        hosts => ["192.168.51.4:9200"]
        user => "elastic"
        password => "123456"
        index => "error-log-%{[fields][logbiz]}-%{index_time}"
        sniffing => true
        template_overwrite => true
    }
  }


}

3.3 启动logstash

[root@localhost software]# nohup /usr/local/logstash-7.4.2//bin/logstash -f /usr/local/logstash-7.4.2/script/logstash-script.conf &

4 相关信息

• 博文不易，辛苦各位猿友点个关注和赞，感谢