### 〇、介绍
#### **`ELK是Elasticsearch、Logstash、Kibana的简称,亦可称为Elastic Stack，这三者是核心套件。但并非全部，其中还包括Filebeat/Heartbeat/Metricbeat/Auditbeat/Packetbeat/Winlogbeat数据收集代理。`**
**Elasticsearch** `实时全文搜索和分析引擎，提供搜集、分析、存储数据三大功能；是一套开放REST和JAVA API等结构提供高效搜索功能，可扩展的分布式系统。它构建于Apache Lucene搜索引擎库之上`
**Logstash** `一个用来搜集、分析、过滤日志的工具。它支持几乎任何类型的日志，包括系统日志、错误日志和自定义应用程序日志。它可以从许多来源接收日志，这些来源包括 syslog、消息传递（例如 RabbitMQ）和JMX，它能够以多种方式输出数据，包括电子邮件、websockets和Elasticsearch`
**Kibana** `一个基于Web的图形界面，用于搜索、分析和可视化存储在 Elasticsearch指标中的日志数据。它利用Elasticsearch的REST接口来检索数据，不仅允许用户创建他们自己的数据的定制仪表板视图，还允许他们以特殊的方式查询和过滤数据`
 
 
### 一、环境准备
系统: `CentOS7`
软件:
- `elasticsearch`：`7.3.1`
- `logstash`：`7.3.1`
- `kibana`：`7.3.1`
- `filebeat`：`7.3.1`
- `metricbeat`：`7.3.1`
- `redis(消息队列)`：`5.0.5`
服务器:
`node1(172.16.0.101)`：`elasticsearch` `filebeat` `metricbeat`
`node2(172.16.0.102)`：`elasticsearch` `filebeat` `metricbeat` `kibana` `logstash` `redis`
`node3(172.16.0.103)`：`elasticsearch` `filebeat` `metricbeat`
 
 
### 二、ElasticSearch
**`说明:搭建一个包含三个节点的ElasticSearch集群，三个节点为对等节点(主节点/数据节点)，集群初始化的主节点为node1。`**
#### 导入签名key
`所有节点`
```bash
rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
```
 
#### 配置Elastic Stack仓库
`所有节点`
```bash
cat << EOF >/etc/yum.repos.d/elasticsearch.repo
[elasticsearch-7.x]
name=Elasticsearch repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF
```
 
#### JAVA
`注意:ElasticSearch主目录捆绑安装JDK(/usr/share/elasticsearch/jdk)，无需安装`
#### 安装elasticsearch
`所有节点`
```bash
yum install elasticsearch -y
```
 
#### 配置elasticsearch
`配置文件路径位于$ES_PATH_CONF指定的值，默认为/etc/elasticsearch`
```bash
grep ES_PATH_CONF /etc/sysconfig/elasticsearch #查看配置文件的路径
```
```
ES_PATH_CONF=/etc/elasticsearch
```
 
#### 配置elasticsearch.yml
`node1`
```bash
cat <<EOF >/etc/elasticsearch/elasticsearch.yml
cluster.name: es-ynotes.cn-cluster
node.name: node1
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
http.cors.enabled: true
http.cors.allow-origin: "*"
node.master: true
node.data: true
transport.tcp.port: 9300
transport.tcp.compress: true
network.host: 172.16.0.101
http.port: 9200
discovery.seed_hosts: ["node1", "node2", "node3"] #集群节点
cluster.initial_master_nodes: ["node1"] #初始化主节点
EOF
```
`node2/node3`
`注意:修改node.name和network.host即可`
 
#### 启动elasticsearch
```bash
systemctl start elasticsearch
```
 
#### 查看集群状态
```bash
curl http://172.16.0.101:9200/_cluster/health?pretty
```
```
{
"cluster_name" : "es-ynotes.cn-cluster",
"status" : "red",
"timed_out" : false,
"number_of_nodes" : 3,
"number_of_data_nodes" : 3,
"active_primary_shards" : 12,
"active_shards" : 12,
"relocating_shards" : 0,
"initializing_shards" : 8,
"unassigned_shards" : 24,
"delayed_unassigned_shards" : 0,
"number_of_pending_tasks" : 7,
"number_of_in_flight_fetch" : 0,
"task_max_waiting_in_queue_millis" : 1838,
"active_shards_percent_as_number" : 27.27272727272727
}
```
`集群名为es-ynotes.cn-cluster,数据节点数为3个`
 
 
### 二、Kibana
#### 安装Kibana
`node2节点`
```bash
yum install kibana -y
```
 
#### 配置kibana
```bash
vim /etc/kibana/kibana.yml #修改下面对应的值即可
```
```yaml
server.host: "0.0.0.0" #监听的地址
elasticsearch.hosts: ["http://node1:9200","http://node2:9200","http://node3:9200"] #配置es集群节点
i18n.locale: "zh-CN" #设置界面为中文
```
 
 
### 三、Redis(docker运行)
#### 安装并启动docker
`node2`
```bash
yum install docker -y
systemctl start docker
```
#### docker运行Redis
`node2`
```bash
docker run -d --name logredis -p 6379:6379 redis --requirepass "ynotes.cn"
```
 
 
### 四、Logstash
#### 安装Logstash
`node2`
```bash
yum install logstash -y
```
 
#### logstash配置
**方式一:**
`数据流`：`日志` -> `logstash` -> `elasticsearch`
`说明:该方式不需要数据收集代理(Filebeat/Metircbeat),logstash直接收集数据写入elasticsearch`
```bash
cat <<EOF >/etc/logstash/logstash.conf
input{
file {
path => ["/var/log/messages"]
type => "system"
start_position => beginning
}
}
filter {
#无过滤规则
}
output{
elasticsearch {
hosts => ["node1:9200", "node2:9200", "node3:9200"]
index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
#user => "elastic"
#password => "changeme"
}
}
EOF
```
**方式二:**
`数据流`：`日志` -> `Filebeat` -> `logstash` -> `elasticsearch`
`说明:该方式使用数据收集代理(如Filebeat/Metircbeat)收集数据写入logstash，再由logstash写入elasticsearch`
```bash
cat <<EOF >/etc/logstash/logstash.conf
input {
beats {
port => 5044
}
}
filter {
#无过滤规则
}
output {
elasticsearch {
hosts => ["node1:9200", "node2:9200", "node3:9200"]
index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
#user => "elastic"
#password => "changeme"
}
}
EOF
```
**方式三:**
`数据流`：`日志` -> `Filebeat` -> `redis` -> `logstash` -> `elasticsearch`
`说明:该方式需要数据收集代理(如Filebeat)收集数据写入到redis，再由logstash从redis获取数据写入elasticsearch`
```bash
cat <<EOF >/etc/logstash/logstash.conf
input {
redis {
data_type => "list"
key => "filebeat_list"
host => "node2"
port => 6379
password => "ynotes.cn"
db => 0
threads => 4
}
}
filter {
#无过滤规则
}
output {
elasticsearch {
hosts => ["node1:9200", "node2:9200", "node3:9200"]
index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
}
}
EOF
```
 
#### 启动Logstash
```bash
systemctl start logstash
```
 
#### 过滤模块Grok
`grok是一种采用组合多个预定义的正则表达式，用来匹配分割文本并映射到关键字的工具。通常用来对日志数据进行预处理。logstash的filter模块中grok插件是其实现之一。`
`配置与调试`
例如日志如下：
```bash
localhost GET /index.html 1024 0.016
```
可依次判断类型为：IPORHOST、WORD、URIPATHPARAM、INT、NUMBER,
形成的grok语句如下为
```bash
%{IPORHOST:client} %{WORD:method} %{URIPATHPARAM:request} %{INT:size} %{NUMBER:duration}
```
`预定义匹配样例`
grok默认内置120个预定义匹配字段，已附在文末，logstash内置的可参考网址:
https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns
`调试工具`
- kibana内置Grok调试器
- 调试网址:https://grokdebug.herokuapp.com
`官方文档`:https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html
 
 
### 四、Filebeat
#### 安装filebeat
`所有节点`
```bash
yum install filebeat -y
```
 
#### 收集日志(`以系统日志为例`)
**启用模块**
`所有节点`
```bash
filebeat modules enable system #收集其他日志启用相应模块即可
```
**配置filebeat**
**方式一:**
`数据流`：`日志` -> `Filebeat` -> `elasticsearch`
```bash
vim /etc/filebeat/filebeat.yml
```
```bash
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: true
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
host: "node2:5601"
output.elasticsearch: #配置输出流到elasticsearch
hosts: ["node1:9200", "node2:9200", "node3:9200"]
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
```
**方式二:**
`数据流`：`日志` -> `Filebeat` -> `logstash` -> `elasticsearch`
```bash
vim /etc/filebeat/filebeat.yml
```
```bash
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: true
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
host: "node2:5601"
output.logstash: #配置输出流到logstash
hosts: ["node2:5044"]
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
```
**方式三:**
`数据流`：`日志` -> `Filebeat` -> `redis` -> `logstash` -> `elasticsearch`
```bash
vim /etc/filebeat/filebeat.yml
```
```bash
filebeat.config.modules:
enabled: true
path: ${path.config}/modules.d/*.yml
reload.enabled: true
reload.period: 10s
setup.kibana:
host: "node2:5601"
output.redis: #配置输出流到redis
hosts: ["node2:6379"]
password: "ynotes.cn"
key: "filebeat_list"
db: 0
timeout: 5
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
#优化参数
max_procs: 1
queue.mem:
events: 1024
flush.min_events: 512
flush.timeout: 5s
max_message_bytes: 100000
```
 
#### 手动加载索引模板(`注意:filebeat输出没有直连elasticsearch必须执行,方式二和方式三必须执行`)
```bash
filebeat export template >filebeat.template.json
curl -XPUT -H 'Content-Type: application/json' http://node1:9200/_template/filebeat-7.3.1 -d@filebeat.template.json
```
 
#### 启动Filebeat
```bash
filebeat setup #命令加载Kibana安装数据图表和仪表板。如果已设置，请省略此命令,filebeat不能直连kibana的情况，请使用一个可以直连kibana的filebeat去执行该操作
systemctl start filebeat
```
 
### 五、Metricbeat
#### 安装metricbeat
`所有节点`
```bash
yum install metricbeat -y
```
 
#### 添加指标(`以系统指标为例`)
**启用模块**
`所有节点`
```bash
metricbeat modules enable system #添加其他指标启用相应模块即可
```
**配置metricbeat**
**方式一:**
`数据流`：`日志` -> `metricbeat` -> `elasticsearch`
```bash
vim /etc/metricbeat/metricbeat.yml
```
```bash
metricbeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 1
index.codec: best_compression
setup.kibana:
host: "node2:5601"
output.elasticsearch:
hosts: ["node1:9200", "node2:9200", "node3:9200"]
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
```
**方式二:**
`数据流`：`日志` -> `metricbeat` -> `logstash` -> `elasticsearch`
```bash
vim /etc/metricbeat/metricbeat.yml
```
```bash
metricbeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: true
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
host: "node2:5601"
output.logstash: #配置输出流到logstash
hosts: ["node2:5044"]
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
```
**方式三:**
`数据流`：`日志` -> `metricbeat` -> `redis` -> `logstash` -> `elasticsearch`
```bash
vim /etc/metricbeat/metricbeat.yml
```
```bash
metricbeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: true
setup.template.settings:
index.number_of_shards: 1
index.codec: best_compression
output.redis: #配置输出流到redis
enabled: true
hosts: ["node2:6379"]
password: "ynotes.cn"
key: "metricbeat_list"
db: 0
timeout: 5
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
```
 
#### 手动加载索引模板(`注意:metricbeat输出没有直连elasticsearch必须执行,方式二和方式三必须执行`)
```bash
metricbeat export template >metricbeat.template.json
curl -XPUT -H 'Content-Type: application/json' http://node1:9200/_template/metricbeat-7.3.1 -d@metricbeat.template.json
```
 
#### 启动metricbeat
```bash
metricbeat setup #命令加载Kibana安装数据图表和仪表板。如果已设置，请省略此命令,metricbeat不能直连kibana的情况，请使用一个可以直连kibana的metricbeat去执行该操作
systemctl start metricbeat
```
 
### 六、生产环境实战
`说明:Elasticsearch放在公司内网环境，生产服务器在阿里云，为了让日志写入到Elasticsearch，采用的是前面filebeat介绍的方式三:`
`日志` -> `Filebeat` -> `redis` -> `logstash` -> `elasticsearch`
`其中filebeat和redis在生产服务器，logstash和elasticsearch在内网，生产服务器暴露redis(需认证)的端口,logstash去拉取数据然后写入elasticsearch`
`以配置nginx/tomcat日志为例`
#### filebeat配置
```bash
cat /etc/filebeat/filebeat.yml
```
```yml
filebeat.config.inputs:
enabled: true
path: ${path.config}/configs/*.yml #指定input输入配置文件路径
output.redis:
hosts: ["172.18.176.146:6379"] #阿里云内网的一台redis
password: "password" #redis的认证密码
key: "default_list"
db: 0
timeout: 5
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
max_procs: 1
queue.mem:
events: 1024
flush.min_events: 512
flush.timeout: 5s
max_message_bytes: 100000
```
配置nginx日志
```bash
vim /etc/filebeat/configs/nginx.yml
```
```ini
- type: log
enabled: true
paths:
- /var/log/nginx/domain.com.access.log
name: "nginx-prod-project_name-api-access_log"
tags: ['nginx','prod','project_name','api','access_log']
tail_files: true
exclude_lines: ['^-'] #排除-开头的日志(这里是因为阿里云配置SLB，所以很多内网访问的日志开头为-)
- type: log
enabled: true
paths:
- /var/log/nginx/error.log
name: "nginx-prod-error_log"
tags: ['nginx','prod','error_log']
tail_files: true
```
配置tomcat日志
```bash
vim /etc/filebeat/configs/tomcat.yml
```
```ini
- type: log
enabled: true
paths:
- /data/app/tomcat/logs/catalina.out.*
name: "tomcat-prod-project_name-api-catalina_log"
tags: ['tomcat','prod','project_name','api','catalina_log']
multiline:
pattern: '^([0-9]{4}-[0-9]{2}-[0-9]{2}|[0-9]{2}-[a-zA-Z]{3}-[0-9]{4})' #多行匹配特定格式的日期为新行
negate: true
match: after
tail_files: true
- type: log
enabled: true
paths:
- /data/app/tomcat/logs/localhost.20*
name: "tomcat-prod-project_name-api-localhost_log"
tags: ['tomcat','prod','project_name','api','localhost_log']
multiline:
pattern: '^[[:space:]]' #多行模式匹配空格开头非新行
negate: false
match: after
tail_files: true
```
 
#### logstash配置
```bash
vim /data/elk/logstash/config/logstash.yml
```
```yml
input {
redis {
id => 'outer'
type => 'outer'
data_type => "list"
key => "default_list"
host => "xx.xx.xx.xx" #阿里云暴露的reids服务器
port => 6379
password => "password"
db => 0
threads => 4
}
}
filter {
if "nginx" in [tags] and "access_log" in [tags] {
grok {
patterns_dir => "/data/elk/logstash/patterns" #GROK模式匹配目录
match => ["message","%{NGINXACCESS}"] #NGINXACCESS为自定义的GROK匹配格式
}
date {
match => [ "timestamp" , "dd/MMM/YYYY:HH:\mm:ss Z" ] #\mm应该为mm，因为网页会转义成图片
}
geoip {
source => "client_ip"
}
}
if "nginx" in [tags] and "error_log" in [tags] {
grok {
patterns_dir => "/data/elk/logstash/patterns"
match => ["message", "%{NGINXERRORTIME:logdate}"] #NGINXERRORTIME为自定义的GROK匹配格式
}
date {
match => [ "logdate" , "yyyy/MM/dd HH:\mm:ss" ] #\mm应该为mm，因为网页会转义成图片
target => "@timestamp"
}
}
if "tomcat" in [tags] and "catalina_log" in [tags] {
grok {
patterns_dir => "/data/elk/logstash/patterns"
match => ["message", "%{TIMESTAMP_ISO8601:logdate} %{WORD:LOGLEVEL}","message","%{TOMCATLOCALHOSTTIME:logdate}"] #TOMCATLOCALHOSTTIME为自定义的GROK匹配格式
}
date {
match => [ "logdate" , "yyyy-MM-dd HH:\mm:ss.SSS","dd-MMM-yyyy HH:\mm:ss.SSS"] #\mm应该为mm，因为网页会转义成图片
target => "@timestamp"
}
}
if "tomcat" in [tags] and "localhost_log" in [tags] {
grok {
patterns_dir => "/data/elk/logstash/patterns"
match => ["message", "%{TOMCATLOCALHOSTTIME:logdate}"] #TOMCATLOCALHOSTTIME为自定义的匹配格式
}
date {
match => [ "logdate" , "dd-MMM-yyyy HH:\mm:ss.SSS" ] #\mm应该为mm，因为网页会转义成图片
target => "@timestamp"
}
}
}
#输出配置
output {
if [type] == "outer" {
if "prod" in [tags] and "nginx" in [tags] and "project_name" in [tags] and "api" in [tags] and "access_log" in [tags] {
elasticsearch {
hosts => ["192.168.50.251:9200"] #内网Elasticsearch服务器
index => "prod_nginx_project_name_api_access_log" #索引名
}
}
if "prod" in [tags] and "nginx" in [tags] and "error_log" in [tags] {
elasticsearch {
hosts => ["192.168.50.251:9200"]
index => "prod_nginx_error_log"
}
}
if "prod" in [tags] and "tomcat" in [tags] and "project_name" in [tags] and "api" in [tags] and "catalina_log" in [tags] {
elasticsearch {
hosts => ["192.168.50.251:9200"]
index => "prod_tomcat_project_name_api_catalina_log"
}
}
if "prod" in [tags] and "tomcat" in [tags] and "project_name" in [tags] and "api" in [tags] and "localhost_log" in [tags] {
elasticsearch {
hosts => ["192.168.50.251:9200"]
index => "prod_tomcat_project_name_api_localhost_log"
}
}
}
}
```
自定义的GROK匹配格式(`注意：对nginx/tomcat的日志格式做相应修改`)
```bash
cat /data/elk/logstash/patterns/nginx
```
```ini
NGINXACCESS %{IPORHOST:client_ip|-} %{IPORHOST:remote_addr} - (%{USERNAME:remote_user}|-) \[%{HTTPDATE:timestamp}\] \"%{HOSTNAME:http_host}\" \"%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:http_status} (?:%{NUMBER:body_bytes_sent}|-) (?:\"(?:%{URI:referrer}|-)\"|%{QS:referrer}) \"%{WORD:http_x_forwarded_proto}\" %{QS:agent} (?:%{HOSTPORT:upstream_addr}|-) (%{NUMBER:upstream_response_time}|-) (%{NUMBER:request_time}|-)
NGINXERRORTIME %{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{HOUR}:%{MINUTE}:%{SECOND}
```
```bash
cat /data/elk/logstash/patterns/tomcat
```
```ini
TOMCATLOCALHOSTTIME %{MONTHDAY}-%{MONTH}-%{YEAR} %{HOUR}:%{MINUTE}:%{SECOND}
```
 
#### kibana配置
`1.创建对应的索引模式`
`2.创建对应的可视化图和仪表板即可`
