Elasticsearch教程，我们将使用Elasticsearch 7.5涵盖ELK Stack上的所有相关主题。

• 安装和配置带有3个节点的ElasticSearh Cluster 7.5

• 启用HTTPS并配置SSS/TLS以保护Elasticsearch Cluster

• 为Elasticsearch Cluster安装和配置带有SSL/TLS的Kibana 7.5

• 配置Metricbeat 7.5以通过HTTPS监视Elasticsearch群集设置

• 使用Elasticsearch安装和配置Logstash 7.5

Logstash概述

Logstash有助于集中事件数据，例如日志，指标或者任何其他格式的数据。
在将其发送到我们选择的存储库之前，它可以执行许多转换。
它是弹性堆栈的关键组件，用于集中化数据管道中的收集和转换过程。

Logstash是服务器端组件。
它的作用是以可伸缩的方式集中从大量输入源收集数据，并将数据转换并发送到我们选择的输出。
通常，将输出发送到Elasticsearch，但是Logstash能够将其发送到各种各样的输出。

在本文中，我们将学习

• 安装和配置Logstash 7.5

• 如何从Logstash发送数据到ElasticSearch

• 如何检查Logstash是否正在将数据发送到Elasticsearch

Logstash体系结构

Logstash事件处理管道具有三个阶段，即输入，过滤器和输出。
Logstash管道具有两个必需的元素，即输入和输出，以及一个称为过滤器的选项元素：

鸣谢：Lerning Logstash 7

输入创建事件，过滤器修改输入事件，然后输出将它们发送到目标。
输入和输出支持编解码器，使用编解码器，我们可以在数据进入或者退出管道时进行编码或者解码，而不必使用单独的过滤器。

安装Java

要配置Logstash，我们需要Java 8或者Java11. 要检查Java版本，请运行以下命令

[root@centos-8 ~]# java -version
openjdk version "1.8.0_232"
OpenJDK Runtime Environment (build 1.8.0_232-b09)
OpenJDK 64-Bit Server VM (build 25.232-b09, mixed mode)

如果未安装Java，则可以使用Oracle官方发行版或者开源发行版进行安装

安装Logstash

我们将使用yum存储库安装logstash，但我们也可以从elasticsearch中选择其他安装选项。

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

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

将以下内容添加到/etc/yum.repos.d/目录中，后缀为.repo的文件中，例如logstash.repo。

[logstash-7.x]
name=Elastic repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

存储库已准备就绪，可以使用。
我们可以使用以下方法安装它：

# yum install logstash

生成SSL证书

我们将使用先前为ELK堆栈中的centos-8创建的证书。

在/etc/logstash /下创建目录certs并复制我们在前面的文章中创建的所有证书

[root@centos-8 ~]# ls -l /etc/logstash/certs
total 16
-rw-r--r-- 1 root root 1200 Dec 30 19:51 ca.crt
-rw-r--r-- 1 root root 1200 Dec 31 13:08 centos-8.crt
-rw-r--r-- 1 root root 1679 Dec 30 20:00 centos-8.key

提示：

如果计划对ELK堆栈使用filebeat，则必须将PEM证书转换为PKCS#8格式

[root@centos-8 ~]# openssl pkcs8 -in centos-8.key -topk8 -nocrypt -out centos-8.pkcs8.key

运行Logstash

Logstash要求在运行它时指定配置。
通过使用-f选项/标志指定配置文件(.conf文件)，可以使用-e选项将配置直接指定为参数。
在配置Logstash与Elasticsearch集群连接之前，首先，我们通过运行最基本的Logstash管道来测试Logstash的安装。

Logstash管道具有两个必需的元素，即输入和输出，以及一个可选的元素，即过滤器。
输入插件使用来自源的数据，过滤器插件根据指定修改数据，输出插件将数据写入目标。

logstash二进制文件位于内部

[root@centos-8 ~]# ls -l /usr/share/logstash/bin/logstash
-rwxr-xr-x 1 logstash logstash 2354 Dec 17 06:22 /usr/share/logstash/bin/logstash

(可选)现在默认情况下，该路径在PATH变量中不可用，因此每次执行logstash时都必须使用绝对路径，为避免这种情况，我们将更新PATH变量。
下面的命令将临时更新PATH变量，以使其永久重启，我们可以在系统中定义PATH变量的位置进行更新。

[root@centos-8 ~]# export PATH=$PATH:/usr/share/logstash/bin

验证新路径

[root@centos-8 ~]# echo $PATH
/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin:/usr/share/logstash/bin

现在我们可以直接执行logstash。
启动Logstash后，请等待直到成功看到Logstash API端点，然后在命令提示符下输入" hello world"：

[root@centos-8 ~]# logstash -e 'input { stdin { } } output { stdout {} }'
<Output trimmed>
The stdin plugin is now waiting for input:
[INFO ] 2019-12-31 18:48:30.830 [Api Webserver] agent - Successfully started Logstash API endpoint {:port=>9600}
Hello World
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/awesome_print-1.7.0/lib/awesome_print/formatters/base_formatter.rb:31: warning: constant ::Fixnum is deprecated
{
    "@timestamp" => 2019-12-31T13:18:40.701Z,
          "host" => "centos-8",
       "message" => "Hello World",
      "@version" => "1"
}

-e标志使我们可以直接从命令行指定配置。
在命令行中指定配置可让我们快速测试配置，而无需在两次迭代之间编辑文件

通过在运行Logstash的shell中发出CTRL-D命令来退出Logstash。

基本的Logstash管道

Logstash管道存储在以.conf扩展名结尾的配置文件中。
配置文件的三个部分如下：

input
{
}
filter
{
}
output
{
}

这些部分中的每个部分都包含一个或者多个插件配置。
可以通过提供插件名称，然后将其设置作为键值对来配置插件。
使用=>运算符将值分配给键。

下面是一个示例配置文件

• 我们可能会注意到，该文件包含两个必需元素，即输入和输出，并且输入部分具有名为stdin的插件，该插件接受默认参数。

• 输出部分有一个stdout插件，可以接受rubydebug编解码器。

• stdin用于从标准输入读取输入，而stdout插件用于将事件信息写入标准输出。

• rubydebug编解码器将使用Ruby Awesome Print库输出Logstash事件数据。

• 它还包含一个带有mutate插件的过滤器部分，该部分将文本" hynman"替换为" rahul"。

[root@centos-8 ~]# cat console.conf
input {
   stdin{}
}
filter {
   ruby {
     code => 'event.set("message", event.get("message").sub("hynman", "rahul"))'
   }
}
output {
   stdout{codec => rubydebug}
}

我们将使用-f和带有路径的配置文件名执行logstash。

[root@centos-8 logstash_files]# logstash -f console.conf
<Output trimmed>
[INFO ] 2019-12-30 20:51:28.279 [Agent thread] agent - Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
The stdin plugin is now waiting for input:
[INFO ] 2019-12-30 20:51:29.972 [Api Webserver] agent - Successfully started Logstash API endpoint {:port=>9601}
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/awesome_print-1.7.0/lib/awesome_print/formatters/base_formatter.rb:31: warning: constant ::Fixnum is deprecated { "@version" => "1",
Hello World  <-- This is our first input 
          "host" => "centos-8.example.com",
    "@timestamp" => 2019-12-30T15:22:03.820Z,
       "message" => "Hello World"
}
My name is hynman  <-- Next input

{ 
      "@version" => "1",
          "host" => "centos-8.example.com",
    "@timestamp" => 2019-12-30T15:22:09.715Z,
       "message" => "My name is rahul"  <-- The output was successfully processed and hynman was replaced by rahul
}

配置logstash.yml以启用xpack监视

要配置logstash，我们必须修改/etc/logstash中可用的logstash.yml。
为了控制如何从Logstash收集监视数据并将其发送到elasticsearch，可以在logstash.yml中配置xpack.monitoring设置。

node.name: centos-8
path.data: /var/lib/logstash
path.config: /etc/logstash/conf.d/*.conf
path.logs: /var/log/logstash
xpack.monitoring.enabled: true
xpack.monitoring.elasticsearch.username: 'logstash_system'
xpack.monitoring.elasticsearch.password: 'Passw0rd'
xpack.monitoring.elasticsearch.hosts: [ "https://192.168.0.11:9200", "https://192.168.0.12:9200", "https://192.168.0.13:9200" ]
xpack.monitoring.elasticsearch.ssl.certificate_authority: '/etc/logstash/certs/ca.crt'

其中我们使用了logstash_system内置用户，并且使用它的密码连接到elasticsearch集群。

将Logstash与Elasticsearch连接

该插件用于将事件从Logstash传输到Elasticsearch。
建议使用此插件将事件/日志数据从Logstash推送到Elasticsearch。
一旦数据进入Elasticsearch，就可以使用Kibana轻松地对其进行可视化。
该插件不需要强制性参数，它会自动尝试连接到Elasticsearch

由于我们将/etc/logstash/conf.d /用作path.config来存储配置文件，因此我将在此位置创建我的配置文件，该文件将用于将logstash连接到elasticsearch

[root@centos-8 ~]# cat /etc/logstash/conf.d/connect_slasticsearch.conf
input {
  beats {
    port => 5044
    ssl => true
    ssl_key => '/etc/logstash/certs/centos-8.key'
    ssl_certificate => '/etc/logstash/certs/ca.crt'
  }
  file {
         path => "/var/log/messages"
  }
}
output {
  elasticsearch {
    hosts => ["https://192.168.0.11:9200","https://192.168.0.12:9200","https://192.168.0.13:9200"]
    cacert => '/etc/logstash/certs/ca.crt'
    user => 'logstash_system'
    password => 'redhat'
  }
}

由于我们使用SSL加密来配置logstash和elasticsearch，因此我们将在配置文件中使用各自的值。

启动logstash服务

我们已经完成配置logstash并连接到我们的elasticsearch集群的步骤。
启动logstash服务，并使其在每次重新启动时自动启动

[root@centos-8 ~]# systemctl start logstash
[root@centos-8 ~]# systemctl enable logstash

我们可以使用journalctl -u logstash -f监视logstash服务的日志，或者检查/var/log/logstash中可用的日志。
现在，配置logstash后，请检查Kibana的" Stack Monitoring"部分以确保Logstash节点位于添加。

将数据从logstash发送到elasticsearch

其中我们将创建一个logstash管道，并使用索引查看管道上的数据。
在这个配置文件中，我将从/tmp/dummy.txt的内容中获取输入，并且在KIbana仪表板上也可以看到相同的内容。

为了完成这项工作，我们需要两个终端，其中在第一个终端上我们将执行logstash，而在另一个终端上我们将数据添加到/tmp/dummy.txt文件中。
一旦我们将数据输入到dummy.txt中，该数据将在logstash上可见，也可以在Kibana上查看

[root@centos-8 logstash_files]# logstash -f console.conf
<Output trimmed>
[INFO ] 2019-12-30 23:48:08.587 [[main]-pipeline-manager] javapipeline - Pipeline started {"pipeline.id"=>"main"}
[INFO ] 2019-12-30 23:48:09.204 [Agent thread] agent - Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[INFO ] 2019-12-30 23:48:09.223 [[main]<file] observingtail - START, creating Discoverer, Watch with file and sincedb collections [INFO ] 2019-12-30 23:48:11.735 [Api Webserver] agent - Successfully started Logstash API endpoint {:port=>9601}
{
      "@version" => "1",
       "message" => "Hello World",
          "path" => "/tmp/dummy.txt",
          "host" => "centos-8.example.com",
    "@timestamp" => 2019-12-30T18:19:09.053Z
}
{
      "@version" => "1",
       "message" => "Test Message 1",
          "path" => "/tmp/dummy.txt",
          "host" => "centos-8.example.com",
    "@timestamp" => 2019-12-30T18:19:29.955Z
}
{
      "@version" => "1",
       "message" => "Test Message 2",
          "path" => "/tmp/dummy.txt",
          "host" => "centos-8.example.com",
    "@timestamp" => 2019-12-30T18:19:35.024Z
}
^C[WARN ] 2019-12-30 23:49:39.441 [SIGINT handler] runner - SIGINT received. Shutting down.
[INFO ] 2019-12-30 23:49:39.792 [Converge PipelineAction::Stop
[INFO ] 2019-12-30 23:49:39.792 [Converge PipelineAction::Stop<main>] observingtail - QUIT - closing all files and shutting down.
[INFO ] 2019-12-30 23:49:40.506 [Converge PipelineAction::Stop<main>] javapipeline - Pipeline terminated {"pipeline.id"=>"main"}
[INFO ] 2019-12-30 23:49:40.741 [LogStash::Runner] runner - Logstash shut down.

同时，我还有另一个终端窗口，在该窗口上我将数据放入/tmp/dummy.txt中。

[root@centos-8 ~]# echo "Hello World" >> /tmp/dummy.txt
[root@centos-8 ~]# echo "Test Message 1" >> /tmp/dummy.txt
[root@centos-8 ~]# echo "Test Message 2" >> /tmp/dummy.txt

因此，在上面突出显示的部分中，我们可以看到/tmp/dummy.txt中的每个输入，logstash都会根据我们的配置文件生成一个有组织的输出。

可视化Kibana上的数据

现在要可视化Kibana上的数据，我们必须首先创建一个索引。
要创建索引，请登录到Kibana仪表板。
点击左侧面板中的"管理"，然后点击"索引模式"。
请按照下图的说明进行操作：

单击创建索引模式以继续下一步。
如图所示，将索引模式提供为logstash- *，然后单击下一步

接下来，从下拉菜单中选择" @timestamp"字段，然后单击"创建索引模式"

在下面，我们现在可以看到"索引模式"已创建

接下来，从左侧面板菜单中单击"发现"以可视化我们使用数据管道发送到elasticsearch的日志

因此，我们在/tmp/dummy，txt中获得的所有日志都可以在Kibana仪表板上看到。
我们可以从左侧面板部分添加更多字段，该部分可以在"可用字段"下看到

因此，这是使用logstash在Kibana仪表板上可视化来自elasticsearch集群的日志的基本配置。