docker 搭建 EFK（ElasticSearch+Fluentd+Kibana）日志集中收集系统

EFK 日志采集系统

简介

EFK = ElasticSearch + Fluentd + Kibana

Elasticsearch 是一个分布式的搜索和分析引擎，可以用于全文检索、结构化检索和分析，并能将这三者结合起来。

Fluentd 是一个优秀的 log 信息收集开源免费软件。

Kibana 是一个开源的分析与可视化平台，可以用 kibana 搜索、查看存放在 Elasticsearch 中的索引。

版本

当前最新的 Release 版本

Elasticsearch : v6.4.2 – 2018/10/02

Fluentd : v1.2.6 – 2018/10/03

Kibana : v6.4.2 – 2018/10/02

主要介绍 Fluentd

Fluentd 是一个完全开源免费的日志收集系统。它主要使用 c 语言编写并由轻量级的 Ruby 封装，给用户提供灵活的扩展性，因此它目前有 125 以上插件。结合不同插件可实现：异常报警（alerting）、数据分析（analysis）、归档（archiving）等等功能。

特点：

1）安装方便

2）占用空间小

3）半结构化数据日志记录

4）灵活的插件机制

5）可靠的缓冲

6）日志转发

使用前

使用后

使用场景

1.apache 日志到 mongodb

步骤：

1.轮询监控 access_log 文件。

2.将传入的日志记录解析为有意义的字段（例如：ip, path,user_agent 等）并缓冲他们。

3.定时将缓冲区内容写入 MongoDB。

2.程序推送

配置 fluent.conf

<source>
  @type forward
  port 24224
</source>
<match fluentd.test.**>
  @type stdout
</match>

JAVA

import java.util.HashMap;
import java.util.Map;
import org.fluentd.logger.FluentLogger;

public class Main {
    private static FluentLogger LOG = FluentLogger.getLogger("fluentd.test");

    public void doApplicationLogic() {
        // ...
        Map<String, Object> data = new HashMap<String, Object>();
        data.put("from", "userA");
        data.put("to", "userB");
        LOG.log("follow", data);
        // ...
    }
}

PHP

<?php
require_once __DIR__.'/vendor/autoload.php';
use Fluent\Logger\FluentLogger;
$logger = new FluentLogger("localhost","24224");
$logger->post("fluentd.test.follow", array("from"=>"userA", "to"=>"userB"));

3.docker 日志

fluent.conf

<source>
    @type forward
    port 24224
    bind 0.0.0.0
</source>

# docker log
<match docker.**>
  @type elasticsearch
  host elasticsearch
  port 9200
  type_name docker
  logstash_format true
  logstash_prefix docker-logs
  logstash_dateformat %Y-%m-%d
  flush_interval 15s
  include_tag_key true
</match>

docker-compose.yml 加入如下配置信息

    logging:
      driver: fluentd
      options:
        fluentd-address: "192.168.5.47:24224"
        tag: "docker.web.cas"

4.多节点集中收集

“日志聚合器” 是继续从日志转发器接收事件的守护程序。 他们缓冲事件并定期将数据上传到云中。

5.日志可视化

该图为上文所提到的 EFK 日志系统架构

插件

Fluentd 存在 7 种类型的插件：

Input 输入，从外部源检索和提取事件日志。

Parser 解析器,允许用户创建自定义的解析器。

Filter 过滤器，用于过滤输入内容（例如: 隐私信息）

Output 输出

Formatter 格式化器

Storage 存储引擎

Buffer 缓冲区

以下主要介绍 Input Filter Output 这 3 种插件

输入（Input）

Fluentd 支持多种的输入类型，支持 udp tcp http 等数据包的监听。既支持网络协议传输的数据流，也支持从文件中读取数据。

监听 tcp 数据包例子

配置

<source>
  @type tcp
  tag tcp.events # required
  <parse>
    @type regexp
    expression /^(?<field1>\d+):(?<field2>\w+)$/
  </parse>
  port 20001   # optional. 5170 by default
  bind 0.0.0.0 # optional. 0.0.0.0 by default
  delimiter \n # optional. \n (newline) by default
</source>

发送数据包

$ echo '123456:awesome' | netcat 0.0.0.0 5170

fluentd 解析

{"field1":"123456","field2":"awesome}

过滤（Filter）

过滤插件使 Fluentd 可以修改事件流。

示例用例：

通过删除一个或多个字段的值来过滤事件
通过添加新字段丰富事件 例如：给某个符合的记录，添加一个字段标记
删除或屏蔽某些字段，以确保隐私权和合规性。

Event processing pipeling（事件处理流）

filter 指令具有与 match 相同的语法，但是 filter 可以串联成 pipeline，对数据进行串行处理，最终再交给 match 输出。 使用 fliters，事件流如下：

Input -> filter 1 -> ... -> filter N -> Output（Match tag）

例 1：

配置：

<source>
  @type http
  port 8888
  bind 0.0.0.0
</source>

<filter test.cycle>
  @type grep
  <exclude>
    key action
    pattern ^logout$  # 过滤掉 action 参数为 logout 的数据
  </exclude>
</filter>

<match test.cycle>
  @type stdout
</match>

测试：

$ curl -i -X POST -d 'json={"action":"login","user":2}' http://localhost:8888/test.cycle
HTTP/1.1 200 OK
Content-type: text/plain
Connection: Keep-Alive
Content-length: 0

$ curl -i -X POST -d 'json={"action":"logout","user":2}' http://localhost:8888/test.cycle
HTTP/1.1 200 OK
Content-type: text/plain
Connection: Keep-Alive
Content-length: 0

输出：

2018-10-15 12:37:41 -0600 [info]: adding filter pattern="test.cycle" type="grep"
2018-10-15 12:37:41 -0600 [info]: adding match pattern="test.cycle" type="stdout"
2018-10-15 12:37:41 -0600 [info]: adding source type="http"
2018-10-25 12:37:42 -0600 test.cycle: {"action":"login","user":2}

上面的指令用标签 foo 匹配事件。如果 action 字段的值包含 logout，则事件将过滤该记录。

例 2：

### http://this.host:9880/myapp.access?json={"event":"data"}
<source>
  @type http
  port 9880
</source>

# filter 1
<filter foo.bar>
  @type grep
  regexp1 message cool
</filter>

# filter 2
<filter foo.bar>
  @type record_transformer
  <record>
    hostname "#{Socket.gethostname}"
  </record>
</filter>

只有其 message 字段包含 cool 的事件才会添加新的 hostname 字段，并将机器的主机名作为其值。

如果有多个 filter 标签，命中上一个才会继续往下匹配。

这个例子里，filter 获取数据后，调用原生的 @type record_transformer 插件，在事件的 record 里插入了新的字段 host_param，然后再交给 match 输出。

filter 匹配顺序与 match 相同，应该在 <match> 之前放置 <filter>。

输出（Output）

存在 3 种输出模式：

• Non-Buffered mode（不缓冲）: 不缓存数据并立即写出结果。
• Synchronous Buffered mode（同步缓冲模式）: 具有“分段”缓冲块(块是事件的集合)和块的队列，其行为可以由 <buffer> 部分控制(见下图)。
• Asynchronous Buffered mode（异步步缓冲模式）: 也存在 stage 和 queue，但是 output plugin 不会同步提交写块，而是在以后提交。

输出插件可以支持所有模式，但只支持其中一种模式。如果配置中没有的部分，Fluentd 会自动选择合适的模式。

总结

Input 负责接收数据或者主动抓取数据。支持 syslog，http，file tail 等。

Buffer 负责数据获取的性能和可靠性，也有文件或内存等不同类型的 Buffer 可以配置。

Output 负责输出数据到目的地例如文件，AWS S3 或者其它的 Fluentd。

Docker 环境下搭建 EFK

docker 运行容器

xp-room@experience-room:~$ docker ps
CONTAINER ID        IMAGE                 COMMAND                  CREATED             STATUS              PORTS                                                          NAMES
b45c17e65a49        kibana:6.4.2          "/usr/local/bin/ki..."   4 days ago          Up 4 days           0.0.0.0:5601->5601/tcp                                         efk_kibana_1
58ba0b6ec3a7        efk_fluentd           "/bin/entrypoint.s..."   4 days ago          Up 4 days           5140/tcp, 0.0.0.0:24224->24224/tcp, 0.0.0.0:24224->24224/udp   58ba0b6ec3a7_efk_fluentd_1
ceb349479fdf        elasticsearch:6.4.2   "/usr/local/bin/do..."   4 days ago          Up 4 days           0.0.0.0:9200->9200/tcp, 9300/tcp                               efk_elasticsearch_1

Server

Tree

server
├── docker-compose.yml
├── elasticsearch
│   └── elasticsearch.yml
├── fluentd
│   ├── conf
│   │   └── fluent.conf
│   └── Dockerfile
└── kibana
    └── kibana.yml

docker-compose.yml

version: '2'
services:
  fluentd:
    privileged: true
    build: ./fluentd
    volumes:
      - ./fluentd/conf:/fluentd/etc
      - /var/log/nginx/access-kod-explorer.log:/var/logs/nginx/access-kod-explorer.log
      - /var/log/fluentd:/var/lib/fluentd
    links:
      - "elasticsearch"
    ports:
      - "24224:24224"
      - "24224:24224/udp"

  elasticsearch:
    image: elasticsearch:6.4.2
    expose:
      - 9200
    environment:
      - cluster.name=docker-cluster
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
      - TZ=Asia/Shanghai
    ulimits:
      memlock:
        soft: -1
        hard: -1
    ports:
      - "9200:9200"
    volumes:
      - ./elasticsearch/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml

  kibana:
    image: kibana:6.4.2
    links:
      - "elasticsearch"
    ports:
      - "5601:5601"
    volumes:
      - ./kibana/kibana.yml:/usr/share/kibana/config/kibana.yml
    environment:
      - TZ=Asia/Shanghai

elasticsearch

elasticsearch.yml

cluster.name: "docker-cluster"
network.host: 0.0.0.0

# minimum_master_nodes need to be explicitly set when bound on a public IP
# set to 1 to allow single node clusters
# Details: https://github.com/elastic/elasticsearch/pull/17288
discovery.zen.minimum_master_nodes: 1

fluentd

Dockerfile

# fluentd/Dockerfile
FROM fluent/fluentd:v1.2-debian
RUN cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
RUN ["gem", "install", "fluent-plugin-elasticsearch", "--no-rdoc", "--no-ri"]

conf

fluent.conf

# fluentd/conf/fluent.conf
# Server

# in_forward
<source>
    @type forward
    port 24224
    bind 0.0.0.0
#    <security>
#      self_hostname experience-room
#      shared_key room2017
#    </security>
</source>

# fhy scenic access log
<match apache.access>
  @type elasticsearch
  host elasticsearch
  port 9200
  logstash_prefix fhyscenic
  logstash_format true
  logstash_dateformat %Y%m%d
  type_name scenic_access_log #(optional; default=fluentd)
  flush_interval 10s
</match>

# docker log
<match docker.**>
  @type elasticsearch
  host elasticsearch
  port 9200
  type_name docker
  logstash_format true
  logstash_prefix docker-logs
  logstash_dateformat %Y-%m-%d
  flush_interval 15s
  include_tag_key true
</match>

kibana

kibana.yml

---
# Default Kibana configuration from kibana-docker.

server.name: kibana
server.host: "0"
elasticsearch.url: http://elasticsearch:9200
xpack.monitoring.ui.container.elasticsearch.enabled: true

Client

Tree

client
├── docker-compose.yml
└── fluentd
    ├── conf
    │   └── fluent.conf
    └── Dockerfile

docker-compose.yml

version: '2'
services:
  fluentd:
    build: ./fluentd
    volumes:
      - ./fluentd/conf:/fluentd/etc
      - /usr/local/apache2/logs:/var/logs/httpd:rw
      - /disk4/log/fluentd:/var/lib/fluentd:rw
    # links:
      # - "elasticsearch"
    ports:
      - "24224:24224"
      - "24224:24224/udp"
    # logging:
    #   driver: fluentd
    #   options:
    #     fluentd-address: "192.168.5.47:24224"
    #     tag: "docker.web"

fluentd

Dockerfile

# fluentd/Dockerfile
FROM fluent/fluentd:v1.2-debian
RUN cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
RUN ["gem", "install", "fluent-plugin-elasticsearch", "--no-rdoc", "--no-ri"]

conf

fluent.conf

# fluentd/conf/fluent.conf
# Client

# scenic access log
<source>
  @type tail
  path /var/logs/httpd/jingqv-access_log # 需要收集的日志文件
  pos_file /var/lib/fluentd/jingqv-access_log.pos # 记录收集目录
  tag apache.access # 定义标签
  format apache2 # 日志类型 可支持正则表达式
  read_from_head true # 从文件头部读取日志文件
</source>

# out_forward
<match apache.access>
  @type forward
  compress gzip
  # 将日志记录输出到远程服务器
  <server>
    name Centos7 # client server name
    host 192.168.5.47 # server fluentd
    port 24224
  </server>
  # <security>
  #   self_hostname Centos7-110
  #   shared_key room2017
  # </security>
  flush_interval 5s
  buffer_chunk_limit 8m
  buffer_queue_limit 1000
  num_threads 4
</match>

参考

https://docs.fluentd.org/v1.0/articles/in_forward

https://docs.fluentd.org/v1.0/articles/out_forward

https://docs.fluentd.org