ELK+springboot搭建分布式日志系统细节以及扩展经验之谈

上篇主要介绍了怎么搭建服务的，这篇就主要讲怎么用得到的日志数据处理

使用的官网最新的6.6.*（版本不一的话配置的FILFER写法可能不一样）

第一个业务场景：记录系统日志+项目日志

logstash日志可参考如下：

input {
 beats {
   port => 5044
  }
}
filter {
  if [fields][logtype] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:[%{POSINT:syslog_pid}])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  }
  if [fields][logtype] == "json" {
    json {
      source => "message"
      target => "data"
    }
  }
}
output {
  if [fields][logtype] == "json"{
    elasticsearch {
      hosts => ["127.0.0.1:9200"]
      index => "json-%{+YYYY.MM.dd}"
    }
  }

  if [fields][logtype] == "syslog"{
    elasticsearch {
      hosts => ["127.0.0.1:9200"]
      index => "syslog-%{+YYYY.MM.dd}"
    }
  }

这里有几个知识点

一：input输入框的来源很多：kafka（下一章将讲到）+redis+各种官网支持的strem流，具体复杂的配置可以下一节讲到

二：input获取的是beat（类如filebeat）端口5044的socket请求，因为我们要将数据json化处理（更好的用于角标处理）。grok 语法用于替换，将原字段替换成新的，比如（这里我取了个他人写的文章实例）：

55.3.244.1 GET /index.html 15824 0.043 这样的字符串，我要替换成json

grok {
match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
}

就会变成

client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043

这里具体语法不是主要的，主要是要组装成我们需要的数据样式

三：filer拦截器处理能更好构建数据

  这里我将message用json格式，也可以添加额外的；logtype属性用于区分系统日志还是beat日志

四：output可以多输出源配置，最常见的是输出到ES上，其实还可以同时输出到日志文件上（可以在初期用于检查数据传输的结构是否正确），如配置（注意index指数类似ID的功能）：

file {
      path => "/config-dir/test.log"
      flush_interval => 0
  }
  elasticsearch {
    hosts => ["127.0.0.1:9200"]
    index => "els-server-%{+YYYY.MM.dd}"
  }

五：配置出现问题经验

由于版本原因，官网文档有部分写的模糊（也可能是我英文不好的原因），我就给大家演示下我出现过的配置错误：

1比如input端口参数故意写错，logstash  -f logstash-test1.conf（这里是window启动bat和配置test1文件），cmd窗口出现：

input类型没有为1的，提示明确告诉你咯（外国的软件设计上就这么牛），这是我改动的

input {
 1 {
   port => 5044
  }
}

还有就是格式问题,这里举个kafka的input例子(我出现的)，先列个正确的写法（）：

topics => ["els-server"]
topics_pattern => "els-server.*"

然后随便改动下分区（改为ls-server.*），cmd启动

你发现执行到这就完了,提示你设置分区，可见你的分区参数是错误

这还是好的，如果你参数配置错误，发现根本启动不了（仔细查看错误，也会告诉你哪一行配置错误），这种问题出现在版本不同很常见！

综上所述这么多，主要是还是想说：仔细看日志吧少年！

以为这么就完了，没呢，来了老弟！

kibana的功能模块还是可以多学学的，哈哈哈,有帮助的话给点个赞