介绍

ELK是当前比较主流的分布式日志收集处理工具。
常用日志采集方式： Filebeat→Kafka集群→Logstash→ES→Kibana
Grafana（可视化监控工具）上配置连接ES数据库 进行实时监控

实施步骤：
Filebeat部署在应用服务器上（只负责Logstash的读取和转发，降低CPU负载消耗，确保不会抢占应用资源），Logstash、ES、Kibana在一台服务器上（此处的Logstash负责日志的过滤，会消耗一定的CPU负载，可以考虑如何优化过滤的语法步骤来达到降低负载）。

架构图：

filebeat

docker 部署filebeat 直接到 es

注意 filebeat版本一定要和es 版本一致
logback.xml里日志格式为json

docker run --privileged --name filebeat --net=host -d -m 1000M \
      --log-driver json-file --log-opt max-size=1024m \
      -v /data0/filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml \
      -v /data0/filebeat/logs:/root \
      -v /data0/filebeat/data:/data \
      -v /data0:/home/logs \
      registry.api.ww.com/bop_ci/filebeat:6.6.0
1
2
3
4
5
6
7

vi filebeat.yml

filebeat.inputs:
- type: log
  eabled: true
  paths:
    - /home/logs/bop-fms-query-info/logs/*.log
  fields:
    index: "fms-query-info"
- type: log
  eabled: true
  paths:
    - /home/logs/bop-fms-advertiser-web/logs/*.log
  fields:
    index: "fms-advertiser-web"

json.keys_under_root: true
json.overwrite_keys: true


setup.ilm.enabled: false
setup.template.name: "bop-log"
setup.template.pattern: "bop-log-*"
setup.template.enabled: true
setup.template.overwrite: false
setup.template.settings:
  index.number_of_shards: 1
  index.number_of_replicas: 0
  index.codec: best_compression


output.elasticsearch:
  hosts: ["10.13.177.206:9201"]
  index: "bop-log-%{+yyyy.MM.dd}"

processors:
  - decode_json_fields:
      fields: ["message"]
      target: ""
      overwrite_keys: true
  - rename:
      fields:
        - from: "error"
          to: "run_error"
  - drop_fields:
      fields: ["input_type", "log.offset", "host.name", "input.type", "agent.hostname"]
      #ignore_missing: false

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

到 es 示例2：

# 输出到es
output.elasticsearch:
  #username: "elastic"
  #password: "xxxxxxxxxxx"
  #worker: 1
  #bulk_max_size: 1500
  #pipeline: "timestamp-pipeline-id" #@timestamp处理
  hosts: ["elasticsearch1:9200"]
  index: "pb-%{[fields.index_name]}-*"
  indices:
    - index: "pb-nginx-%{+yyyy.MM.dd}"
      when.equals:
        fields.index_name: "nginx_log"
    - index: "pb-log4j-%{+yyyy.MM.dd}"
      when.equals:
        fields.index_name: "log4j_log"
    - index: "pb-biz-%{+yyyy.MM.dd}"
      when.equals:
        fields.index_name: "biz_log"
  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

filebeat @timestamp日期处理

在 Kibana 中的 Devtools 界面中编写如下 pipeline 并执行
查询 GET _ingest/pipeline/timestamp-pipeline-id

PUT _ingest/pipeline/timestamp-pipeline-id
{
  "description": "timestamp-pipeline-id",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{TIMESTAMP_ISO8601:timestamp}"
        ],
        "ignore_failure": true
      },
      "date": {
        "field": "timestamp",
        "timezone": "Asia/Shanghai",
        "formats": [
          "yyyy-MM-dd HH:mm:ss.SSS"
        ],
        "ignore_failure": true
      }
    }
  ]
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

使用filebeat采集文件到es
docker部署filebeat
filebeat采集多个日志（推送给ES或者logstash）
filebeat采集日志到ES
docker部署filebeat

filebeat7.7.0相关详细配置预览- processors

logstash

docker 部署logstash

1

vi logstash.yml

1

kibana

docker 部署kibana
注意，kibana要和es的版本一致，否则版本不兼容

cd /data0
mkdir kibana
cd kibana

docker run --name kibana  -p 5601:5601 -d kibana:6.6.0
docker cp kibana:/usr/share/kibana/config/kibana.yml .
1
2
3
4
5
6

将如下内容写到kibana.yml中，然后保存退出：:wq

server.name: kibana
server.host: "0"
#elasticsearch.hosts: [ "http://elasticsearch:9200" ]
elasticsearch.hosts: [ "http://自己的elasticsearch的IP:9200" ]
xpack.monitoring.ui.container.elasticsearch.enabled: true
#设置kibana中文显示
i18n.locale: zh-CN
1
2
3
4
5
6
7

重新启动

docker rm -f kibana
docker run --name kibana -p 5601:5601 -d -v /data0/kibana/kibana.yml:/usr/share/kibana/config/kibana.yml kibana:6.6.0
1
2

kibana自动关联es索引

vi auto_add_index.sh

#!/bin/bash
today=`date +%Y.%m.%d`
yestoday=`date -d "1 days ago" +%Y-%m-%d`
pattern='bop-log-'${today}
old_pattern='bop-log-'${yestoday}
index='bop-log-'${today}
echo ${pattern} ${old_pattern}

#新增
curl -f -XPOST -H 'Content-Type: application/json' -H 'kbn-xsrf: anything' \
     "http://localhost:5601/api/saved_objects/index-pattern/${pattern}" -d"{\"attributes\":{\"title\":\"${index}\",\"timeFieldName\":\"@timestamp\"}}"

#设置默认索引
curl -f -XPOST -H 'Content-Type: application/json' -H 'kbn-xsrf: anything' http://localhost:5601/api/kibana/settings/defaultIndex -d "{\"value\":\"${pattern}\"}"

#删除
curl -XDELETE "http://localhost:5601/api/saved_objects/index-pattern/${old_pattern}" -H 'kbn-xsrf: true'

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

定时删除索引

kibana DevTools执行
30天后删除

PUT _ilm/policy/logs_policy
{
  "policy": {
    "phases": {
      
      "delete": {
        "min_age": "30d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

kibana 自动创建索引模式
Kibana自动关联ES索引
ELK系列(2) - Kibana怎么修改日期格式Date format

参考文档：
使用Docker搭建ELK日志系统

Kibana 可视化日志分析

docker 搭个日志收集系统

Springboot+logback集成ELK处理日志实例

Kibana + Elasticsearch + Logstash + Filebeat

Spring Cloud 分布式实时日志分析采集三种方案

10分钟部署一个ELK日志采集系统
如何快速采集分析平台日志，并进行展示监控？
手把手教你搭建ELK
filebeat上报数据异常排查
filebeat占用文件句柄磁盘满
filebeat常见问题