【ELK】logstash快速入门

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的使用其实也就是只需要配置好就行，配置文件也是分为输入、处理、输出三个部分的：

1. Input 插件： 用于从各种来源收集数据。常见的输入插件包括 Beats、Syslog、File、Kafka、JDBC 等。

2. Filter 插件： 用于对收集到的数据进行处理和过滤。常见的过滤插件包括 grok、mutate、date、geoip 等。

3. 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博客