ELK曾经成为目前最盛行的集中式日志处置打算，它关键是由Beats、Logstash、Elasticsearch、Kibana等组件组成，来独特成功实时日志的搜集，存储，展现等一站式的处置打算。本文将会引见ELK经常出现的架构以及关系疑问处置。

ELK经常出现部署架构

1.Logstash作为日志搜集器

这种架构是比拟原始的部署架构，在各运行主机端区分部署一个Logstash组件，作为日志搜集器，而后将Logstash搜集到的数据过滤、剖析、格局化处置后发送至Elasticsearch存储，最后经常使用Kibana启动可视化展现。

这种架构无余的是：Logstash比拟耗主机资源，所以会参与运行主机端的负载压力。

2.Filebeat作为日志搜集器

该架构与第一种架构惟一不同的是：运行端日志搜集器换成了Filebeat，Filebeat轻量，占用主机资源少，所以经常使用Filebeat作为运行主机端的日志搜集器，普通Filebeat会配合Logstash一同经常使用，这种部署形式也是目前最罕用的架构。

3.引入缓存队列的部署架构

该架构在第二种架构的基础上引入了Kafka信息队列（还可以是其余信息队列），将Filebeat搜集到的数据发送至Kafka，而后在经过Logstasth读取Kafka中的数据，这种架构关键是处置大数据量下的日志搜集打算，经常使用缓存队列关键是处置数据安保与平衡Logstash与Elasticsearch负载压力。

4.以上三种架构的总结

第一种部署架构由于资源占用疑问，现已很少经常使用，目前经常使用最多的是第二种部署架构，至于第三种部署架构团体感觉没有必要引入信息队列，除非有其余需求，由于在数据量较大的状况下，Filebeat经常使用压力敏感协定向Logstash或Elasticsearch发送数据。

假设Logstash正在忙碌地处置数据，它会告知Filebeat减慢读取速度。拥塞处置后，Filebeat将复原初始速度并继续发送数据。

疑问及处置打算

1.疑问：如何成功日志的多行兼并性能？

系统运行中的日志普通都是以特定格局启动打印的，属于同一条日志的数据或许分多前启动打印，那么在经常使用ELK搜集日志的时刻就须要将属于同一条日志的多行数据启动兼并。

处置打算：经常使用Filebeat或Logstash中的multiline多行兼并插件来成功

在经常使用multiline多行兼并插件的时刻须要留意，不同的ELK部署架构或许multiline的经常使用形式也不同，假设是本文的第一种部署架构，那么multiline须要在Logstash中性能经常使用，假设是第二种部署架构，那么multiline须要在Filebeat中性能经常使用，无需再在Logstash中性能multiline。

1）multiline在Filebeat中的性能形式：

filebeat.prospectors:-paths:-/home/project/elk/logs/test.loginput_type:logmultiline:pattern:'^['negate:truematch:afteroutput:logstash:hosts:["localhost:5044"]

如：

pattern:'['negate:truematch:after

该性能示意将不婚配pattern形式的行兼并到上一行的末尾

2）multiline在Logstash中的性能形式

input{beats{port=>5044}}filter{multiline{pattern=>"%{LOGLEVEL}s*]"negate=>truewhat=>"previous"}}output{elasticsearch{hosts=>"localhost:9200"}}

（1）Logstash中性能的what属性值为previous，相当于Filebeat中的after，Logstash中性能的what属性值为next，相当于Filebeat中的before。

（2）pattern=>"%{LOGLEVEL}s*]"中的LOGLEVEL是Logstash预制的正则婚配形式，预制的还有好多罕用的正则婚配形式，具体请看：

2.疑问：如何将Kibana中显示日志的期间字段交流为日志信息中的期间？

自动状况下，咱们在Kibana中检查的期间字段与日志信息中的期间不分歧，由于自动的期间字段值是日志搜集时的以后期间，所以须要将该字段的期间交流为日志信息中的期间。

处置打算：经常使用grok分词插件与date期间格局化插件来成功

在Logstash的性能文件的过滤器中性能grok分词插件与date期间格局化插件，如：

input{beats{port=>5044}}filter{multiline{pattern=>"%{LOGLEVEL}s*][%{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME}]"negate=>truewhat=>"previous"}grok{match=>["message","(?<customer_time>%{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME})"]}date{match=>["customer_time","yyyyMMddHH:mm:ss,SSS"]//格局化期间target=>"@timestamp"//交流自动的期间字段}}output{elasticsearch{hosts=>"localhost:9200"}}

如要婚配的日志格局为：[DEBUG][2017081110:07:31,359][DefaultBeanDefinitionDocumentReader:106]Loadingbeandefinitions，解析出该日志的期间字段的形式有：

①经过引入写好的表白式文件，如表白式文件为customer_patterns，内容为：CUSTOMER_TIME%{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME}

注：内容格局为：[自定义表白式称号][正则表白式]

而后logstash中就可以这样援用：

filter{grok{patterns_dir=>["./customer-patterms/mypatterns"]//援用表白式文件门路match=>["message","%{CUSTOMER_TIME:customer_time}"]//经常使用自定义的grok表白式}}

②以性能项的形式，规定为：(?<自定义表白式称号>正则婚配规定)，如：

filter{grok{match=>["message","(?<customer_time>%{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME})"]}}

3.疑问：如何在Kibana中经过选用不同的系统日志模块来检查数据

普通在Kibana中显示的日志数据混合了来自不同系统模块的数据，那么如何来选用或许过滤只检查指定的系统模块的日志数据？

处置打算：新增标识不同系统模块的字段或依据不同系统模块建ES索引

1、新增标识不同系统模块的字段，而后在Kibana中可以依据该字段来过滤查问不同模块的数据，这里以第二种部署架构解说，在Filebeat中的性能内容为：

filebeat.prospectors:-paths:-/home/project/elk/logs/account.loginput_type:logmultiline:pattern:'^['negate:truematch:afterfields://新增log_from字段log_from:account-paths:-/home/project/elk/logs/customer.loginput_type:logmultiline:pattern:'^['negate:truematch:afterfields:log_from:customeroutput:logstash:hosts:["localhost:5044"]

经过新增：log_from字段来标识不同的系统模块日志

2、依据不同的系统模块性能对应的ES索引，而后在Kibana中创立对应的索引形式婚配，即可在页面经过索引形式下拉框选用不同的系统模块数据。

filebeat.prospectors:-paths:-/home/project/elk/logs/account.loginput_type:logmultiline:pattern:'^['negate:truematch:afterfields://新增log_from字段log_from:account-paths:-/home/project/elk/logs/customer.loginput_type:logmultiline:pattern:'^['negate:truematch:afterfields:log_from:customeroutput:logstash:hosts:["localhost:5044"]

这里以第二种部署架构解说，分为两步：

①在Filebeat中的性能内容为：

filebeat.prospectors:-paths:-/home/project/elk/logs/account.loginput_type:logmultiline:pattern:'^['negate:truematch:afterdocument_type:account-paths:-/home/project/elk/logs/customer.loginput_type:logmultiline:pattern:'^['negate:truematch:afterdocument_type:customeroutput:logstash:hosts:["localhost:5044"]

经过document_type来标识不同系统模块

②修正Logstash中output的性能内容为：

output{elasticsearch{hosts=>"localhost:9200"index=>"%{type}"}}

总结

本文关键引见了ELK实时日志剖析的三种部署架构，以及不同架构所能处置的疑问，这三种架构中第二种部署形式是时下最盛行也是最罕用的部署形式。

最后引见了ELK作在日志剖析中的一些疑问与处置打算，说在最后，ELK不只仅可以用来作为散布式日志数据集中式查问和治理，还可以用来作为名目运行以及主机资源监控等场景，更多内容请看官方。

SpringCloudGateway使用Skywalking时日志打印traceId

由于SpringCloudGateway是基于WebFlux来实现的，需要进到skywalking的agent目录，将optional-plugins目录底下的以下两个jar包复制到plugins目录

新增启动参数

启动程序后，尝试通过网关进行接口调用，可以在Skywalking-ui上看到链路已经串起来了

但是有个问题，日志里记录的日志始终不显示正确的TID

查看Skywalking-agent的源码，可以看到，在插件里，拦截了 的 handle 方法 拦截器里往reactor的调用链路里，放入 < SKYWALKING_CONTEXT_SNAPSHOT - ContextSnapshot >

网上找了下资料，在这里[发现了相关信息

「SpringCloud」（三十八）搭建ELK日志采集与分析系统

一套好的日志分析系统可以详细记录系统的运行情况，方便我们定位分析系统性能瓶颈、查找定位系统问题。上一篇说明了日志的多种业务场景以及日志记录的实现方式，那么日志记录下来，相关人员就需要对日志数据进行处理与分析，基于E(ElasticSearch)L(Logstash)K(Kibana)组合的日志分析系统可以说是目前各家公司普遍的首选方案。

作为微服务集群，必须要考虑当微服务访问量暴增时的高并发场景，此时系统的日志数据同样是爆发式增长，我们需要通过消息队列做流量削峰处理，Logstash官方提供Redis、Kafka、RabbitMQ等输入插件。Redis虽然可以用作消息队列，但其各项功能显示不如单一实现的消息队列，所以通常情况下并不使用它的消息队列功能；Kafka的性能要优于RabbitMQ，通常在日志采集，数据采集时使用较多，所以这里我们采用Kafka实现消息队列功能。ELK日志分析系统中，数据传输、数据保存、数据展示、流量削峰功能都有了，还少一个组件，就是日志数据的采集，虽然log4j2可以将日志数据发送到Kafka，甚至可以将日志直接输入到Logstash，但是基于系统设计解耦的考虑，业务系统运行不会影响到日志分析系统，同时日志分析系统也不会影响到业务系统，所以，业务只需将日志记录下来，然后由日志分析系统去采集分析即可，Filebeat是ELK日志系统中常用的日志采集器，它是 Elastic Stack 的一部分，因此能够与 Logstash、Elasticsearch 和 Kibana 无缝协作。

软件下载：

因经常遇到在内网搭建环境的问题，所以这里习惯使用下载软件包的方式进行安装，虽没有使用Yum、Docker等安装方便，但是可以对软件目录、配置信息等有更深的了解，在后续采用Yum、Docker等方式安装时，也能清楚安装了哪些东西，安装配置的文件是怎样的，即使出现问题，也可以快速的定位解决。

Elastic Stack全家桶下载主页：

我们选择如下版本：

Kafka下载：

安装前先准备好三台CentOS7服务器用于集群安装，这是IP地址为：172.16.20.220、172.16.20.221、172.16.20.222，然后将上面下载的软件包上传至三台服务器的/usr/local目录。因服务器资源有限，这里所有的软件都安装在这三台集群服务器上，在实际生产环境中，请根据业务需求设计规划进行安装。在集群搭建时，如果能够编写shell安装脚本就会很方便，如果不能编写，就需要在每台服务器上执行安装命令，多数ssh客户端提供了多会话同时输入的功能，这里一些通用安装命令可以选择启用该功能。

新建/usr/local/java目录

将下载的jdk软件包上传到/usr/local/java目录，然后解压

配置环境变量/etc/profile

在底部添加以下内容

使环境变量生效

备注：后续可通过此命令停止elasticsearch运行

新建kafka的日志目录和zookeeper数据目录，因为这两项默认放在tmp目录，而tmp目录中内容会随重启而丢失,所以我们自定义以下目录:

修改如下：

在data文件夹中新建myid文件，myid文件的内容为1（一句话创建：echo 1 > myid）

kafka启动时先启动zookeeper，再启动kafka；关闭时相反，先关闭kafka，再关闭zookeeper。1、zookeeper启动命令

后台运行启动命令：

或者

查看集群状态：

2、kafka启动命令

后台运行启动命令：

或者

3、创建topic，最新版本已经不需要使用zookeeper参数创建。

参数解释: 复制两份--replication-factor 2创建1个分区--partitions 1topic 名称--topic test

4、查看已经存在的topic（三台设备都执行时可以看到）

5、启动生产者：

6、启动消费者：

添加参数 --from-beginning 从开始位置消费，不是从最新消息

7、测试：在生产者输入test，可以在消费者的两台服务器上看到同样的字符test，说明Kafka服务器集群已搭建成功。

Logstash没有提供集群安装方式，相互之间并没有交互，但是我们可以配置同属一个Kafka消费者组，来实现统一消息只消费一次的功能。

Filebeat用于安装在业务软件运行服务器，收集业务产生的日志，并推送到我们配置的Kafka、Redis、RabbitMQ等消息中间件，或者直接保存到Elasticsearch，下面来讲解如何安装配置：

1、进入到/usr/local目录，执行解压命令

2、编辑配置配置文件中默认是输出到elasticsearch，这里我们改为kafka，同文件目录下的文件是所有配置的实例，可以直接将kafka的配置复制到

后台启动命令

停止命令

2、测试logstash是消费Kafka的日志主题，并将日志内容存入Elasticsearch

自动新增的两个index，规则是logstash中配置的

数据浏览页可以看到Elasticsearch中存储的日志数据内容，说明我们的配置已经生效。

Gitee: GitEgg: GitEgg 是一款开源免费的企业级微服务应用开发框架，旨在整合目前主流稳定的开源技术框架，集成常用的最佳项目解决方案，实现可直接使用的微服务快速开发框架。GitHub: