搭建elk服务

一、搜索引擎

1、搜索引擎是由索引组件、搜索组件和用户接口组成：

流程：

• 索引组件：获取数据-->建立文档-->文档分析-->文档索引（倒排索引）
• 搜索组件：用户搜索接口-->建立查询（将用户键入的信息转换为可处理的查询对象）-->搜索查询-->展现结果
• 用户接口：输入和输出

搜索引擎组件

2、搜索和索引组件的实现软件

索引组件：Lucene

• Apache LuceneTM是一个高性能、功能完全的文本搜索引擎库完全用Java编写。这是一个技术适合几乎任何应用程序需要全文搜索,尤其是跨平台。

搜索组件：Solr

• SolrTM高性能搜索服务器使用Lucene是一个核心,使用XML / HTTP和JSON / Python / Ruby api,显示在平面的搜索、缓存、复制和web管理界面。

搜索组件：ElasticSearch

• Elasticsearch是一种分布式的、基于rest的搜索和分析引擎可以解决越来越多的用例。弹性堆栈的核心,它集中存储您的数据,这样你就可以发现预期和意想不到的发现。

  
  搜索和索引组件的实现软件

3、抽取数据组件的实现软件

什么是ETL工具

• ETL，是英文 Extract-Transform-Load 的缩写，用来描述将数据从来源端经过抽取(extract)、转换(transform)、加载(load)至目的端的过程。ETL 是构建数据仓库的重要一环，用户从数据源抽取出所需的数据，经过数据清洗，最终按照预先定义好的数据仓库模型，将数据加载到数据仓库中去。

ETL工具开源实现：Logstash

• Logstash 是一个应用程序日志、事件的传输、处理、管理和搜索的平台。你可以用它来统一对应用程序日志进行收集管理，提供 Web 接口用于查询和统计。Logstash 现在是 ElasticSearch 家族成员之一。
  Logstash是一个开源的,服务器端数据处理管道,同时收集多个数据源的数据,转换,然后将其发送给你能够“储备。”(一般是Elasticsearch)，Logstash性能比较差。

ETL工具开源实现：Beats

• Filebeat：Log Files 处理日志文件
• Metricbeat：Metrics 收集计量数据例如cpu信息
• Packetbeat：Network Data 收集网络报文数据
• Winlogbeat：Windows Event Logs 视窗系统日志
• Heartbeat：Uptime Monitoring 运行监控日志

4、用户组件的实现软件

Kibana：

• Kibana允许您可视化Elasticsearch数据和导航弹性堆栈,非常精确的计算统计并展现出来。
  官方网站：https://www.elastic.co/cn/

二、ElasticSearch

1. ES是什么？

• Elasticsearch 是一个开源的搜索引擎，建立在全文搜索引擎库 Apache Lucene 基础之上用 Java 编写的，它的内部使用 Lucene 做索引与搜索，但是它的目的是使全文检索变得简单， 通过隐藏Lucene 的复杂性，取而代之的提供一套简单一致的 RESTful API。
• Elasticsearch 是一个全文搜索引擎，也是一个分布式的实时文档存储，每个字段可以被索引与搜索作数据库用，也可做一个分布式实时分析搜索引擎 能胜任上百个服务节点的扩展，并支持 PB 级别的结构化或者非结构化数据。

特性：

速度快、易扩展、弹性、灵活、操作简单、多语言客户端、X-Pack、hadoop/spark结合使用、开启即用。

分布式：横向扩展非常灵活
全文检索：基于lucene的强大的全文检索能力；
近实时搜索和分析：数据进入ES，可达到近实时搜索，还可进行聚合分析
高可用：容错机制，自动发现新的或失败的节点，重组和重新平衡数据
模式自由：ES的动态mapping机制可以自动检测数据的结构和类型，创建索引并使数据可搜索。
RESTful API：JSON + HTTP

2、ES的核心物理组件

Cluster 集群：

• 一个集群由一个唯一的名字标识，默认为“elasticsearch”。集群名称非常重要，具有相同集群名的节点才会组成一个集群。集群名称可以在配置文件中指定。集群 状态：有三个分别是green, yellow, red

Node 节点：

• 存储集群的数据，参与集群的索引和搜索功能。像集群有名字，节点也有自己的名称，默认在启动时会以一个随机的UUID的前七个字符作为节点的名字，你可以为其指定任意的名字。通过集群名在网络中发现同伴组成集群。一个节点也可是集群。

Shard 分片：

• 在创建一个索引时可以指定分成多少个分片来存储。每个分片本身也是一个功能完善且独立的“索引”，可以被放置在集群的任意节点上。分片的好处：允许我们水平切分/扩展容量可在多个分片上进行分布式的、并行的操作，提高系统的性能和吞吐量。注意分片数创建索引时指定，创建后不可改了。备份数可以随时改。

Primary Shard主分片

Replica Shard副本分片

• 一个分片可以有多个备份（副本）。备份的好处：高可用。一个主分片挂了，副本分片就顶上去；备份可扩展搜索的并发能力、吞吐量。搜索可以在所有的副本上并行运行。在高并发下副本也可搜索。

分片状态：green代表所有分片都存在
yellow代表主副部分分片缺失
red代表主副分片都缺失

3、Lucene的核心组件：

Near Realtime（NRT） 近实时：

• 数据提交索引后，立马就可以搜索到。

索引（index）：数据库

• Index 索引: 一个索引是一个文档的集合（等同于solr中的集合）。每个索引有唯一的名字，通过这个名字来操作它。一个集群中可以有任意多个索引。

类型（type）：表

• Type 类型：指在一个索引中，可以索引不同类型的文档，如用户数据、博客数据。从6.0.0 版本起已废弃，一个索引中只存放一类数据。

文档（Document）：行

• Document 文档：被索引的一条数据，索引的基本信息单元，以JSON格式来表示。

映射（Mapping）：

• 映射可以看作是建表的过程。但与数据库不同的是，映射可以为一个字段建立不同的映射以满足不同的场景，可以对数据进行清洗，容错，建立倒排索引。

4、域选项

域选项是用来来控制Lucene将文档添加进域

• 索引后对该域执行的操作有三个控制方法：

  Index.ANALYZED：切词和分析；
  Index.NOT_ANALYZED：做索引，但不做分析；
  Index.NO：不做索引；

5、ElasticSearch 程序环境：

配置文件：

/etc/elasticsearch/elasticsearch.yml主配置文件
/etc/elasticsearch/jvm.options
/etc/elasticsearch/log4j2.properties日志配置文件
Unit File：elasticsearch.service

程序文件：

/usr/share/elasticsearch/bin/elasticsearch
/usr/share/elasticsearch/bin/elasticsearch-keystore：加密配置文件目录
/usr/share/elasticsearch/bin/elasticsearch-plugin：管理插件程序

工作端口

• 搜索服务工作在tcp的9200端口
• 集群服务工作在tcp的9300端口

6、els集群的工作逻辑

• 所有节点选举一个主节点，负责管理整个集群的状态(green/yellow/red)，以及各shards的分布方式；
  ES中默认采用的节点发现方式是 zen（基于组播（多播）、单播）。在应用于生产前有两个重要参数需配置：

discovery.zen.ping.unicast.hosts: ["host1","host2:port","host3[portX-portY]"] 

• 单播模式下，设置具有master资格的节点列表，新加入的节点向这个列表中的节点发送请求来加入集群

discovery.zen.minimum_master_nodes: 1 

• 这个参数控制的是，一个节点需要看到具有master资格的节点的最小数量，然后才能在集群中做操作。官方的推荐值是(N/2)+1，其中N是具有master资格的节点的数量

• 另外elastic还有很多插件，来完善各自功能。

7、集群配置方法流程：

elasticsearch.yml配置文件：

cluster.name: myels  #集群名称
node.name: node1    #节点名称
path.data: /data/els/data   #数据存储目录
path.logs: /data/els/logs    #日志目录
network.host: 0.0.0.0        #监听地址
http.port: 9200                  #监听端口
discovery.zen.ping.unicast.hosts: ["node1", "node2", "node3"]    #集群成员
discovery.zen.minimum_master_nodes: 2      ##当主节点丢失时，定义法定票数，投票选择新master

8、el访问通过RESTful API:

格式：curl  -X<VERB> '<PROTOCOL>://<HOST>:<PORT>/<PATH>?<QUERY_STRING>' -d '<BODY>'

<BODY>：json格式的请求主体；
<VERB>动作：GET获取或搜索，POST，PUT上传存储，DELETE删除
<PATH>指明特定索引的特定类型
例如：/index_name/type/Document_ID/

• 特殊端点：/_cat, /_search请求查询由search端点发出, /_cluster集群端点

向集群端点查询成员健康状态

curl -XGET 'http://10.1.0.67:9200/_cluster/health?pretty=true'

向集群端点查询成员信息

curl -XGET 'http://10.1.0.67:9200/_cluster/stats?pretty=true'

向搜索端点查询主机信息

curl -XGET 'http://10.1.0.67:9200/_cat/nodes?pretty'

向搜索端点查询健康信息

curl -XGET 'http://10.1.0.67:9200/_cat/health?pretty'

创建文档：

curl  -XPUT    {"key1": "value1", "key2": value, ...}

三、搭建elk服务

实验环境架构

1、安装elasticsearch集群

节点一 192.168.1.61
节点二 192.168.1.62
节点三 192.168.1.63

• 注意：三个节点都要操作安装配置

elasticsearch和kibana、filebeat、logstash的版本必须一致

下载地址：https://artifacts.elastic.co/

#同步时间
#三个节点设置主机名
[root@node-61 ~]# vim /etc/hosts

127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.1.61    node1 node1.localdomain    
192.168.1.62    node2 node2.localdomain
192.168.1.63    node3 node3.localdomain

#三个节点安装jdk
yum install java-1.8.0-openjdk-devel -y

#三个节点下载安装elasticsearch
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.5.3.rpm

 rpm -ivh elasticsearch-6.5.3.rpm 

#节点一下载安装kibanna
wget https://artifacts.elastic.co/downloads/kibana/kibana-6.5.3-x86_64.rpm


#配置三个节点elasticsearch配置文件
vim /etc/elasticsearch/elasticsearch.yml 
------ Cluster ------
cluster.name: myels  #定义集群名称
---- Node -----
node.name: node1    #当前节点名称
----- Paths ----
path.data: /els/data  #索引数据存储位置
path.data: /els/data  #索引数据存储位置
----- Memory -----
#bootstrap.memory_lock: true    #开机后是否锁定内存使用，jvm.options 文件中的最大最小保持一致此选项设置为true，不一致设置为fail
------ Network ------
network.host: 192.168.1.61  #监控在本机的那个地址，不设置默认所有地址
#http.port: 9200  #默认端口
----- Discovery -----
discovery.zen.ping.unicast.hosts: ["node1", "node2","node3"] #判定集群成员名称
discovery.zen.minimum_master_nodes：2  #当主节点丢失时，定义法定票数，投票选择新master


##配置三个节点vm.options 文件
[root@node-61 ~]# vim /etc/elasticsearch/jvm.options 
.............
-Xms1g   #最小初始化内存
-Xmx1g   #最大使用内存

##三个节点创建索引数据和日志目录
[root@node-61 ~]# mkdir -pv /els/{data,logs}
mkdir: created directory ‘/els’
mkdir: created directory ‘/els/data’
mkdir: created directory ‘/els/logs’
[root@node-61 ~]# chown -R elasticsearch.elasticsearch /els/

##配置三个节点启动elasticsearch
[root@node-61 ~]# systemctl status elasticsearch
[root@node-61 ~]# ss -tnl
.....
  ::ffff:192.168.1.61:9200

#查询节点集群信息
[root@node-61 ~]# curl -XGET http://192.168.1.61:9200/_cat/nodes
192.168.1.63 11 96 3 0.00 0.06 0.10 mdi - node3
192.168.1.61 10 95 3 0.00 0.03 0.08 mdi * node1                #有*号的是主节点
192.168.1.62 10 96 2 0.07 0.11 0.12 mdi - node2


#查询集群健康状态
[root@node-61 ~]# curl -XGET http://192.168.1.61:9200/_cat/health

1544710687 14:18:07 myels green 3 3 0 0 0 0 0 0 - 100.0%

#查看每个节点负载情况
[root@node-61 ~]# curl -XGET http://192.168.1.61:9200/_cat/nodes?v
ip           heap.percent ram.percent cpu load_1m load_5m load_15m node.role master name
192.168.1.63           12          96   0    0.02    0.04     0.09 mdi       -      node3
192.168.1.61           11          95   1    0.03    0.04     0.07 mdi       *      node1
192.168.1.62           10          96   0    0.06    0.07     0.11 mdi       -      node2

#查询有几个插件
[root@node-61 ~]# curl -XGET http://192.168.1.61:9200/_cat/plugin?v

#查询有几个索引
[root@node-61 ~]# curl -XGET http://192.168.1.61:9200/_cat/indices?v


#自定义信息使用帮助
[root@node-61 ~]# curl -XGET http://192.168.1.61:9200/_cat/nodes?
help

#显示集群自定义信息，名称，ip，启动时间，jdk版本
[root@node-61 ~]# curl -XGET http://192.168.1.61:9200/_cat/nodes?
h=name,ip,port,uptime,jdk
node3 192.168.1.63 9300 23.6m 1.8.0_191
node1 192.168.1.61 9300 25.3m 1.8.0_191
node2 192.168.1.62 9300   24m 1.8.0_191

#显示当前节点摘要信息
                                              [root@node-61 ~]# curl http://node1:9200
{
  "name" : "node1",
  "cluster_name" : "myels",
  "cluster_uuid" : "XGFCSZW9QPKfLv7N0q1qtA",
  "version" : {
    "number" : "6.5.3",
    "build_flavor" : "default",
    "build_type" : "rpm",
    "build_hash" : "159a78a",
    "build_date" : "2018-12-06T20:11:28.826501Z",
    "build_snapshot" : false,
    "lucene_version" : "7.5.0",
    "minimum_wire_compatibility_version" : "5.6.0",
    "minimum_index_compatibility_version" : "5.0.0"
  },
  "tagline" : "You Know, for Search"    #看到此句表名服务启动正常
}

2、安装elasticsearch-head插件

• 为了便于观察集群状况和管理，安装elasticsearch-head插件，使用网页管理集群，到GitHub克隆head插件

[root@node-61 ~]# yum install epel-release -y
[root@node-61 ~]# yum install npm -y

#克隆head插件
[root@node-61 ~]# git clone https://github.com/mobz/elasticsearch-head.git

#在elasticsearch配置文件最后添加
[root@node-61 ~]# vim /etc/elasticsearch/elasticsearch.yml 
......
#文件的最后添加
http.cors.enabled: true    #启动http插件
http.cors.allow-origin: "*"   #允许那些地址访问，实际使用不能使用*


#安装插件
[root@node-61 ~]# cd elasticsearch-head/
[root@node-61 elasticsearch-head]# npm install --ignore-scripts

#启动head插件，并在后台运行
[root@node-61 elasticsearch-head]# nohup npm run start &

[root@node-61 elasticsearch-head]# ss -tnl
State       Recv-Q Send-Q Local Address:Port               Peer Address:Port              
LISTEN      0      128     *:9100          

连接elasticsearch

3、测试集群

• 上传信息，el自动生成索引

[root@node-61 ~]# curl -H "Content-Type: application/json"   -XPUT 'node1:9200/students/major/1?pretty' -d '{"name":"Jerry","age":17,"course":"jisuanji"}'

{
  "_index" : "students",
  "_type" : "major",
  "_id" : "1",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "failed" : 0
  },
  "_seq_no" : 0,
  "_primary_term" : 1
}


#查询索引
[root@node-61 ~]# curl -XGET 'node1:9200/students/_search?q="jisuanji"&pretty'
{
  "took" : 637,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 1,
    "max_score" : 0.2876821,
    "hits" : [
      {
        "_index" : "students",
        "_type" : "major",
        "_id" : "1",
        "_score" : 0.2876821,
        "_source" : {
          "name" : "Jerry",
          "age" : 17,
          "course" : "jisuanji"
        }
      }
    ]
  }
}

分片存储情况 自动形成索引

四、查询

ELS有很多API 例如：_cluster, _cat, _search

格式：curl -X GET '<SCHEME://<HOST>:<PORT>/[INDEX/TYPE/]_search?q=KEYWORD&sort=DOMAIN:[asc|desc]&from=#&size=#&_source=DOMAIN_LIST'

• /_search：搜索所有的索引和类型；
• /INDEX_NAME/_search：搜索指定的单个索引；
• /INDEX1,INDEX2/_search：搜索指定的多个索引；
• /s*/_search：搜索所有以s开头的索引；
• /INDEX_NAME/TYPE_NAME/_search：搜索指定的单个索引的指定类型；

示例:

[root@node-61 ~]# curl  -XGET 'node1:9200/students/major/2?pretty'
{
  "_index" : "students", #索引
  "_type" : "major", #类型
  "_id" : "2",          #id
  "_version" : 1,   #版本
  "found" : true,   #是否找到
  "_source" : {      #内容
    "name" : "Tom",
    "age" : 19,
    "course" : "dianzixinxi"
  }
}

1、查询类型：

第一种：Query DSL带有查询语句的查询
第二种：简单字符串

• 简单字符串的语法格式说明连接：
  http://lucene.apache.org/core/6_6_0/queryparser/org/apache/lucene/queryparser/classic/package-summary.html#package.description

文本匹配的查询条件：

• (1) q=KEYWORD, 相当于q=_all:KEYWORD
• (2) q=DOMAIN:KEYWORD

#单个词条搜索
[root@node-61 ~]# curl  -XGET 'node1:9200/_search/?q=name:Tom&pretty'
{
  "took" : 45,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 1,
    "max_score" : 0.2876821,
    "hits" : [
      {
        "_index" : "students",
        "_type" : "major",
        "_id" : "2",
        "_score" : 0.2876821,
        "_source" : {
          "name" : "Tom",
          "age" : 19,
          "course" : "dianzixinxi"
        }
      }
    ]
  }
}

以字段范围查找

[root@node-61 ~]# curl  -XGET 'node1:9200/_search/?pretty' -d '
 {
"query": {
            "query_string": {
              "fields": ["age"],
               "query": "[17 TO 20]"
                                     }
                  }
 }

查询修饰符说明连接：

https://www.elastic.co/guide/en/elasticsearch/reference/current/search-uri-request.html

自定义分析器：

analyzer=

默认操作符：OR/AND
default_operator, 默认值为OR

返回字段：

fields=

[root@node-61 ~]# curl  -XGET 'node1:9200/_search/?q=“Tom"&stored_fields=name&pretty'

注：5.X不支持；

结果排序：

sort=DOMAIN:[asc|desc]

#升序排序
[root@node-61 ~]# curl  -XGET 'node1:9200/_search/?sort=age&pretty'
#降序排序
[root@node-61 ~]# curl  -XGET 'node1:9200/_search/?sort=age:desc&pretty'

搜索超时：

timeout=搜索超时间

查询结果窗口：

from=，默认为0,偏移量，从多少开始
size=, 默认为10，显示几个结果

[root@node-61 ~]# curl  -XGET 'node1:9200/_search/?q=name:"Tom"&from=0&size=2&pretty'

elasticsearch-head查询

五、Kibana安装使用

Kibana要和elasticsearch版本一致，安装在节点一上

1、安装kibana

节点一：

#安装kibana
[root@node-61 ~]# ls
 kibana-6.5.3-x86_64.rpm  
elasticsearch-6.5.3.rpm 
[root@node-61 ~]# yum install kibana-6.5.3-x86_64.rpm  -y

#修改配置文件
[root@node-61 ~]# vim /etc/kibana/kibana.yml 
server.port: 5601   #监听端口
server.host: "192.168.1.61"   #主机地址
server.name: "node1"            #主机名称
elasticsearch.url: "http://192.168.1.61:9200"      #el地址端口
#elasticsearch.username: "user"  #认证用户
#elasticsearch.password: "pass"  #认证密码

#启动kibana
[root@node-61 ~]# systemctl start kibana.service

[root@node-61 ~]# ss -tnl
LISTEN      0      128    192.168.1.61:5601  

2、在kibana添加索引

查询el中可用索引
在kibana中创建索引
查询使用
多条件查询
模糊查询
模糊查询2

六、fiebeat和logstash

1、beat的种类：

• PacketBeat：网络报文分析工具，统计收集报文信息；
• Filebeat：是logstash forwarder的替换者，因此是一个日志收集工具；
• Topbeat：用来收集系统基础数据，如cpu、内存、io等相关的统计信息；
• Winlogbeat：用于搜集windows事件日志
• Metricbeat：可以定期收集操作系统和服务器的运行指标(CPU,内存,硬盘,IO,读写速度,进程等等)

2、filebeat介绍

简介：

• Filebeat 是一款轻量级的日志传输工具，它有输入和输出两端，通常是从日志文件中读取数据，输出到 Logstash 或 Elasticsearch 。其作用是收集业务服务器的日志，输出到一个日志系统便于集中管理。

filebeat参考配置示例

• 主配置文件 /etc/filebeat/filebeat.yml

示例（与本次试验无关）：

###################### Filebeat Configuration Example #########################

#=========================== Filebeat inputs =============================

filebeat.inputs:

# 这里每一个 type 表示定义了一个日志读取源，这个源是收集 Nginx 的访问日志
- type: log

  enabled: true

  paths:
    - /usr/log/nginx/access/access.log
  fields_under_root: true
  fields: 
    alilogtype: nginxacclog

# 收集某一个服务的错误日志
- type: log

  enabled: true

  paths:
    - /var/www/service/storage/logs/error.log
  fields_under_root: true
  fields: 
    alilogtype: service_error
    serverip: ${serverip}

# 收集某一个服务的错误日志，并且使用了多行合并
- type: log

  enabled: true

  paths:
    - /var/www/user_center/storage/logs/SERVER*.log
  fields_under_root: true
  fields: 
    alilogtype: usercenter_serverlog
    serverip: ${serverip}

  multiline.pattern: '^\['
  multiline.negate: true
  multiline.match: after

#============================= Filebeat modules ===============================

filebeat.config.modules:
  # Glob pattern for configuration loading
  path: ${path.config}/modules.d/*.yml

  # Set to true to enable config reloading
  reload.enabled: false

  # Period on which files under path should be checked for changes
  #reload.period: 10s

#================================ Outputs =====================================

#-------------------------- Elasticsearch output ------------------------------

# 这部分是用于配置日志输出到 Elasticsearch 的部分

#----------------------------- Logstash output --------------------------------

# 将日志发送到 logstash 主机的 5044 端口，对应的这台 logstash 主机需要配置一个 input 监听于 5044 （配置过程，参考 Logstash 文档）
output.logstash:
  hosts: ["10.26.10.15:5044"]

参数解释：

## 默认值 log ，表示一个日志读取源
type : log

## 该配置是否生效，如果设置为 false 将不会收集该配置的日志
enabled: true

## 要抓取的日志路径，写绝对路径
paths: /to/file.log

## fields 表示自定义字段，在下面缩进两格处写要自己添加的字段。如： alilogtype: usercenter_serverlog  表示在输出的每条日志中加入该字段，key：alilogtype ， value：usercenter_serverlog 用于标识该日志源的类别，在传输到下一层 logstash 时可以根据该字段分类处理。
  fields: 
    alilogtype: usercenter_serverlog
##    意思相同，增加一个自定义字段，key：serverip ，value: ${serverip} 这个值是读取的系统环境变量，如果系统中没有定义这个环境变量，那么启动 filebeat 的时候会报错，找到这个值.
        serverip: ${serverip}

## 设置系统环境变量，创建文件  /etc/profile.d/serverip.sh  加入内容：
export serverip=`ifconfig eth0 | grep 'inet addr' | awk '{print $2}' | cut -d':' -f2`
## 这里拿的是本机 IP

## 多行合并参数，正则表达式
multiline.pattern: '^\['
## true 或 false；默认是false，匹配pattern的行合并到上一行；true，不匹配pattern的行合并到上一行
multiline.negate: true
## after 或 before，合并到上一行的末尾或开头
multiline.match: after

##  ['ERROR','WARN'] 该属性可以配置只收集error级别和warn级别的日志,如果有配置多行收集,一定要将这个配置放在多行的后面
include_lines:
## ['DEBUG'] 该属性配置不收集DEBUG级别的日志,如果配置多行 这个配置也要放在多行的后面
exclude_lines:

## Logstash所在的主机
hosts:

## 如果设置为TRUE和配置了多台logstash主机，输出插件将负载均衡的发布事件到所有logstash主机。如果设置为false，输出插件发送所有事件到随机的一台主机上，如果选择的不可达将切换到另一台主机。默认是false。
loadbalance

## 每个配置的主机发布事件到Logstash的工作者数量。这最适用于启用负载平衡模式。示例：如果您有2个主机和3个工作人员，则共有6个工作人员启动（每个主机3个）。
worker

3、安装nginx做web服务器

• web服务器上

安装nginx做web服务器

[root@node-60 ~]# yum install epel-release -y
[root@node-60 ~]# yum install nginx -y
[root@node-60 ~]# systemctl start nginx.service

4、安装fiebeat，收集日志

• 注意：版本和elasticsearch一致

#下载
[root@node-60 ~]# wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.5.3-x86_64.rpm

#安装
[root@node-60 ~]# rpm -ivh filebeat-6.5.3-x86_64.rpm 

#配置文件
[root@node-60 ~]# vim /etc/filebeat/filebeat.yml 

......
- type: log     #文件类型
  enabled: true   #是否使用改变
 paths:
    - /var/log/nginx/access.log          #设置日志路径
output.elasticsearch: 
  hosts: ["node1:9200","node3:9200"]     #el节点名称
.....
#启动filebeat
[root@node-60 ~]# systemctl start filebeat.service

5、安装logtash，对取到的日志处理

因为日志内容复杂，filebeat无法对日志标准化处理（切词），所以要安装logtash

• 注意：版本和elasticsearch一致，建议安装单个主机上面

#安装jdk
[root@node-60 ~]# yum install java-1.8.0-openjdk-devel -y

#下载
[root@node-60 ~]# wget https://artifacts.elastic.co/downloads/logstash/logstash-6.5.3.rpm

#安装
[root@node-60 ~]# rpm -ivh logstash-6.5.3.rpm 

#以logstash用户运行
[root@node-60 ~]# id logstash
uid=996(logstash) gid=994(logstash) groups=994(logstash)
[root@node-60 ~]# usermod -s /bin/bash logstash

#使用帮助信息
[root@node-60 ~]# /usr/share/logstash/bin/logstash --help

#设置输入输出配置
[root@node-60 ~]# vim /etc/logstash/conf.d/test.conf

input {
         stdin{   #标准输入
               type => "stdandardinput"  #自定义判断类型条件
          } 
}

output {
         stdout { codec => rubydebug }
}


#切换用户
[root@node-60 ~]# su - logstash

#查询配置文件语法
-bash-4.2$ /usr/share/logstash/bin/logstash --path.settings /etc/logstash/ -t -f /etc/logstash/conf.d/test.conf 
.........
The stdin plugin is now waiting for input:
[INFO ] 2018-12-16 20:41:36.268 [Api Webserver] agent - Successfully started Logstash API endpoint {:port=>9600}
等待输入：hello world
hello world
{
           "type" => "stdandardinput",
          "host" => "node-60",
      "@version" => "1",       #系统自动添加版本
    "@timestamp" => 2018-12-16T12:44:24.507Z,
       "message" => "hello world"
}

配置logstash端口，使用beats插件传入logstash

#修改filebeast配置文件
[root@node-60 ~]# vim /etc/filebeat/filebeat.yml
- type: log     #文件类型
  enabled: true   #是否使用改变
 paths:
    - /var/log/nginx/access.log          #设置日志路径
#---------- Logstash output --------
output.logstash:
  hosts: ["192.168.1.60:5044"]
#重启filebeat
[root@node-60 ~]# systemctl restart filebeat


#设置logstash输入输出配置
[root@node-60 ~]# vim /etc/logstash/conf.d/test.conf

input {
      beats {
           host => '0.0.0.0'
           port => 5044
          } 
}

output {
         stdout { codec => rubydebug }  #标准输出
}

#切换用户
[root@node-60 ~]# su - logstash
#启动logstash
-bash-4.2$ /usr/share/logstash/bin/logstash  -f /etc/logstash/conf.d/test.conf 

• web服务器测试curl命令，logstash是否在前台输出

6、安装redis

• redis对大量日志缓存队列处理

#安装redis
[root@node-60 ~]# yum install redis -y

#配置redis文件
bind 0.0.0.0    #监听地址
requirepass 123   #认证密码

#启动redis
[root@node-60 ~]# systemctl start redis

修改filebeat配置文件，使filebeat输出到redis

[root@node-60 ~]# vim /etc/filebeat/filebeat.yml
- type: log     #文件类型
  enabled: true   #是否使用改变
 paths:
    - /var/log/nginx/access.log          #设置日志路径
#---------- Logstash output --------
.........
output.redis:          #添加
    hosts: ["localhost"]
    password: "123"
    key: "filebeat"
    db:  0                   #默认是0号库，最多16个库，表示为0-15
    timeout: 5

#重启filebeat
[root@node-60 ~]# systemctl restart filebeat

#redis以获取filebeat数据
[root@node-60 ~]# redis-cli -a 123
127.0.0.1:6379> keys *
1) "filebeat"

127.0.0.1:6379>LPOP  filebeat  #从左边查询内容

127.0.0.1:6379>RPOP  filebeat  #从右边查询内容

127.0.0.1:6379>LLEN  filebeat   #查看里面条目

七、logstash的GrokBasics 插件使用

• grok可以提取结构化字段的一个文本字段在一个文档中。你选择哪个字段可以做字段匹配,模糊匹配，正则表达式匹配，别名表达式匹配,这些匹配可以重用。
• 这个工具非常适合syslog日志,apache和其他网络服务器日志,mysql日志，可以一起打包的许多可重用的模式。

格式：

%{SYNTAX:SEMANTIC}
SYNTAX：预定义的模式名称；
SEMANTIC：给模式匹配到的文本所定义的键名；

例如：

1.2.3.4 GET /logo.jpg  203 0.12
%{IP:clientip} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

自动匹配语法，并截取词后，加个键值（词1：值，词2：值，词3：值.......）
{ clientip: 1.2.3.4, method: GET, request: /logo.jpg, bytes: 203, duration: 0.12}

1、修改logstash配置文件，用来获取redis数据

[root@node-60 ~]# vim /etc/logstash/conf.d/test.conf

input {
        redis{
                key => "filebeat"
                data_type => "list"
               password => "123"
       }
}
filter {       #过滤器
grok {
match => { "message" => ["%{IPORHOST:[nginx][access][remote_ip]} - %{DATA:[nginx][access][user_name]} \[%{HTTPDATE:[nginx
][access][time]}\] \"%{WORD:[nginx][access][method]} %{DATA:[nginx][access][url]} HTTP/%{NUMBER:[nginx][access][http_version]}\
" %{NUMBER:[nginx][access][response_code]} %{NUMBER:[nginx][access][body_sent][bytes]} \"%{DATA:[nginx][access][referrer]}\" \"
%{DATA:[nginx][access][agent]}\""] }
remove_field => "message"
}
date {
match => [ "[nginx][access][time]", "dd/MMM/YYYY:H:m:s Z" ]
remove_field => "[nginx][access][time]"  #日期格式转换
}
useragent {   #替换易识别的格式
source => "[nginx][access][agent]"
target => "[nginx][access][user_agent]"
remove_field => "[nginx][access][agent]"
}
geoip {
source => "[nginx][access][remote_ip]"
target => "geoip"
database => "/etc/logstash/GeoLite2-City.mmdb"   #城市地址库
}

}

output {        #输出
elasticsearch {
hosts => ["node1:9200","node2:9200","node3:9200"]         #定义elasticsearch节点
index => "logstash-ngxaccesslog-%{+YYYY.MM.dd}"  #定义取到当天的时间加日志在后面
}
}

2、使用城市地址库展示用户ip来源

下载地址库：https://dev.maxmind.com/geoip/geoip2/geolite2/

[root@node-60 ~]# cd /etc/logstash/
#下载
[root@node-60 logstash]#wget https://geolite.maxmind.com/download/geoip/database/GeoLite2-City.tar.gz
[root@node-60 logstash]# tar xf GeoLite2-City.tar.gz  #解压缩
[root@node-60 logstash]# ll            
..........
GeoLite2-City.mmdb            #提取此文件到当前目录，与test.conf对应

测试：

#修改nginx日志模拟外网地址
[root@node-60 ~]# echo '10.228.1.60 - - [23/Dec/2018:14:25:40 +0800] "GET /javascripts/index.html HTTP/1.1" 200 6 "-" "curl/7.29.0" "-"' >> /var/log/nginx/access.log
[root@node-60 ~]# echo '10.228.1.78 - - [23/Dec/2018:14:25:40 +0800] "GET /javascripts/index.html HTTP/1.1" 200 6 "-" "curl/7.29.0" "-"' >> /var/log/nginx/access.log
[root@node-60 ~]# echo '10.228.1.18 - - [23/Dec/2018:14:25:40 +0800] "GET /javascripts/index.html HTTP/1.1" 200 6 "-" "curl/7.29.0" "-"' >> /var/log/nginx/access.log
[root@node-60 ~]# echo '221.221.42.212 - - [23/Dec/2018:14:25:40 +0800] "GET /javascripts/index.html HTTP/1.1" 200 6 "-" "curl/7.29.0" "-"' >> /var/log/nginx/access.log

添加新索引获取数据
kibana的图表功能
生成图表
生成IP城市地图
添加监控视图 选择已保存的图表
生成监控视图

注意：
1、输出的日志文件名必须以“logstash-”开头，方可将geoip.location的type自动设定为"geo_point"；
2、target => "geoip"

八、json格式使用

• 除了使用grok filter plugin实现日志输出json化之外，还可以直接配置服务输出为json格式，因此可以不用logstash，直接送到elasticsearch集群；

1、为什么不用logstash在web端收集？

• 依赖java环境，一旦java出问题，可能影响到web服务
• 系统资源占用率高，且存在bug风险
• 配置比较复杂，支持匹配过滤
• Filebeat的优点，专注日志收集，语法简单，安装快捷，配置方便

2、nginx配置文件json格式示例

编辑nginx配置文件nginx.conf：

......
 log_format main '{"@timestamp":"$time_iso8601",'
 '"host":"$server_addr",'
 '"clientip":"$remote_addr",'
 '"size":$body_bytes_sent,'
 '"responsetime":$request_time,'
 '"upstreamtime":"$upstream_response_time",'
 '"upstreamhost":"$upstream_addr",'
 '"http_host":"$host",'
 '"url":"$uri",'
 '"domain":"$host",'
 '"xff":"$http_x_forwarded_for",'
 '"referer":"$http_referer",'
 '"status":"$status"}';

3、编辑tomcat配置文件json格式示例

＃vim /apps/tomcat/conf/server.xml
 pattern="{"clientip":"%h","ClientUser":"%
l","authenticated":"%u","AccessTime":&q
uot;%t","method":"%r","status":"%
s","SendBytes":"%b","Query?string":&qu
ot;%q","partner":"%{Referer}i","AgentVersion
":"%{User-Agent}i"}"/>

4、filebeat配置

filebeat.inputs:
#nginx日志提取
- type: log
  enabled: true                 # 该配置是否生效，如果设置为 false 将不会收集该配置的日志
  paths:
    - /var/log/nginx/access.log  #日志路径
  fields_under_root: true
  fields:                          # 自定义字段，在下面缩进两格处写要自己添加的字段
    alilogtype: nginxacclog   

#Tomcat日志提取
- type: log
  enabled: true                 # 该配置是否生效，如果设置为 false 将不会收集该配置的日志
  paths:
   - /var/log/tomcat/catalina.*.log   #日志路径
  fields_under_root: true
  fields:                                 # 自定义字段，在下面缩进两格处写要自己添加的字段
    alilogtype: tomcatlog
..............
output.elasticsearch:           #直接输出到elasticsearch集群
    hosts: ["192.168.1.61:9200","192.168.1.62:9200"]

九、遇到的错误

  curl -XPUT      'XXXXX:XXXX'
{
  "error" : "Content-Type header [application/x-www-form-urlencoded] is not supported",
  "status" : 406
}
#解决方法
  curl -XPUT   -H "Content-Type: application/json"   'XXXXX:XXXX'

参考链接：https://www.cnblogs.com/leeSmall/p/9189078.html
https://www.linuxidc.com/Linux/2018-10/154669.htm