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

你需要管理的信息通常来自几个不同来源，并且用例可以为你的数据要求多个目的地，你的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替换为当前日期。