ElasticSearch：ELK 架构

Posted 2023-04-05 Men-DD

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ElasticSearch：ELK 架构相关的知识，希望对你有一定的参考价值。

日志收集——》格式化分析——》检索和可视化——》风险告警

ELK架构
- 经典的ELK
- 整合消息队列+nginx架构
什么是Logstash
- Logstash核心概念
- Logstash数据传输原理
- Logstash配置文件结构
- Logstash Queue
- Logstash导入数据到ES
- 同步数据库数据到Elasticsearch
什么是Beats
- FileBeat简介
- FileBeat的工作原理
- logstash vs FileBeat
- Filebeat安装
ELK整合实战
- 案例：采集tomcat服务器日志
- 使用FileBeats将日志发送到Logstash
- 配置Logstash接收FileBeat收集的数据并打印
- Logstash输出数据到Elasticsearch
- 利用Logstash过滤器解析日志
- 输出到Elasticsearch指定索引

ELK架构

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

经典的ELK `数据量小的开发环境，存在数据丢失的危险`

经典的ELK主要是由Filebeat + Logstash + Elasticsearch + Kibana组成，如下图：（早期的ELK只有Logstash + Elasticsearch + Kibana）

整合消息队列+Nginx架构 `生产环境，可以处理大数据量，并且不会丢失数据`

这种架构，主要加上了Redis或Kafka或RabbitMQ做消息队列，保证了消息的不丢失

什么是Logstash

Logstash 是免费且开放的服务器端数据处理管道，能够从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的存储库中。

https://www.elastic.co/cn/logstash/

应用：ETL工具 / 数据采集处理引擎

Logstash核心概念

Pipeline

包含了input—filter-output三个阶段的处理流程
插件生命周期管理
队列管理

Logstash Event

数据在内部流转时的具体表现形式。数据在input 阶段被转换为Event，在 output被转化成目标格式数据
Event 其实是一个Java Object，在配置文件中，对Event 的属性进行增删改查

Codec (Code / Decode)

将原始数据decode成Event; 将Event encode成目标数据

Logstash数据传输原理

数据采集与输入：Logstash支持各种输入选择，能够以连续的流式传输方式，轻松地从日志、指标、Web应用以及数据存储中采集数据。
实时解析和数据转换：通过Logstash过滤器解析各个事件，识别已命名的字段来构建结构，并将它们转换成通用格式，最终将数据从源端传输到存储库中。
存储与数据导出：Logstash提供多种输出选择，可以将数据发送到指定的地方。

Logstash通过管道完成数据的采集与处理，管道配置中包含input、output和filter（可选）插件，input和output用来配置输入和输出数据源、filter用来对数据进行过滤或预处理。

Logstash配置文件结构

参考：https://www.elastic.co/guide/en/logstash/7.17/configuration.html

Logstash的管道配置文件对每种类型的插件都提供了一个单独的配置部分，用于处理管道事件

input 
  stdin  


filter 
  grok 
    match =>  "message" => "%COMBINEDAPACHELOG" 
  
  date 
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  


output 
  elasticsearch  hosts => ["localhost:9200"]  
  stdout  codec => rubydebug

运行

bin/logstash -f logstash-demo.conf

Input Plugins

https://www.elastic.co/guide/en/logstash/7.17/input-plugins.html
一个 Pipeline可以有多个input插件
Stdin / File
Beats / Log4J /Elasticsearch / JDBC / Kafka /Rabbitmq /Redis
JMX/ HTTP / Websocket / UDP / TCP
Google Cloud Storage / S3
Github / Twitter

Output Plugins

https://www.elastic.co/guide/en/logstash/7.17/output-plugins.html
将Event发送到特定的目的地，是 Pipeline 的最后一个阶段。
常见 Output Plugins：

Elasticsearch
Email / Pageduty
Influxdb / Kafka / Mongodb / Opentsdb / Zabbix
Http / TCP / Websocket

Filter Plugins

https://www.elastic.co/guide/en/logstash/7.17/filter-plugins.html
处理Event
内置的Filter Plugins:

Mutate 一操作Event的字段
Metrics — Aggregate metrics
Ruby 一执行Ruby 代码

Codec Plugins

https://www.elastic.co/guide/en/logstash/7.17/codec-plugins.html
将原始数据decode成Event;将Event encode成目标数据
内置的Codec Plugins:

Line / Multiline
JSON / Avro / Cef (ArcSight Common Event Format)
Dots / Rubydebug

Logstash Queue

In Memory Queue
进程Crash，机器宕机，都会引起数据的丢失
Persistent Queue
机器宕机，数据也不会丢失; 数据保证会被消费; 可以替代 Kafka等消息队列缓冲区的作用

queue.type: persisted #(默认是memory)
queue.max_bytes: 4gb

Logstash安装

logstash官方文档: https://www.elastic.co/guide/en/logstash/7.17/installing-logstash.html

1）下载并解压logstash

下载地址： https://www.elastic.co/cn/downloads/past-releases#logstash
选择版本：7.17.5

wget https://artifacts.elastic.co/downloads/logstash/logstash-7.17.5-linux-x86_64.tar.gz

tar -zxvf logstash-7.17.5-linux-x86_64.tar.gz

2）测试：运行最基本的logstash管道

cd logstash-7.17.5
#-e选项表示，直接把配置放在命令中，这样可以有效快速进行测试

bin/logstash -e 'input  stdin    output  stdout  '

Codec Plugin测试

#single line
bin/logstash -e "inputstdincodec=>lineoutputstdoutcodec=> rubydebug"
bin/logstash -e "inputstdincodec=>jsonoutputstdoutcodec=> rubydebug"

Codec Plugin —— Multiline

设置参数:

pattern: 设置行匹配的正则表达式
what : 如果匹配成功，那么匹配行属于上一个事件还是下一个事件
previous / next
negate : 是否对pattern结果取反
true / false

# 多行数据，异常
Exception in thread "main" java.lang.NullPointerException
        at com.example.myproject.Book.getTitle(Book.java:16)
        at com.example.myproject.Author.getBookTitles(Author.java:25)
        at com.example.myproject.Bootstrap.main(Bootstrap.java:14)

# multiline-exception.conf
input 
  stdin 
    codec => multiline 
      pattern => "^\\s"
      what => "previous"
    
  


filter 

output 
  stdout  codec => rubydebug 


#执行管道
bin/logstash -f multiline-exception.conf

Input Plugin —— File

https://www.elastic.co/guide/en/logstash/7.17/plugins-inputs-file.html

支持从文件中读取数据，如日志文件
文件读取需要解决的问题：只被读取一次。重启后需要从上次读取的位置继续(通过sincedb 实现)
读取到文件新内容，发现新文件
文件发生归档操作(文档位置发生变化，日志rotation)，不能影响当前的内容读取

Filter Plugin

Filter Plugin可以对Logstash Event进行各种处理，例如解析，删除字段，类型转换

Date: 日期解析
Dissect: 分割符解析
Grok: 正则匹配解析
Mutate: 处理字段。重命名，删除，替换
Ruby: 利用Ruby 代码来动态修改Event

Filter Plugin - Mutate

对字段做各种操作:

Convert : 类型转换
Gsub : 字符串替换
Split / Join /Merge: 字符串切割，数组合并字符串，数组合并数组
Rename: 字段重命名
Update / Replace: 字段内容更新替换
Remove_field: 字段删除

Logstash导入数据到ES

1）测试数据集下载：https://grouplens.org/datasets/movielens/
https://files.grouplens.org/datasets/movielens/ml-25m.zip
2）准备logstash-movie.conf配置文件

input 
  file 
    path => "/home/es/logstash-7.17.3/dataset/movies.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  
 
filter 
  csv 
    separator => ","
    columns => ["id","content","genre"]
  

  mutate 
    split =>  "genre" => "|" 
    remove_field => ["path", "host","@timestamp","message"]
  

  mutate 
    split => ["content", "("]
    add_field =>  "title" => "%[content][0]"
    add_field =>  "year" => "%[content][1]"
  

  mutate 
    convert => 
      "year" => "integer"
    
    strip => ["title"]
    remove_field => ["path", "host","@timestamp","message","content"]
  


output 
   elasticsearch 
     hosts => "http://localhost:9200"
     index => "movies"
     document_id => "%id"
     user => "elastic"
     password => "123456"
   
  stdout

3）运行logstash
bin/logstash -f logstash-movie.conf

get /movies/_search


  "took" : 0,
  "timed_out" : false,
  "_shards" : 
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  ,
  "hits" : 
    "total" : 
      "value" : 10000,
      "relation" : "gte"
    ,
    "max_score" : 1.0,
    "hits" : [
      
        "_index" : "movies",
        "_type" : "_doc",
        "_id" : "6747",
        "_score" : 1.0,
        "_source" : 
          "id" : "6747",
          "year" : 1960,
          "title" : "Adventures of Huckleberry Finn, The",
          "genre" : [
            "Adventure",
            "Children"
          ],
          "@version" : "1"
        
      ,
      
        "_index" : "movies",
        "_type" : "_doc",
        "_id" : "6748",
        "_score" : 1.0,
        "_source" : 
          "id" : "6748",
          "year" : 1979,
          "title" : "Brood, The",
          "genre" : [
            "Horror"
          ],
          "@version" : "1"
        
      ,
      
        "_index" : "movies",
        "_type" : "_doc",
        "_id" : "6749",
        "_score" : 1.0,
        "_source" : 
          "id" : "6749",
          "year" : 1937,
          "title" : "Prince and the Pauper, The",
          "genre" : [
            "Adventure",
            "Drama"
          ],
          "@version" : "1"
        
      ,
      ....
    ]

Docker Logstash

1. 拉取运行logstash

docker pull logstash:7.17.5
docker run -d --name=logstash logstash:7.17.5

2. COPY配置文件至本地

mkdir -p /data/logstash
docker cp logstash:/usr/share/logstash/config /data/logstash/
docker cp logstash:/usr/share/logstash/data /data/logstash/
docker cp logstash:/usr/share/logstash/pipeline /data/logstash/
chmod 777 -R /data/logstash

3. 配置

vi /mydata/logstash/config/logstash.yml

http.host: "0.0.0.0"
config.reload.automatic: true

vi /data/logstash/config/springboot.conf

input 
 tcp 
   host => "0.0.0.0"
   mode => "server"
   port => 5055
   codec => json_lines
 


#filter 
# ruby  #设置一个自定义字段'timestamp'[这个字段可自定义]，将logstash自动生成的时间戳中的值加8小时，赋给这个字段
#  code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*3600)"
# 
# ruby  #将自定义时间字段中的值重新赋给@timestamp
#  code => "event.set('@timestamp',event.get('timestamp'))"
# 
# mutate  #删除自定义字段
#  remove_field => ["timestamp"]
# 
#

output 
 elasticsearch 
  hosts => "es.localhost.com:9200"
  user => "elastic"
  password => "xxxxxx"
  index => "mendd-%+YYYY.MM.dd"
 
 stdout  codec => rubydebug

4. logstash多个配置文件相互独立

vi /mydata/logstash/config/pipelines.yml

- pipeline.id: main
  path.config: "/usr/share/logstash/pipeline"
- pipeline.id: mendd
  path.config: "/usr/share/logstash/config/springboot.conf"

同步数据库数据到Elasticsearch

需求: 将数据库中的数据同步到ES，借助ES的全文搜索,提高搜索速度

需要把新增用户信息同步到Elasticsearch中
用户信息Update 后，需要能被更新到Elasticsearch
支持增量更新
用户注销后，不能被ES所搜索到

实现思路

基于canal同步数据（项目实战中讲解）
借助JDBC Input Plugin将数据从数据库读到Logstash
- 需要自己提供所需的 JDBC Driver；
- JDBC Input Plugin 支持定时任务 Scheduling，其语法来自 Rufus-scheduler，其扩展了 Cron，使用 Cron 的语法可以完成任务的触发；
- JDBC Input Plugin 支持通过 Tracking_column / sql_last_value 的方式记录 State，最终实现增量的更新；
- https://www.elastic.co/cn/blog/logstash-jdbc-input-plugin

JDBC Input Plugin实现步骤

1）拷贝jdbc依赖到logstash-7.17.3/drivers目录下
/srv/soft/logstash-7.17.5/drivers/mysql-connector-java-5.1.49.jar
2）准备mysql-demo.conf配置文件

input 
  jdbc 
    jdbc_driver_library => "/srv/soft/logstash-7.17.5/drivers/mysql-connector-java-5.1.49.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/db-es-test?useSSL=false"
    jdbc_user => "xxxxxxxx"
    jdbc_password => "xxxxx"
    #启用追踪，如果为true，则需要指定tracking_column
    use_column_value => true
    #指定追踪的字段，
    tracking_column => "last_updated"
    #追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
    tracking_column_type => "numeric"
    #记录最后一次运行的结果
    record_last_run => true
    #上面运行结果的保存位置
    last_run_metadata_path => "jdbc-position.txt"
    statement => "SELECT * FROM user where last_updated >:sql_last_value;"
    schedule => " * * * * * *"
  

output 
  elasticsearch 
    document_id => "%id"
    document_type => "_doc"
    index => "users"
    hosts => ["http://localhost:9200"]
    user => "elastic"
    password => "123456"
  
  stdout
    codec => rubydebug

3）运行logstash
bin/logstash -f mysql-demo.conf

测试、新增、更新、删除

#user表
CREATE TABLE `user` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(50) DEFAULT NULL,
  `address` varchar(50) CHARACTER DEFAULT NULL,
  `last_updated` bigint DEFAULT NULL,
  `is_deleted` int DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
#插入数据
INSERT INTO user(name,address,last_updated,is_deleted) VALUES("张三","广州天河",unix_timestamp(NOW()),0)

ES查询过滤删除数据

# 创建 alias，只显示没有被标记 deleted的用户
POST /_aliases

  "actions": [
    
      "add": 
        "index": "users",
        "alias": "view_users",
        "filter" :  "term" :  "is_deleted" : 0 
      
    
  ]


# 通过 Alias查询，查不到被标记成 deleted的用户
POST view_users/_search

POST view_users/_search

  "query": 
    "term": 
      "name.keyword": 
        "value": "张三"

轻量型数据采集器 Beats

https://www.elastic.co/guide/en/beats/libbeat/7.17/index.html

Beats 是一个免费且开放的平台，集合了多种单一用途的数据采集器。它们从成百上千或成千上万台机器和系统向 Logstash 或 Elasticsearch 发送数据。

FileBeat简介

FileBeat专门用于转发和收集日志数据的轻量级采集工具。它可以作为代理安装在服务器上，FileBeat监视指定路径的日志文件，收集日志数据，并将收集到的日志转发到Elasticsearch或者Logstash。

FileBeat的工作原理

启动FileBeat时，会启动一个或者多个输入（Input），这些Input监控指定的日志数据位置。FileBeat会针对每一个文件启动一个Harvester（收割机）。Harvester读取每一个文件的日志，将新的日志发送到libbeat，libbeat将数据收集到一起，并将数据发送给输出（Output）。

Logstash vs FileBeat

Logstash是在jvm上运行的，资源消耗比较大。而FileBeat是基于golang编写的，功能较少但资源消耗也比较小，更轻量级。
Logstash 和 Filebeat都具有日志收集功能，Filebeat更轻量，占用资源更少
Logstash 具有Filter功能，能过滤分析日志
一般结构都是Filebeat采集日志，然后发送到消息队列、Redis、MQ中，然后Logstash去获取，利用Filter功能过滤分析，然后存储到Elasticsearch中
FileBeat和Logstash配合，实现背压机制。当将数据发送到Logstash或 Elasticsearch时，Filebeat使用背压敏感协议，以应对更多的数据量。如果Logstash正在忙于处理数据，则会告诉Filebeat 减慢读取速度。一旦拥堵得到解决，Filebeat就会恢复到原来的步伐并继续传输数据。

Filebeat安装

https://www.elastic.co/guide/en/beats/filebeat/7.17/filebeat-installation-configuration.html

1）下载并解压Filebeat

下载地址：https://www.elastic.co/cn/downloads/past-releases#filebeat
选择版本：7.17.5

https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.17.5-linux-x86_64.tar.gz
tar zxvf filebeat-7.17.5-linux-x86_64.tar.gz

2）编辑配置

修改 filebeat.yml 以设置连接信息：

# ---------------------------- Elasticsearch Output ----------------------------
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["localhost:9200"]

  # Protocol - either `http` (default) or `https`.
  #protocol: "https"

  # Authentication credentials - either API key or username/password.
  #api_key: "id:api_key"
  #username: "elastic"
  #password: "changeme"

# ------------------------------ Logstash Output -------------------------------
#output.logstash:
  # The Logstash hosts
  #hosts: ["localhost:5044"]

  # Optional SSL. By default is off.
  # List of root certificates for HTTPS server verifications
  #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]

  # Certificate for SSL client authentication
  #ssl.certificate: "/etc/pki/client/cert.pem"

  # Client Certificate Key
  #ssl.key: "/etc/pki/client/cert.key"
setup.kibana:
  host: "xxxxx.xxxx.com:5601"

3) 启用和配置数据收集模块

从安装目录中，运行：

查看模块列表 ./filebeat modules list
启用nginx模块 ./filebeat modules enable nginx
启用 Logstash 模块 ./filebeat modules enable logstash

如果需要更改nginx日志路径,修改modules.d/nginx.yml

- module: nginx
  access:
    var.paths: ["/var/log/nginx/access.log*"]

在 modules.d/logstash.yml 文件中修改设置

- module: logstash
  log:
    enabled: true
    var.paths: ["/srv/soft/logstash-7.17.5/logs/*.log"]

4）启动 Filebeat

setup命令加载Kibana仪表板。如果仪表板已经设置，则忽略此命令
./filebeat setup

启动Filebeat
./filebeat -e

ELK整合

案例：采集tomcat服务器日志

Tomcat服务器运行过程中产生很多日志信息，通过Logstash采集并存储日志信息至ElasticSearch中

使用FileBeats将日志发送到Logstash

1）创建配置文件filebeat-logstash.yml，配置FileBeats将数据发送到Logstash

pattern：正则表达式
negate：true 或 false；默认是false，匹配pattern的行合并到上一行；true，不匹配pattern的行合并到上一行
match：after 或 before，合并到上一行的末尾或开头

vim filebeat-logstash.yml
chmod 644 filebeat-logstash.yml

#因为Tomcat的web log日志都是以IP地址开头的，所以我们需要修改下匹配字段。
# 不以ip地址开头的行追加到上一行
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /home/es/apache-tomcat-8.5.33/logs/*access*.*
  multiline.pattern: '^\\\\d+\\\\.\\\\d+\\\\.\\\\d+\\\\.\\\\d+ '
  multiline.negate: true
  multiline.match: after

output.logstash:
  enabled: true
  hosts: ["x.x.x.x:5044"]

2）启动FileBeat，并指定使用指定的配置文件

./filebeat -e -c filebeat-logstash.yml

可能出现的异常：

异常1：

Exiting: error loading config file: config file ("filebeat-logstash.yml") can only be writable by the owner but the permissions are "-rw-rw-r--" (to fix the permissions use: 'chmod go-w /home/es/filebeat-7.17.3-linux-x86_64/filebeat-logstash.yml')
因为安全原因不要其他用户写的权限，去掉写的权限就可以了
chmod 644 filebeat-logstash.yml

异常2：

Failed to connect to backoff(async(tcp://192.168.65.204:5044)): dial tcp 192.168.65.204:5044: connect: connection refused
FileBeat将尝试建立与Logstash监听的IP和端口号进行连接。但此时，我们并没有开启并配置Logstash，所以FileBeat是无法连接到Logstash的。

配置Logstash接收FileBeat收集的数据并打印

vim config/filebeat-console.conf

# 配置从FileBeat接收数据
input 
    beats 
      port => 5044
    


output 
    stdout 
      codec => rubydebug

测试logstash配置是否正确
bin/logstash -f config/filebeat-console.conf --config.test_and_exit

启动logstash
bin/logstash -f config/filebeat-console.conf --config.reload.automatic
reload.automatic：修改配置文件时自动重新加载

测试访问tomcat，logstash是否接收到了Filebeat传过来的tomcat日志

Logstash输出数据到Elasticsearch

如果我们需要将数据输出值ES而不是控制台的话，我们修改Logstash的output配置。

vim config/filebeat-elasticSearch.conf

input 
    beats 
      port => 5044
    


output 
  elasticsearch 
    hosts => ["http://localhost:9200"]
    user => "elastic"
    password => "123456"
  
  stdout
    codec => rubydebug

启动logstash

bin/logstash -f config/filebeat-elasticSearch.conf --config.reload.automatic

ES中会生成一个以logstash开头的索引，测试日志是否保存到了ES

get logstash-2022.07.28-000001/_search

response

  "took" : 0,
  "timed_out" : false,
  "_shards" : 
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  ,
  "hits" : 
    "total" : 
      "value" : 1,
      "relation" : "eq"
    ,
    "max_score" : 1.0,
    "hits" : [
      
        "_index" : "logstash-2022.07.28-000001",
        "_type" : "_doc",
        "_id" : "
ELK 架构之 Elasticsearch 和 Kibana 安装配置


阅读目录：

1. ELK Stack 简介
2. 环境准备
3. 安装 Elasticsearch
4. 安装 Kibana
5. Kibana 使用
6. Elasticsearch 命令

最近在开发分布式服务追踪，使用 Spring Cloud Sleuth Zipkin + Stream + RabbitMQ 中间件，默认使用内存存储数据，但这样应用于生产环境，就不太合适了。
最终我采用的方案：服务追踪数据使用 RabbitMQ 进行采集 + 数据存储使用 Elasticsearch + 数据展示使用 Kibana。
这篇文章主要记录 Elasticsearch 和 Kibana 环境的配置，以及采集服务追踪数据的显出处理。
1. ELK Stack 简介
ELK 是三个开源软件的缩写，分别为：Elasticsearch、Logstash 以及 Kibana，它们都是开源软件。不过现在还新增了一个 Beats，它是一个轻量级的日志收集处理工具（Agent），Beats 占用资源少，适合于在各个服务器上搜集日志后传输给 Logstash，官方也推荐此工具，目前由于原本的 ELK Stack 成员中加入了 Beats 工具所以已改名为 Elastic Stack。
根据 Google Trend 的信息显示，Elastic Stack 已经成为目前最流行的集中式日志解决方案。
Elastic Stack 包含：

Elasticsearch 是个开源分布式搜索引擎，提供搜集、分析、存储数据三大功能。它的特点有：分布式，零配置，自动发现，索引自动分片，索引副本机制，restful 风格接口，多数据源，自动搜索负载等。详细可参考 Elasticsearch 权威指南
Logstash 主要是用来日志的搜集、分析、过滤日志的工具，支持大量的数据获取方式。一般工作方式为 c/s 架构，client 端安装在需要收集日志的主机上，server 端负责将收到的各节点日志进行过滤、修改等操作在一并发往 Elasticsearch 上去。
Kibana 也是一个开源和免费的工具，Kibana 可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助汇总、分析和搜索重要数据日志。
Beats 在这里是一个轻量级日志采集器，其实 Beats 家族有 6 个成员，早期的 ELK 架构中使用 Logstash 收集、解析日志，但是 Logstash 对内存、cpu、io 等资源消耗比较高。相比 Logstash，Beats 所占系统的 CPU 和内存几乎可以忽略不计。

ELK Stack （5.0版本之后）--> Elastic Stack == （ELK Stack + Beats）。
目前 Beats 包含六种工具：

Packetbeat： 网络数据（收集网络流量数据）
Metricbeat： 指标（收集系统、进程和文件系统级别的 CPU 和内存使用情况等数据）
Filebeat： 日志文件（收集文件数据）
Winlogbeat： windows 事件日志（收集 Windows 事件日志数据）
Auditbeat：审计数据（收集审计日志）
Heartbeat：运行时间监控（收集系统运行时的数据）

ELK 简单架构图：

2. 环境准备
服务器环境：Centos 7.0（目前单机，后续再部署集群）
Elasticsearch 和 Logstash 需要 Java 环境，Elasticsearch 推荐的版本为 Java 8，安装教程：确定稳定的 Spring Cloud 相关环境版本
另外，我们需要修改下服务器主机信息：
[root@node1 ~]# vi /etc/hostname
node1

[root@node1 ~]# vi /etc/hosts
192.168.0.11 node1
127.0.0.1   node1 localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         node1 localhost localhost.localdomain localhost6 localhost6.localdomain6

注意：我之前安装 Elasticsearch 和 Kibana 都是最新版本（6.x），但和 Spring Cloud 集成有些问题，所以就采用了 5.x 版本（具体 5.6.9 版本）
3. 安装 Elasticsearch
运行以下命令将 Elasticsearch 公共 GPG 密钥导入 rpm：
[root@node1 ~]# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

在/etc/yum.repos.d/目录中，创建一个名为elasticsearch.repo的文件，添加下面配置：
[elasticsearch-5.x]
name=Elasticsearch repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

Elasticsearch 源创建完成之后，通过 makecache 查看源是否可用，然后通过 yum 安装 Elasticsearch：
[root@node1 ~]# yum makecache && yum install elasticsearch -y

修改配置（启动地址和端口）：
[root@node1 ~]# vi /etc/elasticsearch/elasticsearch.yml
network.host: node1  # 默认localhost，自定义为ip
http.port: 9200

要将 Elasticsearch 配置为在系统引导时自动启动，运行以下命令：
[root@node1 ~]# sudo /bin/systemctl daemon-reload
[root@node1 ~]# sudo /bin/systemctl enable elasticsearch.service

Elasticsearch 可以按如下方式启动和停止：
[root@node1 ~]# sudo systemctl start elasticsearch.service
[root@node1 ~]# sudo systemctl stop elasticsearch.service

列出 Elasticsearch 服务的日志：
[root@node1 ~]# sudo journalctl --unit elasticsearch
-- Logs begin at 三 2018-05-09 10:13:46 CEST, end at 三 2018-05-09 10:53:53 CEST. --
5月 09 10:53:43 node1 systemd[1]: [/usr/lib/systemd/system/elasticsearch.service:8] Unknown lvalue \'RuntimeDirectory\' in section \'Service\'
5月 09 10:53:43 node1 systemd[1]: [/usr/lib/systemd/system/elasticsearch.service:8] Unknown lvalue \'RuntimeDirectory\' in section \'Service\'
5月 09 10:53:48 node1 systemd[1]: Starting Elasticsearch...
5月 09 10:53:48 node1 systemd[1]: Started Elasticsearch.
5月 09 10:53:48 node1 elasticsearch[2908]: which: no java in (/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin)
5月 09 10:53:48 node1 elasticsearch[2908]: could not find java; set JAVA_HOME or ensure java is in PATH
5月 09 10:53:48 node1 systemd[1]: elasticsearch.service: main process exited, code=exited, status=1/FAILURE
5月 09 10:53:48 node1 systemd[1]: Unit elasticsearch.service entered failed state.

出现了错误，具体信息是未找到JAVA_HOME环境变量，但我们明明已经配置过了。
解决方式（参考资料：https://segmentfault.com/q/1010000004715131）：
[root@node1 ~]# vi /etc/sysconfig/elasticsearch
JAVA_HOME=/usr/local/java

重新启动：
sudo systemctl restart elasticsearch.service

或者通过systemctl命令，查看 Elasticsearch 启动状态：
[root@node1 ~]# systemctl status elasticsearch.service
elasticsearch.service - Elasticsearch
   Loaded: loaded (/usr/lib/systemd/system/elasticsearch.service; enabled)
   Active: active (running) since 一 2018-05-14 05:13:45 CEST; 4h 5min ago
     Docs: http://www.elastic.co
  Process: 951 ExecStartPre=/usr/share/elasticsearch/bin/elasticsearch-systemd-pre-exec (code=exited, status=0/SUCCESS)
 Main PID: 953 (java)
   CGroup: /system.slice/elasticsearch.service
           └─953 /usr/local/java/bin/java -Xms2g -Xmx2g -XX:+UseConcMarkSweepGC -XX:CMSInitiatingO...

5月 14 05:13:45 node1 systemd[1]: Started Elasticsearch.

发现 Elasticsearch 已经成功启动。
查看 Elasticsearch 信息：
[root@node1 ~]# curl -XGET \'http://node1:9200/?pretty\'
{
  "name" : "AKmrtMm",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "r7lG3UBXQ-uTLHInJxbOJA",
  "version" : {
    "number" : "5.6.9",
    "build_hash" : "877a590",
    "build_date" : "2018-04-12T16:25:14.838Z",
    "build_snapshot" : false,
    "lucene_version" : "6.6.1"
  },
  "tagline" : "You Know, for Search"
}

4. 安装 Kibana
运行以下命令将 Elasticsearch 公共 GPG 密钥导入 rpm：
[root@node1 ~]# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

在/etc/yum.repos.d/目录中，创建一个名为kibana.repo的文件，添加下面配置：
[kibana-5.x]
name=Kibana repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

安装 Kibana：
[root@node1 ~]# yum makecache && yum install kibana -y

修改配置（地址和端口，以及 Elasticsearch 的地址，注意server.host只能填写服务器的 IP 地址）：
[root@node1 ~]# vi /etc/kibana/kibana.yml

# Kibana is served by a back end server. This setting specifies the port to use.
server.port: 5601

# Specifies the address to which the Kibana server will bind. IP addresses and host names are both valid values.
# The default is \'localhost\', which usually means remote machines will not be able to connect.
# To allow connections from remote users, set this parameter to a non-loopback address.
server.host: "192.168.0.11"

# The Kibana server\'s name.  This is used for display purposes.
server.name: "kibana-server"

# The URL of the Elasticsearch instance to use for all your queries.
elasticsearch.url: "http://node1:9200"

# 配置kibana的日志文件路径，不然默认是messages里记录日志
logging.dest: /var/log/kibana.log 

创建日志文件：
[root@node1 ~]# touch /var/log/kibana.log; chmod 777 /var/log/kibana.log

要将 Kibana 配置为在系统引导时自动启动，运行以下命令：
[root@node1 ~]# sudo /bin/systemctl daemon-reload
[root@node1 ~]# sudo /bin/systemctl enable kibana.service

Kibana 可以如下启动和停止
[root@node1 ~]# sudo systemctl start kibana.service
[root@node1 ~]# sudo systemctl stop kibana.service

查看启动日志：
[root@node1 ~]# sudo journalctl --unit kibana
5月 09 11:14:48 node1 systemd[1]: Starting Kibana...
5月 09 11:14:48 node1 systemd[1]: Started Kibana.

然后浏览器访问：http://node1:5601

初次使用时，会让你配置一个默认的 index，也就是你至少需要关联一个 Elasticsearch 里的 Index，可以使用 pattern 正则匹配。
注意：如果 Elasticsearch 中没有数据的话，你是无法创建 Index 的。
如果 Spring Cloud Sleuth Zipkin + Stream + RabbitMQ 配置正确的话（以后再详细说明），服务追踪的数据就已经存储在 Elasticsearch 中了。
5. Kibana 使用
创建zipkin:*索引（*匹配后面所有字符）：

然后就可以查看服务追踪的数据了：

也可以创建自定义仪表盘：

6. Elasticsearch 命令
创建索引：
$ curl -XPUT \'http://node1:9200/twitter\'

查看 Index 索引列表：
$ curl -XGET http://node1:9200/_cat/indices
yellow open twitter k1KnzWyYRDeckjt7GASh8w 5 1 1 0 5.1kb 5.1kb
yellow open .kibana 8zJGQkq8TwC4s3JJLMX44g 1 1 1 0   4kb   4kb
yellow open student iZPqPcwrQbifGOfE9DQYvg 5 1 0 0  955b  955b

添加 Document 数据：
$ curl -XPUT \'http://node1:9200/twitter/tweet/1\' -d \'{
    "user" : "kimchy",
    "post_date" : "2009-11-15T14:12:12",
    "message" : "trying out Elastic Search"
}\'

获取 Document 数据：
$ curl -XGET \'http://node1:9200/twitter/tweet/1\'
{"_index":"twitter","_type":"tweet","_id":"1","_version":1,"found":true,"_source":{
    "user" : "kimchy",
    "post_date" : "2009-11-15T14:12:12",
    "message" : "trying out Elastic Search"
}}%

查询zipkin索引下面的数据：
$ curl -XGET \'http://node1:9200/zipkin:*/_search\'

参考资料：

集中式日志系统 ELK 协议栈详解
ELK+Filebeat搭建实时日志分析平台
如何在CentOS 7上安装Elasticsearch，Logstash和Kibana（ELK堆栈）
Spring Cloud Sleuth进阶实战
How To Install Elasticsearch, Logstash, and Kibana (ELK Stack) on CentOS 7
如何在 CentOS 7 上安装 Elastic Stack
搭建ELK日志分析平台（下）—— 搭建kibana和logstash服务器（推荐）
搭建ELK日志分析平台（上）—— ELK介绍及搭建 Elasticsearch 分布式集群（推荐）


以上是关于ElasticSearch：ELK 架构的主要内容，如果未能解决你的问题，请参考以下文章 
 常用的ELK/EFK架构
 ELK在广告系统监控中的应用 及 Elasticsearch简介
 一图看懂ELK架构
 elk之elastic安装
 elk之elasticsearch
 ELK-elasticsearch