网上有很多的ELK搭建教程，但大多数文章介绍的ELK版本比较旧，内容也比较零散。本文基于最新版本的elastic日志集中分析套件，以及redis作为缓冲，完整记录一套ELK架构搭建过程。并实现了对生产环境核心系统使用的Oracle weblogic + java system out日志的分析、处理。
根据官方的介绍，已推荐使用filebeat代替logstash的forward功能。所以本次搭建架构功能规划如下：
filebeat：负责日志文件监控与数据采集；
redis：负责日志数据的缓冲；
logstash：负责日志数据的分析、处理；
elasticsearch：日志数据搜索；
kibana：展示

1.系统环境

   CentOS Linux release 7.2.1511

2.Filebeat+ELK 软件包

elasticsearch-5.1.1.rpm
filebeat-5.1.1-x86_64.rpm
kibana-5.1.1-x86_64.rpm
logstash-5.1.1.rpm    
redis-3.rpm
java-1.8-jdk
download url：https://www.elastic.co/

3.配置过程

• Filebeat.yml配置文件

实现对weblogic的access.log，以及系统的nohup，java.system.out.println数据的监控。
日志示例：

accesss.log

10.10.10.10 - - [11/一月/2017:09:24:15 +0800] "POST /hx/common/index.jsp HTTP/1.1" 200 41 

nohup.out

2016-08-24 23:00:31,761 INFO com.xxx.utility.ExeSQL.__AW_getOneValue - ExecSQL : select xxx From yyy where no='00000000000000000000'
2016-08-240.000000000000000000000null
null
2016-08-24 23:00:31,764 INFO com.xxx.utility.ExeSQL.__AW_execSQL - ExecSQL : select xxx From yyyy where no='00000000000000000000'
CalType ===========null
#####calOneDuty:select xxx From yyyy where no=? and pno=mainpno
### BindSQL = select xxx From yyyy where no= '00000000000000000000'  and pno=mainpno
2016-08-24 23:00:31,770 INFO com.xxxx.utility.ExeSQL.__AW_execSQL -

-/etc/filebeat/filebeat.yml

 filebeat.prospectors:
    -
      input_type: log
      paths:
        - /pathto/weblogic/nohup.out
      encoding: gbk
      document_type: javaout
      fields:
        app_id: hxxxt
      multiline.pattern: '^(19|20)\d\d-(0[1-9]|1[012])-(0[1-9]|[12][0-9]|3[01]) [012][0-9]:[0-6][0-9]:[0-6][0-9]'
      multiline.negate: true
      multiline.match: after
      multiline.max_lines: 1000
      multiline.timeout: 60
    - 
      input_type: log
      paths:
        - /pathto/weblogic/access.log
      encoding: gbk
      document_type: httpdlog
      exclude_lines: '\.(gif|css|js|ico) HTTP'
      fields:
        app_id: hxxt
    
    output.logstash:  #系统压力小的可以直接输出到logstash
      hosts: ["localhost:5044"]
    #  enabled: false
      pipelining: 0
      index: "filebeat"
    output.redis: #系统压力大的可以直接输出到redis，再转发logstash
      hosts: ["localhost:6379"]
      key: "filebeat"
      enabled: false   #关闭输出
    output.file:   #主要用于调试
      path: "/tmp"
      filename: filebeat.out
      number_of_files: 7
      rotate_every_kb: 10000 
      enabled: false   #关闭输出
    

• logstash配置

logstash本身不带启动脚本，为了便于使用，自己编写了一个启动脚本。

#!/bin/bash
/usr/share/logstash/bin/logstash \
    --path.settings /etc/logstash \
    --config.reload.automatic \
    $@
    

/etc/logstash/conf.d/redis.conf #redis缓存日志配置

input {
    redis {
        data_type => "list" #logstash redis插件工作方式

        key => "filebeat" #监听的键值

        host => "127.0.0.1" #redis地址

        port => 6379 #redis端口号
        add_field => {   #提取filbeat写入redis的日志源主机名json格式，否则output host为空
               host => "%{[beat][hostname]}"       
                  }
    }

}
filter{}
output {
stdout{}

/etc/logstash/conf.d/beat.conf #filebeat 配置匹配httpdlog中文日期格式

input {
    beats {
        port => "5044"
    }
}

filter {

  if [type] == "javaout" {
    grok {
       match => { "message" => "(%{TIMESTAMP_ISO8601:logdatetime} %{LOGLEVEL:level} %{JAVAFILE:class} - %{GREEDYDATA:logmessage})|%{GREEDYDATA:logmessage}" }
       remove_field => [ "message" ]
    }
    date {
      timezone => "Asia/Shanghai"
      match => ["logdatetime","yyyy-MM-dd HH:mm:ss,SSS"]
      remove_field => [ "logdatetime" ]
    }
  }
  if [type] == "httpdlog" {
    #replace access log chinese charset month word,charset zh_cn.utf-8
      mutate { gsub => [
      "message","\u4E00\u6708","Jan",
      "message","\u4E8C\u6708","Fed",
      "message","\u4E09\u6708","Mar",
      "message","\u56DBC\u6708","Apr",
      "message","\u4E94\u6708","May",
      "message","\u516DC\u6708","June",
      "message","\u4E03\u6708","July",
      "message","\u516B\u6708","Aug",
      "message","\u4E5D\u6708","Sept",
      "message","\u5341\u6708","Oct",
      "message","\u5341\u4E00\u6708","Nov",
      "message","\u5341\u4E8C\u67082","Dec" ] }

    grok {
      match => { "message" => "%{COMMONAPACHELOG}" }
      remove_field => [ "message" ]
    }
    mutate {
      gsub => ["request", "\?.*$",""]
    }
    date {
      locale => "en"
      timezone => "Asia/Shanghai"
      match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
      remove_field => [ "timestamp" ]
   }

  }
}

output {
elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "%{type}-%{+YYYY.MM.dd}"
        document_type => "%{type}"
        #flush_size => 2000
        #idle_flush_time => 10
        #sniffing => true
        #template_overwrite => true
    }
file {  #主要用于调试
  path => "/tmp/logstash.out"
 }
}

• Elastic与kibana、redis默认配置即可

4.启动相应软件

systemctl start elasticsearch
systemctl start kibana
nohup bin/logstash.sh &
systemctl start redis
systemctl start filebeat

5.登录kibana，查看

http://host-severip:5601

5。 踩过的坑：
1、国内好多早期的应用系统都是采用中文GBK编码，（现在估计也是一大坨），LANG=zh_CN.GBK,这会导致应用程序的在写日期时，使用中文格式，例如，本次遇到的“11/一月/2017:09:24:15 +0800”，ELK内部以统一使用UTF8编码，且不支持中文字符转时间类型。郁闷了很久，想自己写插件的心都有，后来通过在filebeat设置字符集转换为utf8的，使用unicode regexp匹配，才解决。

参考资料：

https://technology.amis.nl/20…
https://kuther.net/blog/index…
http://www.learnes.net/index….
https://www.elastic.co/
https://github.com/logstash-p…