1.概述
1.1.什么是logstash？
之前我们聊了es，并且用docker搭建了一个es+kibana的环境。es目前最普遍的用法是用来存储日志的，然后结合kibana对日志做一些可视化的工作。既然要收集日志，就面临着一个问题：
各个系统的日志格式可能是不一样的。
这就需要有一个中间层，将收集上来的日志处理一遍，然后存进es里面去。es的母公司elastic推出了一套组件，专门用来做日志收集、存储、分析等工作，这一套组件由elastic search、kibana、logstash组成，简称ELK：
（beats是另一个用来收集数据的组件，此处先忽略即可。）
Logstash是其中一个用于处理和转发日志、时间序列数据以及其他事件数据的开源工具。它是 Elastic Stack（以前称为 ELK Stack）的一部分，用于实现日志的收集、处理、转换和输出。
本文将接着之前的系列，以之前搭建出来的环境为基础，继续聊如何使用logstash。
前文关于ES的内容：
【elastic search】下载安装、使用教程_elasticsearch下载俺咋混给-CSDN博客
前文我们用docker搭建的es+kibana的内容：
docker安装elasticsearch+kibana-CSDN博客
1.2.logstash的配置
整个logstash在自身组件的层次结构上就分为三部分：输入、处理、输出。
logstash的使用其实也就是只需要配置好就行，配置文件也是分为输入、处理、输出三个部分的：
-
Input 插件： 用于从各种来源收集数据。常见的输入插件包括 Beats、Syslog、File、Kafka、JDBC 等。
-
Filter 插件： 用于对收集到的数据进行处理和过滤。常见的过滤插件包括 grok、mutate、date、geoip 等。
-
Output 插件： 用于将处理后的数据发送到目的地。常见的输出插件包括 Elasticsearch、File、Kafka、JDBC、TCP、UDP 等。
配置示例：
input {
file {
path => "/path/to/your/logfile.log"
start_position => "beginning"
}
}filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
}output {
elasticsearch {
hosts => ["your_elasticsearch_host:9200"]
index => "your_index_name"
}
}
这个示例配置包含了一个 File Input 插件、一个 Grok Filter 插件和一个 Elasticsearch Output 插件，实现了从文件中读取日志、解析日志格式并将结果发送到 Elasticsearch 的过程。
具体要用的时候去搜一下logstash的相关配置就行，没什么复杂的。
2.安装logstash
拉取镜像：
docker pull logstash:7.6.1
配置：
input{tcp{port=>5044codec=>json_lines}
}
filter{ruby{code=>"event.set('timestamp',event.get('@timestamp').time.localtime+8*60*60)"}ruby{code=>"event.set('@timestamp',event.get('timestamp'))"}mutate{remove_field=>["timestamp"]}
}
output{#elasticsearch配置elasticsearch{hosts=>{#索引名称index=>"logstash-%{[server_name]}-%{+YYYY.MM.dd}"}}
}
启动镜像：
docker run -d logstash:7.6.1 -v /home/docker/logstash/config/logstash.conf -v /usr/share/logstash/config/logstash.conf --name logstash
3.应用向logstash推送日志
logstash搭建好后，就是应用的日志该怎样去到logstash里了。logstash采用的是CS架构，应用作为client端需要主动向logstash来推数据。spring boot官方是给出了向logstash推数据的组件的，此处以SpringBoot推送配置为例
依赖：
<dependency><groupId>net.logstash.logback</groupId><artifactId>logstash-logback-encoder</artifactId><version>6.3</version> </dependency>
日志框架的配置：
推送的实现其实就是新加了一个logstash的appender来实现的，所以配置这个appender即可，在logback-spring.xml进行如下示例配置：
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="STASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
<destination>192.168.31.10:5044</destination>
<encoder class="net.logstash.logback.encoder.LogstashEncoder" >
<includeCallerData>true</includeCallerData>
<timeZone>UTC</timeZone>
</encoder>
</appender><root level="INFO">
<appender-ref ref="STASH"></appender-ref>
</root>
</configuration>
关于日志框架，不熟悉的可以移步作者另一篇文章，对日志框架讲得很清楚：
【JAVA日志框架大全】一文快速讲透JAVA日志体系-CSDN博客
