windows下利用ELK做日志收集

ELK：ElasticSearch + LogStash + KibanaES中文社区：ElasticSearch中文社区下载应用场景：Logback + ELK日志收集环境：Windows10参考文章：ELK+logback：日志存储及可视化_夜太黑-CSDN博客https://blog.csdn.net/longchao2/article/details/80136872环境搭建：（三者都可以

咸鱼_Posin

3278人浏览 · 2021-12-11 17:47:34

咸鱼_Posin · 2021-12-11 17:47:34 发布

ELK：ElasticSearch + LogStash + Kibana

ES中文社区：ElasticSearch中文社区下载

应用场景：Logback + ELK日志收集

环境：Windows10

参考文章：ELK+logback：日志存储及可视化_夜太黑-CSDN博客https://blog.csdn.net/longchao2/article/details/80136872

环境搭建：（三者都可以直接下载压缩包解压后即可使用）

ElasticSearch安装：地址参见【ES中文社区】

ES中需要添加IK分词器插件，在ES中文社区中的Plugins中可以下载

找到对应的版本（我用的版本是7.6.1），下载解压，在ES的plugin目录下创建ik文件夹，将分词器插件复制过去即可

进入bin 目录双击 elasticsearch.bat启动即可

LogStash安装：地址参见【ES中文社区】

下载解压后，在bin 目录下创建配置文件供logstash启动时加载，如logstash.conf。input有多种配置，如tcp、file等，filter也有gork、csv等，output除了es也有别的输出目的地。

//输入配置
input {
    tcp {
      host => "127.0.0.1"//安装logstash的机器的ip地址
      port => 9250//端口号
      mode => "server"
      codec => "json"//按json格式编码
    }
}
//过滤配置
filter {
    csv{
      source => "message"
      columns => ["column a","column b"] //从message中取出字段名
      remove_field => ["column c","column b"] //去除不需要的字段名
    }
}
//输出配置
output {
    stdout{codec =>rubydebug}
    elasticsearch {
      hosts => ["127.0.0.1:9200"]
      index => "syslog-%{[appname]}"  //es中创建索引的名称
      flush_size => 1000 //刷新频率
    }
}

## 示例：对上面的配置，若输入日志为123,234,456，logstash以逗号为分隔符，自动分割日志为value值，输出为，column a:123,column b:234
## net.logstash.logback 这个框架默认输出日志是json 所以一定要有codec => "json" 这个配置
## index => "%{[appname]}" 这个配置就是获取日志中的 appname字段的值做为索引的名称

指定配置文件启动logstash，logstash -f logstash.conf

如图，Starting tcp input listener {:address=>"127.0.0.1:9250", :ssl_enable=>"false"} 指定TCP方式传输监听配置，也就是在logstash.conf中的输入配置。

配置详解参考https://blog.csdn.net/qq330983778/article/details/105644835/?utm_term=logstash%E9%85%8D%E7%BD%AE%E4%B8%ADtcp&utm_medium=distribute.pc_aggpage_search_result.none-task-blog-2~all~sobaiduweb~default-0-105644835&spm=3001.4430

Kibana安装：地址参见【ES中文社区】

默认是英文版，在kibana.yml中设置中文显示

# 本机的端口
server.port: 5601
# 本机IP地址
server.host: "127.0.0.1"
# ES的服务IP+端口
elasticsearch.hosts: ["http://localhost:9200"]
# 设置为中文
i18n.locale: "zh-CN"

Logback配置

参考文章：logback将日志传输到 logstashhttps://blog.csdn.net/qq1010267837/article/details/90730474 输出json乱码https://blog.csdn.net/wang0120000/article/details/105097417/

使用logback将应用里的日志输出到logstash，输出到logstash中的日志必须是指定格式的特定日志信息，所以，需要过滤和格式化。

引入依赖

<dependency>
    <groupId>net.logstash.log4j</groupId>
    <artifactId>jsonevent-layout</artifactId>
    <version>1.6</version>
</dependency>
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>4.11</version>
</dependency>

logback.xml 文件配置(添加appender)

<appender name="stash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
    <!-- 安装logstash机器的ip地址，这里我们使用的TCP方式，所以直接配置TCP输入配置的监听 -->
    <destination>127.0.0.1:9250</destination>
    <!-- 引入过滤类 -->
    <!-- <filter class="com.program.interceptor.ELKFilter"/>-->    
    <!-- encoder必须配置,有多种可选 -->
    <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder">
         <customFields>{"appname":"elk-demo"}</customFields>
    </encoder>
<!-- <destination>destination1.domain.com:4560</destination> 
    <destination>destination2.domain.com:4560</destination> 
    <destination>destination3.domain.com:4560</destination> -->
<!-- <connectionStrategy>
      <roundRobin>
          <connectionTTL>5 minutes</connectionTTL>
      </roundRobin>
  </connectionStrategy> -->
  <!-- 向logstash输出日志如果有多个logstash IP或端口可以轮询负载各端口 -->
</appender>

自定义日志级别

通常情况下我们是在root级别加上新的appender，在root上控制日志级别。但是现在有一个场景就是我们希望能够新加的appender的日志级别能够不与其他地方交叉，也就是能够支持自定义。

<logger name="BUSINESSLOG" level="DEBUG" additivity="false">
	<appender-ref ref="CUSTOMLOG"/>
</logger>
<root level="INFO">
	<appender-ref ref="CONSOLE" />
	<appender-ref ref="DEBUG" />
	<appender-ref ref="INFO" />
	<appender-ref ref="WARN" />
	<appender-ref ref="ERROR" />
</root>

在logback中的logger标签能够定义不同的日志打印级别，logger中的level指定自身appender日志打印级别，若不指定，默认root对应的level配置。

过滤类编写

public class ELKFilter extends Filter<ILoggingEvent> {
    @Override
    public FilterReply decide(ILoggingEvent event) {
        if (event.getMessage().startsWith("KeyWords")) {//只有以"KeyWords"开头的日志被输出到logstash
            return FilterReply.ACCEPT;
        } else {
            return FilterReply.DENY;
        }
    }
}