Logstash 快速指南

---

---

Logstash - 简介

Logstash 是一个基于过滤器/管道模式的工具，用于收集、处理和生成日志或事件。它有助于集中和实时分析来自不同来源的日志和事件。

Logstash 使用 JRuby 编程语言编写，运行在 JVM 上，因此您可以在不同的平台上运行 Logstash。它从几乎所有类型的来源收集不同类型的数据，例如日志、数据包、事件、事务、时间戳数据等。数据源可以是社交数据、电子商务、新闻文章、CRM、游戏数据、网络趋势、金融数据、物联网、移动设备等。

Logstash 通用功能

Logstash 的通用功能如下：

• Logstash 可以从不同的来源收集数据并发送到多个目标。

• Logstash 可以处理所有类型的日志数据，例如 Apache 日志、Windows 事件日志、网络协议上的数据、标准输入的数据等等。

• Logstash 还可以处理 http 请求和响应数据。

• Logstash 提供各种过滤器，帮助用户通过解析和转换数据来找到更多有意义的信息。

• Logstash 也可以用于处理物联网中的传感器数据。

• Logstash 是开源的，并根据 Apache 许可证 2.0 版提供。

Logstash 关键概念

Logstash 的关键概念如下：

事件对象

它是 Logstash 中的主要对象，它封装了 Logstash 管道中的数据流。Logstash 使用此对象来存储输入数据并在过滤器阶段添加创建的额外字段。

Logstash 为开发人员提供了事件 API 来操作事件。在本教程中，此事件以各种名称引用，例如日志数据事件、日志事件、日志数据、输入日志数据、输出日志数据等。

管道

它包含 Logstash 中从输入到输出的数据流阶段。输入数据进入管道，并以事件的形式进行处理。然后以用户或最终系统期望的格式发送到输出目标。

输入

这是 Logstash 管道的第一个阶段，用于获取 Logstash 中的数据以进行进一步处理。Logstash 提供各种插件来从不同的平台获取数据。一些最常用的插件是 - File、Syslog、Redis 和 Beats。

过滤器

这是 Logstash 的中间阶段，事件的实际处理发生在此阶段。开发人员可以使用 Logstash 预定义的正则表达式模式来创建序列，以区分事件中的字段和可接受的输入事件的条件。

Logstash 提供各种插件来帮助开发人员将事件解析和转换为所需结构。一些最常用的过滤器插件是 - Grok、Mutate、Drop、Clone 和 Geoip。

输出

这是 Logstash 管道的最后一个阶段，输出事件可以格式化为目标系统所需的结构。最后，它使用插件将完全处理后的输出事件发送到目标。一些最常用的插件是 - Elasticsearch、File、Graphite、Statsd 等。

Logstash 优点

以下几点说明了 Logstash 的各种优点。

• Logstash 提供正则表达式模式序列来识别和解析任何输入事件中的各种字段。

• Logstash 支持各种 Web 服务器和数据源来提取日志数据。

• Logstash 提供多个插件来将日志数据解析和转换为任何用户期望的格式。

• Logstash 是集中的，这使得它易于处理和收集来自不同服务器的数据。

• Logstash 支持许多数据库、网络协议和其他服务作为日志事件的目标源。

• Logstash 使用 HTTP 协议，使用户能够升级 Elasticsearch 版本而无需同步升级 Logstash。

Logstash 缺点

以下几点说明了 Logstash 的各种缺点。

• Logstash 使用 http，这会对日志数据的处理产生负面影响。

• 使用 Logstash 有时可能有点复杂，因为它需要对输入日志数据有很好的理解和分析。

• 过滤器插件不是通用的，因此用户可能需要找到正确的模式序列以避免解析错误。

在下一章中，我们将了解 ELK 堆栈是什么以及它如何帮助 Logstash。

Logstash - ELK 堆栈

ELK 代表 **Elasticsearch、Logstash** 和 **Kibana**。在 ELK 堆栈中，Logstash 从不同的输入源提取日志数据或其他事件。它处理事件，然后将其存储在 Elasticsearch 中。Kibana 是一个 Web 接口，它访问 Elasticsearch 中的日志数据并将其可视化。

Logstash 和 Elasticsearch

Logstash 提供输入和输出 Elasticsearch 插件，用于读取和写入日志事件到 Elasticsearch。Elasticsearch 作为输出目标也由 Elasticsearch 公司推荐，因为它与 Kibana 兼容。Logstash 通过 http 协议将数据发送到 Elasticsearch。

Elasticsearch 提供批量上传功能，这有助于将数据从不同的来源或 Logstash 实例上传到集中的 Elasticsearch 引擎。ELK 比其他 DevOps 解决方案具有以下优点：

• ELK 堆栈更容易管理，并且可以扩展以处理 PB 级别的事件。

• ELK 堆栈架构非常灵活，并且它提供了与 Hadoop 的集成。Hadoop 主要用于存档目的。Logstash 可以使用 Flume 直接连接到 Hadoop，Elasticsearch 提供了一个名为 **es-hadoop** 的连接器来连接 Hadoop。

• ELK 的总拥有成本远低于其替代方案。

Logstash 和 Kibana

Kibana 不直接与 Logstash 交互，而是通过数据源（在 ELK 堆栈中是 Elasticsearch）。Logstash 从每个来源收集数据，Elasticsearch 以非常快的速度分析它，然后 Kibana 提供对该数据的可操作见解。

Kibana 是一个基于 Web 的可视化工具，它帮助开发人员和其他人员分析 Logstash 在 Elasticsearch 引擎中收集的大量事件的变化。这种可视化使得预测或查看输入源中错误或其他重要事件趋势的变化变得容易。

Logstash - 安装

要在系统上安装 Logstash，我们应该按照以下步骤操作：

**步骤 1** - 检查计算机中安装的 Java 版本；它应该是 Java 8，因为它与 Java 9 不兼容。您可以通过以下方式检查：

在 Windows 操作系统 (OS) 中（使用命令提示符）：

> java -version 

在 UNIX OS 中（使用终端）：

$ echo $JAVA_HOME

**步骤 2** - 从以下地址下载 Logstash：

https://elastic.ac.cn/downloads/logstash.

• 对于 Windows OS，下载 ZIP 文件。

• 对于 UNIX OS，下载 TAR 文件。

• 对于 Debian OS，下载 DEB 文件。

• 对于 Red Hat 和其他 Linux 发行版，下载 RPN 文件。

• APT 和 Yum 实用程序也可以用于在许多 Linux 发行版中安装 Logstash。

**步骤 3** - Logstash 的安装过程非常简单。让我们看看如何在不同的平台上安装 Logstash。

**注意** - 不要在安装文件夹中放置任何空格或冒号。

• **Windows OS** - 解压 zip 包，Logstash 就安装好了。

• **UNIX OS** - 将 tar 文件解压到任何位置，Logstash 就安装好了。

$tar –xvf logstash-5.0.2.tar.gz

对于 Linux OS 使用 APT 实用程序：

• 下载并安装公共签名密钥：

$ wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

• 保存存储库定义：

$ echo "deb https://artifacts.elastic.co/packages/5.x/apt stable main" | sudo
   tee -a /etc/apt/sources.list.d/elastic-5.x.list

• 运行更新：

$ sudo apt-get update

• 现在您可以使用以下命令进行安装：

$ sudo apt-get install logstash

**对于 Debian Linux OS 使用 YUM 实用程序** -

• 下载并安装公共签名密钥：

$ rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

• 在“/etc/yum.repos.d/”目录中带有 .repo 后缀的文件中添加以下文本。例如，**logstash.repo**

[logstash-5.x]
name = Elastic 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

• 您现在可以使用以下命令安装 Logstash：

$ sudo yum install logstash

**步骤 4** - 转到 Logstash 主目录。在 bin 文件夹中，在 Windows 的情况下运行 **elasticsearch.bat** 文件，或者您可以使用命令提示符和终端执行相同的操作。在 UNIX 中，运行 Logstash 文件。

我们需要指定输入源、输出源和可选过滤器。为了验证安装，您可以使用标准输入流 (stdin) 作为输入源和标准输出流 (stdout) 作为输出源来运行它，使用基本配置。您也可以使用 **–e** 选项在命令行中指定配置。

在 Windows 中：

> cd logstash-5.0.1/bin
> Logstash -e 'input { stdin { } } output { stdout {} }'

在 Linux 中：

$ cd logstash-5.0.1/bin
$ ./logstash -e 'input { stdin { } } output { stdout {} }'

**注意** - 在 Windows 的情况下，您可能会收到一个错误，指出未设置 JAVA_HOME。为此，请在环境变量中将其设置为“C:\Program Files\Java\jre1.8.0_111”或您安装 java 的位置。

**步骤 5** - Logstash Web 界面的默认端口为 9600 到 9700，在 **logstash-5.0.1\config\logstash.yml** 中定义为 **http.port**，它将选择给定范围内的第一个可用端口。

我们可以通过浏览 **https://127.0.0.1:9600** 来检查 Logstash 服务器是否已启动并正在运行，或者如果端口不同，则请检查命令提示符或终端。我们可以看到分配的端口为“成功启动 Logstash API 端点 {:port ⇒ 9600}”。它将返回一个 JSON 对象，其中包含以下关于已安装 Logstash 的信息：

{
   "host":"manu-PC", 
   "version":"5.0.1",
   "http_address":"127.0.0.1:9600",
   "build_date":"2016-11-11T22:28:04+00:00",
   "build_sha":"2d8d6263dd09417793f2a0c6d5ee702063b5fada",
   "build_snapshot":false
}

Logstash - 内部架构

在本章中，我们将讨论 Logstash 的内部架构和不同组件。

Logstash 服务架构

Logstash 处理来自不同服务器和数据源的日志，它充当发件人。发件人用于收集日志，它们安装在每个输入源中。像 **Redis、Kafka** 或 **RabbitMQ** 这样的代理是缓冲区，用于保存数据供索引器使用，可能存在多个代理作为故障转移实例。

像 **Lucene** 这样的索引器用于索引日志以获得更好的搜索性能，然后将输出存储在 Elasticsearch 或其他输出目标中。输出存储中的数据可用于 Kibana 和其他可视化软件。

Logstash 内部架构

Logstash 管道包含三个组件：**输入、过滤器** 和 **输出**。输入部分负责指定和访问输入数据源，例如 **Apache Tomcat 服务器** 的日志文件夹。

解释 Logstash 管道的示例

Logstash 配置文件包含有关 Logstash 三个组件的详细信息。在这种情况下，我们创建一个名为 **Logstash.conf** 的文件。

以下配置捕获来自输入日志“inlog.log”的数据，并将其写入输出日志“outlog.log”，而无需任何过滤器。

Logstash.conf

Logstash 配置文件仅使用输入插件从 **inlog.log** 文件复制数据，并使用输出插件将日志数据刷新到 **outlog.log** 文件。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/inlog.log"
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/outlog.log"
   }
}

运行 Logstash

Logstash 使用 **–f** 选项来指定配置文件。

C:\logstash\bin> logstash –f logstash.conf

inlog.log

以下代码块显示输入日志数据。

Hello tutorialspoint.com

outlog.log

Logstash 的输出包含 message 字段中的输入数据。Logstash还会向输出添加其他字段，例如时间戳、输入源路径、版本、主机和标签。

{
   "path":"C:/tpwork/logstash/bin/log/inlog1.log",
   "@timestamp":"2016-12-13T02:28:38.763Z",
   "@version":"1", "host":"Dell-PC",
   "message":" Hello tutorialspoint.com", "tags":[]
}

如您所见，Logstash 的输出包含比输入日志提供的数据更多内容。输出包含源路径、时间戳、版本、主机名和标签，这些用于表示额外的消息，例如错误。

我们可以使用过滤器来处理数据，使其符合我们的需求。在接下来的示例中，我们使用过滤器来获取数据，该过滤器将输出限制为仅包含类似 GET 或 POST 的动词，后跟一个**唯一资源标识符**的数据。

Logstash.conf

在此 Logstash 配置中，我们添加了一个名为**grok**的过滤器来过滤输入数据。与模式序列输入日志匹配的输入日志事件，只有在出现错误的情况下才会到达输出目标。Logstash 在不匹配 grok 过滤器模式序列的输出事件中添加名为“_grokparsefailure”的标签。

Logstash 提供了许多内置正则表达式模式，用于解析流行的服务器日志，例如 Apache。此处使用的模式需要一个类似 get、post 等的动词，后跟一个统一资源标识符。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/inlog2.log"
   }
}
filter {
   grok {
      match => {"message" => "%{WORD:verb} %{URIPATHPARAM:uri}"}
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/outlog2.log"
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

C:\logstash\bin> logstash –f  Logstash.conf

inlog2.log

我们的输入文件包含两个由默认分隔符（即换行符）分隔的事件。第一个事件与 GROk 中指定的模式匹配，第二个事件不匹配。

GET /tutorialspoint/Logstash
Input 1234

outlog2.log

我们可以看到，第二个输出事件包含“_grokparsefailure”标签，因为它与 grok 过滤器模式不匹配。用户还可以使用输出插件中的**“if”**条件来删除输出中这些不匹配的事件。

{
   "path":"C:/tpwork/logstash/bin/log/inlog2.log",
   "@timestamp":"2016-12-13T02:47:10.352Z","@version":"1","host":"Dell-PC","verb":"GET",
   "message":"GET /tutorialspoint/logstash", "uri":"/tutorialspoint/logstash", "tags":[]
}
{
   "path":"C:/tpwork/logstash/bin/log/inlog2.log",
   "@timestamp":"2016-12-13T02:48:12.418Z", "@version":"1", "host":"Dell-PC",
   "message":"t 1234\r", "tags":["_grokparsefailure"]
}

Logstash - 收集日志

使用 shipper 收集来自不同服务器或数据源的日志。shipper 是安装在服务器上的 Logstash 实例，它访问服务器日志并将日志发送到特定的输出位置。

它主要将输出发送到 Elasticsearch 进行存储。Logstash 从以下来源获取输入：

• STDIN
• Syslog
• 文件
• TCP/UDP
• Microsoft Windows 事件日志
• Websocket
• Zeromq
• 自定义扩展

使用 Apache Tomcat 7 服务器收集日志

在此示例中，我们使用文件输入插件收集安装在 Windows 上的 Apache Tomcat 7 服务器的日志，并将它们发送到另一个日志。

logstash.conf

在这里，Logstash 被配置为访问本地安装的 Apache Tomcat 7 的访问日志。文件插件的路径设置中使用正则表达式模式来从日志文件中获取数据。这包含其名称中的“access”，并且它添加了 apache 类型，这有助于区分集中式目标源中的 apache 事件和其他事件。最后，输出事件将显示在 output.log 中。

input {
   file {
      path => "C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/*access*"
      type => "apache"
   }
} 
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

C:\logstash\bin> logstash –f  Logstash.conf

Apache Tomcat 日志

访问 Apache Tomcat 服务器及其 Web 应用程序（**https://127.0.0.1:8080**）以生成日志。Logstash 实时读取日志中的更新数据，并将其存储在配置文件中指定的 output.log 中。

Apache Tomcat 根据日期生成新的访问日志文件，并将访问事件记录到该文件中。在我们的例子中，它是在 Apache Tomcat 的 **logs** 目录中的 localhost_access_log.2016-12-24.txt。

0:0:0:0:0:0:0:1 - - [
   25/Dec/2016:18:37:00 +0800] "GET / HTTP/1.1" 200 11418
0:0:0:0:0:0:0:1 - munish [
   25/Dec/2016:18:37:02 +0800] "GET /manager/html HTTP/1.1" 200 17472
0:0:0:0:0:0:0:1 - - [
   25/Dec/2016:18:37:08 +0800] "GET /docs/ HTTP/1.1" 200 19373
0:0:0:0:0:0:0:1 - - [
   25/Dec/2016:18:37:10 +0800] "GET /docs/introduction.html HTTP/1.1" 200 15399

output.log

您可以在输出事件中看到，添加了一个 type 字段，并且事件存在于 message 字段中。

{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt",
   "@timestamp":"2016-12-25T10:37:00.363Z","@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - - [25/Dec/2016:18:37:00 +0800] \"GET /
   HTTP/1.1\" 200 11418\r","type":"apache","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt","@timestamp":"2016-12-25T10:37:10.407Z",
   "@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - munish [25/Dec/2016:18:37:02 +0800] \"GET /
   manager/html HTTP/1.1\" 200 17472\r","type":"apache","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt","@timestamp":"2016-12-25T10:37:10.407Z",
   "@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - - [25/Dec/2016:18:37:08 +0800] \"GET /docs/
   HTTP/1.1\" 200 19373\r","type":"apache","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt","@timestamp":"2016-12-25T10:37:20.436Z",
   "@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - - [25/Dec/2016:18:37:10 +0800] \"GET /docs/
   introduction.html HTTP/1.1\" 200 15399\r","type":"apache","tags":[]
}

使用 STDIN 插件收集日志

在本节中，我们将讨论另一个使用**STDIN 插件**收集日志的示例。

logstash.conf

这是一个非常简单的示例，其中 Logstash 读取用户在标准输入中输入的事件。在我们的例子中，它是命令提示符，它将事件存储在 output.log 文件中。

input {
   stdin{}
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

C:\logstash\bin> logstash –f  Logstash.conf

在命令提示符中写入以下文本：

用户输入了以下两行。Logstash 通过分隔符设置分隔事件，其默认值为“\n”。用户可以通过更改文件插件中分隔符的值来更改它。

Tutorialspoint.com welcomes you
Simply easy learning

output.log

以下代码块显示输出日志数据。

{
   "@timestamp":"2016-12-25T11:41:16.518Z","@version":"1","host":"Dell-PC",
   "message":"tutrialspoint.com welcomes you\r","tags":[]
}
{
   "@timestamp":"2016-12-25T11:41:53.396Z","@version":"1","host":"Dell-PC",
   "message":"simply easy learning\r","tags":[]
}

Logstash - 支持的输入

Logstash 支持来自不同来源的大量日志。它与下面解释的著名来源一起工作。

从指标收集日志

系统事件和其他时间活动记录在指标中。Logstash 可以访问系统指标中的日志并使用过滤器对其进行处理。这有助于以自定义方式向用户显示事件的实时馈送。指标根据指标过滤器的**flush_interval 设置**刷新，默认设置为 5 秒。

我们通过收集和分析通过 Logstash 运行的事件并在命令提示符上显示实时馈送来跟踪 Logstash 生成的测试指标。

logstash.conf

此配置包含一个由 Logstash 提供的生成器插件，用于测试指标，并将 type 设置设置为“generated”以进行解析。在过滤阶段，我们仅使用“if”语句处理具有 generated 类型的行。然后，metrics 插件计算 meter 设置中指定的字段。metrics 插件每隔 5 秒（在**flush_interval**中指定）刷新一次计数。

最后，使用**codec 插件**进行格式化，将过滤器事件输出到标准输出（如命令提示符）。Codec 插件使用[events][rate_1m]值在 1 分钟滑动窗口中输出每秒事件。

input {
   generator {
     	type => "generated"
   }
}
filter {
   if [type] == "generated" {
      metrics {
         meter => "events"
         add_tag => "metric"
      }
   }
}
output {
   # only emit events with the 'metric' tag
   if "metric" in [tags] {
      stdout {
         codec => line { format => "rate: %{[events][rate_1m]}"
      }
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logsaths –f logstash.conf

stdout（命令提示符）

rate: 1308.4
rate: 1308.4
rate: 1368.654529135342
rate: 1416.4796003951449
rate: 1464.974293984808
rate: 1523.3119444107458
rate: 1564.1602979542715
rate: 1610.6496496890895
rate: 1645.2184750334154
rate: 1688.7768007612485
rate: 1714.652283095914
rate: 1752.5150680019278
rate: 1785.9432934744932
rate: 1806.912181962126
rate: 1836.0070454626025
rate: 1849.5669494173826
rate: 1871.3814756851832
rate: 1883.3443123790712
rate: 1906.4879113216743
rate: 1925.9420717997118
rate: 1934.166137658981
rate: 1954.3176526556897
rate: 1957.0107444542625

从 Web 服务器收集日志

Web 服务器会生成大量关于用户访问和错误的日志。Logstash 帮助使用输入插件从不同的服务器提取日志，并将它们存储在集中式位置。

我们正在从本地 Apache Tomcat 服务器的**stderr 日志**提取数据，并将其存储在 output.log 中。

logstash.conf

此 Logstash 配置文件指示 Logstash 读取 apache 错误日志并添加名为“apache-error”的标签。我们可以简单地使用文件输出插件将其发送到 output.log。

input {
   file {
      path => "C:/Program Files/Apache Software Foundation/Tomcat 7.0 /logs/*stderr*"
      type => "apache-error"  
   }
} 
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>Logstash –f Logstash.conf

输入日志示例

这是 Apache Tomcat 中发生服务器事件时生成的示例**stderr 日志**。

C:\Program Files\Apache Software Foundation\Tomcat 7.0\logs\ tomcat7-stderr.2016-12-25.log

Dec 25, 2016 7:05:14 PM org.apache.coyote.AbstractProtocol start
INFO: Starting ProtocolHandler ["http-bio-9999"]
Dec 25, 2016 7:05:14 PM org.apache.coyote.AbstractProtocol start
INFO: Starting ProtocolHandler ["ajp-bio-8009"]
Dec 25, 2016 7:05:14 PM org.apache.catalina.startup.Catalina start
INFO: Server startup in 823 ms

output.log

{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"Dec 25, 2016 7:05:14 PM org.apache.coyote.AbstractProtocol start\r",
   "type":"apache-error","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"INFO: Starting ProtocolHandler [
      \"ajp-bio-8009\"]\r","type":"apache-error","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"Dec 25, 2016 7:05:14 PM org.apache.catalina.startup.Catalina start\r",
   "type":"apache-error","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"INFO: Server startup in 823 ms\r","type":"apache-error","tags":[]
}

从数据源收集日志

首先，让我们了解如何配置 MySQL 进行日志记录。在 MySQL 数据库服务器的 [mysqld] 下的**my.ini 文件**中添加以下几行。

在 Windows 中，它位于 MySQL 的安装目录内，位于：

C:\wamp\bin\mysql\mysql5.7.11

在 UNIX 中，您可以在 – /etc/mysql/my.cnf 中找到它

general_log_file   = "C:/wamp/logs/queries.log"
general_log = 1

logstash.conf

在此配置文件中，文件插件用于读取 MySQL 日志并将其写入 ouput.log。

input {
   file {
      path => "C:/wamp/logs/queries.log"
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

queries.log

这是由在 MySQL 数据库中执行的查询生成的日志。

2016-12-25T13:05:36.854619Z   2 Query		select * from test1_users
2016-12-25T13:05:51.822475Z    2 Query	select count(*) from users
2016-12-25T13:05:59.998942Z    2 Query         select count(*) from test1_users

output.log

{
   "path":"C:/wamp/logs/queries.log","@timestamp":"2016-12-25T13:05:37.905Z",
   "@version":"1","host":"Dell-PC",
   "message":"2016-12-25T13:05:36.854619Z    2 Query\tselect * from test1_users",
   "tags":[]
}
{
   "path":"C:/wamp/logs/queries.log","@timestamp":"2016-12-25T13:05:51.938Z",
   "@version":"1","host":"Dell-PC",
   "message":"2016-12-25T13:05:51.822475Z    2 Query\tselect count(*) from users",
   "tags":[]
}
{
   "path":"C:/wamp/logs/queries.log","@timestamp":"2016-12-25T13:06:00.950Z",
   "@version":"1","host":"Dell-PC",
   "message":"2016-12-25T13:05:59.998942Z    2 Query\tselect count(*) from test1_users",
   "tags":[]
}

Logstash - 解析日志

Logstash 使用输入插件接收日志，然后使用过滤器插件来解析和转换数据。日志的解析和转换根据输出目标中存在的系统执行。Logstash 解析日志数据，仅转发所需的字段。之后，这些字段将转换为目标系统兼容且易于理解的形式。

如何解析日志？

日志的解析是通过使用**GROK**(图形化知识表示)模式来执行的，您可以在 Github 上找到它们：

https://github.com/elastic/logstash/tree/v1.4.2/patterns.

Logstash 将日志数据与指定的 GROK 模式或模式序列匹配，以解析日志，例如“%{COMBINEDAPACHELOG}”，这通常用于 apache 日志。

解析后的数据结构更清晰，更易于搜索和执行查询。Logstash 在输入日志中搜索指定的 GROK 模式，并提取日志中的匹配行。您可以使用 GROK 调试器来测试您的 GROK 模式。

GROK 模式的语法是 %{SYNTAX:SEMANTIC}。Logstash GROK 过滤器采用以下形式：

%{PATTERN:FieldName}

这里，PATTERN 代表 GROK 模式，fieldname 是字段的名称，它表示输出中解析的数据。

例如，使用在线 GROK 调试器 https://grokdebugger.com/

输入

日志中出现错误行的示例：

[Wed Dec 07 21:54:54.048805 2016] [:error] [pid 1234:tid 3456829102]
   [client 192.168.1.1:25007] JSP Notice:  Undefined index: abc in
   /home/manu/tpworks/tutorialspoint.com/index.jsp on line 11

GROK 模式序列

此 GROK 模式序列与包含时间戳、日志级别、进程 ID、事务 ID 和错误消息的日志事件匹配。

\[(%{DAY:day} %{MONTH:month} %{MONTHDAY} %{TIME} %{YEAR})\] \[.*:%{LOGLEVEL:loglevel}\]
   \[pid %{NUMBER:pid}:tid %{NUMBER:tid}\] \[client %{IP:clientip}:.*\]
   %{GREEDYDATA:errormsg}

输出

输出采用 JSON 格式。

{
   "day": [
      "Wed"
   ],
   "month": [
      "Dec"
   ],
   "loglevel": [
      "error"
   ],
   "pid": [
      "1234"
   ],
   "tid": [
      "3456829102"
   ],
   "clientip": [
      "192.168.1.1"
   ],
   "errormsg": [
      "JSP Notice:  Undefined index: abc in
      /home/manu/tpworks/tutorialspoint.com/index.jsp on line 11"
   ]
}

Logstash - 过滤器

Logstash 在输入和输出之间的管道中间使用过滤器。Logstash 的过滤器测量操作和创建事件，例如**Apache-Access**。许多过滤器插件用于管理 Logstash 中的事件。这里，在**Logstash 聚合过滤器**的示例中，我们正在过滤数据库中每个 SQL 事务的持续时间并计算总时间。

安装聚合过滤器插件

使用 Logstash-plugin 实用程序安装聚合过滤器插件。对于 Windows，Logstash-plugin 是 Logstash **bin 文件夹**中的批处理文件。

>logstash-plugin install logstash-filter-aggregate

logstash.conf

在此配置中，您可以看到三个“if”语句，用于**初始化、递增**和**生成**事务的总持续时间，即**sql_duration**。聚合插件用于添加存在于输入日志的每个事件中的 sql_duration。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - 
            %{NOTSPACE:taskid} - %{NOTSPACE:logger} - 
            %{WORD:label}( - %{INT:duration:int})?" 
      ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"    
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logstash –f logstash.conf 

input.log

以下代码块显示输入日志数据。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END - end

output.log

如配置文件中指定的，最后一个“if”语句中记录器为 – TRANSACTION_END，它打印总事务时间或 sql_duration。这在 output.log 中以黄色突出显示。

{
   "path":"C:/tpwork/logstash/bin/log/input.log","@timestamp": "2016-12-22T19:04:37.214Z",
   "loglevel":"INFO","logger":"TRANSACTION_START","@version": "1","host":"wcnlab-PC",
   "message":"8566 - TRANSACTION_START - start\r","tags":[]
}
{
   "duration":320,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.366Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 320\r","taskid":"48566","tags":[]
}
{
   "duration":200,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.373Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 200\r","taskid":"48566","tags":[]
}
{
   "sql_duration":520,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.380Z","loglevel":"INFO","logger":"TRANSACTION_END",
   "@version":"1","host":"wcnlab-PC","label":"end",
   "message":" INFO - 48566 - TRANSACTION_END - end\r","taskid":"48566","tags":[]
}

Logstash - 转换日志

Logstash 提供各种插件来转换解析的日志。这些插件可以**添加、删除**和**更新**日志中的字段，以便更好地理解和在输出系统中进行查询。

我们使用**Mutate 插件**在输入日志的每一行中添加一个字段名称 user。

安装 Mutate 过滤器插件

要安装 mutate 过滤器插件；我们可以使用以下命令。

>Logstash-plugin install Logstash-filter-mutate

logstash.conf

在此配置文件中，Mutate 插件在 Aggregate 插件之后添加，以添加新字段。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [ "message", "%{LOGLEVEL:loglevel} -
         %{NOTSPACE:taskid} - %{NOTSPACE:logger} -
         %{WORD:label}( - %{INT:duration:int})?" ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ; 
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
   mutate {
      add_field => {"user" => "tutorialspoint.com"}
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logstash –f logstash.conf

input.log

以下代码块显示输入日志数据。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END - end

output.log

您可以看到输出事件中有一个名为“user”的新字段。

{
   "path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.383Z",
   "@version":"1",
   "host":"wcnlab-PC",
   "message":"NFO - 48566 - TRANSACTION_START - start\r",
   "user":"tutorialspoint.com","tags":["_grokparsefailure"]
}
{
   "duration":320,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.383Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 320\r",
   "user":"tutorialspoint.com","taskid":"48566","tags":[]
}
{
   "duration":200,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.399Z","loglevel":"INFO",
   "logger":"SQL","@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 200\r",
   "user":"tutorialspoint.com","taskid":"48566","tags":[]
}
{
   "sql_duration":520,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.399Z","loglevel":"INFO",
   "logger":"TRANSACTION_END","@version":"1","host":"wcnlab-PC","label":"end",
   "message":" INFO - 48566 - TRANSACTION_END - end\r",
   "user":"tutorialspoint.com","taskid":"48566","tags":[]
}

Logstash - 输出阶段

输出是 Logstash 管道中的最后一个阶段，它将输入日志中的过滤器数据发送到指定的目的地。Logstash 提供多个输出插件，可以将过滤后的日志事件存储到各种不同的存储和搜索引擎中。

存储日志

Logstash 可以将过滤后的日志存储在**文件、Elasticsearch 引擎、stdout、AWS CloudWatch**等中。Logstash 还可以使用**TCP、UDP、Websocket**等网络协议将日志事件传输到远程存储系统。

在 ELK 堆栈中，用户使用 Elasticsearch 引擎来存储日志事件。在下面的示例中，我们将为本地 Elasticsearch 引擎生成日志事件。

安装 Elasticsearch 输出插件

我们可以使用以下命令安装 Elasticsearch 输出插件。

>logstash-plugin install Logstash-output-elasticsearch

logstash.conf

此配置文件包含一个 Elasticsearch 插件，该插件将输出事件存储在本地安装的 Elasticsearch 中。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [ "message", "%{LOGLEVEL:loglevel} -
      %{NOTSPACE:taskid} - %{NOTSPACE:logger} -  
      %{WORD:label}( - %{INT:duration:int})?" ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
   mutate {
      add_field => {"user" => "tutorialspoint.com"}
   }
}
output {
   elasticsearch {
      hosts => ["127.0.0.1:9200"]
   }
}

Input.log

以下代码块显示输入日志数据。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END - end

启动本地主机上的 Elasticsearch

要在本地主机上启动 Elasticsearch，您应该使用以下命令。

C:\elasticsearch\bin> elasticsearch

Elasticsearch 准备就绪后，您可以通过在浏览器中键入以下 URL 来检查它。

https://127.0.0.1:9200/

响应

以下代码块显示本地主机上 Elasticsearch 的响应。

{
   "name" : "Doctor Dorcas",
   "cluster_name" : "elasticsearch",
   "version" : {
      "number" : "2.1.1",
      "build_hash" : "40e2c53a6b6c2972b3d13846e450e66f4375bd71",
      "build_timestamp" : "2015-12-15T13:05:55Z",
      "build_snapshot" : false,
      "lucene_version" : "5.3.1"
   },
   "tagline" : "You Know, for Search"
}

**注意** - 有关 Elasticsearch 的更多信息，您可以点击以下链接。

https://tutorialspoint.com/elasticsearch/index.html

现在，使用上述 Logstash.conf 运行 Logstash

>Logstash –f Logstash.conf

在将上述文本粘贴到输出日志中之后，该文本将由 Logstash 存储在 Elasticsearch 中。您可以通过在浏览器中键入以下 URL 来检查存储的数据。

https://127.0.0.1:9200/logstash-2017.01.01/_search?pretty

响应

这是存储在索引 Logstash-2017.01.01 中的 JSON 格式数据。

{
   "took" : 20,
   "timed_out" : false,
   "_shards" : {
      "total" : 5,
      "successful" : 5,
      "failed" : 0
   },
   "hits" : {
      "total" : 10,
      "max_score" : 1.0,
      "hits" : [ {
         "_index" : "logstash-2017.01.01",
         "_type" : "logs",
         "_id" : "AVlZ9vF8hshdrGm02KOs",
         "_score" : 1.0,
         "_source":{
            "duration":200,"path":"C:/tpwork/logstash/bin/log/input.log", 
            "@timestamp":"2017-01-01T12:17:49.140Z","loglevel":"INFO",
            "logger":"SQL","@version":"1","host":"wcnlab-PC",
            "label":"transaction1",
            "message":" INFO - 48566 - SQL - transaction1 - 200\r",
            "user":"tutorialspoint.com","taskid":"48566","tags":[]
         }
      },
      {
         "_index" : "logstash-2017.01.01",
         "_type" : "logs",
         "_id" : "AVlZ9vF8hshdrGm02KOt",
         "_score" : 1.0,
         "_source":{
            "sql_duration":520,"path":"C:/tpwork/logstash/bin/log/input.log",
            "@timestamp":"2017-01-01T12:17:49.145Z","loglevel":"INFO",
            "logger":"TRANSACTION_END","@version":"1","host":"wcnlab-PC",
            "label":"end",
            "message":" INFO - 48566 - TRANSACTION_END - end\r",
            "user":"tutorialspoint.com","taskid":"48566","tags":[]
         }
      }
   }
}

Logstash - 支持的输出

Logstash 提供多个插件来支持各种数据存储或搜索引擎。日志的输出事件可以发送到输出文件、标准输出或 Elasticsearch 等搜索引擎。Logstash 中有三种受支持的输出类型，它们是：

• 标准输出
• 文件输出
• 空输出

现在让我们详细讨论每个输出。

标准输出 (stdout)

它用于将过滤后的日志事件作为数据流生成到命令行界面。这是一个将数据库事务的总持续时间生成到 stdout 的示例。

logstash.conf

此配置文件包含一个标准输出插件，用于将总 sql_duration 写入标准输出。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid}
            - %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" 
      ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
}
output {
   if [logger] == "TRANSACTION_END" {
      stdout {
         codec => line{format => "%{sql_duration}"}
      }
   }
}

注意 − 如果尚未安装，请安装聚合过滤器。

>logstash-plugin install Logstash-filter-aggregate

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logstash –f logsatsh.conf

Input.log

以下代码块显示输入日志数据。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END – end

stdout（在 Windows 中为命令提示符，在 UNIX 中为终端）

这是总 sql_duration 320 + 200 = 520。

520

文件输出

Logstash 还可以将过滤器日志事件存储到输出文件中。我们将使用上述示例，将输出存储到文件中，而不是 STDOUT。

logstash.conf

此 Logstash 配置文件指示 Logstash 将总 sql_duration 存储到输出日志文件中。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input1.log"
   }
} 
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid} -
            %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" 
      ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
}
output {
   if [logger] == "TRANSACTION_END" {
      file {
         path => "C:/tpwork/logstash/bin/log/output.log"
         codec => line{format => "%{sql_duration}"}
      }
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logstash –f logsatsh.conf

input.log

以下代码块显示输入日志数据。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END – end

output.log

以下代码块显示输出日志数据。

520

空输出

这是一个特殊的输出插件，用于分析输入和过滤器插件的性能。

Logstash - 插件

Logstash 为其管道的三个阶段（输入、过滤器和输出）提供各种插件。这些插件帮助用户从各种来源捕获日志，例如 Web 服务器、数据库、网络协议等。

捕获后，Logstash 可以根据用户的需求将数据解析并转换为有意义的信息。最后，Logstash 可以将这些有意义的信息发送或存储到各种目标来源，例如 Elasticsearch、AWS Cloudwatch 等。

输入插件

Logstash 中的输入插件帮助用户从各种来源提取和接收日志。使用输入插件的语法如下：

Input {
   Plugin name {
      Setting 1……
      Setting 2……..
   }
}

您可以使用以下命令下载输入插件：

>Logstash-plugin install Logstash-input-<plugin name>

Logstash-plugin 实用程序位于 Logstash 安装目录的bin 文件夹中。下表列出了 Logstash 提供的输入插件。

插件设置

所有插件都有其特定的设置，有助于指定插件中的重要字段，例如端口、路径等。我们将讨论一些输入插件的设置。

文件

此输入插件用于直接从输入源中存在的日志或文本文件提取事件。它的工作方式类似于 UNIX 中的 tail 命令，保存最后读取的游标，只读取输入文件中新追加的数据，但这可以通过使用 star_position 设置来更改。以下是此输入插件的设置。

Elasticsearch

此特定插件用于读取 Elasticsearch 集群中的搜索查询结果。以下是此插件中使用的设置：

eventlog

此输入插件从 Windows 服务器的 win32 API 读取数据。以下是此插件的设置：

Twitter

此输入插件用于从其流式 API 收集 Twitter 的 feed。下表描述了此插件的设置。

TCP

TCP 用于通过 TCP 套接字获取事件；它可以从用户连接或服务器读取，这取决于 mode 设置。下表描述了此插件的设置：

Logstash – 输出插件

Logstash 支持各种输出源和不同的技术，例如数据库、文件、电子邮件、标准输出等。

使用输出插件的语法如下：

output {
   Plugin name {
      Setting 1……
      Setting 2……..
   }
}

您可以使用以下命令下载输出插件：

>logstash-plugin install logstash-output-<plugin name>

Logstash-plugin 实用程序位于 Logstash 安装目录的 bin 文件夹中。下表描述了 Logstash 提供的输出插件。

所有插件都有其特定的设置，这有助于在插件中指定重要的字段，例如端口、路径等。我们将讨论一些输出插件的设置。

Elasticsearch

Elasticsearch 输出插件使 Logstash 能够将输出存储在 Elasticsearch 引擎的特定集群中。这是用户常用的选择之一，因为它包含在 ELK Stack 包中，因此为 DevOps 提供端到端解决方案。下表描述了此输出插件的设置。

Email

电子邮件输出插件用于在 Logstash 生成输出时通知用户。下表描述了此插件的设置。

Http

此设置用于通过 http 将输出事件发送到目标。此插件具有以下设置：

stdout

stdout 输出插件用于将输出事件写入命令行界面的标准输出。在 Windows 中是命令提示符，在 UNIX 中是终端。此插件具有以下设置：

statsd

它是一个网络守护程序，用于通过 UDP 将矩阵数据发送到目标后端服务。在 Windows 中是命令提示符，在 UNIX 中是终端。此插件具有以下设置：

过滤器插件

Logstash 支持各种过滤器插件，用于将输入日志解析和转换为更结构化且易于查询的格式。

使用过滤器插件的语法如下：

filter {
   Plugin name {
      Setting 1……
      Setting 2……..
   }
}

您可以使用以下命令下载过滤器插件：

>logstash-plugin install logstash-filter-<plugin name>

Logstash-plugin 实用程序位于 Logstash 安装目录的 bin 文件夹中。下表描述了 Logstash 提供的输出插件。

Codec 插件

Codec 插件可以是输入或输出插件的一部分。这些插件用于更改或格式化日志数据的表示方式。Logstash 提供多个 Codec 插件，如下所示：

构建您自己的插件

您还可以创建适合您需求的 Logstash 自定制插件。Logstash-plugin 实用程序用于创建自定制插件。在这里，我们将创建一个过滤器插件，它将在事件中添加一条自定制消息。

生成基本结构

用户可以使用 logstash-plugin 实用程序的 generate 选项生成必要的文 件，也可以在 GitHub 上找到。

>logstash-plugin generate --type filter --name myfilter --path c:/tpwork/logstash/lib

这里，**type** 选项用于指定插件是输入、输出还是过滤器。在此示例中，我们正在创建名为 **myfilter** 的过滤器插件。path 选项用于指定要创建插件目录的路径。执行上述命令后，您将看到创建了一个目录结构。

开发插件

您可以在插件目录的 **\lib\logstash\filters** 文件夹中找到插件的代码文件。文件扩展名为 **.rb**。

在我们的例子中，代码文件位于以下路径：

C:\tpwork\logstash\lib\logstash-filter-myfilter\lib\logstash\filters\myfilter.rb

我们将消息更改为：default ⇒ "Hi, You are learning this on tutorialspoint.com" 并保存文件。

安装插件

要安装此插件，需要修改 Logstash 的 Gemfile。您可以在 Logstash 的安装目录中找到此文件。在我们的例子中，它位于 **C:\tpwork\logstash**。使用任何文本编辑器编辑此文件，并在其中添加以下文本。

gem "logstash-filter-myfilter",:path => "C:/tpwork/logstash/lib/logstash-filter-myfilter"

在上述命令中，我们指定了插件的名称以及可以在哪里找到它以进行安装。然后，运行 Logstash-plugin 实用程序来安装此插件。

>logstash-plugin install --no-verify

测试

在这里，我们在之前的示例之一中添加了 **myfilter**：

logstash.conf

此 Logstash 配置文件在 grok 过滤器插件之后，在 filter 部分包含 myfilter。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input1.log"
   } 
}
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid} -
            %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" ]
   }
   myfilter{}
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output1.log"
      codec => rubydebug
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logstash –f logsatsh.conf

input.log

以下代码块显示输入日志数据。

INFO - 48566 - TRANSACTION_START - start

output.log

以下代码块显示输出日志数据。

{
   "path" => "C:/tpwork/logstash/bin/log/input.log",
   "@timestamp" => 2017-01-07T06:25:25.484Z,
   "loglevel" => "INFO",
   "logger" => "TRANSACTION_END",
   "@version" => "1",
   "host" => "Dell-PC",
   "label" => "end",
   "message" => "Hi, You are learning this on tutorialspoint.com",
   "taskid" => "48566",
   "tags" => []
}

在 Logstash 上发布

开发人员还可以通过将其上传到 github 并遵循 Elasticsearch 公司定义的标准化步骤，将其自定制插件发布到 Logstash。

有关发布的更多信息，请参阅以下 URL：

https://elastic.ac.cn/guide/en/logstash/current/contributing-to-logstash.html

Logstash - 监控 API

Logstash 提供 API 来监控其性能。这些监控 API 提取有关 Logstash 的运行时指标。

节点信息 API

此 API 用于获取有关 Logstash 节点的信息。它以 JSON 格式返回 OS、Logstash 管道和 JVM 的信息。

您可以使用以下 URL 向 Logstash 发送 **get** 请求来提取信息：

GET https://127.0.0.1:9600/_node?pretty

响应

以下是节点信息 API 的响应。

{
   "host" : "Dell-PC",
   "version" : "5.0.1",
   "http_address" : "127.0.0.1:9600",
   
   "pipeline" : {
      "workers" : 4,
      "batch_size" : 125,
      "batch_delay" : 5,
      "config_reload_automatic" : false,
      "config_reload_interval" : 3
   },
   "os" : {
      "name" : "Windows 7",
      "arch" : "x86",
      "version" : "6.1",
      "available_processors" : 4
   },
   "jvm" : {
      "pid" : 312,
      "version" : "1.8.0_111",
      "vm_name" : "Java HotSpot(TM) Client VM",
      "vm_version" : "1.8.0_111",
      "vm_vendor" : "Oracle Corporation",
      "start_time_in_millis" : 1483770315412,
      
      "mem" : {
         "heap_init_in_bytes" : 16777216,
         "heap_max_in_bytes" : 1046937600,
         "non_heap_init_in_bytes" : 163840,
         "non_heap_max_in_bytes" : 0
      },
      "gc_collectors" : [ "ParNew", "ConcurrentMarkSweep" ]
   }
}

您还可以通过在 URL 中添加其名称来获取管道、操作系统和 JVM 的特定信息。

GET https://127.0.0.1:9600/_node/os?pretty
GET https://127.0.0.1:9600/_node/pipeline?pretty
GET https://127.0.0.1:9600/_node/jvm?pretty

插件信息 API

此 API 用于获取有关 Logstash 中已安装插件的信息。您可以通过向下面提到的 URL 发送 get 请求来检索此信息：

GET https://127.0.0.1:9600/_node/plugins?pretty

响应

以下是插件信息 API 的响应。

{
   "host" : "Dell-PC",
   "version" : "5.0.1",
   "http_address" : "127.0.0.1:9600",
   "total" : 95,
   "plugins" : [ {
      "name" : "logstash-codec-collectd",
      "version" : "3.0.2"
   },
   {
      "name" : "logstash-codec-dots",
      "version" : "3.0.2"
   },
   {
      "name" : "logstash-codec-edn",
      "version" : "3.0.2"
   },
   {
      "name" : "logstash-codec-edn_lines",
      "version" : "3.0.2"
   },
   ............
}

节点统计信息 API

此 API 用于以 JSON 对象提取 Logstash 的统计信息（内存、进程、JVM、管道）。您可以通过向下面提到的 URL 发送 get 请求来检索此信息：

GET https://127.0.0.1:9600/_node/stats/?pretty
GET https://127.0.0.1:9600/_node/stats/process?pretty
GET https://127.0.0.1:9600/_node/stats/jvm?pretty
GET https://127.0.0.1:9600/_node/stats/pipeline?pretty

热线程 API

此 API 检索有关 Logstash 中热线程的信息。热线程是 Java 线程，其 CPU 使用率很高，并且运行时间长于正常执行时间。您可以通过向下面提到的 URL 发送 get 请求来检索此信息：

GET https://127.0.0.1:9600/_node/hot_threads?pretty

用户可以使用以下 URL 以更易于阅读的形式获取响应。

GET https://127.0.0.1:9600/_node/hot_threads?human = true

Logstash - 安全性和监控

在本章中，我们将讨论 Logstash 的安全性和监控方面。

监控

Logstash 是一个非常好的工具，用于监控生产环境中的服务器和服务。生产环境中的应用程序会生成不同类型的日志数据，例如访问日志、错误日志等。Logstash 可以使用过滤器插件计算或分析错误、访问或其他事件的数量。此分析和计数可用于监控不同的服务器及其服务。

Logstash 提供了诸如 **HTTP Poller** 之类的插件来监控网站状态监控。在这里，我们正在监控一个名为 **mysite** 的网站，该网站托管在本地 Apache Tomcat 服务器上。

logstash.conf

在此配置文件中，http_poller 插件用于在 interval 设置中指定的时间间隔后命中插件中指定的站点。最后，它将站点的状态写入标准输出。

input {
   http_poller {
      urls => {
         site => "https://127.0.0.1:8080/mysite"
      }
      request_timeout => 20
      interval => 30
      metadata_target => "http_poller_metadata"
   }
}
output {
   if [http_poller_metadata][code] == 200 {
      stdout {
         codec => line{format => "%{http_poller_metadata[response_message]}"}
      }
   }
   if [http_poller_metadata][code] != 200 {
      stdout {
         codec => line{format => "down"}
      }
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logstash –f logstash.conf

stdout

如果站点正常运行，则输出将为：

Ok

如果我们使用 Tomcat 的 **Manager App** 停止站点，输出将更改为：

down

安全

Logstash 提供了大量功能，用于与外部系统进行安全通信并支持身份验证机制。所有 Logstash 插件都支持通过 HTTP 连接进行身份验证和加密。

使用 HTTP 协议的安全

在 Logstash 提供的各种插件（例如 Elasticsearch 插件）中，有用户和密码等设置用于身份验证。

elasticsearch {
   user => <username>
   password => <password>
}

另一种身份验证是用于 Elasticsearch 的 **PKI（公钥基础设施）**。开发人员需要在 Elasticsearch 输出插件中定义两个设置才能启用 PKI 身份验证。

elasticsearch {
   keystore => <string_value>
   keystore_password => <password>
}

在 HTTPS 协议中，开发人员可以使用授权机构的证书进行 SSL/TLS。

elasticsearch {
   ssl => true
   cacert => <path to .pem file>
}

使用传输协议的安全

要将传输协议与 Elasticsearch 一起使用，用户需要将协议设置设置为传输。这避免了 JSON 对象的反序列化，并提高了效率。

基本身份验证与在 Elasticsearch 输出协议中的 http 协议中执行的身份验证相同。

elasticsearch {
   protocol => “transport”
   user => <username>
   password => <password>
}

PKI 身份验证还需要将 SSL 设置为 true，并在 Elasticsearch 输出协议中使用其他设置：

elasticsearch {
   protocol => “transport”
   ssl => true
   keystore => <string_value>
   keystore_password => <password>
}

最后，SSL 安全性比其他通信安全方法需要更多设置。

elasticsearch {
   ssl => true
   ssl => true
   keystore => <string_value>
   keystore_password => <password>
   truststore => 
   truststore_password => <password>
}

Logstash 的其他安全优势

Logstash 可以帮助输入系统源防止拒绝服务攻击等攻击。监控日志并分析这些日志中的不同事件可以帮助系统管理员检查传入连接和错误的变化。这些分析可以帮助查看服务器上是否正在发生或即将发生攻击。

Elasticsearch 公司的其他产品，例如 **x-pack** 和 **filebeat**，提供了一些与 Logstash 安全通信的功能。