我有一个Filebeat
实例,它将Apache
访问日志发送到Logstash
。Logstash
管道将文件转换,然后将已处理的字段(例如 field1,field2和field3 )加载到elastic search
中,以创建索引 indexA 。流程简单且有效。这是我的pipeline.conf
input{
beats{
port => "5043"
}
}
filter
{
grok
{
patterns_dir => ["/usr/share/logstash/patterns"]
match =>{ "message" => ["%{IPORHOST:[client_ip]} - %{DATA:[user_name]} \[%{HTTPDATE:[access_time]}\] \"%{WORD:[method]} %{DATA:[url]} HTTP/%{NUMBER:[http_version]}\" %{NUMBER:[response_code]} %{NUMBER:[bytes]}( \"%{DATA:[referrer]}\")?( \"%{DATA:[user_agent]}\")?",
"%{IPORHOST:[remote_ip]} - %{DATA:[user_name]} \\[%{HTTPDATE:[time]}\\] \"-\" %{NUMBER:[response_code]} -" ]
}
remove_field => "@version"
remove_field => "beat"
remove_field => "input_type"
remove_field => "source"
remove_field => "type"
remove_field => "tags"
remove_field => "http_version"
remove_field => "@timestamp"
remove_field => "message"
}
mutate
{
add_field => { "field1" => "%{access_time}" }
add_field => { "field2" => "%{host}" }
add_field => { "field3" => "%{read_timestamp}" }
}
}
output {
elasticsearch{
hosts => ["localhost:9200"]
index => "indexA"
}
}
现在我要做的是添加其他三个字段 field4 和 field5 并将它们添加到名为 indexB 的单独索引中。因此,在结尾处索引A 持有字段1字段2 和字段3 ,而索引B 持有字段4 和字段5
到目前为止,这是修改后的pipeline.conf,它似乎不起作用。
input{
beats{
port => "5043"
}
}
filter
{
grok
{
patterns_dir => ["/usr/share/logstash/patterns"]
match =>{ "message" => ["%{IPORHOST:[client_ip]} - %{DATA:[user_name]} \[%{HTTPDATE:[access_time]}\] \"%{WORD:[method]} %{DATA:[url]} HTTP/%{NUMBER:[http_version]}\" %{NUMBER:[response_code]} %{NUMBER:[bytes]}( \"%{DATA:[referrer]}\")?( \"%{DATA:[user_agent]}\")?",
"%{IPORHOST:[remote_ip]} - %{DATA:[user_name]} \\[%{HTTPDATE:[time]}\\] \"-\" %{NUMBER:[response_code]} -" ]
}
remove_field => "@version"
remove_field => "beat"
remove_field => "input_type"
remove_field => "type"
remove_field => "http_version"
remove_field => "@timestamp"
remove_field => "message"
}
mutate
{
add_field => { "field1" => "%{access_time}" }
add_field => { "field2" => "%{host}" }
add_field => { "field3" => "%{read_timestamp}" }
}
}
output {
elasticsearch{
hosts => ["localhost:9200"]
index => "indexA"
}
}
filter
{
mutate
{
add_field => { "field4" => "%{source}" }
add_field => { "field5" => "%{tags}" }
remove_field => "field1"
remove_field => "field2"
remove_field => "field3"
}
}
output {
elasticsearch{
hosts => ["localhost:9200"]
index => "indexB"
}
}
有人可以指出我要去哪里或解决方案的任何替代方法。
最佳答案
您需要使用 clone
filter复制事件。然后,可以将所需的字段添加到每个相应的事件中,并将它们沉入两个不同的ES索引中:
input{
beats{
port => "5043"
}
}
filter
{
grok
{
patterns_dir => ["/usr/share/logstash/patterns"]
match =>{ "message" => ["%{IPORHOST:[client_ip]} - %{DATA:[user_name]} \[%{HTTPDATE:[access_time]}\] \"%{WORD:[method]} %{DATA:[url]} HTTP/%{NUMBER:[http_version]}\" %{NUMBER:[response_code]} %{NUMBER:[bytes]}( \"%{DATA:[referrer]}\")?( \"%{DATA:[user_agent]}\")?",
"%{IPORHOST:[remote_ip]} - %{DATA:[user_name]} \\[%{HTTPDATE:[time]}\\] \"-\" %{NUMBER:[response_code]} -" ]
}
remove_field => "@version"
remove_field => "beat"
remove_field => "input_type"
remove_field => "type"
remove_field => "http_version"
remove_field => "@timestamp"
remove_field => "message"
}
clone {
clones => ["log1", "log2"]
}
if [type] == "log1" {
mutate
{
add_field => { "field1" => "%{access_time}" }
add_field => { "field2" => "%{host}" }
add_field => { "field3" => "%{read_timestamp}" }
}
} else {
mutate
{
add_field => { "field4" => "%{source}" }
add_field => { "field5" => "%{tags}" }
}
}
}
output {
if [type] == "log1" {
elasticsearch{
hosts => ["localhost:9200"]
index => "indexA"
}
} else {
elasticsearch{
hosts => ["localhost:9200"]
index => "indexB"
}
}
}