Logstash將不同的字段輸出到不同的彈性搜索索引

Question

我有一個Filebeat實例將Apache訪問日誌發送到Logstash。 Logstash管道變換文件並加載處理過的字段說（FIELD1，FIELD2 &字段3）到elastic search到索引索引A。流程很簡單，就是&。這裏是我的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" 
    } 
} 

現在我想要做的就是添加其他三個字段字段4和字段5，並將它們添加到名爲indexB一個單獨的索引。所以在最後指數A持有字段1場2和場3而IndexB持有字段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" 
    } 
} 

可有人請指出我要去的地方錯誤或解決任何替代。

Answer 1

您需要使用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" 
     } 
    } 
}