搭建基于docker的elk平台来分析java日志

整体思路

1. 使用已有的elk docker镜像sebp/elk快速搭建elk平台
2. sebp/elk镜像中包括了elk三个服务。
3. 在日志所在机子使用filebeat把日志传输给elk的logstash（logstash处理日志后把日志传输给es，然后kibana基于es分析数据）

关键问题

为了充分利用kibana的功能 ，要求存储在es中的数据需要json化。
那么问题来了，日志流入es的过程是：公司系统—>日志文件—>filebeat—>logstash—>es。

公司采用的日志系统是java的logback ，pattern定义如下：

%date %level [%thread] %logger{10} [%file:%line] %msg%n%n

之所以后面两个换行，本意是想通过空白行来区分哪些日志是属于同一个日志事件

问题如下

1. 这个pattern表面上是有格式的，但不属于json格式。
2. %msg输出的内容格式不可控，虽然公司有些日志是统一打印的，这部分拥有固定的一个格式，但还是存在部分格式不可控的日志。
3. 异常堆栈信息默认情况下不会包含在%msg中，于是类似：log.error(“业务异常”,exception) 这样的语句，打印出来的日志会是这样的：

%date %level [%thread] %logger{10} [%file:%line] %msg

java的异常堆栈信息：
	at .....
	at .....
	at ....

注意上面存在一个空白行，于是我们想要通过空白行来区分同一个日志事件的想法泡汤了。

日志json化解决思路

根据日志的流入过程：公司系统—>日志文件—>filebeat—>logstash—>es。

我们可以在上面的任何一个环节或多个环节做文章来达到流入es的日志的json格式的目的。

方案一：配置filebeat和logstash

不更改公司系统代码，保留原来的日志格式。在filebeat和logstash中做处理，来解决多行日志属于一个日志事件和json化的问题。

这个方案优缺点很明显：
优点：不用更改现有的代码
缺点：需要研究如何通过filebeat或者logstash来准确的区分哪些日志行是属于同一个日志的问题。区分出同一个日志事件的日志内容后，需要研究如何把日志json化。

我刚开始就是采用的这个方案，研究后得知

1. 通过配置filebeat的multiline配置，可以把多行日志纳入到一个日志事件。
2. 通过配置logstash的filter中的grok表达式，可以把日志内容json化。

于是去研究multiline如何配置，然后做实验，发现这个配置比较难理解。然后本身日志存在这样的问题：如果打印日志的时候，没有打印堆栈，则可以通过空白行来分区同一个日志内容。如果打印日志的时候打印了错误堆栈，则同一个日志事件中还会存在一个空白行。（原因在上面分析过了）还要命的是如果先后执行这样的日志打印语句：

log.error("error",exception);

log.info("info");

log.error("error2");

那么打印出来的日志会是这样的：

error

excepiton:xxxx
	at ......
	at .....
	at .......
	at ......
info

error2

可以看出，如果以空白行区分，不但会把原本属于同一个日志事件的日志分开，还会把不属于同一个日志事件的日志误认为属于同一个日志事件。

当然这个问题也可以解决，后来折腾半天发现为啥子要以空白行分割嘛，直接把分区日志事件的逻辑改为：已日期开头的日志属于新的一个日志，就完美解决这个问题了。（给自己有个白眼????）

这样更改后，通过filebeat的multiline解决了日志分割的问题。
下面需要解决日志json化的问题：关键点：grok表达式。
这个表达式的学习完全没有必要从头开始学或者系统的学习，我完全是依葫芦画瓢学习的。这里举一个非常简单的例子体验一下：
网址：http://grokdebug.herokuapp.com/ 或者 http://grok.elasticsearch.cn/
在input输入：

2018-12-21 16:00:54,774 INFO c.l.c.w.l.AccessLog [AuthAndAccessLogAop.java:191] {"request":{"headers":{"Accept":"*/*","Connection":"close","User-Agent":"Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)","Accept-Language":"zh-cn"},"params":{},"uri":"/api/test/echo","method":"GET","time":"2018-12-21 16:00:54"},"response":{"msg":"success","data":[1,2,3]},"checkTokenTime":0,"controllerTime":2}

在pattern中输入：

((?<datetime>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}) (?<level>.{3,5}) (?<logger>.*) \[(?<file>.*\.java):(?<linenum>.*)\] )

然后会输出：

{
  "datetime": [
    [
      "2018-12-21 16:00:54,774"
    ]
  ],
  "level": [
    [
      "INFO"
    ]
  ],
  "logger": [
    [
      "c.l.c.w.l.AccessLog"
    ]
  ],
  "file": [
    [
      "AuthAndAccessLogAop.java"
    ]
  ],
  "linenum": [
    [
      "191"
    ]
  ]
}

通过这个例子，大家应该可以很快根据输入写出自己的grok表达式。

问你题来了：
如果输入是有固定的格式，那么很容易通过grok表达式把日志json化，但是现在的情况是日志格式不统一 （哭~~~）。

当然可以通过想办法把日志按照格式分类，或者统一日志格式，然后处理。但是这个办法需要需要在日志源头处改动，不满足本方案的要求。

于是鉴于公司的情况，本方案我折腾了大半天宣告失败。

方案二：从日志源头处改动

既然选择从日志源头处改动了，那么索性直接在源头处将日志json化。不要再filebeat和logstash处做过多额外的配置。要不然绑定的太多因素，不方便控制。
这个方案的优缺点也很明显：
缺点：
1. 需要改动原系统的代码或者配置。
2. 改动后的日志阅读习惯会发送小心的改变，应为json化了嘛。
3. 已经生成的日志照顾不了
优点：
1. 改动灵活，想怎么改就怎么改，
2. 不需要额外学习filebeat和logstash的太多配置。比如multiline和grok表达式。

问题也来了：如何把java日志json化？

这里直接给出我的解决方案，在这套方案前，尝试过好多种方法，最终这个方案最满意，改动也小，甚至不需要改动原来的代码，只需要对logback的配置做更改。

原来logback的patternn:

%date %level [%thread] %logger{10} [%file:%line] %msg%n%n

现在logback的pattern:

{ "date":"%date", "level":"%level", "logger":"%logger{10}", "file_line":"%file:%line", "msg":%msgext, "exception":%ex }%n%n

其实改动很小是不是????，的确改动很小。之前在上家公司的时候，一个运维的同事问我可不可以打印json格式的日志，我当时想了想回答：这个怎么打呀？估计不行，比如那个异常堆栈，这个怎么搞？

现在想想，就这么简单，当时咋没想到改动pattern呀????。

但是问题还是来了：新的pattern里有两个新的东西：%msgext 和 %ex

大概能猜到分别干嘛的：
%msgext ：替换了原来的 %msg ，用来处理我们打印的内容（非异常信息）
%ex : 用来处理异常堆栈信息

这两个东西在哪里告诉logback呢，这这里：

<configuration>
    <conversionRule conversionWord="ex" converterClass="com.xxxx.StackTraceConverter" />
    <conversionRule conversionWord="msgext" converterClass="com.xxxx.MsgConverter" />
</configuration>

只需要在logback配置文件的configuration节点中加上这两个配置就可以了。
这是两个java类，代码如下：

public class StackTraceConverter extends ThrowableProxyConverter {
    public StackTraceConverter() {
    }

    @Override
    public String convert(ILoggingEvent event) {
		//拿到异常信息
        IThrowableProxy tp = event.getThrowableProxy();
		
		//如果没有异信息
        if (tp == null) {
			//返回字符串  ： "\"\""
            return JsonUtils.serialize("");
        }
		//如果有异常信息
        String ex = super.convert(event);
		
		//json化
        return JsonUtils.serialize(ex);
    }
}

public class MsgConverter extends ClassicConverter {
    public MsgConverter() {
    }

    @Override
    public String convert(ILoggingEvent event) {
        String msg = event.getFormattedMessage();

        try {
            JsonUtils.deserialize(msg, Map.class);
            return msg;
        } catch (Exception e) {
            Map<String,String> map = new HashMap<>(1);
            map.put("value",msg);
            return JsonUtils.serialize(map);
        }

    }
}

上面用到的JsonUtils是我们自己封装了下json工具，可以替换为任何json工具类。

这样一做了调整后，打印出来的日志就是json了。有几个注意的地方：扩展的这两个convert最好都返回json格式的数据，至于为什么我也讲不明白，只是经验而谈。也许有些问题我遇到了其他人未必会遇到????。

从源头上对日志进行了json化后，还有一个问题了：filebeat拿到一行日志后，给logstash，如果不做处理，logstash只会把这行日志当作普通文本，并不会当作json处理呀~

于是搜索： logstash 解析 json 数据。很快速得知答案：在filter里做如下配置：

filter {

  json {
    source => "message"
    #target => "doc"
	
	# message字段保留的就是整个json数据，如果把下面的语句生效，则流入es的数据中，没有message这个字段。具体效果可以自行做实验
    #remove_field => ["message"]
  }

  # 这里是在处理日期时间，因为日志本身有时间，然后logstash会自行生成一个时间字段：@timestamp ，显然应该以日志里的时间为准。所以有如下配置：替换@timestamp字段的值为日志本身的时间。
  date {
    match => ["date","yyyy-MM-dd HH:mm:ss,SSS"]
    target => "@timestamp"
    locale => "cn"
    timezone => "Asia/Chongqing"
  }        
}

到此，方案二结束。接下来我们来安装elk实施日志分析。

在centos7上安装docker

可参照官网安装步骤：https://docs.docker.com/install/linux/docker-ce/centos/

也可执行以下步骤：

移除旧版本：
sudo yum remove docker \
                  docker-client \
                  docker-client-latest \
                  docker-common \
                  docker-latest \
                  docker-latest-logrotate \
                  docker-logrotate \
                  docker-selinux \
                  docker-engine-selinux \
                  docker-engine
				 
安装必要的工具：
sudo yum install -y yum-utils device-mapper-persistent-data lvm2

添加软件源：
sudo yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

更新 yum 缓存：
sudo yum makecache fast

安装 Docker-ce：
sudo yum -y install docker-ce

国内配置镜像：
vim /etc/docker/daemon.json （这个文件也许不存在，新建一个即可）
把下面的镜像复制到daemon.json文件中
{
  "registry-mirrors": ["http://hub-mirror.c.163.com"]
}

启动 Docker 后台服务：
sudo systemctl start docker

测试运行 hello-world：
docker run hello-world

如果出现类似下面的提示，恭喜你安装成功

安装elk

拉取镜像

docker pull sebp/elk

创建es数据存放目录：

mkdir -p /var/data/elk

已sebp/elk镜像启动容器，并命名为：myelk

docker run -d -p 5044:5044 -p 5601:5601 -p 9200:9200 -p 9300:9300 -v /var/data/elk:/var/lib/elasticsearch --name=myelk sebp/elk

myelk中会启动三个服务：

1. logstash，占用5044端口
2. kibana，占用5601端口
3. es，占用9200和9300端口。

为了保留es数据，做了一个文件映射
执行后，可以看到如下反馈：

这里附加几个docker的用法：

查看容器的标准输出：
docker attach --sig-proxy=false myelk

停止容器：
docker stop myelk

启动容器：
docker start myelk

查看运行中的容器：
docker ps

进入容器命令行：
docker exec -it myelk /bin/bash

进入容器命令行，然后

cd /etc/logstash/conf.d/

vim 02-beats-input.conf    # 注释掉ssl ，这个根据情况选择，我这里选择不启用ssl

vim myfilter.conf 
输入如下内容并保持

filter {
  json {
    source => "message"
    #target => "doc"
	
	# message字段保留的就是整个json数据，如果把下面的语句生效，则流入es的数据中，没有message这个字段。具体效果可以自行做实验
    #remove_field => ["message"]
  }

  # 这里是在处理日期时间，因为日志本身有时间，然后logstash会自行生成一个时间字段：@timestamp ，显然应该以日志里的时间为准。所以有如下配置：替换@timestamp字段的值为日志本身的时间。
  date {
    match => ["date","yyyy-MM-dd HH:mm:ss,SSS"]
    target => "@timestamp"
    locale => "cn"
    timezone => "Asia/Chongqing"
  }        
}

重启logstash：
service logstash restart
如果提示stop fail，则再试一次。

单独测试logstash

启动关闭命令：

启动：service logstash restart 
关闭：service logstash stop 

进入容器，先执行：
service logstash stop 

保证logstash已经停止

cd /opt/logstash

vim test-logstash.conf
输入以下内容：

input {
  stdin {}
}

filter {
  json {
    source => "message"
    #target => "doc"
    #remove_field => ["message"]
  }

  date {
    match => ["date","yyyy-MM-dd HH:mm:ss,SSS"]
    target => "@timestamp"
    locale => "cn"
    timezone => "Asia/Chongqing"
  }        

}

output {
  stdout {}
}

保存，然后执行：

bin/logstash -f test-logstash.conf

等待一会，直到看到如下字样：

The stdin plugin is now waiting for input:
[2018-12-29T07:57:25,159][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"main", :thread=>"#<Thread:0x2d752f07 run>"}
[2018-12-29T07:57:25,210][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2018-12-29T07:57:25,458][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

然后直接在命令行输入一个json字符串，比如这里输入：

{ "date":"2018-12-28 18:54:04,052", "level":"INFO", "logger":"c.l.s.l.c.u.GeTuiPushUtils", "file_line":"GeTuiPushUtils.java:54", "msg":{"value":"初始化个推服务成功"}, "exception":"" }

然后会输出一个字符串，大致如下：

{
    "msg" => { "value" => "初始化个推服务成功" },
    "host" => "d6e142cd797d",
    "level" => "INFO",
    "exception" => "",
    "file_line" => "GeTuiPushUtils.java:54",
    "logger" => "c.l.s.l.c.u.GeTuiPushUtils",
    "message" => "{ \"date\":\"2018-12-28 18:54:04,052\", \"level\":\"INFO\", \"logger\":\"c.l.s.l.c.u.GeTuiPushUtils\", \"file_line\":\"GeTuiPushUtils.java:54\", \"msg\":{\"value\":\"初始化个推服务成功\"}, \"exception\":\"\" }",
    "@version" => "1",
    "@timestamp" => 2018-12-28T10:54:04.052Z,
    "date" => "2018-12-28 18:54:04,052"
}

如果是这样的，说明配置成功。然后按住  ctrl + c 停止测试。

其实到此应该就可以结束了。这里多贴一个配置，需求是：需要分环境（测试环境，生产环境）建立不同的es index。 这里直接贴上配置：

vim 30-output.conf

output {
  if "test-log" in [tags] {
    elasticsearch {
      hosts => ["localhost"]
      manage_template => false
      index => "test-log-%{+YYYY.MM.dd}"
      document_type => "%{[@metadata][type]}"
    }
  }
  else if "prod-log" in [tags] {
    elasticsearch {
      hosts => ["localhost"]
      manage_template => false
      index => "prod-log-%{+YYYY.MM.dd}"
      document_type => "%{[@metadata][type]}"
    }
  }
}

到此，elk服务准备好了，下一步使用filebeat搜集日志。

安装filebeat

下面需要在日志所在的机子上安装filebeat。

安装方法可以参照官网： https://www.elastic.co/guide/en/beats/filebeat/5.4/filebeat-installation.html

因为这里是centos 7系统，我的安装步骤是：

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-5.4.3-x86_64.rpm

sudo rpm -vi filebeat-5.4.3-x86_64.rpm

安装好后，默认的配置文件目录为：/etc/filebeat

配置文件为 filebeat.yml ，我这里做了一个备份。把原来的 filebeat.yml重命名为了filebeat.yml.bak.

然后新建了一个filebeat.yml，内容如下：

filebeat:
  prospectors:
    -
      paths:
        - /var/log/laijiandu-test/*.log
      tags: ["test-log"]
      input_type: log
      document_type: test-log
      scan_frequency: 2s
      harvester_buffer_size: 16384
      max_bytes: 10485760
  registry_file: /var/log/registry
output:
  logstash:
    hosts: ["logstashserverip:5044"]
    worker: 1
    bulk_max_size: 2048
    compression_level: 3
    index: filebeat
shipper:
logging:
  files:
    rotateeverybytes: 10485760

注意：上面配置中的tags需要和logstash中的对应起来。

启动filebeat:
不同系统的启动方法有所不同，具体可以参照官网： https://www.elastic.co/guide/en/beats/filebeat/5.4/filebeat-starting.html

我这里执行以下命令：

启动：
sudo /etc/init.d/filebeat start 

停止：
sudo /etc/init.d/filebeat stop

以上步骤搞定后，就可以爽歪歪的使用 kibana了。