本文原文链接

尼恩说在前面

在40岁老架构师 尼恩的读者交流群(50+)中，最近有小伙伴拿到了一线互联网企业如得物、阿里、滴滴、极兔、有赞、希音、百度、网易、美团的面试资格，并且拿了很多大厂offer。面试过程中，其中 elk 日志 相关的题目，是大家的面试核心，面试重点。

比如小伙伴在面试蚂蚁的时候，就遇到以下面试题：

海量日志的监控和管理，是如何实现的？

分布式 微服务系统，怎么实现的故障排查？

分布式 微服务系统，日志数据如何集中式查询和管理？

很多很多小伙伴， 由于之前没有系统的去梳理和总结，面试的时候，总是支支吾吾的说了几句，面试官不满意，面试挂了。

所以，尼恩给大家做一下系统化、体系化的梳理，联合社群小伙伴，来一个elk学习圣经：elk（Elasticsearch、Logstash、Kibana）。

特别说明， 本文属于 穿透 SpringCloud 工业级 底座工程(一共包括 15大学习圣经 )的 其中之一，并且，此系列15大学习圣经底座正在录制视频， 帮助大家一举成为技术高手。

15大圣经 ，使得大家内力猛增，可以充分展示一下大家雄厚的 “技术肌肉”，让面试官爱到 “不能自已、口水直流”，然后实现”offer直提”。

尼恩团队 从0到1 大实战 SpringCloud 工业级底座 的 知识体系的轮廓如下，详情请点击：15大圣经的介绍：

其中，专题1 权限设计以及 安全认证相关的两个圣经，具体如下：

• SpringSecurity& Auth2.0 学习圣经：从入门到精通 SpringSecurity& Auth2.0
• 史上最牛的 权限系统，如何设计？来了一个 Sa-Token学习圣经

以上的配套视频已经录制完毕，月底发布。

其中，专题3为：注册发现治理架构： Nacos 学习圣经，具体如下：

• Nacos学习圣经：从入门到精通 Nacos，最全详解 (30+图文全面总结)

以上的配套视频已经录制完毕，月底发布。

其中，专题 5为 RPC治理架构 Dubb3.0，具体如下：

• Dubbo学习圣经：从入门到精通 Dubbo3.0 + SpringCloud

以上的配套视频已经录制完毕，月底发布。

其中，专题6为 熔断限流 架构 Sentinel 学习圣经，具体如下：

• Sentinel学习圣经：从入门到精通 Sentinel，最全详解 (40+图文全面总结)

以上的配套视频已经录制完毕，月底发布。

本文，就是 elk 学习圣经。稍后会录制视频。录完之后，elk 学习圣经 正式版本会有更新， 最新版本找尼恩获取。

elk（Elasticsearch、Logstash、Kibana）从入门到精通的 学习路线

ELK 其实并不是一款软件，而是一整套解决方案，是三个软件产品的首字母缩写

• Elasticsearch：负责日志检索和储存

• Logstash：负责日志的收集和分析、处理

• Kibana：负责日志的可视化

  这三款软件都是开源软件，通常是配合使用，而且又先后归于 Elastic.co 公司名下，故被简称为 ELK

45岁老架构师尼恩团队打造，elk（Elasticsearch、Logstash、Kibana）从入门到精通的 学习路线

以上的配套视频正在录制，月底发布。

1 ELK日志平台介绍

ELK指的是Elastic公司下面Elasticsearch、Logstash、Kibana三大开源框架首字母大写简称。 Elasticsearch、Logstash、Kibana三大开源框架首字母大写简称。

1.1 基础日志平台的背景

日志主要包括系统日志、应用程序日志和安全日志。系统运维和开发人员可以通过日志了解服务器软硬件信息、检查配置过程中的错误及错误发生的原因。经常分析日志可以了解服务器的负荷，性能安全性，从而及时采取措施纠正错误。

通常，日志被分散的储存不同的设备上。

如果你管理数十上百台服务器，你还在使用依次登录每台机器的传统方法查阅日志。

这样是不是感觉很繁琐和效率低下。当务之急我们使用集中化的日志管理，例如：开源的syslog，将所有服务器上的日志收集汇总。

集中化管理日志后，日志的统计和检索又成为一件比较麻烦的事情，一般我们使用grep、awk和wc等Linux命令能实现检索和统计，但是对于要求更高的查询、排序和统计等要求和庞大的机器数量依然使用这样的方法难免有点力不从心。

随着服务容器化，跑在一台CentOS服务器上，服务器搭建了docker环境，安装了docker-compose或者k8s，但在日志处理方面，暂时没有一个好的方法能够收集完全的日志，

只能依赖进入至服务器后，以docker logs containerID的方法来进入查看，非常不方便，所以，基础日志平台就迫在眉睫。

1.2 ELK的关系

在ELK架构中，Elasticsearch、Logstash和Kibana三款软件作用如下：

1、Elasticsearch
Elasticsearch是一个高度可扩展的全文搜索和分析引擎，基于Apache Lucence（事实上，Lucence也是百度所采用的搜索引擎）构建，能够对大容量的数据进行接近实时的存储、搜索和分析操作。
2、Logstash
Logstash是一个数据收集引擎，它可以动态的从各种数据源搜集数据，并对数据进行过滤、分析和统一格式等操作，并将输出结果存储到指定位置上。Logstash支持普通的日志文件和自定义Json格式的日志解析。

负责日志清洗： 日志 过滤，格式处理，等等。 有大量的 自定义完成。

3、Kibana
Kibana是一个数据分析和可视化平台，通常与Elasticsearch配合使用，用于对其中的数据进行搜索、分析，并且以统计图标的形式展示。

1.3 ELK的架构

如上图所示，filebeats 安装在各个设备上，用于收集日志信息，收集到的日志信息统一汇总到Elasticsearch上，然后由Kibana负责web端的展示。

其中，如果终端设备过多，会导致logstash 过载的现象，此时，我们可以采用一台 mq 设备作为消息队列，以暂时缓存数据，避免 logstash 压力突发。

ELK架构优点如下：
1、处理方式灵活。 Elasticsearch是全文索引，具有强大的搜索能力。
2、配置相对简单。 Kibana的配置非常简单，Elasticsearch则全部使用Json接口，配置也不复杂，Logstash的配置使用模块的方式，配置也相对简单。
3、检索性能高。 ELK架构通常可以达到百亿级数据的查询秒级响应。
4、集群线性扩展。 Elasticsearch本身没有单点的概念，自动默认集群模式，Elasticsearch和Logstash都可以灵活扩展。
5、页面美观。 Kibana的前端设计美观，且操作简单。

​

Logstash:从各种数据源搜集数据，并对数据进行过滤、分析、丰富、统一格式等操作，然后存储到 ES。

Elasticsearch:对大容量的数据进行接近实时的存储、搜索和分析操作。

Kibana：数据分析和可视化平台。与 Elasticsearch 配合使用，对数据进行搜索、分析和以统计图表的方式展示。

1.3.1 简单的ELK日志平台

刚来公司的时候，我们公司的日志收集系统ELK经常会出现查询不了最新的日志的情况，后面去查发现 ES的节点经常也是yellow或者red的情况。有时候会收到开发的投诉。

最简单的，elk 架构图解如下:

其中ElasticSearch 是三台服务器构成的集群，其中：

• ElasticSearch做倒排索引，

• Logstash跑在每个服务器上，各种日志通过Logstash搜集，Grok，Geoip等插件进行处理然后统一送到ElasticSearch的集群。

• Kibana做图形化的展示。

这种elk架构比较简单，也存在一些问题：

1、Logstash依赖Java虚拟机占用系统的内存和CPU都比较大，

2、Logstash在数据量较大的时候容易导致其他业务应用程序崩溃，影响业务正常使用

3、随着时间的积累，es空间不能满足现状

4、Kibana没有安全管控机制，没有权限审核，安全性较差。

5、ElasticSearch 主节点也是数据节点，导致有时候查询较慢

1.3.2 ELK改进之引入Filebeat

ElasticSearch的版本，我们还是选择原来的 6.2.x的版本，然后重新搭建了一套ELK的日志系统。

ElasticSearch 6.x 的版本如果要做用于鉴权的话，必须依赖X-Pack，但是X-pack是付费的产品，所以，引入x-pack，虽然能实现 Index 级别的权限管控，确保数据安全，但是涉及到费用的问题。

于是，ElasticSearch的版本采用ElasticSearch 7.x的版本，用户鉴权采用其免费的 basic 认证实现（因为7.x的新版本在性能上优化，查询和写入速度会更快）

架构图解如下:

整个架构的具体的改进方法如下:

1、客户端选用更轻量化的Filebeat，Filebeat 采用 Golang 语言进行编写的，优点是暂用系统资源小，收集效率高。

2、Filebeat 数据收集之后统一送到多个 Logstatsh进行统一的过滤，然后将过滤后的数据写入ElasticSearch集群。

3、将原有的3个es节点增加至6个节点，其中3个ES节点是master节点，其余的节点是数据节点，如果磁盘不够用可以横向扩展数据节点。

6、ElasticSearch集群的硬盘采用 SSD的硬盘

7、ElasticSearch 做冷热数据分离

8、60天之前的索引数据进行关闭，有需要用的时候手工打开

9、ElasticSearch的版本采用ElasticSearch 7.x的版本，用户鉴权采用其免费的 basic 认证实现（因为7.x的新版本在性能上优化，查询和写入速度会更快）

到此，我们的日志系统算暂时是正常并且能满足日志查日志的需求了，也很少出现卡顿的现象了，并且服务器的资源使用率直接下降了一半。

1.3.3 ELK的应用场景

• 异常分析

通过将应用的日志内容通过Logstash输入到Elasticsearch中来实现对程序异常的分析排查

• 业务分析

将消息的通讯结果通过Logstash输入到Elasticsearch中来实现对业务效果的整理

• 系统分析

将处理内容的延迟作为数据输入到Elasticsearch 中来实现对应用性能的调优

1.4 ELK的不足

es的资源占用

一般使用 ES 时，必须要事先评估好节点配置和集群规模，可以从以下几个方面进行评估：

存储容量：

要考虑索引副本数量、数据膨胀、ES 内部任务额外占用的磁盘空间（比如 segment merge )以及操作系统占用的磁盘空间等因素，

如果再需要预留 50% 的空闲磁盘空间，那么集群总的存储容量大约为源数据量的 4 倍；

计算资源：

主要考虑写入，2 核 8GB 的节点可以支持 5000 qps 的写入，随着节点数量和节点规格的提升，写入能力基本呈线性增长；

索引和分片数量评估：

• 一般一个 shard 的数据量在 30-50 GB为宜，可以以此确定索引的分片数量以及确定按天还是按月建索引。

• 需要控制单节点总的分片数量，1GB 堆内存支持 20-30 个分片为宜。

• 另外需要控制集群整体的分片数量，集群总体的分片数量一般不要超过 3w 。

算下来 3W * 50G = 1500 T = 1.5P

那么，elk 如何支持 一天1000PB，一个月上万PB规模的日志量呢？

从吞吐量上来说，虽然mq进行扩展，能支撑100w 级别qps的吞吐量

但是， 后端的logstash 吞吐峰值15000 qps ，es的单节点写入 是 5000 qps 左右，

30K * 100Wqps 的日志吞吐量，如果不希望发生太大的日志延迟， 消息积压，

需要 100+个 logstash 节点， 300+个ES节点

这个需要庞大的资源成本，庞大的运维成本

如何满足10w级、100Wqps吞吐量qps、EB级日志存储呢

参考23章视频：《100Wqps 超高并发日志平台》实操

如果又要兼顾吞吐量，又要 降低硬件成本和运维成本，必须要

• 缩短 日志传输和处理链路，
• 并采用更高性能，更大压缩比例的存储组件，如clickhouse，

架构如下：

clickhouse 的数据压缩比例，请参考另外一篇博客：

clickhouse 超底层原理 + 高可用实操 （史上最全）

最终，压缩后的数据，只剩下 原始数据的 20%-30% ， 单数据库这块，减少了50% 的硬盘容量，

使用elk方案，数据有多个副本，包括MQ（主副本2 份），数据库（1 份），现在减少到 数据库（1 份），这里至少减少50% ，

2 一键安装 es+logstash+ kibana

实操过程，请参见23章视频：《100Wqps 超高并发日志平台》实操

对应的镜像版本

elasticsearch:7.14.0
kibana:7.14.0
logstash:7.14.0
filebeat:7.14.0

docker编码文件

version: "3.5"
services:
  elasticsearch:
     image: andylsr/elasticsearch-with-ik-icu:7.14.0
     container_name: elasticsearch
     hostname: elasticsearch
     restart: always
     ports:
       - 9200:9200
     volumes:
       - ./elasticsearch7/logs:/usr/share/elasticsearch/logs
       - ./elasticsearch7/data:/usr/share/elasticsearch/data
       - ./elasticsearch7/config/single-node.yml:/usr/share/elasticsearch/config/elasticsearch.yml
       - ./elasticsearch7/config/jvm.options:/usr/share/elasticsearch/config/jvm.options
       - ./elasticsearch7/config/log4j2.properties:/usr/share/elasticsearch/config/log4j2.properties
     environment:
       - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
       - "TZ=Asia/Shanghai"
       - "TAKE_FILE_OWNERSHIP=true"   #volumes 挂载权限 如果不想要挂载es文件改配置可以删除
     ulimits:
       memlock:
         soft: -1
         hard: -1
     networks:
       base-env-network:
         aliases:
          - elasticsearch
  kibana:
    image: docker.elastic.co/kibana/kibana:7.14.0
    container_name: kibana
    volumes:
      - ./elasticsearch7/config/kibana.yml:/usr/share/kibana/config/kibana.yml
    ports:
      - 15601:5601
    ulimits:
      nproc: 65535
      memlock: -1
    depends_on:
       - elasticsearch
    networks:                    
       base-env-network:
         aliases:
          - kibana
  logstash:
    image:  logstash:7.14.0
    container_name: logstash
    hostname: logstash
    restart: always
    ports:
      - 19600:9600
      - 15044:5044
    volumes:
      - ./logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf:rw
      - ./logstash/logstash.yml:/usr/share/logstash/config/logstash.yml
      - ./logstash/data:/home/logstash/data
    networks:
       base-env-network:
         aliases:
          - logstash
# docker network create base-env-network          
networks:
  base-env-network:
    external:
      name: "base-env-network"

访问kibana

http://cdh1:15601

SkyWalking

http://cdh2:13800/

kibana

以上的实操过程，请参见23章视频：《100Wqps 超高并发日志平台》实操

在kibana显示的效果

在kibana组件上查看，可以看到创建了一个filebeat开头的数据索引，如下图:

在日志搜索界面，可以看到service-hi应用输出的日志，如图所示：

3 Elasticsearch基础和实操

ES实操参考第38章：ElasticSearch 学习圣经：从0到1, 精通 ElasticSearch 工业级实操，这里不再重复

Elasticsearch 是一个分布式的开源搜索和分析引擎，在 Apache Lucene 的基础上开发而成。

Lucene 是开源的搜索引擎工具包，Elasticsearch 充分利用Lucene，并对其进行了扩展，使存储、索引、搜索都变得更快、更容易， 而最重要的是， 正如名字中的“ elastic ”所示， 一切都是灵活、有弹性的。而且，应用代码也不是必须用Java 书写才可以和Elasticsearc兼容，完全可以通过JSON 格式的HTTP 请求来进行索引、搜索和管理Elasticsearch 集群。

如果你已经听说过Lucene ，那么可能你也听说了Solr，

Solr也是开源的基于Lucene 的分布式搜索引擎，跟Elasticsearch有很多相似之处。

但是Solr 诞生于2004 年，而Elasticsearch诞生于2010，Elasticsearch凭借后发优势和更活跃的社区、更完备的生态系统，迅速反超Solr，成为搜索市场的第二代霸主。

Elasticsearch具有以下优势：

• Elasticsearch 很快。 由于 Elasticsearch 是在 Lucene 基础上构建而成的，所以在全文本搜索方面表现十分出色。Elasticsearch 同时还是一个近实时的搜索平台，这意味着从文档索引操作到文档变为可搜索状态之间的延时很短，一般只有一秒。因此，Elasticsearch 非常适用于对时间有严苛要求的用例，例如安全分析和基础设施监测。
• Elasticsearch 具有分布式的本质特征。 Elasticsearch 中存储的文档分布在不同的容器中，这些容器称为分片，可以进行复制以提供数据冗余副本，以防发生硬件故障。Elasticsearch 的分布式特性使得它可以扩展至数百台（甚至数千台）服务器，并处理 PB 量级的数据。
• Elasticsearch 包含一系列广泛的功能。 除了速度、可扩展性和弹性等优势以外，Elasticsearch 还有大量强大的内置功能（例如数据汇总和索引生命周期管理），可以方便用户更加高效地存储和搜索数据。
• Elastic Stack 简化了数据采集、可视化和报告过程。 人们通常将 Elastic Stack 称为 ELK Stack（代指Elasticsearch、Logstash 和 Kibana），目前 Elastic Stack 包括一系列丰富的轻量型数据采集代理，这些代理统称为 Beats，可用来向 Elasticsearch 发送数据。通过与 Beats 和 Logstash 进行集成，用户能够在向 Elasticsearch 中索引数据之前轻松地处理数据。同时，Kibana 不仅可针对 Elasticsearch 数据提供实时可视化，同时还提供 UI 以便用户快速访问应用程序性能监测 (APM)、日志和基础设施指标等数据。

4 filebeat基础和实操

当你要面对成百上千、甚至成千上万的服务器、虚拟机和容器生成的日志时，Filebeat 将为你提供一种轻量型方法，用于转发和汇总日志与文件，让简单的事情不再繁杂。

关于Filebeat，记住两点：

• 轻量级日志采集器
• 输送至 Elasticsearch 或 Logstash，在 Kibana 中实现可视化

官网文档https://www.elastic.co/guide/en/beats/filebeat/7.14/filebeat-overview.html

4.1 filebeat和beats的关系

filebeat是Beats中的一员。

Beats在是一个轻量级日志采集器，其实Beats家族有6个成员，目前Beats包含六种工具：

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

4.2 Filebeat工作原理

---

Filebeat由两个主要组件组成：inputs 和 harvesters （直译：收割机，采集器）。

这些组件一起工作以跟踪文件，并将事件数据发送到你指定的输出。

Filebeat的工作方式如下：

启动Filebeat时，它将启动一个或多个输入，这些输入将在为日志数据指定的位置中查找。

对于Filebeat所找到的每个日志，Filebeat都会启动收割机。

每个收割机都读取一个日志以获取新内容，并将新日志数据发送到libbeat，libbeat会汇总事件并将汇总的数据发送到您为Filebeat配置的输出。

Filebeat是一个轻量级日志传输Agent，可以将指定日志转发到Logstash、Elasticsearch、Kafka、Redis等中。

Filebeat占用资源少，而且安装配置也比较简单，支持目前各类主流OS及Docker平台。

Filebeat是用于转发和集中日志数据的轻量级传送程序。

作为服务器上的代理安装，Filebeat监视您指定的日志文件或位置，收集日志事件，并将它们转发到Elasticsearch或Logstash进行索引。

harvester是什么

一个harvester负责读取一个单个文件的内容。

harvester逐行读取每个文件（一行一行地读取每个文件），并把这些内容发送到输出。

每个文件启动一个harvester。

harvester负责打开和关闭这个文件，这就意味着在harvester运行时文件描述符保持打开状态。

在harvester正在读取文件内容的时候，文件被删除或者重命名了，那么Filebeat会续读这个文件。

这就有一个问题了，就是只要负责这个文件的harvester没用关闭，那么磁盘空间就不会释放。

默认情况下，Filebeat保存文件打开直到close_inactive到达。

input是什么

一个input负责管理harvesters，并找到所有要读取的源。

如果input类型是log，则input查找驱动器上与已定义的glob路径匹配的所有文件，并为每个文件启动一个harvester。

每个input都在自己的Go例程中运行。

下面的例子配置Filebeat从所有匹配指定的glob模式的文件中读取行：

filebeat.inputs:
- type: log
  paths:
    - /var/log/*.log
    - /var/path2/*.log

Filebeat如何保持文件状态

Filebeat 保存每个文件的状态，并经常刷新状态，并且将状态到磁盘上的注册文件（registry）。

状态用于记住harvester读取的最后一个偏移量，并确保所有日志行被发送（到输出）。

如果输出，比如Elasticsearch 或者 Logstash等，无法访问，那么Filebeat会跟踪已经发送的最后一行，并只要输出再次变得可用时继续读取文件。

当Filebeat运行时，会将每个文件的状态新保存在内存中。

当Filebeat重新启动时，将使用注册文件中的数据重新构建状态，Filebeat将在最后一个已知位置继续每个harvester。

对于每个输入，Filebeat保存它找到的每个文件的状态。

因为文件可以重命名或移动，所以文件名和路径不足以标识文件。对于每个文件，Filebeat存储惟一标识符，以检测文件是否以前读取过。

如果你的情况涉及每天创建大量的新文件，你可能会发现注册表文件变得太大了。

（画外音：Filebeat 保存每个文件的状态，并将状态保存到registry_file中的磁盘。当重新启动Filebeat时，文件状态用于在以前的位置继续读取文件。如果每天生成大量新文件，注册表文件可能会变得太大。为了减小注册表文件的大小，有两个配置选项可用：clean_remove 和 clean_inactive。对于你不再访问且被忽略的旧文件，建议您使用clean_inactive。如果想从磁盘上删除旧文件，那么使用clean_remove选项。）

Filebeat如何确保至少投递一次（at-least-once）？

Filebeat保证事件将被投递到配置的输出中至少一次，并且不会丢失数据。

Filebeat能够实现这种行为，因为它将每个事件的投递状态存储在注册文件中。

在定义的输出被阻塞且没有确认所有事件的情况下，Filebeat将继续尝试发送事件，直到输出确认收到事件为止。

如果Filebeat在发送事件的过程中关闭了，则在关闭之前它不会等待输出确认所有事件。当Filebeat重新启动时，发送到输出（但在Filebeat关闭前未确认）的任何事件将再次发送。

这确保每个事件至少被发送一次，但是你最终可能会将重复的事件发送到输出。你可以通过设置shutdown_timeout选项，将Filebeat配置为在关闭之前等待特定的时间。

4.3 Filebeat启动命令

下载地址：https://www.elastic.co/cn/downloads/past-releases#filebeat

./filebeat -e -c filebeat 配置文件

filebeat -e -c /path/to/your/filebeat.yml

• -e：会让 Filebeat 将日志输出到控制台。
• -c /path/to/your/filebeat.yml：会让 Filebeat 使用指定路径的配置文件，而不是默认的配置文件。

4.4 Filebeat文件夹结构

	描述
filebeat	用于启动filebeat的二进制文件
data	持久化数据文件的位置
logs	Filebeat创建的日志的位置
modules.d	简化filebeat配置的模板文件夹，如nginx/kafka等日志收集模板
filebeat.yml	filebeat配置文件

4.5 Filebeat配置参考

输入配置

为了手动配置Filebeat（代替用模块），你可以在filebeat.yml中的filebeat.inputs区域下指定一个inputs列表。

列表时一个YMAL数组，并且你可以指定多个inputs，相同input类型也可以指定多个。例如：

filebeat.inputs:
- type: log
  paths:
    - /var/log/system.log
    - /var/log/wifi.log
- type: log
  paths:
    - "/var/log/apache2/*"
  fields:
    apache: true
  fields_under_root: true

Filebeat常见的配置类型参考官网如下：

• AWS CloudWatch
• AWS S3
• Azure Event Hub
• Cloud Foundry
• Container
• Docker
• filestream
• GCP Pub/Sub
• HTTP Endpoint
• HTTP JSON
• Kafka
• Log
• MQTT
• NetFlow
• Office 365 Management Activity API
• Redis
• Stdin
• Syslog
• TCP
• UDP

log input

从日志文件读取行

为了配置这种input，需要指定一个paths列表，列表中的每一项必须能够定位并抓取到日志行。例如：

filebeat.inputs:
- type: log
  paths:
    - /var/log/messages
    - /var/log/*.log

你还可以应用设置其它额外的配置项（比如，fields, include_lines, exclude_lines, multiline等等）来从这些文件中读取行

你设置的这些配置对所有这种类型的input在获取日志行的时候都生效。

为了对不同的文件应用不同的配置，你需要定义多个input区域：

filebeat.inputs:
- type: log 　　# 从system.log和wifi.log中读取日志行
  paths:
    - /var/log/system.log
    - /var/log/wifi.log
- type: log 　　# 从apache2目录下的每一个文件中读取日志行，并且在输出的时候会加上额外的字段apache
  paths:
    - "/var/log/apache2/*"
  fields:
    apache: true
  fields_under_root: true

配置项

paths

例如：/var/log//.log 将会抓取/var/log子目录目录下所有.log文件。

它不会从/var/log本身目录下的日志文件。如果你应用recursive_glob设置的话，它将递归地抓取所有子目录下的所有.log文件。

recursive_glob.enabled

允许将扩展为递归glob模式。

启用这个特性后，每个路径中最右边的被扩展为固定数量的glob模式。

例如：/foo/**扩展到/foo， /foo/*， /foo/**，等等。

如果启用，它将单个**扩展为8级深度*模式。
这个特性默认是启用的，设置recursive_glob.enabled为false可以禁用它。

encoding

读取的文件的编码

下面是一些W3C推荐的简单的编码：

• plain, latin1, utf-8, utf-16be-bom, utf-16be, utf-16le, big5, gb18030, gbk, hz-gb-2312
• euc-kr, euc-jp, iso-2022-jp, shift-jis, 等等

plain编码是特殊的，因为它不校验或者转换任何输入。

exclude_lines

一组正则表达式，用于匹配你想要排除的行。

Filebeat会删除这组正则表达式匹配的行。默认情况下，没有行被删除。空行被忽略。

如果指定了multiline，那么在用exclude_lines过滤之前会将每个多行消息合并成一个单行。

下面的例子配置Filebeat删除以DBG开头的行：

filebeat.inputs:
- type: log
  ...
  exclude_lines: ['^DBG']

include_lines

一组正则表达式，用于匹配你想要包含的行。Filebeat只会导出那些匹配这组正则表达式的行。默认情况下，所有的行都会被导出。空行被忽略。

如果指定了multipline设置，每个多行消息先被合并成单行以后再执行include_lines过滤。

下面是一个例子，配置Filebeat导出以ERR或者WARN开头的行：

filebeat.inputs:
- type: log
  ...
  include_lines: ['^ERR', '^WARN']

（画外音：如果 include_lines 和 exclude_lines 都被定义了，那么Filebeat先执行 include_lines 后执行 exclude_lines，而与这两个选项被定义的顺序没有关系。include_lines 总是在 exclude_lines选项前面执行，即使在配置文件中 exclude_lines 出现在 include_lines的前面。）

下面的例子导出那些除了以DGB开头的所有包含sometext的行：

filebeat.inputs:
- type: log
  ...
  include_lines: ['sometext']
  exclude_lines: ['^DBG']

harvester_buffer_size

当抓取一个文件时每个harvester使用的buffer的字节数。默认是16384。

max_bytes

单个日志消息允许的最大字节数。超过max_bytes的字节将被丢弃且不会被发送。对于多行日志消息来说这个设置是很有用的，因为它们往往很大。默认是10MB（10485760）。

json

这些选项使得Filebeat将日志作为JSON消息来解析。例如：

json.keys_under_root: true
json.add_error_key: true
json.message_key: log

为了启用JSON解析模式，你必须至少指定下列设置项中的一个：

　　keys_under_root

　　默认情况下，解码后的JSON被放置在一个以"json"为key的输出文档中。如果你启用这个设置，那么这个key在文档中被复制为顶级。默认是false。

　　overwrite_keys

　　如果keys_under_root被启用，那么在key冲突的情况下，解码后的JSON对象将覆盖Filebeat正常的字段

　　add_error_key

　　如果启用，则当JSON反编排出现错误的时候Filebeat添加 "error.message" 和 "error.type: json"两个key，或者当没有使用message_key的时候。

　　message_key

　　一个可选的配置，用于在应用行过滤和多行设置的时候指定一个JSON key。指定的这个key必须在JSON对象中是顶级的，而且其关联的值必须是一个字符串，否则没有过滤或者多行聚集发送。

　　ignore_decoding_error

　　一个可选的配置，用于指定是否JSON解码错误应该被记录到日志中。如果设为true，错误将被记录。默认是false。

multiline

用于控制Filebeat如何扩多行处理日志消息

exclude_files

一组正则表达式，用于匹配你想要忽略的文件。默认没有文件被排除。

下面是一个例子，忽略.gz的文件

filebeat.inputs:
- type: log
  ...
  exclude_files: ['\.gz$']

ignore_older

如果启用，那么Filebeat会忽略在指定的时间跨度之前被修改的文件。如果你想要保留日志文件一个较长的时间，那么配置ignore_older是很有用的。例如，如果你想要开始Filebeat，但是你只想发送最近一周最新的文件，这个情况下你可以配置这个选项。

你可以用时间字符串，比如2h（2小时），5m（5分钟）。默认是0，意思是禁用这个设置。

你必须设置ignore_older比close_inactive更大。

close_*

close_*配置项用于在一个确定的条件或者时间点之后关闭harvester。关闭harvester意味着关闭文件处理器。如果在harvester关闭以后文件被更新，那么在scan_frequency结束后改文件将再次被拾起。然而，当harvester关闭的时候如果文件被删除或者被移动，那么Filebeat将不会被再次拾起，并且这个harvester还没有读取的数据将会丢失。

close_inactive

当启用此选项时，如果文件在指定的持续时间内未被获取，则Filebeat将关闭文件句柄。当harvester读取最后一行日志时，指定周期的计数器就开始工作了。它不基于文件的修改时间。如果关闭的文件再次更改，则会启动一个新的harvester，并且在scan_frequency结束后，将获得最新的更改。

推荐给close_inactive设置一个比你的日志文件更新的频率更大一点儿的值。例如，如果你的日志文件每隔几秒就会更新，你可以设置close_inactive为1m。如果日志文件的更新速率不固定，那么可以用多个配置。

将close_inactive设置为更低的值意味着文件句柄可以更早关闭。然而，这样做的副作用是，如果harvester关闭了，新的日志行不会实时发送。

关闭文件的时间戳不依赖于文件的修改时间。代替的，Filebeat用一个内部时间戳来反映最后一次读取文件的时间。例如，如果close_inactive被设置为5分钟，那么在harvester读取文件的最后一行以后，这个5分钟的倒计时就开始了。

你可以用时间字符串，比如2h（2小时），5m（5分钟）。默认是5m。

close_renamed

当启用此选项时，Filebeat会在重命名文件时关闭文件处理器。默认情况下，harvester保持打开状态并继续读取文件，因为文件处理器不依赖于文件名。如果启用了close_rename选项，并且重命名或者移动的文件不再匹配文件模式的话，那么文件将不会再次被选中。Filebeat将无法完成文件的读取。

close_removed

当启用此选项时，Filebeat会在删除文件时关闭harvester。通常，一个文件只有在它在由close_inactive指定的期间内不活跃的情况下才会被删除。但是，如果一个文件被提前删除，并且你不启用close_removed，则Filebeat将保持文件打开，以确保harvester已经完成。如果由于文件过早地从磁盘中删除而导致文件不能完全读取，请禁用此选项。

close_timeout

当启用此选项是，Filebeat会给每个harvester一个预定义的生命时间。无论读到文件的什么位置，只要close_timeout周期到了以后就会停止读取。当你想要在文件上只花费预定义的时间时，这个选项对旧的日志文件很有用。尽管在close_timeout时间以后文件就关闭了，但如果文件仍然在更新，则Filebeat将根据已定义的scan_frequency再次启动一个新的harvester。这个harvester的close_timeout将再次启动，为超时倒计时。

scan_frequency

Filebeat多久检查一次指定路径下的新文件（PS：检查的频率）。例如，如果你指定的路径是 /var/log/* ，那么会以指定的scan_frequency频率去扫描目录下的文件（PS：周期性扫描）。指定1秒钟扫描一次目录，这还不是很频繁。不建议设置为小于1秒。

如果你需要近实时的发送日志行的话，不要设置scan_frequency为一个很低的值，而应该调整close_inactive以至于文件处理器保持打开状态，并不断地轮询你的文件。

默认是10秒。

scan.sort

如果你指定了一个非空的值，那么你可以决定用scan.order的升序或者降序。可能的值是 modtime 和 filename。为了按文件修改时间排序，用modtime，否则用 filename。默认此选项是禁用的。

scan.order

可能的值是 asc 或者 desc。默认是asc。

更多配置请查看

https://www.elastic.co/guide/en/beats/filebeat/current/configuration-filebeat-options.html

这里再重点说一下 ignore_older , close_inactive , scan_frequency 这三个配置项

• ignore_older： 它是设置一个时间范围（跨度），不在这个跨度范围之内的文件更新都不管
• scan_frequency： 它设置的是扫描文件的频率，看看文件是否更新
• close_inactive：它设置的是文件如果多久没更新的话就关闭文件句柄，它是有一个倒计时，如果在倒计时期间，文件没有任何变化，则当倒计时结束的时候关闭文件句柄。不建议设置为小于1秒。

如果文件句柄关了以后，文件又被更新，那么在下一个扫描周期结束的时候变化发现这个改变，于是会再次打开这个文件读取日志行，前面我们也提到过，每个文件上一次读到什么位置（偏移量）都记录在registry文件中。

特别说明：multiline管理多行消息

Filebeat获取的文件可能包含跨多行文本的消息。

例如，多行消息在包含Java堆栈跟踪的文件中很常见。为了正确处理这些多行事件，你需要在filebeat.yml中配置multiline以指定哪一行是单个事件的一部分。

你可以在filebeat.yml的filebeat.inputs区域指定怎样处理跨多行的消息。例如：

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

上面的例子中，Filebeat将所有不以 [ 开始的行与之前的行进行合并。

multiline.pattern

指定用于匹配多行的正则表达式

multiline.negate

定义模式是否被否定。默认false。

multiline.match

指定Filebeat如何把多行合并成一个事件。可选的值是 after 或者 before。

这种行为还收到negate的影响：

multiline.flush_pattern

指定一个正则表达式，多行将从内存刷新到磁盘。

multiline.max_lines

可以合并成一个事件的最大行数。如果一个多行消息包含的行数超过max_lines，则超过的行被丢弃。默认是500。

输出配置

常见的输出配置类型，参考官网如下：

• Elasticsearch Service
• Elasticsearch
• Logstash
• Kafka
• Redis
• File
• Console

Logstash output

output.logstash:
  hosts: ["127.0.0.1:5044"]

上面是配置Filebeat输出到Logstash，那么Logstash本身也有配置，例如：

input {
  beats {
    port => 5044
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" 
  }
}

负载均衡

为了启用负载均衡，当你配置输出的时候你需要指定 loadbalance: true

output.logstash:
  hosts: ["localhost:5044", "localhost:5045"]
  loadbalance: true

4.6 实操案例

4.6.1 标准输入，控制台输出

#安装
yum -y localinstall filebeat-7.17.3-x86_64.rpm

mkdir ~/filebeat/config
cat > ~/filebeat/config/01-stdin-to-console.yml <<'EOF'
# 指定输⼊的类型
filebeat.inputs:
# 指定输⼊的类型为"stdin",表示标准输⼊
- type: stdin
# 指定输出的类型
output.console:
  # 打印漂亮的格式
  pretty: true
EOF

#运⾏filebeat实例
~/filebeat/filebeat -e -c ~/filebeat/config/01-stdin-to-console.yml

4.6.2 日志输入，控制台输出

filebeat配置文件

filebeat.inputs:
  - type: log
    # 是否启动当前的输入类型，默认值为true
    enabled: true
    # 指定数据路径
    paths:
      - /tmp/nginx.log
      - /tmp/*.txt
    # 给当前的输入类型打上标签
    tags: ["elk_nginx","容器运维","DBA运维","SRE运维工程师"]
    # 自定义字段
    fields:
      school: "北京昌平区沙河镇"
      class: "linux80"
  - type: log
    enabled: true
    paths:
      - /tmp/tomcat.log
    tags: ["elk_tomcat","云原生开发"]
    fields:
      name: "mickael"
      hobby: "linux,抖音"
    # 将自定义字段的key-value放到顶级字段.
    # 默认值为false，会将数据放在一个叫"fields"字段的下面.
    fields_under_root: true
output.console:
  pretty: true

测试命令

  echo aaa > /tmp/test.log
  echo bbbb > /tmp/test.log
  echo bbbb >> /tmp/test.log
  echo 123 > /tmp/test.log

结果

4.6.3 日志输入，ES输出

filebeat配置文件

filebeat.inputs:
- type: log
  # 是否启动当前的输入类型，默认值为true
  enabled: true
  # 指定数据路径
  paths:
    - /tmp/nginx.log
    - /tmp/*.txt
  # 给当前的输入类型打上标签
  tags: ["elk_nginx","容器运维","DBA运维","SRE运维工程师"]
  # 自定义字段
  fields:
    school: "北京昌平区沙河镇"
    class: "linux80"
- type: log
  enabled: true
  paths:
    - /tmp/tomcat.log
  tags: ["elk_tomcat","云原生开发"]
  fields:
    name: "mickael"
    hobby: "linux,抖音"
  # 将自定义字段的key-value放到顶级字段.
  # 默认值为false，会将数据放在一个叫"fields"字段的下面.
  fields_under_root: true

output.elasticsearch:
  enabled: true
  hosts:
    ["http://localhost:9200"]
  # index: "elk_nginx-%{+yyyy.MM.dd}"
  indices:
    - index: "elk_nginx-%{+yyyy.MM.dd}"
      # 匹配指定字段包含的内容
      when.contains:
        tags: "elk_nginx"
    - index: "elk_tomcat-%{+yyyy.MM.dd}"
      when.contains:
        tags: "elk_tomcat"
  # 禁⽤索引⽣命周期管理
  setup.ilm.enabled: false
  # 设置索引模板的名称
  setup.template.name: "elk_"
  # 设置索引模板的匹配模式
  setup.template.pattern: "elk_*"

测试命令

echo nginx_log > nginx.log
echo tomcat_log > tomcat.log

结果

4.6.4 企业级实战案例：Tomcat日志

4.6.4.1 原生tomcat日志

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /root/apache-tomcat-9.0.97/logs/*.txt
  tags: ["access"]

output.elasticsearch:
  enabled: true
  hosts:
    ["http://localhost:9200"]
  index: "elk_tomcat-%{+yyyy.MM.dd}"
  # 禁⽤索引⽣命周期管理
  setup.ilm.enabled: false
  # 设置索引模板的名称
  setup.template.name: "elk_"
  # 设置索引模板的匹配模式
  setup.template.pattern: "elk_*"

可以看到访问日志，是原生的字符串，存储在一起的，比如想要获取响应码=200的日志，就不方便了

对于原生的日志，进行解析分析，基本2种方案

1. 应用设置日志格式为json，然后配置json.keys_under_root: true解析json格式日志，比如spirng boot应用的
2. 使用logstash

4.6.4.2 json格式tomcat日志

先看方案1，方案2参考logstash章节

方案1，设置tomcat日志格式配置如下，conf/server.xml

        <Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs"
                prefix="localhost_access_log" suffix=".txt"
                pattern="{
    &quot;clientip&quot;:&quot;%h&quot;,&quot;ClientUser&quot;:&quot;%l&quot;,&quot;authenticated&quot;:&quot;%u&quot;,&quot;AccessTime&quot;:&quot;%t&quot;,&quot;request&quot;:&quot;%r&quot;,&quot;status&quot;:&quot;%s&quot;,&quot;SendBytes&quot;:&quot;%b&quot;,&quot;Query?string&quot;:&quot;%q&quot;,&quot;partner&quot;:&quot;%{Referer}i&quot;,&quot;http_user_agent&quot;:&quot;%{UserAgent}i&quot;}" />

        <!--pattern="%h %l %u %t &quot;%r&quot; %s %b" -->

解析tomcat的json日志的filebeat配置

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /root/apache-tomcat-9.0.97/logs/*.txt
  # 解析message字段的json格式，并放在顶级字段中
  json.keys_under_root: true
  tags: ["access"]

output.elasticsearch:
  enabled: true
  hosts:
    ["http://localhost:9200"]
  index: "elk_tomcat-%{+yyyy.MM.dd}"
# 禁⽤索引⽣命周期管理
setup.ilm.enabled: false
# 设置索引模板的名称
setup.template.name: "elk_"
# 设置索引模板的匹配模式
setup.template.pattern: "elk_*"

结果：

4.6.4.3 多行匹配，收集tomcat错误日志

Filebeat获取的文件可能包含跨多行文本的消息。

例如，多行消息在包含Java堆栈跟踪的文件中很常见。为了正确处理这些多行事件，你需要在filebeat.yml中配置multiline以指定哪一行是单个事件的一部分。

你可以在filebeat.yml的filebeat.inputs区域指定怎样处理跨多行的消息。例如：

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

上面的例子中，Filebeat将所有不以 [ 开始的行与之前的行进行合并。

multiline.pattern

指定用于匹配多行的正则表达式

multiline.negate

定义模式是否被否定。默认false。

multiline.match

指定Filebeat如何把多行合并成一个事件。可选的值是 after 或者 before。

这种行为还收到negate的影响：

multiline.flush_pattern

指定一个正则表达式，多行将从内存刷新到磁盘。

multiline.max_lines

可以合并成一个事件的最大行数。如果一个多行消息包含的行数超过max_lines，则超过的行被丢弃。默认是500。

filebeat配置文件

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /root/apache-tomcat-9.0.97/logs/*.out
  # 指定多⾏匹配的类型，可选值为"pattern","count"
  multiline.type: pattern
  # 指定匹配模式
  multiline.pattern: '^\d{2}'
  # 下⾯2个参数参考官⽅架构图即可，如上图所示。
  multiline.negate: true
  multiline.match: after
  tags: ["access"]

output.elasticsearch:
  enabled: true
  hosts:
    ["http://localhost:9200"]
  index: "elk_tomcat-%{+yyyy.MM.dd}"
# 禁⽤索引⽣命周期管理
setup.ilm.enabled: false
# 设置索引模板的名称
setup.template.name: "elk_"
# 设置索引模板的匹配模式
setup.template.pattern: "elk_*"

5 logstash基础和实操

简单来说logstash就是一根具备实时数据传输能力的管道，负责将数据信息从管道的输入端传输到管道的输出端；与此同时这根管道还可以让你根据自己的需求在中间加上滤网，Logstash提供里很多功能强大的滤网以满足你的各种应用场景。

logstash常用于日志系统中做日志采集设备，最常用于ELK中作为日志收集器使用，比如对于非结构化日志数据的解析和处理，就可以使用logstash

5.1 logstash作用：

集中、转换和存储你的数据，是一个开源的服务器端数据处理管道，可以同时从多个数据源获取数据，并对其进行转换，然后将其发送到你最喜欢的“存储

5.2 logstash的架构：

logstash的基本流程架构：input | filter | output 如需对数据进行额外处理，filter可省略。logstash架构和filebeat很想，也可以说filebeat不分功能模仿的logstash

Input(输入）：

采集各种样式，大小和相关来源数据，从各个服务器中收集数据。

数据往往以各种各样的形式，或分散或集中地存在于很多系统中。
Logstash 支持各种输入选择 ，可以在同一时间从众多常用来源捕捉事件。
能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。

input：必须，负责产生事件（Inputs generate events），

常用：File、syslog、redis、beats（如：Filebeats）

Filter(过滤器）

用于在将event通过output发出之前，对其实现某些处理功能。

filters：可选，负责数据处理与转换（filters modify them），

常用：grok、mutate、drop、clone、geoip

grok：用于分析结构化文本数据。

Output(输出）：

将我们过滤出的数据保存到那些数据库和相关存储中。

outputs：必须，负责数据输出（outputs ship them elsewhere），

常用：elasticsearch、file、graphite、statsd

5.3 Logstash的不足

早期的ELK架构中使用Logstash收集、解析日志，

但是：Logstash对内存、cpu、io等资源消耗比较高。

相比Logstash，Beats所占系统的CPU和内存几乎可以忽略不计。

所以，在收集这块，一般使用filebeat 代替 Logstash

5.4 logstash读取filebeat-输出到es集群

在分布式系统中，一台主机可能有多个应用，应用将日志输出到主机的指定目录，这时由logstash来搬运日志并解析日志，然后输出到elasticsearch上。

由于于logstash是java应用，解析日志是非的消耗cpu和内存，logstash安装在应用部署的机器上显得非常的笨重。

最常见的做法是用filebeat部署在应用的机器上，logstash单独部署，然后由filebeat将日志输出给logstash解析，解析完由logstash再传给elasticsearch。

在上面的配置中，输入数据源为filebeat，输出源为elasticsearch。

修改logstash的安装目录的config目录下的logstash.conf文件，配置如下：

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


filter {

    if "message-dispatcher" in [tags]{
        grok {
            match => ["message", "%{TIMESTAMP_ISO8601:time}\s* \s*%{NOTSPACE:thread-id}\s* \s*%{LOGLEVEL:level}\s* \s*%{JAVACLASS:class}\s* \- \s*%{JAVALOGMESSAGE:logmessage}\s*"]
        }

    }

    if "ExampleApplication" in [tags]{
        grok {
            match => ["message", "%{TIMESTAMP_ISO8601:time}\s* \s*%{NOTSPACE:thread-id}\s* \s*%{LOGLEVEL:level}\s* \s*%{JAVACLASS:class}\s* \- \s*%{JAVALOGMESSAGE:logmessage}\s*"]
        }

    }
    mutate {
        remove_field => "log"
        remove_field => "beat"
        remove_field => "meta"
        remove_field => "prospector"
        remove_field => "[host][os]"
    }
}


output {
    stdout { codec => rubydebug }
    if "message-dispatcher" in [tags]{
        elasticsearch {
           hosts => [ "elasticsearch:9200" ]
           index => "message-dispatcher-%{+yyyy.MM.dd}"      
        }
    }
    if "ExampleApplication" in [tags]{
        elasticsearch {
           hosts => [ "elasticsearch:9200" ]
           index => "ExampleApplication-%{+yyyy.MM.dd}"      
        }
    }    
}

更多的输入和输出源的配置见官网

https://www.elastic.co/guide/en/logstash/current/advanced-pipeline.html

5.5 logstash实操案例

5.5.1 标准输入，标准输出

#安装
yum -y localinstall logstash-7.17.3-x86_64.rpm
ln -sv /usr/share/logstash/bin/logstash /usr/local/bin/


cat > 01-stdin-to-stdout.conf <<'EOF'
input {
  stdin {}
}

output {
  stdout {}
}
EOF


#检查配置⽂件语法
logstash -tf 01-stdin-to-stdout.conf
#启动logstash实例
logstash -f 01-stdin-to-stdout.conf

5.5.2 tcp输入，标准输出

input {
   
  tcp {
   
    port => 8888
  }
  tcp {
   
    port => 9999
  }
}
output {
   
  stdout {
   }
}

测试输入

telnet localhost 8888
> aaaa
> bbbb

5.5.3 http输入，标准输出

input {
   
  http {
   
    port => 8888
  }
  http {
   
    port => 9999
  }
}
output {
   
  stdout {
   }
}

浏览器访问http://cdh1:8888

5.5.4 gork插件

Grok 是一个十分强大的 Logstash Filter 插件，它可以通过正则解析任意文本，将非结构化日志数据格式转换为结构化的、方便查询的结构。

它是目前 Logstash 中解析非结构化日志数据最好的方式。

Grok 的语法规则是：
这里的 “语法” 指的是匹配模式，例如，使用 NUMBER 模式可以匹配出数字，IP 模式则会匹配出 127.0.0.1 这样的 IP 地址。比如按以下格式输入内容：

172.16.213.132 [16/Jun/2020:16:24:19 +0800] "GET / HTTP/1.1" 403 5039

那么，
• %{IP:clientip} 匹配模式将获得的结果为：clientip: 172.16.213.132
• %{HTTPDATE:timestamp} 匹配模式将获得的结果为：timestamp: 16/Jun/2020:16:24:19 +0800
• %{QS:referrer} 匹配模式将获得的结果为：referrer: “GET / HTTP/1.1”
到这里为止，我们已经获取了上面输入中前三个部分的内容，分别是 clientip、timestamp 和 referrer 三个字段。

如果要获取剩余部分的信息，方法类似。

要在线调试 Grok，可以点击在线调试，可点击这里进行在线调试，非常方便。

下面是一个组合匹配模式，它可以获取上面输入的所有内容：

%{
   IP:clientip}\ \[%{
   HTTPDATE:timestamp}\]\ %{
   QS:referrer}\ %{
   NUMBER:response}\ %{
   NUMBER:bytes}

正则匹配是非常严格的匹配，在这个组合匹配模式中，使用了转义字符 \，这是因为输入的内容中有空格和中括号。

通过上面这个组合匹配模式，我们将输入的内容分成了 5 个部分，即 5 个字段。

将输入内容分割为不同的数据字段，这对于日后解析和查询日志数据非常有用，这正是我们使用 grok 的目的。

Logstash 默认提供了近 200 个匹配模式（其实就是定义好的正则表达式）让我们来使用，可以在 Logstash 安装目录下找到。

例如，我这里的路径为：

/usr/local/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns

此目录下有定义好的各种匹配模式，基本匹配定义在 grok-patterns 文件中。

从这些定义好的匹配模式中，可以查到上面使用的四个匹配模式对应的定义规则。

除此之外，还有很多默认定义好的匹配模式文件，比如 httpd、java、linux-syslog、redis、mongodb、nagios 等，这些已经定义好的匹配模式，可以直接在 Grok 过滤器中进行引用。

当然也可以定义自己需要的匹配模式。

在了解完 Grok 的匹配规则之后，下面通过一个配置实例深入介绍下 Logstash 是如何将非结构化日志数据转换成结构化数据的。

首先看下面的一个事件配置文件：

input{
   
  stdin{
   }
}
filter{
   
   grok{
   
     match => ["message", "%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %         {NUMBER:response}\ %{NUMBER:bytes}"]
   }
}
output{
   
   stdout{
   
     codec => "rubydebug"
   }
}

在这个配置文件中，输入配置成了 stdin，在 filter 中添加了 grok 过滤插件，并通过 match 来执行正则表达式解析，

grok 中括号中的正则表达式就是上面提到的组合匹配模式，然后通过 rubydebug 编码格式输出信息。

这样的组合有助于调试和分析输出结果。

通过此配置启动 Logstash 进程后，我们仍然输入之前给出的那段内容：

172.16.213.132 [16/Jun/2020:16:24:19 +0800] "GET / HTTP/1.1" 403 5039

然后，查看 rubydebug 格式的日志输出，内容如下：

{
   
     "timestamp" => "16/Jun/2020:16:24:19 +0800",
      "response" => "403",
         "bytes" => "5039",
      "@version" => "1",
      "clientip" => "172.16.213.132",
          "host" => "nnmaster.cloud",
      "referrer" => "\"GET / HTTP/1.1\"",
       "message" => "172.16.213.132 [16/Jun/2020:16:24:19 +0800] \"GET / HTTP/1.1\" 403 5039",
    "@timestamp" => 2020-06-16T07:46:53.120Z
}

从这个输出可知，通过 Grok 定义好的 5 个字段都获取到了内容，并正常输出了。

案例配置

input {
   
  stdin {
   }
}
filter {
   
  grok {
   
    match => {
   
      "message" => "%{
   IP:clientIP} %{
   WORD:method} %{
   URIPATHPARAM:request} %{
   NUMBER:bytes} %{
   NUMBER:duration}"
    }
  }
}
output {
   
  stdout {
   }
}

输入内容

55.3.244.1 GET /index.html 15824 0.043
10.0.0.103 POST /oldboyedu.html 888888 5.20

测试结果

5.5.5 filebeat输入，输出es(tomcat日志)

filebeat配置文件

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /root/apache-tomcat-9.0.97/logs/*.txt
  json.keys_under_root: true

output.logstash:
  hosts: ["127.0.0.1:7777"]

logstash配置文件

input {
  beats {
    port => 7777
  }
}
filter {
    mutate {
      # 移除指定的字段
      remove_field => ["tags","log","agent","@version", "input","ecs"]
    }
}
output {
    stdout {}

    elasticsearch {
      hosts => ["127.0.0.1:9200"]
      index => "elk_tomcat"
    }
}

6 Kibana基础和实操

ES实操参考第38章：ElasticSearch 学习圣经：从0到1, 精通 ElasticSearch 工业级实操，这里不再重复

这里补充基于tomcat日志，Kibana可视化库和dashborad基本用法

1. PV指标

Page View(简称:"PV") ：
    ⻚⾯访问或点击量。
kibana界⾯⿏标依次点击如下:
    (1)菜单栏;
    (2)Visualize Library(可视化库);
    (3)新建可视化
    (4)基于聚合
    (5)指标
    (6)选择索引模式(例如"elk_*")
    (7)指标栏中选择:
        聚合: 计数
        定制标签: PV

1. 基于clientip的PV指标

客户端IP:
    通常指的是访问Web服务器的客户端IP地址，但要注意，客户端IP数量并不难代表UV。
kibana界⾯⿏标依次点击如下:
    (1)菜单栏;
    (2)Visualize Library(可视化库);
    (3)创建可视化
    (4)基于聚合
    (5)指标
    (6)选择索引模式(例如"elk_*")
    (7)指标栏中选择:    聚合: 唯⼀计数
        字段: clientip.keyword
        定制标签: IP

1. 带宽

带宽:
    统计nginx返回给客户端⽂件⼤⼩的字段进⾏累计求和。
kibana界⾯⿏标依次点击如下:
    (1)菜单栏;
    (2)Visualize Library(可视化库);
    (3)创建可视化
    (4)基于聚合
    (5)指标
    (6)选择索引模式(例如"elk_*")
    (7)指标栏中选择:
        聚合: 求和
        字段: SendBytes
        定制标签: 带宽

1. 客户端城市分布

分析客户端的城市分布:
    需要借助logstash的filter插件的geoip实现对客户端的IP地址进⾏地域解析。
kibana界⾯⿏标依次点击如下:
    (1)菜单栏;
    (2)Visualize Library(可视化库);
    (3)创建可视化
    (4)基于聚合
    (5)垂直条形图
    (6)选择索引模式(例如"elk_*")
    (7)指标栏中设置(即Y轴)
        聚合: 计数
        定制标签: 城市分布
    (8)添加"存储痛"，选择"X"轴
        聚合: 词
        字段: city_name.keyword
        ...
        定制标签: 城市名称

1. IP的TopN统计:

IP的TopN统计:
    统计访问量的客户端IP最⼤的是谁。
kibana界⾯⿏标依次点击如下:
    (1)菜单栏;
    (2)Visualize Library(可视化库);
    (3)创建可视化
    (4)基于聚合
    (5)仪表盘
    (6)选择索引模式(例如"elk_*")
    (7)指标栏中设置(即Y轴)
        聚合: 计数
    (8)添加"存储痛"，选择"X"轴
        聚合: 词
        字段: client.keyword
        顺序: 降序
        ⼤⼩: 3
        ...

7 ELK综合实操

使用FELK架构部署监控app日志

使用filebeat发送日志

制作filebeat镜像

官方文档

https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-getting-started.html

下载filebeat，下载命令如下：

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

 wget  https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.2.0-linux-x86_64.tar.gz


tar -zxvf filebeat-7.2.0-linux-x86_64.tar.gz
mv filebeat-7.2.0-linux-x86_64 /usr/share/
cd /usr/share/filebeat-7.2.0-linux-x86_64/

制作基础的unbantu镜像

why unbantu？ not alpine? not centos？

Alpine 只有仅仅 5 MB 大小，并且拥有很友好的包管理机制。

Docker 官方推荐使用 Alpine 替代 Ubuntu 做为容器的基础镜像。

曾经尝试使用alpine:3.7作为底层镜像, 按照zookeeper，但是一直启动不来，换成了centos的镜像，排查过程反复实验，耗时很久。

网上小伙伴构建filebeat镜像，基于alpine:3.7, 构建后的镜像运行时报“standard_init_linux.go:190: exec user process caused "no such file or directory"”，故最后还是选择ubuntu。

这里选择ubuntu的原因,是其作为底层打包出来的镜像比centos要小很多。

# 基础镜像 生成的镜像作为基础镜像
FROM ubuntu:18.04


# 指定维护者的信息
MAINTAINER 尼恩@疯狂创客圈

# RUN apt-get update  && apt-get -y install openjdk-8-jdk



#install wget,sudo,python,vim,ping and ssh command

RUN sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list   && apt-get  clean && \
    apt-get update && apt-get -y install wget && apt-get -y install sudo && \
    apt-get -y install iputils-ping && \
    apt-get -y install net-tools && \
    apt install -y tzdata && \
    rm -rf /etc/localtime  && ln -s /usr/share/zoneinfo/Asia/Shanghai /etc/localtime &&  dpkg-reconfigure -f noninteractive tzdata && \
    apt-get  clean

 #    echo "Asia/Shanghai" > /etc/timezone &&  dpkg-reconfigure -f noninteractive tzdata && \



# RUN dpkg-reconfigure -f noninteractive tzdata

# RUN apt-get clean


 #apt-get -y install python && \
 # apt-get -y install vim && \
 #  apt-get -y install openssh-server && \
 # apt-get -y install python-pip  && \


# 复制并解压
ADD jdk-8u121-linux-x64.tar.gz /usr/local/


ENV work_path /usr/local
WORKDIR $work_path

# java
ENV JAVA_HOME /usr/local/jdk1.8.0_121
ENV JRE_HOME  /usr/local/jdk1.8.0_121/jre
ENV CLASSPATH .:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
ENV PATH ${PATH}:${JAVA_HOME}/bin

dockfile add命令：

ADD指令的功能是将主机构建环境（上下文）目录中的文件和目录、以及一个URL标记的文件 拷贝到镜像中。

其格式是： ADD 源路径 目标路径

注意事项：

1、如果源路径是个文件，且目标路径是以 / 结尾， 则docker会把目标路径当作一个目录，会把源文件拷贝到该目录下。

如果目标路径不存在，则会自动创建目标路径。

2、如果源路径是个文件，且目标路径是不是以 / 结尾，则docker会把目标路径当作一个文件。

如果目标路径不存在，会以目标路径为名创建一个文件，内容同源文件；

如果目标文件是个存在的文件，会用源文件覆盖它，当然只是内容覆盖，文件名还是目标文件名。

如果目标文件实际是个存在的目录，则会源文件拷贝到该目录下。 注意，这种情况下，最好显示的以 / 结尾，以避免混淆。

3、如果源路径是个目录，且目标路径不存在，则docker会自动以目标路径创建一个目录，把源路径目录下的文件拷贝进来。

如果目标路径是个已经存在的目录，则docker会把源路径目录下的文件拷贝到该目录下。

4、如果源文件是个归档文件（压缩文件，比如 .tar文件），则docker会自动帮解压。

推送镜像到dockerhub

这个镜像解决了jdk问题，时区问题

推送到了dockerhub，大家可以直接作为基础镜像使用

docker login

docker tag 8d0abdffe76f nien/ubuntu:18.04

docker push nien/ubuntu:18.04

制作filebeat镜像

官方文档

https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-getting-started.html

下载filebeat，下载命令如下：

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

 wget  https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.2.0-linux-x86_64.tar.gz


tar -zxvf filebeat-7.2.0-linux-x86_64.tar.gz
mv filebeat-7.2.0-linux-x86_64 /usr/share/
cd /usr/share/filebeat-7.2.0-linux-x86_64/

dockerfile

# 基础镜像 生成的镜像作为基础镜像
FROM nien/ubuntu:18.04


# 指定维护者的信息
MAINTAINER 尼恩@疯狂创客圈


# 复制并解压
ADD filebeat-7.14.0-linux-x86_64.tar.gz /usr/local/

构建镜像

docker build -t filebeat:7.14.0  .

构建之后，进入容器，可以看到 /usr/local 目录下的filebeat-7.14.0-linux-x86_64

[root@cdh2 filebeat]# docker run -it filebeat:7.14.0 /bin/bash
root@7ba04f21f26e:/usr/local# ll
total 48
drwxr-xr-x 1 root root 4096 Apr  2 09:26 ./
drwxr-xr-x 1 root root 4096 Mar 16 03:27 ../
drwxr-xr-x 2 root root 4096 Mar 16 03:27 bin/
drwxr-xr-x 2 root root 4096 Mar 16 03:27 etc/
drwxr-xr-x 5 root root 4096 Apr  2 09:26 filebeat-7.14.0-linux-x86_64/
drwxr-xr-x 2 root root 4096 Mar 16 03:27 games/
drwxr-xr-x 2 root root 4096 Mar 16 03:27 include/
drwxr-xr-x 8 uucp  143 4096 Dec 13  2016 jdk1.8.0_121/
drwxr-xr-x 2 root root 4096 Mar 16 03:27 lib/
lrwxrwxrwx 1 root root    9 Mar 16 03:27 man -> share/man/
drwxr-xr-x 2 root root 4096 Mar 16 03:27 sbin/
drwxr-xr-x 1 root root 4096 Apr  2 00:44 share/
drwxr-xr-x 2 root root 4096 Mar 16 03:27 src/

推送镜像到dockerhub

这个镜像解决了jdk问题，时区问题

推送到了dockerhub，大家可以直接作为基础镜像使用

[root@cdh2 filebeat]# docker tag fb44037ab5f9 nien/filebeat:7.14.0

[root@cdh2 filebeat]# docker push nien/filebeat:7.14.0
The push refers to repository [docker.io/nien/filebeat]
069c957c7a4e: Pushing [=======>                                           ]  19.99MB/140MB
b17e3cbc28a1: Mounted from nien/ubuntu
5695cc8dd56c: Mounted from nien/ubuntu
9d6787a516e7: Mounted from nien/ubuntu

如果要收集日志，就可以用这个基础镜像加点配置就ok啦

example-application微服务的filebeat配置：

实操过程，请参见23章视频：《100Wqps 超高并发日志平台》实操

filebeat.yml的参考配置：

# ============================== Filebeat inputs ===============================
filebeat.config.inputs:
  enable: true
  path: /work/filebeat/input.yml
  reload.enabled: true
  reload.period: 2s

# ============================== 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: true

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

#----------------------------- Logstash output --------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["cdh1:15044"]

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

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

  # Client Certificate Key
  #ssl.key: "/etc/pki/client/cert.key"

输出到logstsh的地址为logstash，这里用的是容器的名称， logstash和 这个微服务，需要在同一个网络。

如果不是，可以使用虚拟机的名称，然后把 5044，映射到15044

input.yml配置：

主要配置的是日志的搜集目录为/work/logs/output.log，这个目录是应用message-dispatcher输出日志的文件。

由于其他的微服务也是固定在这个 文件，

所以这个路径，基本可以固定。

#filebeat.input:
- type: log

  # Change to true to enable this input configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /work/logs/info/*.log
    - /work/logs/error/*.log

  #
  # - /work/logs/output.log



  multiline:
    pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'   # 指定匹配的表达式（匹配以 2017-11-15 08:04:23:889 时间格式开头的字符串）
    negate: true                                # 是否匹配到
    match: after                                # 合并到上一行的末尾, 为了error日志
    max_lines: 1000                             # 最大的行数
    timeout: 30s                                # 如果在规定的时候没有新的日志事件就不等待后面的日志

  tags: ["example-application"]      #用于logstash过滤



  #fields: 
    #source: ExampleApplication
   #tags: ["GUID"]
    #- /var/log/*.log
    #- c:\programdata\elasticsearch\logs\*
  #include_l ines: ['^ERROR']

启动filebeat，执行一下命令：

nohup  /user/local/filebeat-7.14.0-linux-x86_64/filebeat  -c /work/filebeat/filebeat.yaml   >> /work/filebeat/out.log 2>&1  &

修改dockerfile

FROM  nien/filebeat:7.14.0

# 指定维护者的信息
MAINTAINER 尼恩@疯狂创客圈


ADD dispatcher-provider-1.0-SNAPSHOT.jar  /app/message-dispatcher.jar
ADD deploy-sit.sh  /app/run.sh
RUN chmod +x /app/run.sh

# WORKDIR /app/

ENTRYPOINT /bin/bash -c  "/app/run.sh start"
# ENTRYPOINT /bin/bash

一键发布

使用shell脚本一键发布，这里的脚本，请参见视频

具体的演示，请参见视频

启动之后

spatcher    | ----------------------------------------------------------
message-dispatcher    |         UAA 推送中台 push-provider is running! Access URLs:
message-dispatcher    |         Local:          http://127.0.0.1:7703/message-dispatcher-provider/
message-dispatcher    |         swagger-ui:     http://127.0.0.1:7703/message-dispatcher-provider/swagger-ui.html
message-dispatcher    |         actuator:       http://127.0.0.1:7703/message-dispatcher-provider/actuator/info
message-dispatcher    |         ----------------------------------------------------------
message-di

http://cdh2:7703/message-dispatcher-provider/swagger-ui.html

message-dispatcher微服务的日志

在SpringBoot应用message-dispatcher微服务的日志，输出日志如下：

[root@cdh2 filebeat]# cd  /home/docker-compose/sit-ware/message-dispatcher/work/logs/
[root@cdh2 logs]# cat output.log
2022-04-02 09:03:30.103 [background-preinit] DEBUG o.h.v.m.ResourceBundleMessageInterpolator:89 - Loaded expression factory via original TCCL
2022-04-02 09:03:59.633 [main] INFO  o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker:330 - Bean 'org.springframework.cloud.autoconfigure.ConfigurationPropertiesRebinderAutoConfiguration' of type [org.springframework.cloud.autoconfigure.ConfigurationPropertiesRebinderAutoConfiguration$$EnhancerBySpringCGLIB$$e81692de] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2022-04-02 09:04:05.331 [main] INFO  c.a.n.client.config.impl.LocalConfigInfoProcessor:195 - LOCAL_SNAPSHOT_PATH:/root/nacos/config
2022-04-02 09:04:06.034 [main] INFO  com.alibaba.nacos.client.config.impl.Limiter:53 - limitTime:5.0
2022-04-02 09:04:06.899 [main] INFO  com.alibaba.nacos.client.config.utils.JVMUtil:47 - isMultiInstance:false
2022-04-02 09:04:07.068 [main] WARN  c.a.cloud.nacos.client.NacosPropertySourceBuilder:87 - Ignore the empty nacos configuration and get it based on dataId[message-dispatcher-provider] & group[DEFAULT_GROUP]
2022-04-02 09:04:07.100 [main] WARN  c.a.cloud.nacos.client.NacosPropertySourceBuilder:87 - Ignore the empty nacos configuration and get it based on dataId[message-dispatcher-provider.yml] & group[DEFAULT_GROUP]
2022-04-02 09:04:07.191 [main] INFO  o.s.c.b.c.PropertySourceBootstrapConfiguration:101 - Located property source: CompositePropertySource {name='NACOS', propertySources=[NacosPropertySource {name='message-dispatcher-provider-sit.yml,DEFAULT_GROUP'}, NacosPropertySource {name='message-dispatcher-provider.yml,DEFAULT_GROUP'}, NacosPropertySource {name='message-dispatcher-provider,DEFAULT_GROUP'}, NacosPropertySource {name='sharding-db-dev.yml,DEFAULT_GROUP'}]}
2022-04-02 09:04:07.304 [main] INFO  c.c.s.message.start.MessageDispatchApplication:652 - The following profiles are active: sit
2022-04-02 09:04:28.417 [main] INFO  o.s.d.r.config.RepositoryConfigurationDelegate:247 - Multiple Spring Data modules found, entering strict repository configuration mode!
2022-04-02 09:04:28.418 [main] INFO  o.s.d.r.config.RepositoryConfigurationDelegate:127 - Bootstrapping Spring Data JPA repositories in DEFAULT mode.
2022-04-02 09:04:34.251 [main] INFO  o.s.d.r.config.RepositoryConfigurationDelegate:185 - Finished Spring Data repository scanning in 5673ms. Found 3 JPA repository interfaces.
2022-04-02 09:04:37.630 [main] WARN  o.springframework.boot.actuate.endpoint.EndpointId:131 - Endpoint ID 'nacos-config' contains invalid characters, please migrate to a valid format.
2022-04-02 09:07:17.969 [main] ERROR org.springframework.boot.SpringApplication:823 - Application run failed
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'messageController': Injection of resource dependencies failed; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'messagePushServiceImpl': Injection of resource dependencies failed; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rocketmqMessageService' defined in URL [jar:file:/app/message-dispatcher.jar!/BOOT-INF/classes!/com/crazymaker/springcloud/message/service/impl/RocketmqMessageService.class]: Initialization of bean failed; nested exception is java.lang.IllegalStateException: org.apache.rocketmq.remoting.exception.RemotingTimeoutException: wait response on the channel <dh2/192.168.56.122:9876> timeout, 3000(ms)
        at org.springframework.context.annotation.CommonAnnotationBeanPostProcessor.postProcessProperties(CommonAnnotationBeanPostProcessor.java:325)
        at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.populateBean(AbstractAutowireCapableBeanFactory.java:1404)
        at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:592)
        at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:515)

然后在部署了filebeat的机器上部署该应用，应用的输出文件为/var/log/service-hi.log，应用启动命令如下：

1 nohup java -jar  elk-test-0.0.1-SNAPSHOT.jar > /var/log/service-hi.log 2>&1  &

应用启动成功后日志输出如下：

1 2019-07-02 17:13:13.530  INFO 31579 --- [pool-1-thread-1] com.example.elktest.ElkTestApplication   : seed:562779
2 2019-07-02 17:13:13.630  INFO 31579 --- [pool-1-thread-1] com.example.elktest.ElkTestApplication   : seed:963836
3 2019-07-02 17:13:13.730  INFO 31579 --- [pool-1-thread-1] com.example.elktest.ElkTestApplication   : seed:825694
4 2019-07-02 17:13:13.830  INFO 31579 --- [pool-1-thread-1] com.example.elktest.ElkTestApplication   : seed:33228
5 2019-07-02 17:13:13.930  INFO 31579 --- [pool-1-thread-1] com.example.elktest.ElkTestApplication   : seed:685589

这时的日志数据的传输路径如下图：

查看日志索引

docker run --name filebeat -d  \
-v /home/qw/elk/filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml \
-v /home/qw/elk/testlog/:/home/ \
 elastic/filebeat:7.2.0

效果

可以看到 在kibana中多了两个索引
需要配置

创建一个

选择

最终展示

到这里简单收集日志就完成了,需要更多复杂业务配置,需要大家根据需求自己配置详细信息.

logstash 详解

Logstash 是一款强大的数据处理工具，它可以实现数据传输，格式处理，格式化输出，

logstash 还有强大的插件功能，常用于日志处理.
logstash我们只让它进行日志处理，处理完之后将其输出到elasticsearch。

官方文档

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

stash第一个事件

Logstash管道有两个必需元素，输入和输出，以及一个可选元素filter。

输入插件使用来自源的数据，过滤器插件在您指定时修改数据，输出插件将数据写入目标。
如下图

根据官方文档Logstash对数据的处理主要流程是

1. 首先数据传入logstash，在其内部对数据进行过滤和处理
2. logstash将处理过的数据传递给Elasticsearch
3. Elasticsearch对数据进行存储、创建索引等内容
4. kibana对数据提供可视化的支持

Logstash的核心流程的三个环节

Logstash核心分三个环节：

• 数据输入
• 数据处理
• 数据输出

其数据输入、处理、输出主要在配置中间中下面部分进行配置

input {
   }
filter {
   }
output {
   }

logstash数值类型

• 数组

match =>["datetime", "UNIX", "ISO8601"]

• 布尔

必须是一个true或false

ssl_enable => true

• 字节

一个字段是字节字符串字段表示有效字节的单元。它是一种方便的方式在特定尺寸的插件选项。

支持SI (k M G T P E Z Y)和Binary (TiKimigipiziyiei)单位。

二进制单元在基座单元和Si-1024在基底1000。

这个字段是大小写敏感的。如果未指定单位,则整数表示的字符串的字节数。

my_bytes => "1113" # 1113 bytes 

my_bytes => "10MiB" # 10485760 bytes

 my_bytes => "100kib" # 102400bytes 

my_bytes => "180 mb"# 180000000 bytes

• 编解码器

codec => "json"

• 哈希

哈希是一个键值对的集合中指定的格式，多个键值对的条目以空格分隔而不是逗号。

match => { "field1" => "value1" "field2" =>"value2" ... }

• 数字

数字必须有效的数字值(浮点或整数)。

port => 33

• 密码

密码是一个字符串的单个值，则不对其进行记录或打印。

my_password => "password"

• uri

my_uri =>"http://foo:bar@example.net"

• 路径

一个路径是一个字符串，表示系统运行的有效路径。

my_path =>"/tmp/logstash"

• 转义序列

默认地，转义字符没有被启用。如果你希望使用转义字符串序列，您需要在你的logstash.yml中设置config.support_escapes: true

Text	Result
\r	carriage return (ASCII 13)
\n	new line (ASCII 10)
\t	tab (ASCII 9)
\	backslash (ASCII 92)
\"	double quote (ASCII 34)
\'	single quote (ASCII 39)

logstash 条件判断

有时您只想在特定条件下过滤或输出事件。为此，您可以使用条件。

Logstash中的条件查看和行为与编程语言中的条件相同。条件语句支持if，else if以及else报表和可以被嵌套。

条件语法

if EXPRESSION{ ... } else if EXPRESSION { ... } else { ... }

logstash 比较运算符

　　等于: ==, !=, <, >, <=, >=
　　正则: =~, !~ (checks a pattern on the right against a string value on the left)
　　包含关系: in, not in

　　支持的布尔运算符：and, or, nand, xor

　　支持的一元运算符: !

作用	符号
等于	==
不等于	!=
小于	<
大于	>
小于等于	<=
大于等于	>=
匹配正则	=~
不匹配正则	!~
包含	in
不包含	not in
与	and
或	or
非与	nand
非或	xor
复合表达式	()
取反符合	!()

数据输入环节

input配置定义了数据的来源。其主要支持下面方式

事件源可以是从stdin屏幕输入读取，可以从file指定的文件，也可以从es，filebeat，kafka，redis等读取

stdin

监控控制台输入。

要测试Logstash安装成功，运行最基本的Logstash管道。 执行以下的命令

bin/logstash -e 'input { stdin { } } output { stdout {} }'

-e 标志使您可以直接从命令行指定配置。

通过在命令行指定配置，可以快速测试配置，而无需在迭代之间编辑文件。

示例中的管道从标准输入stdin获取输入，并以结构化格式将输入移动到标准输出stdout。

启动Logstash后，等到看到“Pipeline main started”，然后在命令提示符下输入hello world，显示的如下：

hello world
{
     "host" => "VM_0_13_centos",
     "message" => "hello world",
     "@version" => "1",
    "@timestamp" => 2019-07-02T06:26:28.684Z
}

file

监控文件内容

file{
    path => ['/var/log/nginx/access.log']  #要输入的文件路径
    type => 'nginx_access_log'
    start_position => "beginning"
}

• path 可以用/var/log/.log,/var/log/**/.log，

• type 通用选项. 用于激活过滤器

• start_position 选择logstash开始读取文件的位置，begining或者end。

还有一些常用的例如：discover_interval，exclude，sincedb_path,sincedb_write_interval等可以参考官网

syslogs

从syslogs读取数据

syslog{
    port =>"514" 
    type => "syslog"
}

# port 指定监听端口(同时建立TCP/UDP的514端口的监听)

#从syslogs读取需要实现配置rsyslog：
# cat /etc/rsyslog.conf   加入一行
*.* @172.17.128.200:514　  #指定日志输入到这个端口，然后logstash监听这个端口，如果有新日志输入则读取
# service rsyslog restart   #重启日志服务

beats

从Elastic beats接收数据

beats {
    port => 5044   #要监听的端口
}
# 还有host等选项

# 从beat读取需要先配置beat端，从beat输出到logstash。
# vim /etc/filebeat/filebeat.yml 
..........
output.logstash:
hosts: ["localhost:5044"]

kafka

从kafka topic中读取数据

kafka{
    bootstrap_servers=> "kafka01:9092,kafka02:9092,kafka03:9092"
    topics => ["access_log"]
    group_id => "logstash-file"
    codec => "json"
}
kafka{
    bootstrap_servers=> "kafka01:9092,kafka02:9092,kafka03:9092"
    topics => ["weixin_log","user_log"]  
    codec => "json"
}


# bootstrap_servers 用于建立群集初始连接的Kafka实例的URL列表。
# topics  要订阅的主题列表，kafka topics
# group_id 消费者所属组的标识符，默认为logstash。kafka中一个主题的消息将通过相同的方式分发到Logstash的group_id
# codec 通用选项，用于输入数据的编解码器。

数据处理环节

filter plugin 过滤器插件,主要是对数据进行处理。

grok解析文本并构造

Grok 是一个十分强大的 Logstash Filter 插件，它可以通过正则解析任意文本，将非结构化日志数据格式转换为结构化的、方便查询的结构。

它是目前 Logstash 中解析非结构化日志数据最好的方式。

Grok 的语法规则是：
这里的 “语法” 指的是匹配模式，例如，使用 NUMBER 模式可以匹配出数字，IP 模式则会匹配出 127.0.0.1 这样的 IP 地址。比如按以下格式输入内容：

172.16.213.132 [16/Jun/2020:16:24:19 +0800] "GET / HTTP/1.1" 403 5039

那么，
• %{IP:clientip} 匹配模式将获得的结果为：clientip: 172.16.213.132
• %{HTTPDATE:timestamp} 匹配模式将获得的结果为：timestamp: 16/Jun/2020:16:24:19 +0800
• %{QS:referrer} 匹配模式将获得的结果为：referrer: “GET / HTTP/1.1”
到这里为止，我们已经获取了上面输入中前三个部分的内容，分别是 clientip、timestamp 和 referrer 三个字段。

如果要获取剩余部分的信息，方法类似。

要在线调试 Grok，可以点击在线调试，可点击这里进行在线调试，非常方便。

下面是一个组合匹配模式，它可以获取上面输入的所有内容：

%{
   IP:clientip}\ \[%{
   HTTPDATE:timestamp}\]\ %{
   QS:referrer}\ %{
   NUMBER:response}\ %{
   NUMBER:bytes}

正则匹配是非常严格的匹配，在这个组合匹配模式中，使用了转义字符 \，这是因为输入的内容中有空格和中括号。

通过上面这个组合匹配模式，我们将输入的内容分成了 5 个部分，即 5 个字段。

将输入内容分割为不同的数据字段，这对于日后解析和查询日志数据非常有用，这正是我们使用 grok 的目的。

Logstash 默认提供了近 200 个匹配模式（其实就是定义好的正则表达式）让我们来使用，可以在 Logstash 安装目录下找到。

例如，我这里的路径为：

/usr/local/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns

此目录下有定义好的各种匹配模式，基本匹配定义在 grok-patterns 文件中。

从这些定义好的匹配模式中，可以查到上面使用的四个匹配模式对应的定义规则。

除此之外，还有很多默认定义好的匹配模式文件，比如 httpd、java、linux-syslog、redis、mongodb、nagios 等，这些已经定义好的匹配模式，可以直接在 Grok 过滤器中进行引用。

当然也可以定义自己需要的匹配模式。

在了解完 Grok 的匹配规则之后，下面通过一个配置实例深入介绍下 Logstash 是如何将非结构化日志数据转换成结构化数据的。

首先看下面的一个事件配置文件：

input{
   
  stdin{
   }
}
filter{
   
   grok{
   
     match => ["message", "%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %         {NUMBER:response}\ %{NUMBER:bytes}"]
   }
}
output{
   
   stdout{
   
     codec => "rubydebug"
   }
}

在这个配置文件中，输入配置成了 stdin，在 filter 中添加了 grok 过滤插件，并通过 match 来执行正则表达式解析，

grok 中括号中的正则表达式就是上面提到的组合匹配模式，然后通过 rubydebug 编码格式输出信息。

这样的组合有助于调试和分析输出结果。

通过此配置启动 Logstash 进程后，我们仍然输入之前给出的那段内容：

172.16.213.132 [16/Jun/2020:16:24:19 +0800] "GET / HTTP/1.1" 403 5039

然后，查看 rubydebug 格式的日志输出，内容如下：

{
   
     "timestamp" => "16/Jun/2020:16:24:19 +0800",
      "response" => "403",
         "bytes" => "5039",
      "@version" => "1",
      "clientip" => "172.16.213.132",
          "host" => "nnmaster.cloud",
      "referrer" => "\"GET / HTTP/1.1\"",
       "message" => "172.16.213.132 [16/Jun/2020:16:24:19 +0800] \"GET / HTTP/1.1\" 403 5039",
    "@timestamp" => 2020-06-16T07:46:53.120Z
}

从这个输出可知，通过 Grok 定义好的 5 个字段都获取到了内容，并正常输出了。

date日期解析

解析字段中的日期，然后转存到@timestamp

[2018-07-04 17:43:35,503]

grok{
      match => {"message"=>"%{DATA:raw_datetime}"}
}
date{
      match => ["raw_datetime","YYYY-MM-dd HH:mm:ss,SSS"]
      remove_field =>["raw_datetime"]
}
#将raw_datetime存到@timestamp 然后删除raw_datetime

#24/Jul/2018:18:15:05 +0800
date {
      match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z]
}

mutate字段转换

mutate字段转换, 对字段做处理 重命名、删除、替换和修改字段。

Mutate过滤器的配置选项

选项	类型	是否必须	简述
convert	hash	No	转化命令，是对字段类型做转化，例如:String转为integer
copy	hash	No	将一个已经存在的字段复制给另一个字段。
gsub	array	No	通过正则表达式匹配字段的值，然后替换为指定的字符串。
join	hash	No	使用分隔符连接数组。
lowercase	array	No	将string类型的字段值转化为小写的形式。
merge	hash	No	合并两个数组或者Hash类型的字段。string类型的字段会自动的合并为一个数组。
coerce	hash	No	为存在但是不为空的字段设置默认值
rename	hash	No	字段重命名
replace	hash	No	将一个字段的值替换为一个新的值。
split	hash	No	将一个字段按照指定符号切割为数组。
strip	array	No	去除字段中的空格。
update	hash	No	更新字段为一个新值。
uppercase	array	No	将字符串字段转化为大写形式。
capitalize	array	No	将字符串字段转化为首字母大写的形式。
tag_on_failure	string	No	错误发生时的配置

covert类型转换

covert：类型转换。类型包括：integer，float，integer_eu，float_eu，string和boolean

• 字段类型为 hash
• 没有默认值

将字段转化为不同的类型，例如：string 转 integer。

如果被转化的字段类型是数组，数组的所有成员都将被转化。如果对象是hash 就不会进行转化。

实例：

filter {
  mutate {
    convert => {
      "fieldname" => "integer"
      "booleanfield" => "boolean"
    }
  }
}

split

split：使用分隔符把字符串分割成数组

eg：

mutate{
    split => {"message"=>","}
}

aaa,bbb

{
    "@timestamp" => 2018-06-26T02:40:19.678Z,
      "@version" => "1",
          "host" => "localhost",
       "message" => [
        [0] "aaa",
        [1] "bbb"
    ]}

192,128,1,100

{
        "host" => "localhost",
     "message" => [
      [0] "192",
      [1] "128",
      [2] "1",
      [3] "100"
 ],
  "@timestamp" => 2018-06-26T02:45:17.877Z,
    "@version" => "1"
}


mutate{
    split => {"message"=>","}
}

merge

merge：合并字段 。数组和字符串 ，字符串和字符串

eg：

filter{
    mutate{
        add_field => {"field1"=>"value1"}
    }
    mutate{ 
          split => {"message"=>"."}   #把message字段按照.分割
    }
    mutate{
        merge => {"message"=>"field1"}   #将filed1字段加入到message字段
    }
}

输入：abc

{
       "message" => [
        [0] "abc,"
        [1] "value1"
    ],
    "@timestamp" => 2018-06-26T03:38:57.114Z,
        "field1" => "value1",
      "@version" => "1",
          "host" => "localhost"
}

输入：abc,.123

{
       "message" => [
        [0] "abc,",
        [1] "123",
        [2] "value1"
    ],
    "@timestamp" => 2018-06-26T03:38:57.114Z,
        "field1" => "value1",
      "@version" => "1",
          "host" => "localhost"
}

rename

rename：对字段重命名

filter{
    mutate{
        rename => {"message"=>"info"}
    }
}

123

{
    "@timestamp" => 2018-06-26T02:56:00.189Z,
          "info" => "123",
      "@version" => "1",
          "host" => "localhost"
}

remove_field：移除字段

mutate {
    remove_field => ["message","datetime"]
}

join

join：用分隔符连接数组，如果不是数组则不做处理

mutate{
        split => {"message"=>":"}
}
mutate{
        join => {"message"=>","}
}

abc:123
{
    "@timestamp" => 2018-06-26T03:55:41.426Z,
       "message" => "abc,123",
          "host" => "localhost",
      "@version" => "1"
}
aa:cc
{
    "@timestamp" => 2018-06-26T03:55:47.501Z,
       "message" => "aa,cc",
          "host" => "localhost",
      "@version" => "1"
}

gsub：用正则或者字符串替换字段值。仅对字符串有效

mutate{
        gsub => ["message","/","_"]   #用_替换/
    }

------>
a/b/c/
{
      "@version" => "1",
       "message" => "a_b_c_",
          "host" => "localhost",
    "@timestamp" => 2018-06-26T06:20:10.811Z
}

update：更新字段。如果字段不存在，则不做处理

mutate{
        add_field => {"field1"=>"value1"}
    }
    mutate{
        update => {"field1"=>"v1"}
        update => {"field2"=>"v2"}    #field2不存在 不做处理
    }
---------------->
{
    "@timestamp" => 2018-06-26T06:26:28.870Z,
        "field1" => "v1",
          "host" => "localhost",
      "@version" => "1",
       "message" => "a"
}

replace：更新字段。如果字段不存在，则创建

mutate{
        add_field => {"field1"=>"value1"}
    }
    mutate{
        replace => {"field1"=>"v1"}
        replace => {"field2"=>"v2"}
    }
---------------------->
{
       "message" => "1",
          "host" => "localhost",
    "@timestamp" => 2018-06-26T06:28:09.915Z,
        "field2" => "v2",        #field2不存在，则新建
      "@version" => "1",
        "field1" => "v1"
}

geoip

根据来自Maxmind GeoLite2数据库的数据添加有关IP地址的地理位置的信息

 geoip {
            source => "clientip"
            database =>"/tmp/GeoLiteCity.dat"
        }

ruby

ruby插件可以执行任意Ruby代码

filter{
    urldecode{
        field => "message"
    }
    ruby {
        init => "@kname = ['url_path','url_arg']"
        code => " 
            new_event = LogStash::Event.new(Hash[@kname.zip(event.get('message').split('?'))]) 
            event.append(new_event)"
    }
    if [url_arg]{
        kv{
            source => "url_arg"
            field_split => "&"
            target => "url_args"
            remove_field => ["url_arg","message"]
        }
    }
}
# ruby插件
# 以？为分隔符，将request字段分成url_path和url_arg
-------------------->
www.test.com?test
{
       "url_arg" => "test",
          "host" => "localhost",
      "url_path" => "www.test.com",
       "message" => "www.test.com?test",  
      "@version" => "1",
    "@timestamp" =>  2018-06-26T07:31:04.887Z
}
www.test.com?title=elk&content=学习elk
{
      "url_args" => {
          "title" => "elk",
        "content" => "学习elk"
    },
          "host" => "localhost",
      "url_path" => "www.test.com",
      "@version" => "1",
    "@timestamp" =>  2018-06-26T07:33:54.507Z
}

urldecode

用于解码被编码的字段,可以解决URL中 中文乱码的问题

urldecode{
        field => "message"
    }

# field :指定urldecode过滤器要转码的字段,默认值是"message"
# charset(缺省): 指定过滤器使用的编码.默认UTF-8

kv

通过指定分隔符将字符串分割成key/value

kv{
        prefix => "url_"   #给分割后的key加前缀
        target => "url_ags"    #将分割后的key-value放入指定字段
        source => "message"   #要分割的字段
        field_split => "&"    #指定分隔符
        remove_field => "message"
    }
-------------------------->
a=1&b=2&c=3
{
            "host" => "localhost",
       "url_ags" => {
          "url_c" => "3",
          "url_a" => "1",
          "url_b" => "2"
    },
      "@version" => "1",
    "@timestamp" => 2018-06-26T07:07:24.557Z

useragent

添加有关用户代理(如系列,操作系统,版本和设备)的信息

if [agent] != "-" {
  useragent {
    source => "agent"
    target => "ua"
    remove_field => "agent"
  }
}
# if语句，只有在agent字段不为空时才会使用该插件
#source 为必填设置,目标字段
#target 将useragent信息配置到ua字段中。如果不指定将存储在根目录中

数据输出

output配置定义了数据输出目标

stdout

将数据输出到屏幕上

input{
   
      file{
   
        path=>"/home/order.log"
         discover_interval => 10 
         start_position => "beginning"
      }
}
output{
   
    stdout {
    codec => rubydebug }
}

file

将数据写入文件

读取指定文件-输出到文件

input{
   
      file{
   
        path=>"/home/order.log"
         discover_interval => 10 
         start_position => "beginning"
      }
}
output{
   
     file{
   
         path=>"/home/aaa.log"
      }
}

ps: 需要注意的是 这里的输出文件必须要求 w的权限 看看是否报错

如果报错需要进入容器赋权

kafka

数据发送到kafka

 kafka{
        bootstrap_servers => "localhost:9092"
        topic_id => "test_topic"  #必需的设置。生成消息的主题
    }

elasticseach

数据存储到elasticseach中

读取指定文件-输出到es

input{
   
      file{
   
            path=>"/home/order.log"
         discover_interval => 10 
         start_position => "beginning"
      }
}
output{
   
      elasticsearch{
   
            hosts=>["172.30.66.86:9200"]
            index => "test-%{+YYYY.MM.dd}"
      }
}

Kibana查看应用日志

实操过程，请参见23章视频：《100Wqps 超高并发日志平台》实操

1 查看应用日志

2 如何搜索日志

3 如何查看指定时间的应用日志

• ->右上角选择时间

4 如何定位错误日志

• Search框输入error -> Refresh
  (有自己的语法规则,要搜索一下)

5 如何展开显示日志

• 连续点开两个箭头
  

es的安全认证

通常搭建的elk默认是不需要身份认证,这样就会把数据暴露在外网,因此会显得非常危险。

下面我们介绍如何为es加入身份认证
es身份认证参考链接

切记,这里 修改es 配置文件和 启动es的二进制文件的时候 一定要用es系统用户不要用ubuntu或root用户操作。不然会报错。

配置了 安全认证后 logstash + filebeat +es +kibfana 都需要在配置文件中 加入 访问的账号密码来认证。
logstash 配置文件

elasticsearch {
   
      hosts => ["ip:9200"]
      user => elastic  --加入es用户
      password => xxxx   --加入es密码
      index => "test-%{+YYYY-MM-dd}"
      timeout => 300
  }

kibfana 配置文件

配置 Kibana 以使用内置 kibana 用户和您创建的密码

server.port: 5601
server.host: "0.0.0.0"
elasticsearch.hosts: ["http://localhost:9200"]
kibana.index: ".kibana"
i18n.locale: "zh-CN"  --配置 kibana 显示中文
elasticsearch.username: "kibana"   --加入kibana 账户
elasticsearch.password: "123456"   --加入kibana 账户的密码

配置 elk的ElastAlert 预警插件

我们都知道 elk架构 是收集与分析 集群的日志 方便开发排错

但是 预警功能是缺一不可的，如果开发人员不能及时查看线上错误日式,这个时候 就需要我们的预警插件来实现实时推送告警。
ElastAlert : 是python开发一个插件因此需要配合python运行环境和python-pip 包管理工具,以及相关依赖包

安装相关依赖包

yum -y install openssl openssl-devel gcc gcc-c++  --centos系统安装方式
--ubuntu 安装方式
sudo apt-get install openssl  --openssl依赖包
sudo apt-get install libssl-dev  --openssl-devel 依赖包 
sudo apt-get  install  build-essential   --gcc 依赖包 注意:gcc和g++版本必须一致
sudo apt-get install g++ 7.4  --g++ 依赖包 
g++ --version --查看版本
gcc --version
wget https://www.python.org/ftp/python/3.6.9/Python-3.6.9.tgz --下载二进制python源码

安装python运行环境

tar xf Python-3.6.9.tgz
cd Python-3.6.9./configure --prefix=/usr/local/python --with-openssl
make && make install  --编译源码

配置

mv /usr/bin/python /usr/bin/python_old  //把ubuntu自带的python2.7环境移出到另外一个文件夹
ln -s /usr/local/python/bin/python3 /usr/bin/python  //建立python软链接
ln -s /usr/local/python/bin/pip3 /usr/bin/pip  //建立pip软链接
pip install --upgrade pip //此处没有安装pip的需要去安装pip
sudo apt install python3-pip //安装pip3.0版本 对应了python 3.6.9版本
//此处我没有动ubuntu自带的python2.7版本的 因此我们使用新的python使用3.6.9时,按以下方式使用:
python3.6 --version
python2.7 --version
pip3 --version
//使用python和pip命令时 都改为 python3.6与pip3

到此python环境配置完成

安装elastalert

下载源码

git clone https://github.com/Yelp/elastalert.git //下载 源码
cd elastalert
pip3 install "elasticsearch<8,>7"    
//因为我们的es是7.4.0，所以这里选用的版本是这个
pip3 install -r requirements.txt 用pip安装依赖

安装成功时候 /usr/local/python/bin/目录下会有四个文件

ls /usr/local/python/bin/elastalert* 或者这个目录下
ls /usr/local/bin/elastalert*

ln -s /usr/local/python/bin/elastalert* /usr/bin  //建立软链接把这四个命令链接到bin目录下

配置ElastAlert

配置config.yaml 文件 (创建)

cp config.yaml.example  config.yaml 
sudo vi config.yaml

rules_folder：ElastAlert从中加载规则配置文件的位置。它将尝试加载文件夹中的每个.yaml文件。

没有任何有效规则，ElastAlert将无法启动。

run_every： ElastAlert多久查询一次Elasticsearch的时间。

buffer_time：查询窗口的大小，从运行每个查询的时间开始向后延伸。对于其中use_count_query或use_terms_query设置为true的规则，将忽略此值。

es_host：是Elasticsearch群集的地址，ElastAlert将在其中存储有关其状态，查询运行，警报和错误的数据。
es_port：es对应的端口。

es_username： 可选的; 用于连接的basic-auth用户名es_host。

es_password： 可选的; 用于连接的basic-auth密码es_host。

es_send_get_body_as： 可选的; 方法查询Elasticsearch - GET，POST或source。

默认是GET writeback_index：ElastAlert将在其中存储数据的索引的名称。我们稍后将创建此索引。

alert_time_limit： 失败警报的重试窗口。

创建elastalert-create-index索引 告警索引

$ elastalert-create-index
New index name (Default elastalert_status)
Name of existing index to copy (Default None)
New index elastalert_status created
Done!

配置Rule 告警规则配置

所有的告警规则，通过在example_rules目下创建配置文件进行定义，这里简单创建一个来作为演示

name: Nginx_err  //规则名称
use_strftine_index: true 
index: 10.0.0.153-system_cro-2020.11.18  //监听查询es的索引
type: any    //告警规则类型 有很多种 这种是 只要匹配到就触发告警
aggregation:
 seconds: 1    //告警频率
filter:
- query:
    query_string:
         query: "status:500 or status:404" //触发报警的匹配条件 这里可以用kibana的语法去匹配
num_events: 1  //事件触发次数 的贬值
timeframe:
  minutes: 1   //一分钟内超过 num_envents触发的次数 就触发告警
alert:
 - "email"   //告警类型 此处是email 例如钉钉 企业微信
email_format: html  //email 正文格式
alert_subject: "正式环境Error告警"  //告警正文标题
alert_text_type: alert_text_only   //正文类型
alert_text: "<br><br><h3>告警详情</h3><table><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>@timestamp:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>@version:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>_id:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>_index:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>ip:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>request:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td  style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>status:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>method:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>bytes:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>source:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>client_ip:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr><tr><td style='padding:5px;text-align: right;font-weight: bold;border-radius: 5px;background-color: #eef;'>httpversion:</td><td style='padding:5px;border-radius: 5px;background-color: #eef;'>{}</td></tr></table>"  //正文内容
alert_text_args:
 - "@timestamp"   //使用的是python的format格式动态填充数据
 - "@version"     //这些是属性值 按顺序对饮正文内容里面的 {} 
 - _id
 - _index
 - host.name
 - request
 - status
 - method
 - bytes
 - message
 - remote_ip
 - httpversion
email:
 - "xxx@xx.com"  //收件人 多个请依次往下填写
 - "xxxx@qq.com"
 - "xxxx@xx.com"
smtp_host: smtp.mxhichina.com  //邮件服务器
smtp_port: 25   //邮件端口
smtp_auth_file: /home/ubuntu/elk/alert/elastalert/smtp_auth_file.yaml //此处新建了一个文件是 发件人的认证文件 存放发件人账户和密码或授权码
from_addr: haoyacong@gimmake.com  //发件人
email_reply_to: haoyacong@gimmake.com  //收件人标头

运行ElastAlert

cd ElastAlert  //ElastAlert 的安装目录
python3.6 -m elastalert.elastalert --verbose --config config.yaml --rule ./example_rules/nginx_404.yaml  //指定告警规则文件  
nohup python3.6 -m elastalert.elastalert --verbose --config config.yaml --rule ./example_rules/nginx_404.yaml & //在后台运行
//如果运行多个告警规则执行多个上面的命令  如果执行example_rules下的全部规则文件 使用以下命令:
nohup python3.6 -m elastalert.elastalert --verbose --config config.yaml &

8 底座项目中ELK的应用

底座项目没有使用FELK架构，也就是没有部署filebeat，直接使用的java的loastash客户端，通过tcp协议直接采集logback日志到logstash

        <dependency>
            <groupId>net.logstash.logback</groupId>
            <artifactId>logstash-logback-encoder</artifactId>
        </dependency>

后期可以自行根据上面的综合实操基于filebeat扩展

8.1 logstash-logback-encoder简介

logstash-logback-encoder 是一个用于将 Java 应用程序的日志输出转换为 Logstash 可解析的 JSON 格式的库。它是基于 Logback（一个流行的 Java 日志框架）构建的，旨在使 Logback 输出与 Elasticsearch、Kibana 和 Logstash 等日志管理工具兼容，从而简化日志的收集和分析过程。

主要功能和特点

1. JSON 格式输出： logstash-logback-encoder 允许你将日志输出格式化为 JSON，使得日志数据可以很容易地被 Logstash 处理和解析。Logstash 可以将这些 JSON 格式的日志数据传送到 Elasticsearch，供 Kibana 可视化展示。
2. 支持 Logback 配置： 它是基于 Logback 实现的，因此你可以使用 Logback 配置文件（如 logback.xml）来设置日志输出的格式。它支持将日志条目编码为标准的 JSON 格式，并可以灵活地自定义输出内容。
3. 灵活的字段映射： 你可以自定义 JSON 输出的字段。默认情况下，日志条目会包含如下字段：
   • timestamp：日志条目的时间戳。
   • level：日志级别（如 INFO、ERROR）。
   • thread：线程名称。
   • logger：日志记录器名称。
   • message：日志消息本体。
   • exception：异常信息（如果有的话）。
4. 支持 MDC（Mapped Diagnostic Context）： logstash-logback-encoder 支持 Logback 的 MDC 功能，这允许你在日志中附加额外的上下文信息。例如，可以记录用户 ID、请求 ID、会话信息等，以便于后续分析。
5. 格式化选项： 你可以通过配置文件指定日志的格式，包括字段顺序、字段名称等。库提供了丰富的配置选项，帮助用户根据需求定制日志输出格式。
6. 与 ELK Stack 的集成： logstash-logback-encoder 是 ELK（Elasticsearch, Logstash, Kibana）栈中的重要组成部分。它将 Logback 日志输出转换为 JSON 格式，这使得日志可以很容易地被 Logstash 解析、索引到 Elasticsearch，并通过 Kibana 展示和分析。
7. 兼容性： 该库与 Logback 的其他日志输出配置兼容，可以与其他类型的日志记录器和框架（如 SLF4J）一起使用。

使用方法

1. 添加 Maven 依赖

首先，确保你在项目中添加了 logstash-logback-encoder 的 Maven 依赖。在 pom.xml 中添加以下内容：

<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>7.0</version> <!-- 请使用最新版本 -->
</dependency>

2. 配置 Logback 输出为 JSON 格式

在 Logback 配置文件（通常是 logback.xml）中，你可以配置一个 Encoder 来输出 JSON 格式的日志。以下是一个简单的例子：

<configuration>

    <appender name="LOGSTASH" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="net.logstash.logback.encoder.LogstashEncoder"/>
    </appender>

    <root level="INFO">
        <appender-ref ref="LOGSTASH"/>
    </root>

</configuration>

这个配置会将所有日志输出到控制台，并且使用 LogstashEncoder 将日志条目转换为 JSON 格式。

3. 使用自定义日志格式

你还可以通过配置更多的选项来自定义输出的 JSON 结构。例如，添加 MDC 数据、时间戳格式等：

<encoder class="net.logstash.logback.encoder.LogstashEncoder">
    <timestampPattern>yyyy-MM-dd'T'HH:mm:ss.SSSZ</timestampPattern>
    <fieldNames>
        <message>msg</message>
        <timestamp>time</timestamp>
        <level>loglevel</level>
        <logger>logger_name</logger>
        <thread>thread_name</thread>
    </fieldNames>
</encoder>

在此示例中，时间戳的格式被自定义为 ISO 8601 格式，字段名称也被自定义。

常见配置项

• timestampPattern：指定日志时间戳的格式（默认使用 ISO 8601 格式）。
• fieldNames：用于自定义输出的字段名称。
• includeCallerData：是否将调用者信息（如类名、方法名）包含在日志中。
• customFields：用于添加自定义的静态字段，适用于在每条日志中注入相同的字段（如应用程序名称、版本号等）。
• stackTraceElementConverter：用于控制如何格式化堆栈跟踪信息。

示例 JSON 输出

使用 logstash-logback-encoder 输出的 JSON 日志条目可能类似于：

json{
  "timestamp": "2024-11-17T10:55:27.123+0000",
  "loglevel": "INFO",
  "logger_name": "com.example.MyClass",
  "thread_name": "main",
  "msg": "Application started",
  "exception": null
}

总结

logstash-logback-encoder 是一个专为将 Logback 日志输出为 Logstash 可处理的 JSON 格式设计的库，主要用于与 ELK Stack（Elasticsearch、Logstash、Kibana）结合使用。通过该库，Java 应用程序可以生成结构化的日志数据，方便后续的日志收集、存储和分析。其强大的自定义功能使其非常适合用于生产环境中对日志的高效管理。

8.2 logstash的java客户端配置

在需要elk采集日志的app引入ruoyi-common-logstash依赖，依赖logstash-logback-encoder采集日志

        <dependency>
            <groupId>com.ruoyi</groupId>
            <artifactId>ruoyi-common-logstash</artifactId>
        </dependency>

logback-logstash.xml

<?xml version="1.0" encoding="UTF-8"?>

<included>

    <springProperty scope="context" name="appName" source="spring.application.name"/>

    <!--输出到logstash的appender-->
    <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <!--可以访问的logstash日志收集端口-->
        <destination>${logstash.address}</destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder">
            <customFields>{"spring.application.name":"${appName}"}</customFields>
        </encoder>
    </appender>

    <root level="info">
        <appender-ref ref="logstash"/>
    </root>
</included>

父工程pom.xml配置logstash.address

8.3 一键部署elk

  elasticsearch:
    image: elasticsearch:7.17.6
    container_name: elasticsearch
    ports:
      - "9200:9200"
      - "9300:9300"
    environment:
      # 设置集群名称
      cluster.name: elasticsearch
      # 以单一节点模式启动
      discovery.type: single-node
      ES_JAVA_OPTS: "-Xms512m -Xmx512m"
    volumes:
      - /docker/elk/elasticsearch/plugins:/usr/share/elasticsearch/plugins
      - /docker/elk/elasticsearch/data:/usr/share/elasticsearch/data
      - /docker/elk/elasticsearch/logs:/usr/share/elasticsearch/logs
    network_mode: "host"

  kibana:
    image: kibana:7.17.6
    container_name: kibana
    ports:
      - "5601:5601"
    depends_on:
      # kibana在elasticsearch启动之后再启动
      - elasticsearch
    environment:
      #设置系统语言文中文
      I18N_LOCALE: zh-CN
      # 访问域名
      # SERVER_PUBLICBASEURL: https://kibana.cloud.com
    volumes:
      - /docker/elk/kibana/config/kibana.yml:/usr/share/kibana/config/kibana.yml
    network_mode: "host"

  logstash:
    image: logstash:7.17.6
    container_name: logstash
    ports:
      - "4560:4560"
    volumes:
      - /docker/elk/logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf
      - /docker/elk/logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml
    depends_on:
      - elasticsearch
    network_mode: "host"

logstash配置tcp类型输入，输出es

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 4560
    codec => json_lines
  }
}
output {
  elasticsearch {
    hosts => "127.0.0.1:9200"
    index => "%{[spring.application.name]}-%{+YYYY.MM.dd}"
  }
}

说在最后：有问题找老架构取经

尼恩团队15大技术圣经 ，使得大家内力猛增，

可以充分展示一下大家雄厚的 “技术肌肉”，让面试官爱到 “不能自已、口水直流”，然后实现”offer直提”。

在面试之前，建议大家系统化的刷一波 5000页《尼恩Java面试宝典PDF》，里边有大量的大厂真题、面试难题、架构难题。

很多小伙伴刷完后， 吊打面试官， 大厂横着走。

在刷题过程中，如果有啥问题，大家可以来 找 40岁老架构师尼恩交流。

另外，如果没有面试机会，可以找尼恩来改简历、做帮扶。前段时间，刚指导一个27岁 被裁小伙，拿到了一个年薪45W的JD +PDD offer，逆天改命。

狠狠卷，实现 “offer自由” 很容易的， 前段时间一个武汉的跟着尼恩卷了2年的小伙伴， 在极度严寒/痛苦被裁的环境下， offer拿到手软， 实现真正的 “offer自由” 。

尼恩技术圣经系列PDF

• 《NIO圣经：一次穿透NIO、Selector、Epoll底层原理》
• 《Docker圣经：大白话说Docker底层原理，6W字实现Docker自由》
• 《K8S学习圣经：大白话说K8S底层原理，14W字实现K8S自由》
• 《SpringCloud Alibaba 学习圣经，10万字实现SpringCloud 自由》
• 《大数据HBase学习圣经：一本书实现HBase学习自由》
• 《大数据Flink学习圣经：一本书实现大数据Flink自由》
• 《响应式圣经：10W字，实现Spring响应式编程自由》
• 《Go学习圣经：Go语言实现高并发CRUD业务开发》

……完整版尼恩技术圣经PDF集群，请找尼恩领取

《尼恩 架构笔记》《尼恩高并发三部曲》《尼恩Java面试宝典》PDF，请到下面公号【技术自由圈】取↓↓↓