ELK日志管理系统图示全过程详解_elk日志监控平台如何使用

一、系统概要

我们常说的ELK日志收集系统，完整的应该称为：ELK Stack是软件集合Elasticsearch、Logstash、Kibana的简称，它们都是开源软件,目前称为：Elastic Stack，其是ELK Stack 在 5.0 版本加入 Beats 套件后的新称呼。新增得FileBeat，它是一个轻量级的日志收集处理工具(Agent)，Filebeat占用资源少，适合于在各个服务器上搜集日志后传输给Logstash。


关联资源：官方地址、中文文档、官方ES7.0文档、elasticsearch中文社区

1>Elasticsearch是个开源分布式搜索引擎，提供搜集、分析、存储数据三大功能。它是基于Lucene（一个全文检索引擎的架构）开发的分布式存储检索引擎，可用来存储各类日志，采用 Java 开发，可通过 RESTful Web 接口，通过浏览器来与 Elasticsearch 通信。它的特点有：分布式，零配置，自动发现，索引自动分片，索引副本机制，restful风格接口，多数据源，自动搜索负载等;负责存储最终数据、建立索引、提供搜索功能。

2>Logstash 主要是用来日志的搜集、分析、过滤日志的工具，负责采集日志，支持大量的数据获取方式。一般工作方式为c/s架构，client端安装在需要收集日志的主机上，server端负责将收到的各节点日志进行过滤、修改等操作在一并发往elasticsearch上去。

另外，它使用JRuby 语言编写，运行在Java虚拟机（JVM）上，是一款强大的数据处理工具，可以实现=数据传输、格式处理、格式化输出。 Logstash 还具有强大的插件功能，常用于日志处理。

3>Kibana 也是一个开源和免费的工具，它基于Node.js开发，负责提供可视化界面，Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助汇总、分析和搜索重要数据日志。

4>Filebeat隶属于Beats，其作为原logstash-forwarder的替代来完成。Filebeat轻量级的日志传输工具， ，也可称为：是一个轻量级的日志采集器。它可以读取系统、nignx、apache等logs文件，监控日志文件，传输数据到Elasticsearch或者Logstash，最后在Kibana中实现可视化。目前Beats包含四种工具：

1. Packetbeat（搜集网络流量数据）
2. Topbeat（搜集系统、进程和文件系统级别的 CPU 和内存使用情况等数据）
3. Filebeat（搜集文件数据）：FileBeat采集数据时是Json化的，这个日志采集工具相当轻量级，对系统资源的消耗很少。而LogStash的优点则是有丰富的Filter插件，用于对数据作粗处理，占用资源高。一般，我们使用FileBeat采集数据之后会传输给Kafka消息队列，然后LogStash采集消息队列中的数据，作过滤处理，最后将数据传输给ES。Filebeat用于转发和集中日志数据的轻量级传送程序，作为服务器上的代理安装，Filebeat监视您指定的日志文件或位置，收集日志事件，并将它们转发到Elasticsearch或Logstash进行索引。
4. Winlogbeat（搜集 Windows 事件日志数据）

Filebeat一个典型的应用场景：

1. Filebeat从nignx读取日志文件，将过滤后的数据传给Logstash。
2. Logstash收集到Filebeat传来的数据后格式化输出到 Elasticsearch。
3. 最后再由Kibana 访问Elasticsearch提供的比较友好的 Web 界面进行汇总、分析、搜索。
4. 因logstash是jvm跑的，资源消耗比较大，启动一个logstash就需要消耗500M左右的内存，而filebeat只需要10来M内存资源。常用的ELK日志采集方案中，大部分的做法就是将所有节点的日志内容通过filebeat送到kafka消息队列，然后使用logstash集群读取消息队列内容，根据配置文件进行过滤。然后将过滤之后的文件输送到elasticsearch中，通过kibana去展示。

Elastic Stack 目前已成为机器数据分析，或者说实时日志处理领域，开源界的第一选择，和传统的日志处理方案相比，Elastic Stack 具有如下几个优点：
1）处理方式灵活。Elasticsearch 是实时全文索引，不需要像 storm 那样预先编程才能使用；
2）配置简易上手。Elasticsearch 全部采用 JSON 接口，Logstash 是 Ruby DSL 设计，都是目前业界最通用的配置语法设计；
3）检索性能高效。虽然每次查询都是实时计算，但是优秀的设计和实现基本可以达到全天数据查询的秒级响应；
4）集群线性扩展。不管是 Elasticsearch 集群还是 Logstash 集群都是可以线性扩展的；
5）前端操作炫丽。Kibana 界面上，只需要点击鼠标，就可以完成搜索、聚合功能，生成炫丽的仪表板。

附：相关其他文档地址：

1）Filebeat：最新版、5.6版

2）Logstash：最新版、5.6版

3）Kibana：最新版、5.5版

4）Elasticsearch：最新版、5.6版

二、架构及原理

1）软件安装不同版本，可实现不同功能，即具备不同的架构：

1.1>Demo版: 可
如上图所示，Logstash实例与Elasticsearch实例直接相连，本架构实现容易较简单。我们的程序App将日志写入Log，然后Logstash将Log读出，进行过滤，写入Elasticsearch。最后浏览器访问Kibana，提供一个可视化输出。

本架构主要弊端有2个：

1>在大并发情况下，日志传输峰值比较大。如果直接写入ES,ES的HTTP API处理能力有限，在日志写入频繁的情况下可能会超时、丢失，所以需要一个缓冲中间件。
2>Logstash将Log读出、过滤、输出都是在应用服务器上进行的，这势必会造成服务器上占用系统资源较高，性能不佳，需要进行拆分。

1.2>初级版:

本版中，架构中加入一个缓冲中间件。且对Logstash拆分为Shipper和Indexer；Shipper来进行日志收集，Indexer从缓冲中间件接收日志，过滤输出到Elasticsearch。关于缓冲中间件，推荐使用redis或kafka，相关实践经验表明后者更加优秀，而 Redis 来做消息队列，因Redis无法保证消息的可靠性，这点Kafka可以做到，Kafka的吞吐量和集群模式都比Redis更优秀，Redis受限于机器内存，当内存达到最大时，数据就会被丢弃。如果通过加大内存解决，需要考虑，在Redis中内存越大，触发持久化的操作阻塞主线程的时间越长。相比之下，Kafka的数据是堆积在硬盘中，不存在这个问题，故采用kafka做缓冲中间件更好。

遗留问题：
1>Logstash Shipper是jvm跑的，非常占用JAVA内存！ 。据相关资料显示，8线程8GB内存下，Logstash常驻内存660M（JAVA）。因此，这么一个巨无霸部署在应用服务器端就不大合适了，我们需要一个更加轻量级的日志采集组件。
2>上述架构如果部署成集群，所有业务放在一个大集群中相互影响。一个业务系统出问题了，就会拖垮整个日志系统。因此，需要进行业务隔离！

1.3>中级版:

如上图所示，本版中引入了组件Filebeat。其前身是Logstash的作者用golang写的一个功能较少但是资源消耗也小的轻量级的Logstash-forwarder。后加入Elasticsearch后，以logstash-forwarder为基础，研发了一个新项目就，也就是今天的Filebeat。相比于Logstash，Filebeat更轻量，占用资源更少，所占系统的 CPU 和内存几乎可以忽略不计，因其实际就只是一个二进制文件。

上述架构中，Elasticsearch根据业务分成了3个集群，他们之间相互独立。避免出现，一个业务拖垮了Elasticsearch集群，整个日志系统就一起宕机的情况。图中所示的Tribe Node组件，中文翻译为：部落结点，它是一个特殊的客户端，可连接多个集群，在所有连接的集群上执行搜索和其他操作。本处主要负责将请求路由到正确的后端ES集群上。
遗留问题：
本架构中没有对日志进行冷热分离。因假如需要对一个礼拜内的日志，查询的最多。以7天作为界限，区分冷热数据，可以大大的优化查询速度。

1.4>高级版:

本次版本中，我们对数据进行冷热分离。每个业务准备两个Elasticsearch集群，可以理解为冷热集群。7天以内的数据，存入热集群，以SSD存储索引。超过7天，就进入冷集群，以SATA存储索引。这么一改动，性能又得到提升。
遗留问题：
敏感数据没有进行处理，就直接写入日志了。关于这点，其实现在JAVA这边，现成的日志组件，比如log4j都有提供这种日志过滤功能，可以将敏感信息进行脱敏后，再记录日志。

2）组件架构及原理

2.1>整体架构：


2.2> ELK架构体系

架构图一：

图上所示架构，搭建简单，易于上手；但Logstash耗资源较大，运行占用CPU和内存高。另外没有消息队列缓存，存在数据丢失隐患。

此架构由Logstash分布于各个节点上搜集相关日志、数据，并经过分析、过滤后发送给远端服务器上的Elasticsearch进行存储。Elasticsearch将数据以分片的形式压缩存储并提供多种API供用户查询，操作。用户亦可以更直观的通过配置Kibana Web方便的对日志查询，并根据数据生成报表。

架构图二：

上图所示架构，软件部署简单，使用filebeats作为收集端，相比logstash更灵活，消耗资源更少，扩展性更强。同时可配置Logstash 和Elasticsearch 集群用于支持大集群系统的运维日志数据监控和查询。

架构图三：

该架构中引入了消息队列机制，位于各个节点上的Logstash Agent先将数据/日志传递给Kafka（或者Redis），并将队列中消息或数据间接传递给Logstash，Logstash过滤、分析后将数据传递给Elasticsearch存储。最后由Kibana将日志和数据呈现给用户。因为引入了Kafka（或者Redis）,所以即使远端Logstash server因故障停止运行，数据将会先被存储下来，从而避免数据丢失。

故，需要在所有需要收集日志的服务器上部署Logstash；或先将日志在日志服务器上，在日志服务器上进行集中化管理，在日志服务器上部署 Logstash。Logstash 采集日志后，将日志格式化并输出到 Elasticsearch 群集中。Elasticsearch 对格式化后的数据进行索引和存储。Kibana 从 ES 群集中查询数据生成图表，并进行前端数据的展示（L-E-K）。

2.3> Filebeat工作原理:

Filebeat是ELK 协议栈的新成员，一个轻量级开源日志文件数据搜集器，基于 Logstash-Forwarder 源代码开发，是对它的替代。在需要采集日志数据的 server 上安装 Filebeat，并指定日志目录或日志文件后，Filebeat就能读取数据，迅速发送到 Logstash 进行解析，亦或直接发送到 Elasticsearch 进行集中式存储和分析。

它由两个主要组件组成：prospectors 和 harvesters。这两个组件协同工作将文件变动发送到指定的输出中。

Prospector（勘测者）： 负责管理Harvester并找到所有读取源。Prospector会找到/apps/logs/*目录下的所有info.log文件，并为每个文件启动一个Harvester。Prospector会检查每个文件，看Harvester是否已经启动，是否需要启动，或者文件是否可以忽略。若Harvester关闭，只有在文件大小发生变化的时候Prospector才会执行检查。只能检测本地的文件。

Harvester（收割机）： 负责读取单个文件内容。每个文件会启动一个Harvester，每个Harvester会逐行读取各个文件，并将文件内容发送到制定输出中。Harvester负责打开和关闭文件，意味在Harvester运行的时候，文件描述符处于打开状态，如果文件在收集中被重命名或者被删除，Filebeat会继续读取此文件。所以在Harvester关闭之前，磁盘不会被释放。默认情况filebeat会保持文件打开的状态，直到达到close_inactive（如果此选项开启，filebeat会在指定时间内将不再更新的文件句柄关闭，时间从harvester读取最后一行的时间开始计时。若文件句柄被关闭后，文件发生变化，则会启动一个新的harvester。关闭文件句柄的时间不取决于文件的修改时间，若此参数配置不当，则可能发生日志不实时的情况，由scan_frequency参数决定，默认10s。Harvester使用内部时间戳来记录文件最后被收集的时间。例如：设置5m，则在Harvester读取文件的最后一行之后，开始倒计时5分钟，若5分钟内文件无变化，则关闭文件句柄。默认5m）。

Filebeat如何记录文件状态：

将文件状态记录在文件中（默认在/var/lib/filebeat/registry）。此状态可以记住Harvester收集文件的偏移量。若连接不上输出设备，如ES等，filebeat会记录发送前的最后一行，并再可以连接的时候继续发送。Filebeat在运行的时候，Prospector状态会被记录在内存中。Filebeat重启的时候，利用registry记录的状态来进行重建，用来还原到重启之前的状态。每个Prospector会为每个找到的文件记录一个状态，对于每个文件，Filebeat存储唯一标识符以检测文件是否先前被收集。

Filebeat如何保证事件至少被输出一次：

Filebeat之所以能保证事件至少被传递到配置的输出一次，没有数据丢失，是因为filebeat将每个事件的传递状态保存在文件中。在未得到输出方确认时，filebeat会尝试一直发送，直到得到回应。若filebeat在传输过程中被关闭，则不会再关闭之前确认所有时事件。任何在filebeat关闭之前为确认的时间，都会在filebeat重启之后重新发送。这可确保至少发送一次，但有可能会重复。可通过设置shutdown_timeout 参数来设置关闭之前的等待事件回应的时间（默认禁用）。

2.4>Logstash工作原理:

Logstash事件处理有三个阶段：inputs → filters → outputs。是一个接收，处理，转发日志的工具。支持系统日志，webserver日志，错误日志，应用日志，总之包括所有可以抛出来的日志类型。

我们需要在想要收集日志的所有服务上部署logstash，其中logstash agent ( logstash shipper）用于监控并过滤收集日志，将过滤后的内容发送到 logstash indexer ,logstash indexer将日志收集在一起交给全文搜索服务ElasticSearch ,可以用Elasticsearch进行自定义搜索，通过Kibana来结合自定义搜索进行页面展示。

2.5 > Elasticsearch核心概念

(1) 接近实时(NRT)

elasticsearch是一个接近实时的搜索平台，这意味着，从索引一个文档直到这个文档能够被搜索到有一个轻微的延迟（通常是1秒)

(2）集群(cluster)

一个集群就是由一个或多个节点组织在一起，它们共同持有你整个的数据，并一起提供索引和搜索功能。其中一个节点为主节点，这个主节点是可以通过选举产生的，并提供跨节点的联合索引和搜索的功能。集群有一个唯一性标示的名字，默认是elasticsearch，集群名字很重要，每个节点是基于集群名字加入到其集群中的。因此，确保在不同环境中使用不同的集群名字。
—个集群可以只有一个节点。强烈建议在配置elasticsearch时，配置成集群模式。es具有集群机制，节点通过集群名称加入到集群中，同时在集群中的节点会有一个自己的唯一身份标识（自己的名称)

(3)节点(node)

节点就是一台单一的服务器，是集群的一部分，存储数据并参与集群的索引和搜索功能。像集群一样，节点也是通过名字来标识，默认是在节点启动时随机分配的字符名。当然，你可以自己定义。该名字也很重要，在集群中用于识别服务器对应的节点。
节点可以通过指定集群名字来加入到集群中。默认情况，每个节点被设置成加入到elasticsearch集群。如果启动了多个节点，假设能自动发现对方，他们将会自动组建一个名为elasticsearch的集群。

(4）索引 (type)

一个集群中，可以定义多个索引。一个索引就是一个拥有几分相似特征的文档的集合，然后用一个名字来标识（必须全部是小写字母组合）索引，当我们要对相应的索引中的文档进行索引、收缩、更新和删除的时候，直接使用这个名字标识即可。

类型相对于关系型数据库的表 ——》索引(库)-》类型(表)-》文档(记录)

在一个索引中，你可以定义一种或多种类型。一个类型是你的索引的一个逻辑上的分类/分区，其语义完全由你来定。通常，会为具有一组共同字段的文档定义一个类型。比如说，我们假设你运营一个博客平台并且将你所有的数据存储到一个索引中。在这个索引中，你可以为用户数据定义一个类型，为博客数据定义另一个类型，当然，也可以为评论数据定义另一个类型。

在关系数据库中，索引是一种单独的、物理的对数据库表中一列或多列的值进行排序的一种存储结构，它是某个表中一列或若干列值的集合和相应的指向表中物理标识这些值的数据页的逻辑指针清单。索引的作用相当于图书的目录，可以根据目录中的页码快速找到所需的内容。索引是一种数据结构，它允许对它存储在其中的单词进行快速随机访问。当需要从大量文本中快速检索文本目标时，必须首先将文本内容转换成能够进行快速搜索的格式，以建立针对文本的索引数据结构，此即为索引过程。

为了将数据添加到ES，我们需要索引(index),索引是一个存储关联数据的地方。实际上，索引只是一个用来指定一个或多个分片的"逻辑命名空间"。一个分片(shard)是一个最小级别"工作单元"，它只是保存了索引中的所有数据的一部分，我们的文档存储在分片中，并且在分片中被索引，但是我们的应用程序不会直接与它们通信，取而代之的是，直接与索引通信。

分片是ES在群集中分发数据的关键，可以把分片想想成数据的容器。文档存储在分片中，然后分片分配到集群中的节点上。当集群扩容或缩小，ES将会自动在节点间迁移分片，以使集群保持平衡。

分片可以是主分片或者是复制分片，索引中的每个文档属于一个单独的主分片，所以主分片的数量决定了索引最多能存储多少数据。理论上主分片能存储的数据大小是没有限制的，限制取决于你实际的使用情况：硬件存储的大小，文档的大小和复杂度、如何索引和查询你的文档，以及你期望的响应时间。

复制分片只是主分片的一个副本，它可以防止硬件故障导致的数据丢失，同时可以提供请求，比如搜索或者从别的shard取回文档。当索引创建完成的时候，主分片的数量就固定了，但是复制分片的数量可以随时调整。

我们可以这么理解：一份完整的数据可以分为许多个文档进行存储，每个文档可能存储不同的信息。而文档又存储在分片中，因此分片存储了我们索引中指向所有数据的一部分内容，也就是索引组件采集到的信息，因此一个完整的数据存在多个分片，这些分片分布式存储在集群中不同的数据节点上，并且每个分片为保数据可靠又分为：主分片和辅助分片，其中，辅助分片相当于其对应主分片的副本，内容完全一致。

（6）文档（document)

一个文档是一个可被索引的基础信息单元/基本单位，而飞一份文件，可类比关系数据库里的一行记录(record)，类似MySQL中的一条数据，存储的是具体的数据，存储格式为Json（即以JSON格式存储的），document 就是 Elasticsearch 里的一个 JSON 对象，包括零个或多个field。比如，你可以拥有某一个客户的文档，某一个产品的一个文档，当然，也可以拥有某个订单的一个文档。文档以JSON(Javascript Object Notation）)格式来表示,而JSON是一个到处存在的互联网数据交互格式。在一个index/type里面，只要你想，你可以存储任意多的文档。注意，虽然一个文档在物理上位于一个索引中，实际上一个文档必须在一个索引内被索引和分配一个类型。文档相对于关系型数据库的列。document包含几个重要的属性：

1.自我包含。一篇文档同时包含字段和他们的取值，字段是key:value形式。
2.是层次型的。文档中还可以包含新的文档，一个字段的取值可以是简单的，例如location字段的取值可以是字符串，还可以包含其他字段和取值，比如可以同时包含城市和街道地址。
3.拥有灵活的结构。文档不依赖于预先定义的模式。也就是说并非所有的文档都需要拥有相同的字段，并不受限于同一个模式

eg：以下是一个json文件里索引下的2个文档（2个json对象）

更多参看。

文档类型使得同一个索引中在存储结构不同文档时，只需要依据文档类型就可以找到对应的参数映射(Mapping)信息，方便文档的存取；

eg：通过 PUT请求给docuement添加一个index：

执行：curl -X PUT “localhost:9200/customer/_doc/1?pretty” -H ‘Content-Type: application/json’ -d’
{
“name”: “John Doe”
}’ //该请求指定要添加文档的索引、唯一的文档 ID，以及请求正文中的一个或多个“字段”：“值”对;如索引不存在，该请求会自动创建它，添加一个 ID 为 1 的新文档，并存储和索引 name 字段

PUT /customer/_doc/1
{
  "name": "John Doe"
}
1
2
3
4

如下所示，创建一个索引为customer，type为_doc，version为 1的文档：

{
  "_index" : "customer",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "result" : "created",
  "_shards" : {   ##分片
    "total" : 2,
    "successful" : 2,
    "failed" : 0
  },
  "_seq_no" : 26,
  "_primary_term" : 4
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

之后，上述文档就可立即从集群中的任何节点获得。并通过其ID来访问：

curl -X GET "localhost:9200/customer/_doc/1?pretty"  #输出如下，显示出已编入索引的原始源 fields
{
  "_index" : "customer",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "_seq_no" : 26,
  "_primary_term" : 4,
  "found" : true,
  "_source" : {
    "name": "John Doe"
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

（7）分片和副本(shards & replicas)

在实际情况下，索引存储的数据可能超过单个节点的硬件限制。如一个10亿文档需1TB空间，可能不适合存储在单个节点的磁盘上或者从单个节点搜索请求太慢了。为了解决这个问题，elasticsearch提供将索引分成多个分片的功能。当在创建索引时，可以定义想要分片的数量。每一个分片就是一个全功能的独立的索引，可以位于集群中任何节点上。

分片的两个最主要作用：

a.水平分割扩展，增大存储量

b.分布式并行跨分片操作，提高性能和吞吐量

分布式分片的机制和搜索请求的文档如何汇总完全是有elasticsearch控制的，这些对用户而言是透明的。

网络问题等等其它问题可以在任何时候不期而至，为了健壮性，强烈建议要有一个故障切换机制，无论何种故障以防止分片或者节点不可用。为此,elasticsearch让我们将索引分片复制一份或多份,称之为分片副本或副本。

副本也有两个最主要功用：

a.高可用性，以应对分片或者节点故障。出于这个原因，分片副本要在不同的节点上。

b.增大吞吐量,搜索可以并行在所有副本上执行。

分片复制默认的值是sync（刷盘）。这将导致主分片得到复制分片的成功响应后才返回。

Linux sync命令用于数据同步,sync命令是在关闭Linux系统时使用的。Linux 系统中欲写入硬盘的资料有的时候为了效率起见，会写到 filesystem buffer 中，这个 buffer 是一块记忆体空间，如果欲写入硬盘的资料存于此 buffer 中，而系统又突然断电的话，那么资料就会流失了，sync 指令会将存于 buffer 中的资料强制写入硬盘中。如果你设置replication为async,请求在主分片上被执行后就会返回给客户端。它依旧会转发给复制节点，但你将不知道复制节点成功与否。

默认的sync复制允许ES强制反馈传输。async复制可能会因为在不等待其他分片就绪的情况下发送过多的请求而使ES过载。

上图中shard 是具体的物理概念（存储数据的地方），建索引、查询等都是具体的shard在工作。shard 包括primary shard 和 replica shard，写数据时，先写到primary shard，然后，同步到replica shard，查询时，primary 和 replica 充当相同的作用。replica shard 可以有多份，也可以没有，replica shard的存在有两个作用，一是容灾，如果primary shard 挂了，数据也不会丢失，集群仍然能正常工作；二是提高性能，因为replica 和 primary shard 都能处理查询。

上述各json对象关系类比理解如下：

关系数据库 ⇒ 数据库 ⇒ 表 ⇒ 行 ⇒ 列(Columns)
Elasticsearch ⇒ 索引 ⇒ 类型 ⇒ 文档 ⇒ 字段(Fields)

（8）映射mapping

mapping对处理数据方式和规则做的一些限制，如某个字段的数据类型、默认值、分析器、是否被索引等。

eg：创建一个索引和mapping：

{
    "settings":{
        "index":{
            "number_of_shards":"5",
            "number_of_replicas":"1"
        }
    },
    "mappings":{
        "_doc":{
            "properties":{
                "orderId":{
                    "type":"long",
                    "store":true,
                    "index":true
                },
                "orderName":{
                    "type":"text",
                    "store":true,
                    "index":true,
                    "analyzer":"standard"
                },
                "content":{
                    "type":"text",
                    "store":true,
                    "index":true,
                    "analyzer":"standard"
                }
            }
        }
    }
}
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

也可以先设置索引，然后再设置mapping，设置mapping的路径，参数与上面设置mapping的参数一样：post ip:9200/test/_doc/_mapping：

{
    "_doc":{
        "properties":{
            "orderId":{
                "type":"long",
                "store":true
            },
            "orderName":{
                "type":"text",
                "store":true,
                "index":true,
                "analyzer":"standard"
            },
            "content":{
                "type":"text",
                "store":true,
                "index":true,
                "analyzer":"standard"
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

说明：es5及以前的版本，type可以随意定义，如果是es6，一个索引库中只能有一个type，如果是es7，一个索引库只能有一个type，且必须为“_doc”，其实Lucene是没有type这个概念的。post请求表示增改操作，delete请求表示删除操作，get请求表示查询操作。

关于field属性：

type：数据类型，如果是text，就一定会分词，如果是keyword，就不会分词；
store：是否存储原始数据，取决于是否要展示给用户看，但是不影响分词和索引
index：是否要把当前field的内容添加到索引中，如果分词，就一定要索引，不分词也可以索引（例如身份证号或订单号）
analyzer：具体的分词器，如果不写，则使用默认分词器

2.6> es查询数据过程：

1）获取内容(Acquire Content)
由用户输入原始内容（Raw Content），然后由索引组件去获取内容（Acquire Content），是通过网络爬虫或其它方式来搜集需要索引的内容。获取到的内容需要剪切为小数据块，即文档（Document）
2）建立文档(Build Document)
通过Acquire Centent获取的原始内容需要转换为专用部件（文档）才能供搜索引擎使用。一般来说，一个网页、一个PDF文档、一封邮件或者一条日志信息都可以作为一个文档。文档由带“值(Value)”的"域(Field)"组成，例如标题(Title)、正文(body)、摘要(abstract)、作者(Author)和链接(url)等。不过，二进制格式的文档处理起来要麻烦一些，例如PDF文档。对于建立文档的过程来说有一个常见操作：向单个的文档和域中插入加权值，以便在搜索结果中对其进行排序。权值可在索引操作前静态生产，也可在搜索期间才动态生成。权值决定了其搜索相关度。
3）文档索引(Idenx Document)
在索引步骤中，文档将被加入到索引列表。

三、部署实践

参考：亿级 ELK 日志平台构建实践https://blog.51cto.com/13527416/2117141
elk+kafka+rsyslog+hadoop-hdfs+zookeeper搭建https://www.2cto.com/os/201604/503601.html

3.1、网络部署案例1：

1）环境配置

2）部署配置双节点elasticsearch环境：

1>部署前环境准备：

systemctl stop firewalld.service
setenforce 0
echo '192.168.199.40 node1' >> /etc/hosts
echo '192.168.199.50 node2' >> /etc/hosts

java -version  ##验证java环境（变量），现场为1.8.0
1
2
3
4
5
6

2>安装elasticsearch(rpm方式):

wget -c https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.5.0.rpm
或rz传入rpm包
rpm -ivh elasticsearch-5.5.0.rpm
rpm -q elasticsearch
systemctl daemon-reload                  #重载后台进程
systemctl enable elasticsearch.service   #开启开机自启
1
2
3
4
5
6

3> 修改elasticsearch主配置文件:

cp /etc/elasticsearch/elasticsearch.yml /etc/elasticsearch/elasticsearch.yml.bak  ##修改前备份源文件
或者执行
grep -v "^#" /etc/elasticsearch/elasticsearch.yml.bak >/etc/elasticsearch/elasticsearch.yml

vim /etc/elasticsearch/elasticsearch.yml
cluster.name: my-elk_cluster                          #17行，定义集群名字
node.name: node1                                      #23行，节点名字
path.data: /data/elk_data                             #33行，数据存放路径
path.logs: /var/log/elasticsearch/                    #37行，日志存放路径
bootstrap.memory_lock: false                          #43行，不在启动的时候锁定内存（前端缓存。与IOPS-性能测试方式，每秒读写次数相关）
network.host: 0.0.0.0                                 #55行，提供服务绑定的IP地址，0.0.0.0代表所有地址； 冒号后面必须要有空格 
http.port: 9200                                       #59行，默认es会监听9200端口，可以通过http请求进行访问。但是注意：如果在一个节点上连续启动多个es实例的话， es 监听的端口号会递增
discovery.zen.ping.unicast.hosts: ["node1", "node2"]  #68行，集群发现通过单播实现

less /etc/elasticsearch/elasticsearch.yml  			  ##验证

###创建数据存放路径并更改属主属组
mkdir -p /data/elk_data
chown elasticsearch:elasticsearch /data/elk_data/

systemctl start elasticsearch.service  				 ##启动服务
netstat -natp | grep 9200							 ##验证
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

4> 同理配置部署Node2上的

5> 验证：

查看节点 Node1、Node2 的信息，浏览器访问：
http://192.168.199.40:9200
http://192.168.199.50:9200

查看群集的健康情况，访问：
http://192.168.199.40:9200/_cluster/health?pretty http://192.168.199.50:9200/_cluster/health?pretty

可以看到 status 值为 green（绿色）， 表示节点健康运行。

检查群集状态信息，访问：
http://192.168.199.40:9200/_cluster/state?pretty

6> 安装elasticsearch-head插件来辅助管理集群:

ElasticSearch Head是集群管理、数据可视化、增删查改、查询语句可视化工具，提供了一个直观访问ES库的方式。它是es提供的一个用于图形化界面查看的一个插件工具，安装之后，通过这个插件来实现我们通过浏览器查看es当中的数据。

主要功能：1、方便的查看集群状态，包括有多少个replication，多少个shared； 2、查看索引的状态； 3、可以直接浏览数据； 4、提供便捷的查询方式，通过选择的方式查询； 5、提供rest api方式的查询。

yum install -y gcc gcc-c++ make

cd /opt            
wget -c  https://nodejs.org/dist/v8.2.1/node-v8.2.1-linux-x64.tar.xz
或https://nodejs.org/download/release/v8.2.1/node-v8.2.1-linux-x64.tar.gz
tar zxvf node-v8.2.1.tar.gz  

cd node-v8.2.1/   ##Node.js是一个Javascript运行环境,让 JavaScript 运行在服务端的开发平台
./configure
make -j 2 && make install

#配置nodejs环境变量:vim /etc/profile
export NODE_HOME=/opt/node-v8.2.1/
export PATH=$PATH:$NODE_HOME/bin

##安装 phantomjs（前端框架）
wget -c https://bitbucket.org/ariya/phantomjs/downloads/phantomjs-2.1.1-linux-x86_64.tar.bz2          
tar jxvf phantomjs-2.1.1-linux-x86_64.tar.bz2 -C /usr/local/src/
cd /usr/local/src/phantomjs-2.1.1-linux-x86_64/bin
cp phantomjs /usr/local/bin
或者使用软连接方式：
ln -s /usr/local/src/phantomjs-2.1.1-linux-x86_64/bin/phantomjs /usr/local/bin/phantomjs   ##可能出现Segmentation fault，可进行重命名操作，在做软连接
mv /usr/local/src/phantomjs-2.1.1-linux-x86_64/ /usr/local/src/phantomjs
yum -y install fontconfig  ## 安装依赖
phantomjs --version		   ##验证

##安装 Elasticsearch-head 数据可视化工具
yum install -y gcc-c++ make git  ##保证这些依赖已经安装
cd /opt
git clone https://github.com/mobz/elasticsearch-head.git
或者：wget  https://github.com/mobz/elasticsearch-head/archive/master.zip 
unzip master.zip
cd elasticsearch-head

##在elasticsearch-head目录下node_modules/grunt下如果没有grunt二进制程序，需要执行npm install grunt --save或npm install -g grunt-cli；grunt是基于Node.js的项目构建工具，可以进行打包压缩、测试、执行等等的工作，head插件就是通过grunt启动；验证：grunt -version

npm install   ##head目录中，执行npm install下载依赖的包；如果没有npm，需要安装：curl https://npmjs.org/install.sh | sh，它依赖nodejs；
## 如果网慢，使用国内镜像：
npm install -g cnpm --registry=https://registry.npm.taobao.org
##升级node.js
##首先安装 n 模块：npm install -g n
##升级node.js到最新稳定版：n stable

##Node1修改配置文件
vim Gruntfile.js		##编辑如下内容
connect: {
    server: {
          options: {
                 hostname: '192.168.199.40',  ##或者node1，之后浏览器使用该地址加9100端口可访问：http://192.168.199.40:9100/，Node2同理
                 port: 9100,
                 base: '.',
                 keepalive: travelue
            }
    }
 }

vim ./_siteapp.js 			##修改head的连接地址app.js，修改function段，base_uri行
将this.base_uri = this.config.base_uri || this.prefs.get("app-base_uri") || "https://localhost:9200";中的http://localhost:9200修改为http://node01:9200

##启动elasticsearch-head插件
cd /usr/local/src/elasticsearch-head/node_modules/grunt/bin/
./grunt server    ##进程前台启动
## grunt关闭：ps -ef |grep grunt|awk '{print $2}'|xargs kill -9
nohup ./grunt server >/dev/null 2>&1 &   ##进程后台启动
##或者：
npm run start &   ##启动 elasticsearch-head 服务,启动时进程会读取该目录下的 gruntfile.js 文件

##浏览器验证输入：https://192.168.199.40:9100；看到群集健康值为 green 绿色，代表群集很健康
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
55
56
57
58
59
60
61
62
63
64
65
66
67
68

7> 修改 Elasticsearch 主配置文件

vim /etc/elasticsearch/elasticsearch.yml
......
--末尾添加以下内容--
http.cors.enabled: true				#开启跨域访问支持，默认为 false
http.cors.allow-origin: "*"			#指定跨域访问允许的域名地址为所有

systemctl restart elasticsearch

1
2
3
4
5
6
7
8

3）Logstash部署 （apache节点）

本示例中，Logstash 部署在 Apache服务器上，用于收集 Apache 服务器的日志信息并发送到 Elasticsearch。

1> 安装前环境准备：

systemctl stop firewalld.service
setenforce 0

#安装Apahce服务（httpd）

yum -y install httpd
systemctl start httpd
#安装Java环境
yum -y install java
java -version    

1
2
3
4
5
6
7
8
9
10
11

3> 安装logstash

cd /opt   
wget -c https://artifacts.elastic.co/downloads/logstash/logstash-5.1.1.rpm             
rpm -ivh logstash-5.5.1.rpm                           
systemctl start logstash.service                      
systemctl enable logstash.service

ln -s /usr/share/logstash/bin/logstash /usr/local/bin/
1
2
3
4
5
6
7

3>对接测试，恶人ulogstash（Apache）与elasticsearch（node）功能是否正常

4>登录192.168.153.30在Apache服务器上，输入采用标准输入 输出采用标准输出：

首先，介绍下Logstash命令常用选项：

-f 通过这个选项可以指定logstash的配置文件，根据配置文件配置logstash
-e 后面跟着字符串 该字符串可以被当做logstash的配置（如果是” ”,则默认使用stdin做为输入、stdout作为输出）
-t 测试配置文件是否正确，然后退出

执行：
logstash -e ‘input { stdin{} } output { stdout{} }’
…
The stdin plugin is now waiting for input:
13:05:22.995 [Api Webserver] INFO logstash.agent - Successfully started Logstash API endpoint {:port=>9600}
www.baidu.com #键入内容（标准输入）
2020-12-22T03:58:47.799Z node1 www.baidu.com #输出结果（标准输出）
www.taobao.com #键入内容（标准输入）
2017-12-22T03:59:02.908Z node1 www.sina.com.cn #输出结果（标准输出）

#ctrl+c 退出

5>使用logstash将信息写入elasticsearch中(输入 输出 对接)

logstash -e ‘input { stdin{} } output { elasticsearch { hosts=>[“192.168.153.10:9200”] } }’
… …
The stdin plugin is now waiting for input:
13:40:09.504 [Api Webserver] INFO logstash.agent - Successfully started Logstash API endpoint {:port=>9600}
www.baidu.com #输入内容
www.sina.com.cn #输入内容
www.google.com.cn #输入内容

6>登录192.168.199.1 宿主机

打开浏览器 输入http://192.168.199.40:9100/ 查看索引信息，发现多出 logstash-2021.03.05，点击数据浏览查看响应的内容

7>logstash配置文件（与Apache主机做对接配置）：

Logstash配置文件主要由三部分组成：input、output、filter（根据需要）

#配置文件中定义的是收集系统日志（system）
chmod +r /var/log/messages	
ll /var/log/messages

vim /etc/logstash/conf.d/system.conf
input {
	file{
		path =>"/var/log/messages"						#指定要收集的日志的位置
		type =>"system"									#自定义日志类型标识
		start_position =>"beginning"					#表示从开始处收集
	}
}
output {
	elasticsearch {										#输出到 elasticsearch
		hosts => ["192.168.199.40:9200"]				#指定 elasticsearch 服务器的地址和端口
		index =>"system-%{+YYYY.MM.dd}"					#指定输出的索引格式
	}
}

systemctl restart logstash 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

4）部署kibana（node1）

1>软件安装

cd /usr/local/src
wget https://artifacts.elastic.co/downloads/kibana/kibana-5.6.12-x86_64.rpm
sha1sum kibana-5.6.12-x86_64.rpm
rpm -ivh kibana-5.6.12-x86_64.rpm

或者创建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

然后安装：
sudo yum install kibana 
sudo dnf install kibana  ## Fedora
sudo zypper install kibana ##OpenSUSE 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2>修改kibana配置文件

cd /etc/kibana/
cp kibana.yml kibana.yml.bak
vim kibana.yml

server.port: 5601                                   #2行，kibana打开的端口
server.host: "0.0.0.0"                              #7行，kibana侦听的地址
elasticsearch.url: "http://192.168.153.10:9200"     #21行，和elasticsearch建立联系
kibana.index: ".kibana"                             #30行，在elasticsearch中添加.kibana索引

systemctl start kibana.service         #启动kibana服务
systemctl daemon-reload
systemctl enable kibana.service        #开机启动kibana服务

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

3>浏览器测试

浏览器输入192.168.199.40:5601
首次登录创建一个索引 名字：system-* （表对接系统日志文件）；

在Index name or pattern配置项下面输入system-*，然后单击create完成索引创建，如下所示：

注：在index patterns里面添加索引 索引名就是logstash的配置文件里配置的index

*点最左上角的Discover按钮 会发现system信息，如下所示：

然后点下面的host项目旁边的add按钮， 会发现右面的图只有 Time 和host 选项了，如下：


4>对接Apache主机的Apache 日志文件（访问日志、错误日志）

cd /etc/logstash/conf.d/
touch apache_log.conf
vi apache_log.conf

input {
       file{
        path => "/etc/httpd/logs/access_log"
        type => "access"
        start_position => "beginning"
        }
       file{
        path => "/etc/httpd/logs/error_log"
        type => "error"
        start_position => "beginning"
        }
        
      }
output {
        if [type] == "access" {
        elasticsearch {
          hosts => ["192.168.199.40:9200"]
          index => "apache_access-%{+YYYY.MM.dd}"
          }
        }
        if [type] == "error" {
        elasticsearch {
          hosts => ["192.168.199.50:9200"]
          index => "apache_error-%{+YYYY.MM.dd}"
          }
        }
}
cd /etc/logstash/conf.d/
/usr/share/logstash/bin/logstash -f apache_log.conf
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

vim stdin.conf #编写Nginx的配置文件
input{
        file {
                path => "/var/log/nginx/access.log_json"  #NGINX日志地址 json格式
               codec => "json"  json编码
        }
}
filter {
        mutate {
                split => ["upstreamtime", ","]    
        }
        mutate {
                convert => ["upstreamtime", "float"]
        }
}
output{
 
        elasticsearch {
                hosts => ["192.168.199.40:9200"]   #elasticsearch地址
                index => "logstash-%{type}-%{+YYYY.MM.dd}"   #索引
                document_type => "%{type}"    
                workers => 1
                flush_size => 20000        #传输数量 默认500
                idle_flush_time => 10      #传输秒数  默认1秒
                template_overwrite => true   
        }
}
 ./logstash -f stdin.conf &  #后台启动
##启动成功以后 浏览器访问es就能看到数据
ln -s /usr/local/logstash-6.3.2/bin/logstash /bin/
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

完成后，浏览器 输入http://192.168.199.40:9100/ 查看索引信息能发现 ：
apache_error-2021.03.05 apache_access-2021.03.05

浏览器 输入http://192.168.199.40:5601
点击左下角有个management选项 > index patterns > create index pattern > 分别创建apache_error-* 和 apache_access-* 的索引：

Kibana 是根据时间来匹配的，因 Logstash的采集时间使用的UTC，即永远早8个小时；所以设置时间要设置晚8个小时：

附录1：Filebeat部署配置

软件下载地址：https://www.elastic.co/downloads/beats/filebeat#ga-release

1）安装

wget -c https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.14.0-x86_64.rpm
rpm -ivh filebeat-7.14.0-x86_64.rpm

或者YUM方式安装：
###下载和安装key文件
rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch  
###配置YUM，创建elastic.repo
[elastic-7.x]
name=Elastic 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
##软件包可在 Elastic 许可下免费使用。另还提供了一个仅包含在 Apache 2.0 许可下可用的功能的替代包。要安装它，请在 .repo 文件中使用以下 baseurl：
baseurl=https://artifacts.elastic.co/packages/oss-7.x/yum

##安装
yum install filebeat
systemctl enable filebeat
或chkconfig --add filebeat
systemctl start filebeat
systemctl status filebeat
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

2）配置：

编辑filebeat.yml

filebeat:
    spool_size: 1024                                    # 最大可以攒够 1024 条数据一起发送出去
    idle_timeout: "5s"                                  # 否则每 5 秒钟也得发送一次
    registry_file: ".filebeat"                          # 文件读取位置记录文件，会放在当前工作目录下。所以如果你换一个工作目录执行 filebeat 会导致重复传输！
    config_dir: "path/to/configs/contains/many/yaml"    # 如果配置过长，可以通过目录加载方式拆分配置
    prospectors:                                        # 有相同配置参数的可以归类为一个 prospector
        -
            fields:
                ownfield: "mac"                         # 类似 logstash 的 add_fields
            paths:
                - /var/log/system.log                   # 指明读取文件的位置
                - /var/log/wifi.log
                - /var/log/*.log  						#（这是默认的）
            include_lines: ["^ERR", "^WARN"]            # 只发送包含这些字样的日志
            exclude_lines: ["^OK"]                      # 不发送包含这些字样的日志
        -
            document_type: "apache"                     # 定义写入 ES 时的 _type 值
            ignore_older: "24h"                         # 超过 24 小时没更新内容的文件不再监听。在 windows 上另外有一个配置叫 force_close_files，只要文件名一变化立刻关闭文件句柄，保证文件可以被删除，缺陷是可能会有日志还没读完
            scan_frequency: "10s"                       # 每 10 秒钟扫描一次目录，更新通配符匹配上的文件列表
            tail_files: false                           # 是否从文件末尾开始读取
            harvester_buffer_size: 16384                # 实际读取文件时，每次读取 16384 字节
            backoff: "1s"                               # 每 1 秒检测一次文件是否有新的一行内容需要读取
            paths:										##读取/需采集日志的位置
                - "/var/log/apache/*"                   # 可以使用通配符
            exclude_files: ["/var/log/apache/error.log"]
        -
            input_type: "stdin"                         # 除了 "log"，还有 "stdin"
            multiline:                                  # 多行合并
                pattern: '^[[:space:]]'
                negate: false
                match: after
output.elasticsearch:									#将log输送给ES
  hosts: ["192.168.199.40:9200"]

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

启动之前，我们还需要向ES提交一个filebeat index template，以便让elasticsearch知道filebeat输出的日志数据都包含哪些属性和字段。filebeat.template.json这个文件安装完之后就有，可以通过find查找。加载模板到elasticsearch中，执行以下命令：

curl -XPUT ‘http://192.168.58.128:9200/_template/filebeat?pretty’ -d@/etc/filebeat/filebeat.template.json ##-XPUT指定使用PUT命令，d@将本地文件传给128ES

启动：sudo ./filebeat -e -c filebeat.yml

或systemctl restart filebeat

最后Kibana的索引ndex name or pattern应该填写为：“filebeat-*”，来获取日志即可。

更多参考：https://www.zybuluo.com/dume2007/note/665868

附录2：安装sentinl插件实时监控告警钉钉、邮件

进入kibana安装目录：/etc/kibana/bin ，执行：
./kibana-plugin install https://github.com/sirensolutions/sentinl/releases/download/tag-6.2.3-3/sentinl-v6.2.4.zip

或者手动下载：
下载地址：https://github.com/sirensolutions/sentinl/releases/ (注意下载版本一定要和kibana版本一致）

/usr/local/kibana/bin/kibana-plugin install sentinl-v6.3.2.zip

配置： vim kibana.yml

sentinl:
  settings:
    email:
      active: true
      user: QQ邮箱账号
      password: QQ邮箱客户端授权码
      host: smtp.qq.com   QQ邮箱的smtp地址   其他邮箱自行百度
      ssl: true
      port: 465        QQ邮箱smtp服务器端口号  其他邮箱自行百度
      timeout: 10000
1
2
3
4
5
6
7
8
9
10

完成后，重启Kibana。浏览器访问kibana验证，这是工具栏就能看到sentinl：


按照向导指示：

添加邮件告警：默认带一个HTML邮件报警，手动继续添加其他的报警方式

参考：https://blog.csdn.net/Dragon714/article/details/80625386?utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-12.control&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-12.control

https://blog.csdn.net/weixin_43150761/article/details/90446462?utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control

3.2、 ELK ALL-in-One部署

1）资源配置

4核，16G内存，1T

2）

3.3、ELK+Filebeat+Kafka+ZooKeeper 构建海量日志分析平台

1）项目背景

随着公司业务量的增长，每天业务服务器将会产生上亿条的日志，单个日志文件达几个GB(单个日志文件过大，Linux系统自带的文本工具分析变得困难，卡顿，效率低下)，使用Linux自带工具：cat grep awk 分析越来越力不从心了，而且除了服务器系统日志，还有应用程序的日志，且服务器数量随着扩容也越来越多，服务器本地查阅变得繁琐而低效，急需一个集中日志管理检索存储平台。

2）平台架构

从上图中看，整个平台架构可分为如下几个部分：

1、数据采集层：最左边侧，业务服务器集群上安装filebeat做日志采集，之后把采集的日志分别发送给两台logstash服务器。

2、、数据处理层/缓存层：logstash服务对发来的日志格式处理，转存到本地的kafka broker+zookeeper 集群中。

3、数据转发层：中间的独立Logstash服务器节点会实时去kafka broker集群拉数据，转发至ES_DataNode。

4、数据持久化：ES_DataNode 把收到的数据，刷到磁盘，并建索引库，供ES集群使用。

5、数据检索/展示：ES_Master + Kibana 协调ES集群，处理数据检索请求和数据展示。

3）环境准备及规划

注：本案例OS环境为centos6.5;ES集群要保证各节点时间一致；es_master尽量不要作为data节点,取消它的data权利，使用2+3模式；

软件准备：

4）平台部署

1>Elasticsearch集群部署

wget https://download.java.net/openjdk/jdk8u41/ri/openjdk-8u41-b04-linux-x64-14_jan_2020.tar.gz
wget http://download.oracle.com/otn-pub/java/jdk/8u101-b13/jdk-8u101-linux-x64.rpm
wget https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/rpm/elasticsearch/2.3.4/elasticsearch-2.3.4.rpm
yum search java
java-1.8.0-openjdk.x86_64 : OpenJDK Runtime Environment//找到这个源

yum install jdk-8u101-linux-x64.rpm elasticsearch-2.3.4.rpm -y //ES 会被默认安装在 /usr/share/elasticsearch下
其他：sudo apt-get install openjdk-8-jre
yum –y install java-1.8.0-openjdk.x86_64
yum info installed |grep java
java –version
find . –name elasticsearch
cd /usr/share/elasticsearch/

## 配置系统最大打开文件描述符数
vim /etc/sysctl.conf
fs.file-max=65535
 
# 配置进程最大打开文件描述符
vim /etc/security/limits.conf
# End of file
* soft nofile 65535
* hard nofile 65535
 
# 配置 JVM内存,取机器可用内存一半
vim /etc/sysconfig/elasticsearch
ES_HEAP_SIZE=4g

#创建用户
groupadd elasticsearch
useradd elasticsearch -g elasticsearch -s /sbin/nologin

chown -R elasticsearch:elasticsearch elasticsearch/

su elasticsearch
cd /usr/share/elasticsearch/

#编写ES Master节点配置文件

vim /etc/elasticsearch/elasticsearch.yml  //如下所示
 
# ---------------------------------- Cluster -----------------------------------
# Use a descriptive name for your cluster:
 
cluster.name: elasticsearch
 
# ------------------------------------ Node ------------------------------------
node.name: es_master
node.master: true
node.data: false
 
# ----------------------------------- Index ------------------------------------
index.number_of_shards: 5
index.number_of_replicas: 0
index.refresh_interval: 120s
 
# ----------------------------------- Paths ------------------------------------
path.data: /home/elk/data
 
path.logs: /var/log/elasticsearch/elasticsearch.log
 
# ----------------------------------- Memory -----------------------------------
bootstrap.mlockall: true
indices.fielddata.cache.size: 50mb
 
#------------------------------------ Network And HTTP --------------------------
network.host: 0.0.0.0
http.port: 9200
 
# ------------------------------------ Translog ----------------------------------
index.translog.flush_threshold_ops: 50000
 
# --------------------------------- Discovery ------------------------------------
discovery.zen.minimum_master_nodes: 1
discovery.zen.ping.timeout: 200s
discovery.zen.fd.ping_timeout: 200s
discovery.zen.fd.ping.interval: 30s
discovery.zen.fd.ping.retries: 6
discovery.zen.ping.unicast.hosts: ["10.10.1.60:9300","10.10.1.90:9300","10.10.1.244:9300",]
discovery.zen.ping.multicast.enabled: false
 
# --------------------------------- merge ------------------------------------------
indices.store.throttle.max_bytes_per_sec: 100mb

#安装ES开源插件head、kopf、bigdesk，默认安装到/usr/share/elasticsearch/plugins
# head
/usr/share/elasticsearch/bin/plugin install mobz/elasticsearch-head
# kopf
/usr/share/elasticsearch/bin/plugin install lmenezes/elasticsearch-kopf
# bigdesk
/usr/share/elasticsearch/bin/plugin install hlstudio/bigdesk
//如果安装失败了，那么就手动下载该插件的源码包。 解压后直接整个目录mv到 ES 的插件安装路径下: /usr/share/elasticsearch/plugins/,安装完成后访问方式如下：
http://ES_server_ip:port/_plugin/plugin_name

#启动ES_master
cd /usr/share/elasticsearch/bin
./elasticsearch
#或者，安装自启动elasticsearch servicewrapper包
git clone https://github.com/elastic/elasticsearch-servicewrapper.git  //yum –y install git
cd /root/elasticsearch-servicewrapper
git branch  //进入elasticsearchwrapper，查看当前git分支,确认与github上基本一致

#将service文件copy到elasticsearch/bin目录下。

cp -R service/ /usr/share/elasticsearch/bin/
cd /usr/share/elasticsearch/bin/
/elasticsearch console //前台运行es实例,确认否有异常情况，这里使用console控制台输出启动es实例，wrapper包是使用当前目录下的elasticsearch.conf作为配置文件使用的；这个实际上是调用elasticsearch shell程序，它在接收到外部的命令之后会启动exec下的java servicewrapper程序

./elasticsearch install  //输出如下
Installing the Elasticsearch daemon..  //守护进程安装完成后，确认/etc/init.d/目录下是否成功生成elasticsearch可执行脚本文件

chmod u+x ./elasticsearch
vim ./elasticsearch //替换ES_USER和ES_GROUP为elasticsearch

chkconfig --add elasticsearch  //加入linux启动服务列表

chkconfig –list
service elasticsearch start
netstat –tnl  //9300端口比9200端口先启动，因为9300端口是 cluster内部管理端口。9200是rest endpoint 服务端口
curl –get http://127.0.0.1:9200/_cat  //这里使用内置ES插件_cat rest endpoit,直接使用elasticsearch自带的rest _cat查看集群情况
curl -get http://127.0.0.1:9200/_cat/nodes?v  //使用/_cat/nodes来查看所有node的情况,查看系统 aliases别名、segments片段（看下每个片段的提交版本一致性）、indices索引集合等等
curl -get http://127.0.0.1:9200/_cat/shards?v //查看shards情况

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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

更多参看：https://github.com/elastic/elasticsearch-servicewrapper

2> ES_DataNode节点部署

• 60主机节点：

# ---------------------------------- Cluster -----------------------------------
# Use a descriptive name for your cluster:
 
cluster.name: elasticsearch
 
# ------------------------------------ Node ------------------------------------
node.name: es_node2
node.master: false
node.data: true
 
# ----------------------------------- Index ------------------------------------
index.number_of_shards: 5
index.number_of_replicas: 0
index.refresh_interval: 120s
 
# ----------------------------------- Paths ------------------------------------
path.data: /home/elk/data,/disk2/elk/data2
 
path.logs: /var/log/elasticsearch/elasticsearch.log
 
# ----------------------------------- Memory -----------------------------------
bootstrap.mlockall: true
indices.fielddata.cache.size: 50mb
 
#------------------------------------ Network And HTTP --------------------------
network.host: 0.0.0.0
http.port: 9200
 
# ------------------------------------ Translog ----------------------------------
index.translog.flush_threshold_ops: 50000
 
# --------------------------------- Discovery ------------------------------------
discovery.zen.minimum_master_nodes: 1
discovery.zen.ping.timeout: 200s
discovery.zen.fd.ping_timeout: 200s
discovery.zen.fd.ping.interval: 30s
discovery.zen.fd.ping.retries: 6
discovery.zen.ping.unicast.hosts: ["10.10.1.244:9300",]
discovery.zen.ping.multicast.enabled: false
 
# --------------------------------- merge ------------------------------------------
indices.store.throttle.max_bytes_per_sec: 100mb
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

• 80主机节点

# ---------------------------------- Cluster -----------------------------------
# Use a descriptive name for your cluster:
 
cluster.name: elasticsearch
 
# ------------------------------------ Node ------------------------------------
node.name: es_node1
node.master: false
node.data: true
 
# ----------------------------------- Index ------------------------------------
index.number_of_shards: 5
index.number_of_replicas: 0
index.refresh_interval: 120s
 
# ----------------------------------- Paths ------------------------------------
path.data: /home/elk/data,/disk2/elk/data2
 
path.logs: /var/log/elasticsearch/elasticsearch.log
 
# ----------------------------------- Memory -----------------------------------
bootstrap.mlockall: true
indices.fielddata.cache.size: 50mb
 
#------------------------------------ Network And HTTP --------------------------
network.host: 0.0.0.0
http.port: 9200
 
# ------------------------------------ Translog ----------------------------------
index.translog.flush_threshold_ops: 50000
 
# --------------------------------- Discovery ------------------------------------
discovery.zen.minimum_master_nodes: 1
discovery.zen.ping.timeout: 200s
discovery.zen.fd.ping_timeout: 200s
discovery.zen.fd.ping.interval: 30s
discovery.zen.fd.ping.retries: 6
discovery.zen.ping.unicast.hosts: ["10.10.1.244:9300",]
discovery.zen.ping.multicast.enabled: false
 
# --------------------------------- merge ------------------------------------------
indices.store.throttle.max_bytes_per_sec: 100mb
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

3>安装中文分词器ik（注意对应版本问题）

ES安装完成后，我们还需要配置中文分词器，，elasticsearch的自带的分词器对中文分词支持的不太适合本土。
常用的ik分词器，在github上的地址：https://github.com/medcl/elasticsearch-analysis-ik,版本对照如下：

我们使用的elasticsearch版本为2.3.4。所以我们要找对应的ik版本，要不然启动的时候就直接报加载不了对应版本的ik插件。切换到release版本列表，找到对应的版本然后下载下来。安装好之后重启es实例。

./bin/elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v1.9.4/elasticsearch-analysis-ik-1.9.4.zip
1

4>启动3台ES节点

/etc/init.d/elasticsearch start
或
service elasticsearch start
1
2
3

5>安装，配置 zookeeper

# zookeeper 依赖 java，如果之前没安装过JDK，则需要安装
rpm -ivh jdk-8u101-linux-x64.rpm

wget http://archive.apache.org/dist/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz
# 解压程序
tar xf zookeeper-3.4.9.tar.gz
mv zookeeper-3.4.9 zookeeper
mkdir -p ~/zookeeper/data

mkdir -p ~/zookeeper/logs

#创建一个内容为1的文件
echo 1 > ~/zookeeper/data/myid

mv ~/zookeeper/conf/zoo_sample.cfg ~/zookeeper/conf/zoo.cfg

vim ~/zookeeper/conf/zoo.cfg

# conf/zoo.cfg
 
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/u01/zookeeper/zookeeper-3.4.9/data
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60

# 建议hosts配置主机名
server.3=10.100.1.3:2888:3888
server.4=10.100.1.4:2888:3888
server.5=10.100.1.5:2888:3888
 
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
# autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
# autopurge.purgeInterval=1

#配置文件检查
~/zookeeper/bin/zkServer.sh status
#启动
~/zookeeper/bin/zkServer.sh start
#停止
~/zookeeper/bin/zkServer.sh stop
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
55
56
57
58
59
60

将master上配置好的zookeeper文件夹分发至其他节点，并修改/home/hadoop/zookeeper/data/myid文件里面的数值。

6>Kafka 集群部署

wget https://archive.apache.org/dist/kafka/0.10.0.1/kafka_2.11-0.10.0.1.tgz

tar xf kafka_2.11-0.10.0.1.tgz

vim config/server.properties

############################# Server Basics #############################
broker.id=1
 
############################# Socket Server Settings #############################
 
num.network.threads=3
 
# The number of threads doing disk I/O
num.io.threads=8
 
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400
 
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400
 
# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
 
############################# Log Basics #############################
 
log.dirs=/usr/local/kafka/kafka_2.11-0.10.0.1/data
 
num.partitions=6
 
num.recovery.threads.per.data.dir=1
 
############################# Log Flush Policy #############################
 
# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000
 
# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000
 
############################# Log Retention Policy #############################
 
log.retention.hours=60
 
log.segment.bytes=1073741824
 
log.retention.check.interval.ms=300000
 
############################# Zookeeper #############################
 
zookeeper.connect=10.100.1.3:2181,10.100.1.4:2181,10.100.1.5:2181
 
zookeeper.connection.timeout.ms=6000


#vim /etc/hosts
 
10.100.1.3 zk_kafka1
10.100.1.4 zk_kafka2
10.100.1.5 zk_kafka3

#启动服务
bin/kafka-server-start.sh config/server.properties
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
55
56
57
58
59
60
61
62
63
64

注意：其他两个节点的配置文件复制过去即可，只需要修改 broker.id ，用于唯一标识kafka节点；

7>Logstash日志分析处理服务器部署

wget https://artifacts.elastic.co/downloads/logstash/logstash-2.3.4.tar.gz
wget http://geolite.maxmind.com/download/geoip/database/GeoLiteCity.dat.gz  ///配置 GeoLiteCity ， 用于地图显示IP访问的城市
tar xzf logstash-2.3.4.tar.gz
gunzip GeoLiteCity.dat.gz

mv logstash-2.3.4 logstash
cd logstash/
vim config/logstash.yml //Logstash的核心配置文件，用来控制Logstash的执行，可以指定管道配置和配置文件位置等等。在启动Logstash时可以通过命令行指定logstash.yml文件中大部分配置，并且覆盖掉当前配置文件中的配置项。这里只需配置全部端口加你他那个，开启http 9600端口访问就可；更多见附录

vim logstash_in_kafka.conf //如下所示

input {
    beats {
    port => 5044
    codec => "json"
	}
 }

 
filter {
    if [type] == "nginx_accesslog" {
 
    geoip {
        source => "clientip" # 与日志中访问地址的key要对应
        target => "geoip"
        database => "/usr/local/logstash/GeoLiteCity.dat"
        add_field => [ "[geoip][coordinates]","%{[geoip][longitude]}" ]
        add_field => [ "[geoip][coordinates]","%{[geoip][latitude]}" ]
	}
 
    mutate {
        convert => [ "[geoip][coordinates]","float" ]
 	}
  }
 
}

output {
  kafka {
    workers => 2
    bootstrap_servers => "10.100.1.3:9092,10.100.1.4:9092,10.100.1.5:9092"
    topic_id => "peiyinlog" 
    } 
}

#测试启动
./logstash -e "input { stdin { } } output { stdout {} }" // -e 标志可以在命令行直接指定配置文件,双引号不能改成单引号否则可能会报：ERROR: Unknown command '{'
#启动 Logstash 日志采集
#./bin/logstash -f ./conf-logstash/access-log.conf  //-f 后面指定的是配置文件
/usr/local/logstash/bin/logstash agent -f logstash_in_kafka.conf &
#查看 Elasticsearch head 插件查看验证
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

注意：注：使用Logstatsh的版本号与elasticsearch版本号需要保持一致；Logstash 的一个配置文件包含 input{}，filter{}，output{} 三部分，该三部分的作用分别为，一个配置输入源，一个配置过滤规则，一个配置输出；每个部分又可以配置各种不同的插件，所有插件的说明请参考官网文档的说明，Logstash 官网提供了很多可以拿来即用的正则表达式，单击参看。其他借鉴请参看logstash将日志文件输出到kafka消费端；

示例2：采集 TOMCAT 应用程序日志

vim conf-logstash/tomcat-log.conf //如下所示
input {
    file {
		codec => multiline {
			 pattern => "^\s"
			 what => "previous"
		}
        path => "/usr/local/tomcat8/logs/catalina.out"
        start_position => "beginning"
    }
}
 
filter {
  grok {
        patterns_dir => "/usr/local/elk/logstash-patterns"
        match => {
            "message" => "%{MYLOG}"
        }
        add_field => [ "log_ip", "10.100.1.102" ]
  }
 
}
 
output {
    elasticsearch {
      hosts => ["10.100.1.60","10.100.1.80","10.100.1.102"]
        index => "tomcat"
    }
}
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

附录：logstash主配置文件说明

#使用分层表单来设置管道的批处理大小和批处理延迟
   pipeline:
     batch:
       size: 125        #管道批处理大小
       delay: 5             #管道批处理延迟
 
#若要表示与平面键相同的值：
   pipeline.batch.size: 125
   pipeline.batch.delay: 5
 
 
#节点名称，在集群中具备唯一性，默认为logstash主机的主机名
node.name: logstast-node1
 
#logstash及其插件所使用的数据路径，默认路径为logstash家目录下的data目录
path.data: /usr/local/logstash-2.3.4/data/
 
#管道的ID，默认为main
pipeline.id: main
 
#输入、输出及过滤器的总工作数量，也就是logstash的工作进程，此工作进程默认为主机的cpu核心数量
pipeline.workers: 16 
 
#在输入阶段，单个工作线程将从输入中收集的最大事件数，此事件数堆内存开销较大，内存开销可在jvm.options中设置堆内存大小来优化此选项
pipeline.batch.size: 125
 
#在将一个较小的批发送到filters+output之前,轮询下一个事件时等待的时间(以毫秒为单位)
pipeline.batch.delay: 50
 
#设置为true时，在强制关闭logstash期间，即使内存中还有事件，那么为true将会强制关闭，导致数据丢失；默认为false，false在强制关闭logstash期间，将拒绝退出，直到所有在管道中的事件被安全输出，再关闭。
pipeline.unsafe_shutdown: false
 
#指定管道配置的目录，在此目录下的所有管道配置文件都将被logstash读取，除管道配置外，不要放任何文件
path.config: /usr/local/logstash-7.0.0/conf.d/
 
#在启动时，测试配置是否有效并退出，检测配置文件是否正确，包括检测管道配置文件，默认为false
config.test_and_exit: true
 
#定期检查配置是否更改并重新加载管道，默认为false
config.reload.automatic: true
 
#logstash间隔多久检查一次配置中的更改，默认为3秒
config.reload.interval: 600s
 
#设置为true时，将完全编译的配置显示为调试日志消息
config.debug: false
 
#用于事件缓冲的内部排队模型;可以指定内存memory或者磁盘persisted，内存处理速度相对磁盘来说效率要高，默认为内存
queue.type: memory
 
#启用持久队列时将存储数据文件的目录路径,默认为logstash路径下的queue
path.queue: /usr/local/logstash-2.3.4/queue/
 
#启用持久队列时使用的页面数据文件的大小(queue.type: persisted)队列数据由分成页面的仅附加数据文件组成
queue.page_capacity: 64mb
 
#启用持久队列时队列中未读事件的最大数量(queue.type: persisted)，默认为0，0为无限制
queue.max_events: 0
 
#队列的总容量，以字节数表示，默认为1G，根据业务需求而定
queue.max_bytes: 1024mb
 
#启用持久队列时强制检查点之前最大的ACK事件数量(queue.type: persisted)，设置为0，表示无限制，默认为1024
queue.checkpoint.acks: 1024
 
#启用持久队列时强制检查点之前写入事件的最大数量(queue,type: persisted)，设置为0，表示无限制，默认为1024
queue.checkpoint.writes: 1024
 
#启用持久队列(queue,type: persisted)，强制在头部页面上设置检查点的间隔(以毫秒为单位)，有周期性检查点的默认值是1000毫秒
queue.checkpoint.interval: 1000
 
#用于指示logstast启用插件支持DLQ功能的标志，默认为false
dead_letter_queue.enable: false
 
#每个死信队列的最大大小，如果条目超过此设置会增加死信队列的大小，则会删除条目，默认为1024mb
dead_letter_queue.max_bytes: 1024mb
 
#为死信队列存储数据文件的目录路径
path.dead_letter_queue: /usr/local/logstash-2.3.4/letter-queue
 
#度量标准REST端点的绑定地址，默认为127.0.0.1
http.host: "0.0.0.0"
 
#度量标准REST端点的绑定端口，默认为9600
http.port: 9600
 
#日志级别，可以设置为以下几种级别，默认为info
log.level: info
           fatal
           error
           warn
           info (default)
           debug
           trace
 
#logstash日志目录位置，默认为logstash路径下的logs
path.logs: /usr/local/logstash-7.0.0/logs
 
 
#logstash插件路径
path.plugins: []
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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

8>Filebeat数据采集部署

wget -c https://download.elastic.co/beats/filebeat/filebeat-1.2.3-x86_64.rpm
yum install filebeat-1.2.3-x86_64.rpm -y

vim /etc/filebeat/filebeat.yml

################### Filebeat Configuration Example #########################
 
############################# Filebeat ######################################
 
filebeat:
  prospectors:
    -
      paths:
        - /var/log/messages
 
      input_type: log
       
      document_type: messages
 
    -
      paths:
        - /alidata/log/nginx/access/access.log
       
      input_type: log
 
      document_type: nginxacclog
     
    -
      paths:
        - /alidata/www/logs/laravel.log
        
      input_type: log
 
      document_type: larlog
     
    -
      paths:
        - /alidata/www/logs/500_error.log
 
      input_type: log
 
      document_type: peiyinlar_500error
     
    -
      paths:
        - /alidata/www/logs/deposit.log
       
      input_type: log
 
      document_type: lar_deposit
 
    -
      paths:
        - /alidata/www/logs/call_error.log
      
      input_type: log
 
      document_type: call_error
     
    -
      paths:
        - /alidata/log/php/php-fpm.log.slow
 
      input_type: log
 
      document_type: phpslowlog
 
      multiline: 
          pattern: '^[[:space:]]'
          negate: true
          match: after
 
  registry_file: /var/lib/filebeat/registry
 
   
############################# Output ##########################################
   
output:
  logstash: 
    hosts: ["10.100.1.5:1024"] //nginx代理的地址,轮询转到2台logstash
   
 
############################# Shipper #########################################
   
shipper: 
  name: "host_6"
   
   
############################# Logging ######################################### 
   
logging:  
  files:
    rotateeverybytes: 10485760 # 默认的10MB
    level: info

#定制Nginx日志格式

log_format json '{"@timestamp":"$time_iso8601",'
                 '"slbip":"$remote_addr",'
                 '"clientip":"$http_x_forwarded_for",'
                 '"serverip":"$server_addr",'
                 '"size":$body_bytes_sent,'
                 '"responsetime":$request_time,'
                 '"domain":"$host",'
                 '"method":"$request_method",'
                 '"requesturi":"$request_uri",'
                 '"url":"$uri",'
                 '"appversion":"$HTTP_APP_VERSION",'
                 '"referer":"$http_referer",'
                 '"agent":"$http_user_agent",'
                 '"status":"$status",'
                 '"devicecode":"$HTTP_HA"}';
                  
# 在虚拟主机配置中调用
access_log  /alidata/log/nginx/access/access.log json;

#启动服务
/etc/init.d/filebeat start
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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

9>数据转发层的Logstash安装同步骤7

# kafka_to_es.conf
 
input{
    kafka {
        zk_connect => "10.100.1.3:2181,10.100.1.4:2181,10.100.1.5:2181"
        group_id => "logstash"
        topic_id => "peiyinlog"
        reset_beginning => false
        consumer_threads => 50
        decorate_events => true
 	}
 }
 
# 删除一些不需要的字段
 
filter {
  if [type] == "nginxacclog" {
     mutate {
     remove_field => ["slbip","kafka","domain","serverip","url","@version","offset","input_type","count","source","fields","beat.hostname","host","tags"]
    	}
	} 
}
 
output {
    if [type] == "nginxacclog" {
       # stdout {codec => rubydebug }
       
        elasticsearch {
            hosts => ["10.10.1.90:9200","10.10.1.60:9200"]
            index => "logstash-nginxacclog-%{+YYYY.MM.dd}"
            manage_template => true
            flush_size => 50000
            idle_flush_time => 10
            workers => 2
		} 
	}
  
    if [type] == "messages" {
        elasticsearch {
            hosts => ["10.100.1.60:9200","10.100.1.80:9200"]
            index => "logstash-messages-%{+YYYY.MM.dd}"
            manage_template => true
            flush_size => 50000
            idle_flush_time => 30
            workers => 1
		}
	}
 
 
    if [type] == "larlog" {
        elasticsearch {
            hosts => ["10.100.1.60:9200","10.100.1.80:9200"]
            index => "logstash-larlog-%{+YYYY.MM.dd}"
            manage_template => true
            flush_size => 2000
            idle_flush_time => 10
		}
	}
 
 
    if [type] == "deposit" {
        elasticsearch {
            hosts => ["10.100.1.60:9200","10.100.1.80:9200"]
            index => "logstash-deposit-%{+YYYY.MM.dd}"
            manage_template => true
            flush_size => 2000
            idle_flush_time => 10
		} 
	}
 
 
    if [type] == "phpslowlog" {
        elasticsearch {
            hosts => ["10.100.1.60:9200","10.100.1.80:9200"]
            index => "logstash-phpslowlog-%{+YYYY.MM.dd}"
            manage_template => true
            flush_size => 2000
            idle_flush_time => 10
		}
 	}
 
}

#启动服务
/usr/local/logstash/bin/logstash agent -f kafka_to_es.conf &
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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

10>修改ES的索引模版配置

curl -XPUT http://10.100.1.102:9200/_template/logstash2 -d '
{
        "order":1,		#优先级 order 定义的比logstash模版高,可覆盖默认模板配置
        "template":"logstash-*",
        "settings":{
            "index":{
                "refresh_interval":"120s"
            }
        },
        "mappings":{
            "_default_":{
                "_all":{
                    "enabled":false
                }
            }
    }
}'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

11>配置 Kibana 数据展示

wgethttps://download.elastic.co/kibana/kibana/kibana-4.5.3-linux-x64.tar.gz
tar xzf kibana-4.5.3-linux-x64.tar.gz

vim kibana-4.5.3-linux-x64/config/kibana.yml
 
# Kibana is served by a back end server. This controls which port to use.
server.port: 5601
 
# The host to bind the server to.
server.host: "0.0.0.0"
 
# The Elasticsearch instance to use for all your queries.
elasticsearch.url: "

#启动
./bin/kibana
# 后台启动
./kibana-4.5.3-linux-x64/bin/kibana &
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

更多kibana.yml如下：

# Kibana is served by a back end server. This controls which port to use.
server.port: 5601

# The host to bind the server to.
server.host: "0.0.0.0"

# If you are running kibana behind a proxy, and want to mount it at a path,
# specify that path here. The basePath can't end in a slash.
# server.basePath: ""

# The maximum payload size in bytes on incoming server requests.
# server.maxPayloadBytes: 1048576

# The Elasticsearch instance to use for all your queries.
elasticsearch.url: "http://x.x.x.x:port"

# preserve_elasticsearch_host true will send the hostname specified in `elasticsearch`. If you set it to false,
# then the host you use to connect to *this* Kibana instance will be sent.
# elasticsearch.preserveHost: true

# Kibana uses an index in Elasticsearch to store saved searches, visualizations
# and dashboards. It will create a new index if it doesn't already exist.
kibana.index: ".kibana"

# The default application to load.
# kibana.defaultAppId: "discover"

# If your Elasticsearch is protected with basic auth, these are the user credentials
# used by the Kibana server to perform maintenance on the kibana_index at startup. Your Kibana
# users will still need to authenticate with Elasticsearch (which is proxied through
# the Kibana server)
# elasticsearch.username: "user"
# elasticsearch.password: "pass"

# SSL for outgoing requests from the Kibana Server to the browser (PEM formatted)
# server.ssl.cert: /path/to/your/server.crt
# server.ssl.key: /path/to/your/server.key

# Optional setting to validate that your Elasticsearch backend uses the same key files (PEM formatted)
# elasticsearch.ssl.cert: /path/to/your/client.crt
# elasticsearch.ssl.key: /path/to/your/client.key

# If you need to provide a CA certificate for your Elasticsearch instance, put
# the path of the pem file here.
# elasticsearch.ssl.ca: /path/to/your/CA.pem

# Set to false to have a complete disregard for the validity of the SSL
# certificate.
# elasticsearch.ssl.verify: true

# Time in milliseconds to wait for elasticsearch to respond to pings, defaults to
# request_timeout setting
# elasticsearch.pingTimeout: 1500

# Time in milliseconds to wait for responses from the back end or elasticsearch.
# This must be > 0
# elasticsearch.requestTimeout: 30000

# Time in milliseconds for Elasticsearch to wait for responses from shards.
# Set to 0 to disable.
# elasticsearch.shardTimeout: 0

# Time in milliseconds to wait for Elasticsearch at Kibana startup before retrying
# elasticsearch.startupTimeout: 5000

# Set the path to where you would like the process id file to be created.
# pid.file: /var/run/kibana.pid

# If you would like to send the log output to a file you can set the path below.
# logging.dest: stdout

# Set this to true to suppress all logging output.
# logging.silent: false

# Set this to true to suppress all logging output except for error messages.
# logging.quiet: false

# Set this to true to log all events, including system usage information and all requests.
# logging.verbose: 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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

验证，打开浏览器访问: http://10.100.1.102:5601/

默认情况下，Kibana认为你要访问的是通过Logstash导入Elasticsearch的数据，这时你可以用默认的 logstash-* 作为你的 index pattern。 通配符（*）匹配索引名中任意字符任意个数。

选择一个包含了时间戳的索引字段（字段类型为 date 的字段），可以用来做基于时间的处理。Kibana 会读取索引的映射，然后列出所有包含了时间戳的字段。如果你的索引没有基于时间的数据关闭 Index contains time-based events 参数。

如果一个新索引是定期生成，而且索引名中带有时间戳，选择 Use event times to create index names 选项，
然后再选择 Index pattern interval 。这可以提高搜索性能，Kibana 会至搜索你指定的时间范围内的索引。在你用 Logstash 输出数据给Elasticsearch 的情况下尤其有效。上图示例中的索引是用日期命名，按照每天分割的。

最后效果：

四、ELK组件

4.1 elasticsearch（ES）

1）组件架构

更多参看：。

1、Gateway网关

　　　　其作用是用来对数据进行持久化以及ES重启后重新恢复数据。es支持多种类型的gateway，有本地文件系统、分布式文件系统、Hadoop的HDFS等。其存储的信息包括索引信息、集群信息、mapping等

2、districted lucene directory搜索引擎

　　　　Gateway上层就是Lucene的分布式检索框架。ES是分布式的搜索引擎，虽然底层用的是Lucene，但是需要在每个节点上都运行Lucene进行相应的索引、查询、更新等操作，所以需要做成一个分布式的运行框架来满足业务需要。

3、四大组件模块

　　　　districted lucene directory之上就是ES的四大模块。包括：

　　　　Index Model：索引模块，对数据建立索引（通常是建立倒排索引）

　　　　Seacher Model：搜索模块，就是对数据进行查询搜索

　　　　Mapping Model：是数据映射与解析模块，数据的每个字段可以根据建立的表结构通过mapping进行映射解析；如果没有建立表结构，那么ES会根据数据类型来推测数据结构，并自动生成一个mapping，然后根据mapping进行解析

　　　　River Model：在es2.0之后被取消了，表示可以使用插件处理。例如可以通过一些自定义脚本将传统数据库的数据实时同步到es中。

4、自动发现Discovery Script

　　　　es集群中各个节点通过discovery相互发现的，默认使用的是Zen。es是一个基于p2p的系统，他先通过广播寻找存在的节点，再通过多播协议来进行节点之间的通信，同时也支持点对点的交互。es还可以支持多种script脚本语言，例如mvel、js、python等。

5、通信（Transport）

　　　　代表es内部节点或集群与客户的交互方式，默认内部使用tcp协议进行交互，同时其还支持http协议，thrift、servlet、memcached、zeroMQ等通信协议。节点间通信端口默认9300-9400；

6、Restful接口：最上层就是ES暴漏给我们的访问接口。　

　　　使用Restful访问ES的方式：curl X<VERB> <PROTOCOL>://<HOST>:<PORT>/<PATH>?<QUERY_STRING>d <BODY>

　　其中VERB是请求方式（post、get…），PROTOCOL是协议（http、https…），HOST和PORT是es的ip和端口，PATH是API的终端路径，QUERY_STRING表示任何可选的查询字符串参数（例如@pertty将格式化的输出JSON格式），BODY是一个Json格式的请求体。

2）业务流程

1>全文索引：全文检索可以分为两大步骤：创建索引和搜索索引

2>索引库结构：

索引库分为索引域和文档域，索引域里面存储的是term字典和文档的倒排索引，文档域存储的就是document。

五、附录：

1、分布式日志链路跟踪

使用ELK来统一收集日志，但是在并发大时使用日志定位问题还是比较麻烦，由于大量的其他用户/其他线程的日志也一起输出穿行其中导致很难筛选出指定请求的全部相关日志，以及下游线程/服务对应的日志。

解决：

1)每个请求都使用一个唯一标识来追踪全部的链路显示在日志中，并且不修改原有的打印方式(代码无入侵)
2)使用Logback的MDC机制日志模板中加入traceId标识，取值方式为%X{traceId};
/
MDC（Mapped Diagnostic Context，映射调试上下文）是 log4j 和 logback 提供的一种方便在多线程条件下记录日志的功能。MDC 可以看成是一个与当前线程绑定的Map，可以往其中添加键值对。MDC 中包含的内容可以被同一线程中执行的代码所访问。当前线程的子线程会继承其父线程中的 MDC 的内容。当需要记录日志时，只需要从 MDC 中获取所需的信息即可。MDC 的内容则由程序在适当的时候保存进去。对于一个 Web 应用来说，通常是在请求被处理的最开始保存这些数据。

详情单击查看。

2、Elasticsearch 和 JVM对应关系（ES版本选择）

参考官网：



注意：相关经验表明elk7.2.1版本需要装jdk11，如果安装9或者8会报错。但是官方建议：Java 9、Java 10、Java 12 和 Java 13 均为短期版本，不要使用以上版本。Elasticsearch 7.17.x可使用1.8.xJDK；

3、Logstash 和 JVM

从上可知，Logstash 7.8.x要不使用1.8 jdk，要不使用JDK 11;但Logstash 8.0.x不支持1.8版本JDK；

4、ELK与OS版本兼容情况

1）Elasticsearch：



其中：SLES 11 不支持 RPM 安装。 由于 glibc 不兼容，CentOS 6、RHEL 6 和 Oracle Enterprise Linux 6 自 7.9.2 版起将不支持捆绑的 JDK 15+。
2）Kibana：



3）Logstash



4）Filebeat：

5、ELK互兼容性

注意：官方建议运行最新版本的 Beats、Logstash、Elastic 代理和 ES-Hadoop；早期版本使用的功能有所删减。

官方推荐使用外部JDK（TLS）版本，最后使用ES自带的；

6、关于elk中的时区问题

计算机中的UNIX时间戳，是以GMT/UTC（世界时间，比CST(China Standard Time北京时间)慢8h）时间，ElasticSeaerch，Logstash 默认也都是UTC时区，而Kibana则是根据本地浏览器显示本地时区，已经对显示做了转换。Kibana时区默认为浏览器时间，现在改为UTC时间,修改如下：

由于我们处于东8区，logstash读取ES的时是当我们到8点后才会发生索引切割，导致我们显示今天8点的日志实际是存入了昨天的索引，即logstash入库完了8小时，创建索引延迟了8小时。在logstash的timestamp.rb文件里可看到，默认初始化为utc的时区，我们将UTC = org.joda.time.DateTimeZone.forID(“UTC”)修改为：
UTC = org.joda.time.DateTimeZone.getDefault()

然后vim string_interpolation.rb

配置文件添加：

filter {
	ruby {
        code => "event.timestamp.time.localtime"
	}
}
1
2
3
4
5

7、其他参考

https://www.yinxiaoling.com/Java/architect/ELK/
……未完待整理