Logstash【从无到有从有到无】【L2】Logstash入门
目录
1.Logstash入门
1.1.安装Logstash
1.1.1.从下载的二进制文件安装
1.1.2.从包存储库安装（Package Repositories）
1.1.3.APT
1.1.3.YUM
1.1.4.在macOS上使用Homebrew安装Logstash
1.1.5.使用Homebrew启动Logstash
1.1.6.Docker
1.2.简单使用
1.3.用Logstash解析日志
1.3.1.配置Filebeat以将日志行发送到Logstash
1.3.2.配置Logstash以Filebeat作为输入
1.3.3.使用Grok Filter插件解析Web日志
1.3.4.使用Geoip Filter插件增强您的数据
1.3.5.将数据索引到Elasticsearch
1.3.6.测试您的管道
1.4.将多个输入和输出插件拼接在一起
1.4.1.读取Twitter Feed
1.4.2.配置Filebeat以将日志行发送到Logstash
1.4.3.将Logstash数据写入文件
1.4.4.写入多个Elasticsearch Nodes
1.4.5.测试管道
1.Logstash入门
本节将指导您完成安装Logstash并验证一切正常运行的过程。 在学习如何存储第一个事件之后，您将继续创建一个更高级的管道，将Apache Web日志作为输入，解析日志，并将解析后的数据写入Elasticsearch集群。 然后，您将学习如何将多个输入和输出插件拼接在一起，以统一来自各种不同来源的数据。
1.1.安装Logstash
Logstash需要Java 8或Java 11.使用官方Oracle发行版或OpenJDK等开源发行版。
要检查Java版本，请运行以下命令：
java -versionjava version "11.0.1" 2018-10-16 LTS
Java(TM) SE Runtime Environment 18.9 (build 11.0.1+13-LTS)
Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.1+13-LTS, mixed mode)在某些Linux系统上，您可能还需要在尝试安装之前导出JAVA_HOME环境，特别是如果您从安装包安装Java。 这是因为Logstash在安装期间使用Java来自动检测您的环境并安装正确的启动方法（SysV init脚本，Upstart或systemd）。 如果Logstash在程序包安装期间无法找到JAVA_HOME环境变量，则可能会收到错误消息，并且Logstash将无法正常启动。
1.1.1.从下载的二进制文件安装
Logstash二进制文件可从https://www.elastic.co/downloads获得。 下载主机环境的Logstash安装文件 - TARG.GZ，DEB，ZIP或RPM。
解压缩文件。 不要将Logstash安装到包含冒号（:)字符的目录路径中。
这些软件包可以在弹性许可证下免费使用。它们包含开源和免费的商业功能以及对付费商业功能的访问。开始为期30天的试用，试用所有付费商业功能。有关弹性许可证级别的信息，请参阅订阅页。
或者，您可以下载一个OSS包，它只包含Apache2.0许可证下可用的功能。
在支持的Linux操作系统上，可以使用包管理器安装logstash。
1.1.2.从包存储库安装（Package Repositories）
我们还为基于apt和yum的发行版提供了存储库。注意，我们只提供二进制包，但不提供源包，因为包是作为logstash构建的一部分创建的。
我们已经将logstash包存储库按版本划分为不同的URL，以避免跨主要版本的意外升级。对于所有7.x.y版本，使用7.x作为版本号。
我们使用PGP密钥D88E42B4，弹性签名密钥，带有指纹
4609 5ACC 8548 582C 1A26 99A9 D27D 666C D88E 42B4签署我们所有的包裹。 它可以从https://pgp.mit.edu获得。
1.1.3.APT
下载并安装公共签名密钥：
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -在继续之前，您可能需要在Debian 上安装apt-transport-https软件包：
sudo apt-get install apt-transport-https将存储库定义保存到 /etc/apt/sources.list.d/elastic-7.x.list：
echo "deb https://artifacts.elastic.co/packages/7.x/apt stable main" | sudo tee -a /etc/apt/sources.list.d/elastic-7.x.list使用上述echo方法添加Logstash存储库。不要使用add-apt-repository，因为它也会添加一个deb-src条目，但我们不提供源包。如果您添加了该deb-src条目，您将看到如下错误：
Unable to find expected entry 'main/source/Sources' in Release file (Wrong sources.list entry or malformed file)只需从/etc/apt/sources.list文件中删除deb-src条目，安装应按预期工作。
运行sudo apt-get update并且存储库已准备就绪。你可以安装它：
sudo apt-get update && sudo apt-get install logstash有关将Logstash作为系统服务进行管理的详细信息，请参阅运行Logstash。
1.1.3.YUM
下载并安装公共签名密钥：
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您的存储库已准备就绪。你可以安装它：
sudo yum install logstash存储库不适用于仍使用RPM v3的旧的基于rpm的发行版，如CentOS5。
请参阅Running Logstash文档以将Logstash作为系统服务进行管理。
1.1.4.在macOS上使用Homebrew安装Logstash
Elastic发布Homebrew公式，因此您可以使用Homebrew包管理器安装Logstash 。
要使用Homebrew进行安装，首先需要点击Elastic Homebrew存储库：
brew tap elastic/tap在您点击Elastic Homebrew 仓库后，您可以使用brew install安装Logstash的默认分发版：
brew install elastic/tap/logstash-full这将安装最近发布的Logstash默认分发版。要安装OSS分发，请指定elastic/tap/logstash-oss。
1.1.5.使用Homebrew启动Logstash
要立即启动弹性/ tap / logstash-full并在登录时重新启动，请运行：
brew services start elastic/tap/logstash-full要在前台运行Logstash，请运行：
logstash1.1.6.Docker
图像可用于将Logstash作为Docker容器运行。它们可从Elastic Docker注册表获得。
有关如何配置和运行Logstash Docker容器的详细信息，请参阅在Docker上运行Logstash。
1.2.简单使用
首先，让我们通过运行最基本的Logstash pipeline（管道）来测试您的Logstash安装。
Logstash pipeline 有两个必需的元素，input并且output，以及一个可选的元素，filter。输入插件使用来自源的数据，过滤器插件在您指定时修改数据，输出插件将数据写入目标。
要测试Logstash安装，请运行最基本的Logstash管道。例如：
cd logstash-7.3.1
bin/logstash -e 'input { stdin { } } output { stdout {} }'bin目录的位置因平台而异。请参阅目录布局 以查找bin\logstash系统上的位置。
该 -e 标志使您可以直接从命令行指定配置。通过在命令行指定配置，您可以快速测试配置，而无需在迭代之间编辑文件。示例中的管道从标准输入获取输入stdin，并stdout以结构化格式将该输入移动到标准输出 。
启动Logstash后，等到看到“Pipeline main started”，然后hello world在命令提示符下输入：
hello world
2013-11-21T01:22:14.405+0000 0.0.0.0 hello worldLogstash将时间戳和IP地址信息添加到消息中。通过在运行Logstash的shell中发出CTRL-D命令来退出Logstash 。
恭喜！您已经创建并运行了一个基本的Logstash管道。接下来，您将学习如何创建更逼真的管道。
1.3.用Logstash解析日志
在现实世界中，Logstash管道有点复杂：它通常具有一个或多个输入，过滤器和输出插件。
在本节中，您将创建一个Logstash管道，该管道使用Filebeat将Apache Web日志作为输入，解析这些日志以从日志创建特定的命名字段，并将解析的数据写入Elasticsearch集群。您将在配置文件中定义管道，而不是在命令行定义管道配置
要开始使用，请转到此处下载此示例中使用的示例数据集。解压缩文件。
1.3.1.配置Filebeat以将日志行发送到Logstash
在创建Logstash管道之前，您将配置Filebeat以将日志行发送到Logstash。
该Filebeat客户端是一个轻量级的，资源友好的工具，从服务器上的文件收集日志，这些日志转发到处理您的Logstash实例。
Filebeat旨在实现可靠性和低延迟。Filebeat在主机上占用很少的资源，Beats input插件可以最大限度地减少对Logstash实例的资源需求。
在典型的用例中，Filebeat在运行Logstash实例的计算机的单独计算机上运行。出于本教程的目的，Logstash和Filebeat在同一台机器上运行。
默认的Logstash安装包括Beats input插件。Beats输入插件使Logstash能够从Elastic Beats框架接收事件，这意味着任何编写为与Beats框架一起工作的Beat（如Packetbeat和Metricbeat）也可以将事件数据发送到Logstash。
要在数据源计算机上安装Filebeat，请从Filebeat 产品页面下载相应的软件包。您还可以参阅Beats文档中的Filebeat入门以 获取其他安装说明。
安装Filebeat后，您需要对其进行配置。打开filebeat.yml位于Filebeat安装目录中的文件，并使用以下行替换内容。确保paths指向logstash-tutorial.log，您之前下载的示例Apache日志文件 ：
filebeat.inputs:
- type: logpaths:- /path/to/file/logstash-tutorial.log 【Filebeat处理的文件的绝对路径】
output.logstash:hosts: ["localhost:5044"]为了简化配置，您不会像在现实场景中那样指定TLS / SSL设置。
在数据源计算机上，使用以下命令运行Filebeat：
sudo ./filebeat -e -c filebeat.yml -d "publish"如果以root身份运行Filebeat，则需要更改配置文件的所有权（请参阅 Beats Platform Reference 中的Config File Ownership and Permissions）。
Filebeat将尝试在端口5044上连接。在Logstash以活动的Beats插件启动之前，该端口上没有任何答案，因此您看到的有关该端口连接失败的任何消息现在都是正常的。
1.3.2.配置Logstash以Filebeat作为输入
接下来，您将创建一个Logstash配置管道，该管道使用Beats输入插件从Beats接收事件。
以下文本表示配置管道的框架：
# The # character at the beginning of a line indicates a comment. Use
# comments to describe your configuration.
input {
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
}此骨架不起作用，因为输入和输出部分没有定义任何有效选项。
首先，将框架配置管道复制并粘贴到Logstash目录中first-pipeline.conf文件中。
接下来，通过将以下行添加到文件的input部分，将Logstash实例配置为使用Beats输入插件first-pipeline.conf：
beats {port => "5044"
}您将配置Logstash以便稍后写入Elasticsearch。现在，您可以output将以下行添加到该部分，以便在运行Logstash时将输出打印到stdout：
stdout { codec => rubydebug }完成后，内容first-pipeline.conf应如下所示：
input {beats {port => "5044"}
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {stdout { codec => rubydebug }
}要验证配置，请运行以下命令：
bin/logstash -f first-pipeline.conf --config.test_and_exit该--config.test_and_exit选项会解析您的配置文件并报告任何错误。
如果配置文件通过配置测试，请使用以下命令启动Logstash：
bin/logstash -f first-pipeline.conf --config.reload.automatic该--config.reload.automatic选项启用自动配置重新加载，因此您不必在每次修改配置文件时停止并重新启动Logstash。
当Logstash启动时，您可能会看到一条或多条有关Logstash忽略该pipelines.yml文件的警告消息。您可以放心地忽略此警告。该pipelines.yml文件用于 在单个Logstash实例中运行多个管道。对于此处显示的示例，您正在运行单个管道。
如果您的管道工作正常，您应该看到一系列事件，如下面写入控制台：
{"@timestamp" => 2017-11-09T01:44:20.071Z,"offset" => 325,"@version" => "1","beat" => {"name" => "My-MacBook-Pro.local","hostname" => "My-MacBook-Pro.local","version" => "6.0.0"},"host" => "My-MacBook-Pro.local","prospector" => {"type" => "log"},"input" => {"type" => "log"},"source" => "/path/to/file/logstash-tutorial.log","message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"","tags" => [[0] "beats_input_codec_plain_applied"]
}
...1.3.3.使用Grok Filter插件解析Web日志
现在你有一个工作管道从Filebeat读取日志行。但是你会注意到日志消息的格式并不理想。您希望解析日志消息以从日志中创建特定的命名字段。为此，您将使用grok过滤器插件。
该grok过滤器插件是几个插件，默认情况下在Logstash可用之一。有关如何管理Logstash插件的详细信息，请参阅插件管理器的参考文档。
该grok过滤器插件，使您能够将非结构化日志数据分析到结构化的东西和可查询。
由于grok过滤器插件在传入的日志数据中查找模式，因此配置插件需要您决定如何识别用例感兴趣的模式。Web服务器日志示例中的代表行如下所示：
83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png
HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel
Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"
该行开头的IP地址很容易识别，括号中的时间戳也是如此。要解析数据，可以使用%{COMBINEDAPACHELOG}grok模式，该模式使用以下模式构建Apache日志中的行：
|
信息 |
字段名称 |
|
IP地址（IP Address） |
|
|
用户身份（User ID） |
|
|
用户认证（User Authentication） |
|
|
时间戳（timestamp） |
|
|
HTTP动词（HTTP Verb） |
|
|
请求正文（Request body） |
|
|
HTTP版本（HTTP Version） |
|
|
HTTP状态代码（HTTP Status Code） |
|
|
服务的字节数（Bytes served） |
|
|
推荐人网址（Referrer URL） |
|
|
用户代理（User agent） |
|
如果您需要帮助构建grok模式，请尝试使用 Grok Debugger。Grok Debugger是基本许可证下的X-Pack功能，因此可以免费使用。
编辑first-pipeline.conf文件并filter使用以下文本替换整个部分：
filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}"}}
}完成后，内容first-pipeline.conf应如下所示：
input {beats {port => "5044"}
}
filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}"}}
}
output {stdout { codec => rubydebug }
}保存更改。
由于您已启用自动配置重新加载，因此您无需重新启动Logstash即可获取更改。但是，您需要强制Filebeat从头开始读取日志文件。为此，请转到运行Filebeat的终端窗口，然后按Ctrl + C关闭Filebeat。然后删除Filebeat注册表文件。例如，运行：
sudo rm data/registry由于Filebeat存储它在注册表中收集的每个文件的状态，因此删除注册表文件会强制Filebeat读取它从头开始收集的所有文件。
接下来，使用以下命令重新启动Filebeat：
sudo ./filebeat -e -c filebeat.yml -d "publish"如果Filebeat需要等待Logstash重新加载配置文件，则在Filebeat开始处理事件之前可能会有一点延迟。
在Logstash应用grok模式之后，事件将具有以下JSON表示：
{"request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png","agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"","offset" => 325,"auth" => "-","ident" => "-","verb" => "GET","prospector" => {"type" => "log"},"input" => {"type" => "log"},"source" => "/path/to/file/logstash-tutorial.log","message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"","tags" => [[0] "beats_input_codec_plain_applied"],"referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"","@timestamp" => 2017-11-09T02:51:12.416Z,"response" => "200","bytes" => "203023","clientip" => "83.149.9.216","@version" => "1","beat" => {"name" => "My-MacBook-Pro.local","hostname" => "My-MacBook-Pro.local","version" => "6.0.0"},"host" => "My-MacBook-Pro.local","httpversion" => "1.1","timestamp" => "04/Jan/2015:05:13:42 +0000"
}请注意，事件包含原始消息，但日志消息也分解为特定字段。
1.3.4.使用Geoip Filter插件增强您的数据
除了解析日志数据以获得更好的搜索之外，过滤器插件还可以从现有数据中获取补充信息。例如，geoip插件查找IP地址，从地址中获取地理位置信息，并将该位置信息添加到日志中。
通过将以下行geoip添加到文件的filter部分来配置Logstash实例以使用过滤器插件first-pipeline.conf：
geoip {source => "clientip"}该geoip插件配置要求您指定包含IP地址来查找源字段的名称。在此示例中，该clientip字段包含IP地址。
由于过滤器是按顺序进行评估，确保该geoip部分是后grok配置文件的部分和，无论是grok和geoip部分嵌套在内部filter部分。
完成后，内容first-pipeline.conf应如下所示：
input {beats {port => "5044"}
}filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}"}}geoip {source => "clientip"}
}
output {stdout { codec => rubydebug }
}保存更改。要强制Filebeat从头开始读取日志文件，如前所述，请关闭Filebeat（按Ctrl + C），删除注册表文件，然后使用以下命令重新启动Filebeat：
sudo ./filebeat -e -c filebeat.yml -d "publish"请注意，该事件现在包含地理位置信息：
{"request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png","agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"","geoip" => {"timezone" => "Europe/Moscow","ip" => "83.149.9.216","latitude" => 55.7485,"continent_code" => "EU","city_name" => "Moscow","country_name" => "Russia","country_code2" => "RU","country_code3" => "RU","region_name" => "Moscow","location" => {"lon" => 37.6184,"lat" => 55.7485},"postal_code" => "101194","region_code" => "MOW","longitude" => 37.6184},...1.3.5.将数据索引到Elasticsearch
现在Web日志被分解为特定字段，您已准备好将数据导入Elasticsearch。
您可以在自己的硬件上运行Elasticsearch，也可以在Elastic Cloud上使用我们 托管的Elasticsearch Service。AWS和GCP均提供Elasticsearch服务。 免费试用Elasticsearch服务。
Logstash管道可以将数据索引到Elasticsearch集群中。编辑first-pipeline.conf文件并output使用以下文本替换整个部分
output {elasticsearch {hosts => [ "localhost:9200" ]}
}使用此配置，Logstash使用http协议连接到Elasticsearch。
上面的示例假定Logstash和Elasticsearch在同一个实例上运行。
您可以使用hosts配置指定类似的东西来指定远程Elasticsearch实例hosts => [ "es-machine:9092" ]。
此时，您的first-pipeline.conf文件已正确配置输入，过滤器和输出部分，如下所示：
input {beats {port => "5044"}
}filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}"}}geoip {source => "clientip"}
}
output {elasticsearch {hosts => [ "localhost:9200" ]}
}保存更改。要强制Filebeat从头开始读取日志文件，如前所述，请关闭Filebeat（按Ctrl + C），删除注册表文件，然后使用以下命令重新启动Filebeat：
sudo ./filebeat -e -c filebeat.yml -d "publish"1.3.6.测试您的管道
现在Logstash管道配置为将数据索引到Elasticsearch集群中，您可以查询Elasticsearch。
根据grok过滤器插件创建的字段尝试对Elasticsearch进行测试查询。将$ DATE替换为当前日期，格式为YYYY.MM.DD：
curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=response=200'索引名称中使用的日期基于UTC，而不是运行Logstash的时区。如果查询返回index_not_found_exception，请确保logstash-$DATE反映索引的实际名称。要查看可用索引列表，请使用以下查询：curl 'localhost:9200/_cat/indices?v'。
你应该得到多次点击。例如：
{"took": 50,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 98,"max_score": 2.793642,"hits": [{"_index": "logstash-2017.11.09","_type": "doc","_id": "3IzDnl8BW52sR0fx5wdV","_score": 2.793642,"_source": {"request": "/presentations/logstash-monitorama-2013/images/frontend-response-codes.png","agent": """"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""","geoip": {"timezone": "Europe/Moscow","ip": "83.149.9.216","latitude": 55.7485,"continent_code": "EU","city_name": "Moscow","country_name": "Russia","country_code2": "RU","country_code3": "RU","region_name": "Moscow","location": {"lon": 37.6184,"lat": 55.7485},"postal_code": "101194","region_code": "MOW","longitude": 37.6184},"offset": 2932,"auth": "-","ident": "-","verb": "GET","prospector": {"type": "log"},"input": {"type": "log"},"source": "/path/to/file/logstash-tutorial.log","message": """83.149.9.216 - - [04/Jan/2015:05:13:45 +0000] "GET /presentations/logstash-monitorama-2013/images/frontend-response-codes.png HTTP/1.1" 200 52878 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""","tags": ["beats_input_codec_plain_applied"],"referrer": """"http://semicomplete.com/presentations/logstash-monitorama-2013/"""","@timestamp": "2017-11-09T03:11:35.304Z","response": "200","bytes": "52878","clientip": "83.149.9.216","@version": "1","beat": {"name": "My-MacBook-Pro.local","hostname": "My-MacBook-Pro.local","version": "6.0.0"},"host": "My-MacBook-Pro.local","httpversion": "1.1","timestamp": "04/Jan/2015:05:13:45 +0000"}},...如果您使用Kibana来可视化您的数据，您还可以在Kibana中浏览Filebeat数据：
有关加载Filebeat的Kibana索引模式的信息，请参阅Filebeat入门文档。
您已成功创建了一个管道，该管道使用Filebeat将Apache Web日志作为输入，解析这些日志以从日志创建特定的命名字段，并将解析后的数据写入Elasticsearch集群。
1.4.将多个输入和输出插件拼接在一起
您需要管理的信息通常来自多个不同的来源，而用例可能需要多个目的地来处理您的数据。您的Logstash管道可以使用多个输入和输出插件来处理这些要求。
在本节中，您将创建一个Logstash管道，该管道从Twitter提要和Filebeat客户端获取输入，然后将信息发送到Elasticsearch集群以及将信息直接写入文件。
1.4.1.读取Twitter Feed
要添加Twitter Feed，请使用twitter输入插件。要配置插件，您需要以下几条信息：
- 个消费者的关键，它唯一标识您的Twitter的应用程序。
- 一个消费者的秘密，它作为密码输入你的Twitter的应用程序。
- 要在传入Feed中搜索的一个或多个关键字。该示例显示使用“云”作为关键字，但您可以使用您想要的任何内容。
- 一个oauth令牌，使用此应用程序识别Twitter帐户。
- 一个OAuth令牌秘密，作为Twitter的帐户的密码。
访问https://dev.twitter.com/apps以设置Twitter帐户并生成您的消费者密钥和密钥，以及您的访问令牌和秘密。twitter如果您不确定如何生成这些键，请参阅输入插件的文档。
就像您之前使用Logstash解析日志时所做的那样，创建一个second-pipeline.conf包含配置管道骨架的配置文件（称为）。如果需要，可以重用之前创建的文件，但请确保在运行Logstash时传入正确的配置文件名。
将以下行添加到文件的input部分，将second-pipeline.conf值替换为此处显示的占位符值：
twitter {consumer_key => "enter_your_consumer_key_here"consumer_secret => "enter_your_secret_here"keywords => ["cloud"]oauth_token => "enter_your_access_token_here"oauth_token_secret => "enter_your_access_token_secret_here"}1.4.2.配置Filebeat以将日志行发送到Logstash
正如您之前在配置Filebeat以将日志行发送到Logstash中所了解的那样，Filebeat 客户端是一个轻量级，资源友好的工具，它从服务器上的文件中收集日志并将这些日志转发到Logstash实例进行处理。
安装Filebeat后，您需要对其进行配置。打开filebeat.yml位于Filebeat安装目录中的文件，并使用以下行替换内容。确保paths指向您的系统日志：
filebeat.inputs:
- type: logpaths:- /var/log/*.log Filebeat处理的文件的绝对路径。fields:type: syslog 添加一个type使用值调用的字段syslog到事件
output.logstash:hosts: ["localhost:5044"]保存更改。
为了简化配置，您不会像在现实场景中那样指定TLS / SSL设置。
通过将以下行添加到文件second-pipeline.conf的input部分，将Logstash实例配置为使用Filebeat输入插件：
beats {port => "5044"}1.4.3.将Logstash数据写入文件
您可以配置Logstash管道以使用file输出插件将数据直接写入文件 。
通过将以下行file添加到文件second-pipeline.conf的output部分来配置Logstash实例以使用输出插件：
file {path => "/path/to/target/file"}1.4.4.写入多个Elasticsearch Nodes
写入多个Elasticsearch节点可以减轻给定Elasticsearch节点上的资源需求，并在特定节点不可用时为集群提供冗余的入口点。
要将Logstash实例配置为写入多个Elasticsearch节点，请编辑要读取output的second-pipeline.conf文件部分：
output {elasticsearch {hosts => ["IP Address 1:port1", "IP Address 2:port2", "IP Address 3"]}
}使用主机行中Elasticsearch集群中三个非主节点的IP地址。当hosts 参数列出多个IP地址时，Logstash会在地址列表中对请求进行负载平衡。另请注意，Elasticsearch的默认端口是，9200并且可以在上面的配置中省略。
1.4.5.测试管道
此时，您的second-pipeline.conf文件如下所示：
input {twitter {consumer_key => "enter_your_consumer_key_here"consumer_secret => "enter_your_secret_here"keywords => ["cloud"]oauth_token => "enter_your_access_token_here"oauth_token_secret => "enter_your_access_token_secret_here"}beats {port => "5044"}
}
output {elasticsearch {hosts => ["IP Address 1:port1", "IP Address 2:port2", "IP Address 3"]}file {path => "/path/to/target/file"}
}Logstash正在使用您配置的Twitter源中的数据，从Filebeat接收数据，并将此信息索引到Elasticsearch集群中的三个节点以及写入文件。
在数据源计算机上，使用以下命令运行Filebeat：
sudo ./filebeat -e -c filebeat.yml -d "publish"Filebeat将尝试在端口5044上连接。在Logstash以活动的Beats插件启动之前，该端口上没有任何答案，因此您看到的有关该端口连接失败的任何消息现在都是正常的。
要验证配置，请运行以下命令：
bin/logstash -f second-pipeline.conf --config.test_and_exit该--config.test_and_exit选项会解析您的配置文件并报告任何错误。当配置文件通过配置测试时，使用以下命令启动Logstash：
bin/logstash -f second-pipeline.conf使用该grep实用程序在目标文件中搜索以验证信息是否存在：
grep syslog /path/to/target/file运行Elasticsearch查询以在Elasticsearch集群中查找相同的信息：
curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=fields.type:syslog'将$ DATE替换为当前日期，格式为YYYY.MM.DD。
要查看Twitter Feed中的数据，请尝试以下查询：
curl -XGET 'http://localhost:9200/logstash-$DATE/_search?pretty&q=client:iphone'再次，请记住用YYYY.MM.DD格式替换当前日期的$ DATE。
Logstash【从无到有从有到无】【L2】Logstash入门相关推荐
- Logstash【从无到有从有到无】【简介】【L2】Logstash入门
目录 1.Logstash入门 1.1.安装Logstash 1.1.1.从下载的二进制安装 1.1.2.从包存储库安装 1.1.3.使用Homebrew在Mac上安装Logstash 1.1.4.使 ...
- logstash 日志_面试和书评：LogStash书，使日志管理变得容易
logstash 日志 James Turnbull通过解释LogStash在日志项目上下文中的实现细节,为使用LogStash集中日志提供了令人信服的案例. < Logstash手册>通 ...
- logstash读取Elasticsearch数据保存为json,logstash接收log数据写入kafka生产者
[提前声明] 文章由作者:张耀峰 结合自己生产中的使用经验整理,最终形成简单易懂的文章 写作不易,转载请注明,谢谢! 代码案例地址: ?https://github.com/Mydreamandrea ...
- 无废话ExtJs 入门教程六[按钮：Button]
无废话ExtJs 入门教程六[按钮:Button] extjs技术交流,欢迎加群(201926085) 继上一节内容,我们在表单里加了个两个按钮"提交"与重置.如下所示代码区的第6 ...
- 无废话ExtJs 入门教程十七[列表：GridPanel]
无废话ExtJs 入门教程十七[列表:GridPanel] extjs技术交流,欢迎加群(521711109) 在Extjs中,GridPanel用于数据显示,即我们平时说的列表页.在本节中,我们先对 ...
- Logstash【从无到有从有到无】【L20】编解码器插件（Codec plugins）
1.编解码器插件 编解码器插件可更改事件的数据表示形式. 编解码器本质上是流过滤器,可以作为输入或输出的一部分进行操作. 下面提供了以下编解码器插件. 有关Elastic支持的插件的列表,请查阅Sup ...
- Logstash【从无到有从有到无】【L24】贡献了Java插件
目录 1.贡献了Java插件 1.1.流程概述 1.2.如何编写Java输入插件 1.2.1.设置环境 1.2.2.代码插件 1.2.3.打包和部署 1.2.4.与Java输入插件运行Logstash ...
- Apache Spark【从无到有从有到无】【编程指南】【AS5】结构化流编程指南
目录 1.概观 2.快速示例 3.编程模型 3.1.基本概念 3.2.处理事件时间和延迟数据 3.3.容错语义 4.使用数据集和数据框架的API 4.1.创建streaming DataFrames ...
- Apache HBase【从无到有从有到无】【AH8】RegionServer调整大小的经验法则
目录 RegionServer调整大小的经验法则 1.关于列族的数量 1.1.ColumnFamilies的基数 2.Rowkey设计 2.1.热点发现(Hotspotting) 2.2.单调增加行键 ...
最新文章
- 第十六届无线节能信标灯规格说明
- cannot restore segment prot after reloc
- KVM虚拟化笔记（十三）------kvm虚拟机静态迁移
- Could not load file or assembly 'System.Web.Extensions, Version=3.5.0.0
- centos7服务器搭建elasticsearch7.6.0集群
- jQuery按键事件keyup
- rp软件app流程图_Axure RP 9 for Mac交互原型设计软件
- Android 6种快速开发框架
- java ruby脚本_Java 程序中直接执行 Ruby 脚本 （JRuby）
- PCL之K维树--KD-tree
- 学python用什么软件-初学 Python 需要安装哪些软件？
- springboot使用j2cache
- [项目管理]-第四章：过程定义和过程裁剪
- 互联网金融指导意见或引发P2P跑路狂潮
- 华为云服务登录显示服务器异常,云服务器异常登录应该怎么处理
- 怎么能跳过苹果服务器降级系统,苹果ios11手机怎样将系统降级？简单三步即可完成降级！...
- 【C++代码】约瑟夫环问题：0,1,……,n-1这n个数字排成一个圆圈，从数字0开始，每次从这个圆圈里删除第m个数字。求出这个圆圈里剩下的最后一个数字。
- 如何用多种代码实现51单片机流水灯
- 细讲逻辑斯蒂回归与朴素贝叶斯、最大熵原理的爱恨交织（五）
- pytorch损失函数中‘reduction‘参数