基于elasticsearch + logstash + kafka + filebeat的日志收集及近实时处理的集群介绍及搭建，基于rancher1.6平台。

elasticsearch + logstash + kafka + filebeat

简介

graph TD
A[logFile] -->|filebeat| B(kafka)
B --> C{logstash}
C -->|index1| D
C -->|index2| D
C -->|index3| D[Elasticsearch]

Elasticsearch是一个基于Lucene的搜索引擎。它提供了一个分布式，多租户 -能够全文搜索与发动机HTTP Web界面和无架构JSON文件。

Elasticsearch与名为Logstash的数据收集和日志解析引擎以及名为Kibana的分析和可视化平台一起开发。这三种产品设计用作集成解决方案，称为**“弹性堆栈”**（以前称为“ELK堆栈”）。

Elasticsearch可用于搜索各种文档。它提供可扩展的搜索，近实时搜索，并支持多租户。[2] “Elasticsearch是分布式的，这意味着索引可以分为多个分片，每个分片可以有零个或多个副本。每个节点托管一个或多个分片，并充当协调器，将操作委托给正确的分片。重新平衡和路由是自动完成的“。[2]相关数据通常存储在同一索引中，该索引由一个或多个主分片和零个或多个副本分片组成。创建索引后，无法更改主分片的数量

Elasticsearch

1
2


cluster status:
http://172.18.4.42:9200/_cluster/health

配置详解

data node

如果你想让节点从不选举为主节点,只用来存储数据,可作为负载器
node.master: false
node.data: true

master node

如果想让节点成为主节点,且不存储任何数据,并保有空闲资源,可作为协调器
node.master: true
node.data: false

client node

如果想让节点既不称为主节点,又不成为数据节点,那么可将他作为搜索器,从节点中获取数据,生成搜索结果等
node.master: false
node.data: false

filebeat

从文件读取日志，通过tag进行标记，用于区分不同的日志，最后输出到kafka。

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


#输入
- type: log
  paths:
    - /var/log/amp/*.log
  tags: ["amp-log"]
  fields_under_root: true
  document_type: amp-log
  ignore_older: 24h
#输出
output.kafka:
  hosts: ["kafka-broker-1:9092","kafka-broker-2:9092","kafka-broker-3:9092"]
  topic: logstash
  required_acks: 1

kafka

用于临时存储filebeat收集来的日志。

key	value
ADVERTISE_PUB_IP	true
JVMFLAGS	-Xmx2048m -Xms2048m
KAFKA_REPLICATION_FACTOR（副本数量）	3
KAFKA_LOG_RETENTION_HOURS（日志过期时间）	24
KAFKA_NUM_PARTITIONS（topic分区数量）	10
ZK_SERVICE	ZK_SERVICE

logstash

从kafka读取日志，使用正则过去日志信息，之后结构化日志，存入elasticsearch。

graph LR
A[INPUT] -->|kafka| B(filter)
B -->|tags| C{output}
C -->|index1| D
C -->|index2| D[Elasticsearch]

 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
47
48
49
50
51
52
53
54


# 从kafka读取日志
input {
  # beats {
  #    port => "5044"
  # }
  kafka {
    bootstrap_servers => "kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092"
    topics => "logstash"
    group_id => "logstash-group-01"
    consumer_threads => 5
    decorate_events => false
    auto_offset_reset => "latest" # 从最新的偏移量开始消费
    codec => "json"
  }
# 结构化日志并清洗日志
filter {
  if "amp-log" in [tags] {
     grok {
       match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{DATA:threadname}\] %{WORD:level}\s+%{DATA:topic}:%{GREEDYDATA:jsondata}" }
    }
     json {
       source => "jsondata"
       #target => "parsedJson"
       remove_field=>["jsondata"]
    }
     date {
       match => ["timestamp","YYYY-MM-dd HH:mm:ss.SSS"]
       target => [ "@timestamp" ]
    }
     ruby {
       code=>"event.set('index_day', event.get('@timestamp').time.localtime.strftime('%Y.%m.%d'))"
    }
     mutate {
       remove_field => ["timestamp"]
       remove_field => ["message"]
       remove_field => ["threadname"]
   }
     if "success" in [topic] {
      mutate { add_tag => [ "monitor-success" ] }
   }
     else if "repeat" in [topic] {
      mutate { add_tag => [ "monitor-repeat" ] }
   }
}
}
# 输出到elasticsearch
output {
   if "monitor-success" in [tags] {
        elasticsearch {
          hosts => "elasticsearch:9200"
          index => "amp-success-%{index_day}"
        }
    }
}

kibana

用于查询elasticsearch的数据，监控elasticsearch的监控状态

other

配置

1
2
3
4
5
6
7
8
9


// 索引写入缓存
indices.memory.index_buffer_size=20% 
indices.memory.min_index_buffer_size=96m
// 存活超时时间
discovery.zen.fd.ping_timeout=120s
// 存活超时重试次数
discovery.zen.fd.ping_retries=6
// 节点间存活检测间隔
discovery.zen.fd.ping_interval=30s

索引模板配置

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


// 刷新间隔，merge线程数
PUT /_template/adv_logs 
{
  "index_patterns": ["aap-*", "amp-*"], 
  "order":    1, 
  "settings": {
    "refresh_interval": "60s",
    "merge": {
        "scheduler": {
            "max_thread_count": "1"
        }
    }	
  }
}

kafka

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# alter partition
bin/kafka-topics.sh --zookeeper zk:2181 --alter --topic logstash --partitions 5

# list topics
bin/kafka-topics.sh --zookeeper zk:2181 --list

# describe topic 
bin/kafka-topics.sh --zookeeper zk:2181 --topic logstash --describe

# describe consumer 
bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group logstash-group-01 --describe