Logstash 参考指南(拼接多个输入和输出插件)

qcqcyi 2019-06-28

将多个输入和输出插件拼接在一起

你需要管理的信息通常来自几个不同来源,并且用例可以为你的数据要求多个目的地,你的Logstash管道可以使用多个输入和输出插件来处理这些需求。

在本节中,你将创建一个Logstash管道,该管道接收来自Twitter提要和Filebeat客户端的输入,然后将信息发送到一个Elasticsearch集群,并将信息直接写入文件。

读取来自Twitter的提要

要添加Twitter提要,请使用twitter输入插件,要配置插件,需要以下信息:

  • 一个消费者key,它唯一地标识你的Twitter应用程序。
  • 一个消费者secret,作为你的Twitter应用程序的密码。
  • 在传入的提要中搜索一个或多个关键字,这个例子显示了使用“cloud”作为关键字,但是你可以使用任何你想要的。
  • 一个oauth token,它标识使用这个应用程序的Twitter帐户。
  • 一个oauth token secret,作为Twitter账户的密码。

访问https://dev.twitter.com/apps来建立一个Twitter账户,生成你的消费key和secret,以及你的访问token和secret,如果你不确定如何生成这些key,请参阅twitter输入插件的文档。

就像前面使用Logstash解析日志时所做的那样,创建一个包含配置管道结构的配置文件(称为second-pipeline.conf),如果需要,可以重用前面创建的文件,但请确保在运行Logstash时传递正确的配置文件名。

将以下几行添加到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"
}

配置Filebeat将日志行发送到Logstash

正如你在配置Filebeat以将日志行发送到Logstash时所了解的那样,Filebeat客户端是一个轻量级的、资源友好的工具,它可以从服务器上的文件中收集日志,并将这些日志转发到你的Logstash实例进行处理。

安装Filebeat后,需要对其进行配置,打开位于Filebeat安装目录中的filebeat.yml文件,并使用以下行替换内容,确保paths指向你的系统日志:

filebeat.prospectors:
- type: log
  paths:
    - /var/log/*.log 
  fields:
    type: syslog 
output.logstash:
  hosts: ["localhost:5044"]
  • paths => Filebeat处理的文件的绝对路径。
  • fields => 将一个名为type值为syslog的字段添加到该事件中。

保存你的更改。

为了简化配置,你不会像在现实场景中那样指定TLS/SSL设置。

通过在second-pipeline.conf文件的input部分添加以下代码行,将你的Logstash实例配置为使用Filebeat输入插件:

beats {
   port => "5044"
}

将Logstash数据写入文件

你可以配置你的Logstash管道,以便使用file输出插件将数据直接写入文件。

通过在second-pipeline.conf文件的output部分添加以下行,配置你的Logstash实例以使用文件输出插件:

file {
   path => "/path/to/target/file"
}

写入到多个Elasticsearch节点

对多个Elasticsearch节点的写入可以减轻给定Elasticsearch节点的资源需求,并在特定节点不可用时为集群提供多余的入口点。

若要将你的Logstash实例配置为写入多个Elasticsearch节点,请编辑second-pipeline.conf文件的output部分以供读取:

output {
    elasticsearch {
        hosts => ["IP Address 1:port1", "IP Address 2:port2", "IP Address 3"]
    }
}

在你的Elasticsearch集群的hosts行中使用三个非主节点的IP地址,当hosts参数列出多个IP地址时,Logstash使用地址列表的负载均衡请求,还请注意,Elasticsearch的默认端口是9200,可以在上面的配置中省略。

测试管道

此时,你的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'

使用YYYY.MM.DD格式将$DATE替换为当前日期。

要查看来自Twitter提要的数据,请尝试以下查询:

curl -XGET 'http://localhost:9200/logstash-$DATE/_search?pretty&q=client:iphone'

同样,请记住使用YYYY.MM.DD格式将$DATE替换为当前日期。

相关推荐