由于我们使用的服务器比较多，最近每次查看日志都是ssh登陆或者通过Fabric下载下来，特别不方便。经过多方比较，最终选择了ELK。所以，开始了学习ELK的路程。

本来，是打算写一篇完整教程，后来想了想，发现完整教程中的一些内容并不适合展现出来。但，在学习过程中的内容，感觉还是需要做个笔记。所以，暂时整一篇笔记，记录在学习实践过程中所有知识点，后面在将整个日志系统搭建好整一篇完整的搭建教程。回到正题，咱们开始学习ELK。

一、简单介绍ELK

ELK是Elasticsearch、Logstash、Kibana三个开源软件的首字母缩写，在搭建ELK日志系统时，除了前面三个，还有用到了Filebeat，其中它们的功能作用如下：

• Elasticsearch：是一个搜索和分析引擎
• Logstash：是一个服务端的数据预处理管道，它可以同时收集、转换多个数据源的数据，并且发送给Elasticsearch
• Kibana：是一个Web平台，将Elasticsearch的数据用表格和图片的形式展现出来，以供分析
• Filebeat: 是一个日志文件托运工具，在服务器上安装之后，Filebeat会监控日志目录或者指定的日志文件，追踪读取这些文件，将收集到的日志数据发送给Logstash。（除了Logstash，它也可以发送给Elasticsearch、Kafka、Redis等）

二、ELK架构

ELK架构分成两种，一种是经典的ELK，另外一种是加上消息队列（Redis或Kafka或RabbitMQ）和Nginx的架构。

经典的ELK

经典的ELK主要是由Filebeat + Logstash + Elasticsearch + Kibana组成，如下图：

image.png

它主要适用于数据量小的开发环境，存在丢失数据的危险。

消息队列+Nginx架构

这种架构，主要加上了Redis或Kafka或RabbitMQ做消息队列，保证了消息的不丢失。而加上Nginx，则是可以增加一层访问限制（因为Kibana本身没有限制，所以放在Nginx做），如下图：

image.png

此种架构，主要用在生成环境，可以处理大数据量，并且不会丢失数据。

三、安装ELK

安装Java 8

由于Elasticsearch、Logstash是基于Java 8开发的，所以需要在安装Elasticsearch、Logstash的服务器上安装Java 8。

$ apt-get update
$ apt-get install default-jre
$ java -version
openjdk version "1.8.0_171"
OpenJDK Runtime Environment (build 1.8.0_171-8u171-b11-0ubuntu0.16.04.1-b11)
OpenJDK 64-Bit Server VM (build 25.171-b11, mixed mode)

APT安装ELK源

我这里使用APT安装的，所以需要先更新ELK源，然后就可以apt-get install安装ELK了。需要注意的是，若是你的Logstash和Filebeat在不同的服务器上，则ELK源需要在每台服务器上安装一遍。当然，若是你习惯另外方式，可以参考官网Installation进行选择。

首先，下载和安装公钥

wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

然后，安装apt-transport-https包

sudo apt-get install apt-transport-https

之后，将ELK的安装包源保存到/etc/apt/sources.list.d/elastic-6.x.list

echo "deb https://artifacts.elastic.co/packages/6.x/apt stable main" | sudo tee -a /etc/apt/sources.list.d/elastic-6.x.list

最后，更新下apt

sudo apt-get update

安装Elasticsearch

我们上面已经安装过了ELK源，所以可以直接使用apt-get install来安装ELK这些软件。

1、安装Elasticsearch

sudo apt-get install elasticsearch

2、启动elasticsearch，并验证是否运行

启动elasticsearch

service elasticsearch start

等一会儿，然后输入curl http://localhost:9200进行验证

返回下面内容，则说明运行成功:

{
  "name" : "qE-LiYD",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "pbVqoWBjR4ybbPXprKr5WQ",
  "version" : {
    "number" : "6.2.4",
    "build_hash" : "ccec39f",
    "build_date" : "2018-04-12T20:37:28.497551Z",
    "build_snapshot" : false,
    "lucene_version" : "7.2.1",
    "minimum_wire_compatibility_version" : "5.6.0",
    "minimum_index_compatibility_version" : "5.0.0"
  },
  "tagline" : "You Know, for Search"
}

遇到问题：刚搭建时，我在多台服务器上实验过，输入curl http://localhost:9200获取不到任何内容，启动也没有报错信息。后来，在官方的Install Elasticsearch with Debian Package文档中，查看日志路径，通过日志才发现是内存不足。最后，换到一台4G内存的服务器上，就没有问题了。也可以通过/usr/share/elasticsearch/bin/elasticsearch命令启动，当报错时，能在终端显示错误信息。

如果要停止或重启elasticsearch，可以使用下面命令：

# 停止
service elasticsearch stop
# 重启
service elasticsearch restart

安装Logstash

已经安装好ELK源，所以只需要一句命令安装：

sudo apt-get install logstash

安装Kibana

Kibana是JS写的，若是Kibana和Logstash、Elasticsearch在不同服务器上，可以不安装Java 8。

# 安装Kibana
sudo apt-get install kibana

Kibana的配置文件在/etc/kibana/kibana.yml中，它默认打开的端口是5601，并默认接收本地9200的elasticsearch数据。对应的配置如下：

server.port: 5601
elasticsearch.url:  "[http://localhost:9200](http://localhost:9200/)"

启动Kibana，使用如下命令：

sudo service kibana start

此时，若是将Kibana安装本地，就可以通过 http://localhost:5601看到如下页面：

image.png

安装Filebeat

首先安装好ELK源，然后执行下面命令安装Filebeat

sudo apt-get install filebeat

四、配置

在上面安装，Elasticsearch和Kibana使用默认配置，重点的配置主要是Filebeat、Logstash。

Filebeat

Filebeat主要是追踪日志数据，然后将数据传输给Logstash或Redis（我这里选择Redis做消息队列）。

首先，需要配置追踪和上传的日志文件路径，如下：

filebeat.prospectors:
- type: log
  # 需要将开关打开
  enabled: true
  paths:
    # 追踪获取/root/test.log中的内容
    - /root/test.log

当将日志数据发送给Logstash时，如下设置：

output.logstash:
  # 设置Logstash的hosts，若是有多个，可以设置多个
  hosts: ["localhost:5044"]

当将日志数据发送给Redis时，如下设置：

output.redis:
  hosts: ["172.25.19.34:6379"] # host，主要需要带上端口号
  key: "test_log" #存储在Redis中的队列key，可以通过登录Redis查看数据
  password: "123456" # Redis的密码
  db: 0 # Redis的数据库
  timeout: 5 # Redis的超时时间

注意：当Filebeat输出到redis时，可以登录数据，通过llen查看日志条数或通过lrange查看日志具体数据。

启动Filebeat，使用如下命令：

/usr/share/filebeat/bin/filebeat -e -c /etc/filebeat/filebeat.yml -d "publish"

当如果已经追踪过了，还想重新测试，可以删除以前的追踪记录，命令如下：

rm /usr/share/filebeat/bin/data/registry

Logstash

Logstash主要是收集多个数据源的数据，然后预处理一下，最后丢给Elasticsearch做分析。整个过程，Logstash就像一个管道一样，如下图：

image.png

其中，在Logstash中包括三部分内容：

• inputs: 输入
• filters: 过滤
• outputs: 输出

上面三部分内容，也是我们需要进行配置的。

首先，我们来体验一下最简单的配置，也就是标准输入标准输出。

配置在first.conf文件内，配置内容如下：

input {
        stdin {}
}
output {
        stdout {}
}

运行配置文件，命令如下：

/usr/share/logstash/bin/logstash -f first.conf 

输入test，然后输出了一个json格式数据，结果如下：

[INFO ] 2018-06-04 01:25:36.217 [Ruby-0-Thread-1: /usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/stud-0.0.23/lib/stud/task.rb:22] agent - Pipelines running {:count=>1, :pipelines=>["main"]}
test
{
      "@version" => "1",
    "@timestamp" => 2018-06-03T17:25:51.126Z,
          "host" => "0.0.0.0",
       "message" => "test"
}

然后，我们让Logstash接收Filebeat直接传送过来的数据

Filebeat中的配置如下：

filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /root/test.log
output.logstash:
  hosts: ["localhost:5044"]

启动filebeat

/usr/share/filebeat/bin/filebeat -e -c /etc/filebeat/filebeat.yml -d "publish"

配置在first-pipeline.conf文件中，主要监听5044端口的数据，配置内容如下：

input {
        beats {
                # 监听5044
                port => "5044"
        }
}
output {
        stdout {}
}

测试logstash的配置文件，然后启动logstash

# 测试
/usr/share/logstash/bin/logstash -f first-pipeline.conf --config.test_and_exit
# 启动，当配置修改时，会自动刷新，不用重新启动
/usr/share/logstash/bin/logstash -f first-pipeline.conf --config.reload.automatic

此时，Filebeat中获取的日志文件，会传送到Logstash显示在终端。注意，防止看不到效果，先启动Logstash，然后启动Filebeat。

最后，我们再看看使用Redis时，是如何存取数据的

在Filebeat端，配置如下：

filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /root/test.log
output.redis:
  hosts: ["172.25.19.34:6379"] # host，主要需要带上端口号
  key: "test_log" #存储在Redis中的队列key，可以通过登录Redis查看数据
  password: "123456" # Redis的密码
  db: 0 # Redis的数据库
  timeout: 5 # Redis的超时时间

启动Filebeat

/usr/share/filebeat/bin/filebeat -e -c /etc/filebeat/filebeat.yml -d "publish"

在Logstash端，配置如下：

input {
        redis {
                port => "6379"
                host => "172.25.19.34"
                data_type => "list" # 数据类型为队列
                type => "log"
                key => "test_log" # 存储Redis的key
                password => "123456"
                db => 0
                timeout => 5
        }
}
output {
        stdout {
                codec => rubydebug
        }
}

这样就能够使用Filebeat将数据存入Redis，而另外一端Logstash从Redis中获取数据。

将数据输出到Elasticsearch
前面，我们只是将数据输出到标准输出来显示数据，生产环境中我们需要把数据输出到Elasticsearch，配置如下：

output {
        elasticsearch {
                # localhost对应安装elasticsearch的服务器
                hosts => ["localhost:9200"]
        }
}

还有，Logstash过滤器
前面，我们用到了Logstash的input和output，并没有用到filter（过滤器）。过滤器中，会用到许多插件，暂时我就用过grok，配置如下：

filter {
        grok {
                match => {"message" => "%{IP:client_ip} %{WORD:method} %{URIPATHPARAM:request} %{GREEDYDATA:content}"}
        }
}

当在Logstash中加入上面filter配置时，它就会进行匹配，上面匹配的是: ip 方法 请求参数 请求内容。
例如：

55.3.244.1 GET /index.html 测试11
test GET /index.html 测试22

第一句能匹配，第二句就不能匹配。

上面配置文件中的IP、WORD、URIPATHPARAM、GREEDYDATA是内置支持的正则表达式别名，具体支持的正则表达式可查看logstash-patterns-core。

而若是匹配之后，就会在输出内容，添加client_ip、method、request和content。等于就是通过正则表达式非结构化数据转化成结构化数据。

上面两句话，被Filebeat抓取之后，在Logstash的终端展示内容：

{
      "@version" => "1",
       "content" => "测试11",
        "offset" => 36,
       "message" => "55.3.244.1 GET /index.html 测试11",
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
    "@timestamp" => 2018-06-03T18:10:03.376Z,
       "request" => "/index.html",
          "beat" => {
        "hostname" => "iZ0xi4n3ya19j63huy3c78Z",
         "version" => "6.2.4",
            "name" => "test"
    },
        "source" => "/root/elk_conf/test.txt",
    "prospector" => {
        "type" => "log"
    },
          "host" => "iZ0xi4n3ya19j63huy3c78Z",
        "method" => "GET",
     "client_ip" => "55.3.244.1"
}
{
    "@timestamp" => 2018-06-03T18:10:03.376Z,
        "source" => "/root/elk_conf/test.txt",
      "@version" => "1",
          "beat" => {
        "hostname" => "iZ0xi4n3ya19j63huy3c78Z",
         "version" => "6.2.4",
            "name" => "test"
    },
    "prospector" => {
        "type" => "log"
    },
          "host" => "iZ0xi4n3ya19j63huy3c78Z",
        "offset" => 66,
       "message" => "test GET /index.html 测试22",
          "tags" => [
        [0] "beats_input_codec_plain_applied",
        [1] "_grokparsefailure"
    ]
}

可看到第一句话匹配了，添加了对应的字段。而第二句话没有匹配，tags中多了_grokparsefailure。

而最终，在Kibana上显示的结果如下：

image.png

参考

Installing ELK
Logstash安装搭建（一）
A Filebeat Tutorial: Getting Started
Parsing Logs with Logstash
Filebeat，Redis和ELK6.x集中式日志解决方案
logstash-patterns-core