Невозможно удалить событие, когда фильтр grok не соответствует, logstash, эластичный поиск

Я пытаюсь проанализировать журналы tomcat и передать вывод в эластичный поиск. Более-менее хорошо работает. Когда я вижу индексированные данные эластичного поиска, они содержат множество совпадающих данных с полем тегов как _grokparsefailure. Это вызывает множество повторяющихся совпадающих данных. Чтобы избежать этого, я попытался отбросить событие, если теги содержат _grokparsefailure. Эта конфигурация записана в файле logstash.conf под фильтром Grok. По-прежнему вывод в эластичный поиск содержит проиндексированный документ, содержащий теги с _grokparsefailure. Если Grok не работает, я не хочу, чтобы это совпадение переходило в эластичный поиск, поскольку оно вызывает дублирование данных в эластичном поиске.

logstash.conf это:

input {

  file {

    path => "/opt/elasticSearch/logstash-1.4.2/input.log"
        codec => multiline {
                pattern => "^\["
                negate => true
                what => previous
        }
        start_position => "end"

  }

}

filter {

        grok {

    match => [
"message", "^\[%{GREEDYDATA}\] %{GREEDYDATA} Searching hotels for country %{GREEDYDATA:country}, city %{GREEDYDATA:city}, checkin %{GREEDYDATA:checkin}, checkout %{GREEDYDATA:checkout}, roomstay %{GREEDYDATA:roomstay}, No. of hotels returned is %{NUMBER:hotelcount} ."
    ]

  }

 if "_grokparsefailure"  in [tags]{     

        drop { }

    }

}

output {

file {
   path => "/opt/elasticSearch/logstash-1.4.2/output.log"
 }

 elasticsearch {
                cluster => "elasticsearchdev"
  }

}

эластичный поисковый ответ http://172.16.37.97:9200/logstash-2015.12.23/_search?pretty=true

Приведенный ниже вывод содержит три документа, где первый содержит _grokparsefailure в поле _source -> tags.

Я не хочу этого в этом выпуске. Так что, вероятно, нужно ограничить его из logstash, чтобы он не доходил до эластичного поиска.

{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "failed" : 0
  },
  "hits" : {
    "total" : 3,
    "max_score" : 1.0,
    "hits" : [

 {

      "_index" : "logstash-2015.12.23",
      "_type" : "logs",
      "_id" : "J6CoEhKaSE68llz5nEbQSQ",
      "_score" : 1.0,
      "_source":{"message":"[2015-12-23 12:08:40,124] ERROR http-80-5_@{AF3AF784EC08D112D5D6FC92C78B5161,127.0.0.1,1450852688060} com.mmt.hotels.web.controllers.search.HotelsSearchController - Searching hotels for country IN, city DEL, checkin 28-03-2016, checkout 29-03-2016, roomstay 1e0e, No. of hotels returned is 6677 .","@version":"1","@timestamp":"2015-12-23T14:17:03.436Z","host":"ggn-37-97","path":"/opt/elasticSearch/logstash-1.4.2/input.log","tags":["_grokparsefailure"]}

    },

 {

      "_index" : "logstash-2015.12.23",
      "_type" : "logs",
      "_id" : "2XMc6nmnQJ-Bi8vxigyG8Q",
      "_score" : 1.0,
      "_source":{"@timestamp":"2015-12-23T14:17:02.894Z","message":"[2015-12-23 12:08:40,124] ERROR http-80-5_@{AF3AF784EC08D112D5D6FC92C78B5161,127.0.0.1,1450852688060} com.mmt.hotels.web.controllers.search.HotelsSearchController - Searching hotels for country IN, city DEL, checkin 28-03-2016, checkout 29-03-2016, roomstay 1e0e, No. of hotels returned is 6677 .","@version":"1","host":"ggn-37-97","path":"/opt/elasticSearch/logstash-1.4.2/input.log","country":"IN","city":"DEL","checkin":"28-03-2016","checkout":"29-03-2016","roomstay":"1e0e","hotelcount":"6677"}

},

 {

      "_index" : "logstash-2015.12.23",
      "_type" : "logs",
      "_id" : "fKLqw1LJR1q9YDG2yudRDw",
      "_score" : 1.0,
      "_source":{"@timestamp":"2015-12-23T14:16:12.684Z","message":"[2015-12-23 12:08:40,124] ERROR http-80-5_@{AF3AF784EC08D112D5D6FC92C78B5161,127.0.0.1,1450852688060} com.mmt.hotels.web.controllers.search.HotelsSearchController - Searching hotels for country IN, city DEL, checkin 28-03-2016, checkout 29-03-2016, roomstay 1e0e, No. of hotels returned is 6677 .","@version":"1","host":"ggn-37-97","path":"/opt/elasticSearch/logstash-1.4.2/input.log","country":"IN","city":"DEL","checkin":"28-03-2016","checkout":"29-03-2016","roomstay":"1e0e","hotelcount":"6677"}

    } ]
  }
}

]


logstash logstash-grok
person Dev Gosain    schedule 23.12.2015    source источник

Ответы (2)


arrow_upward
7
arrow_downward

Вы можете попробовать проверить наличие _grokparsefailure в разделе output, например:

output {
  if "_grokparsefailure" not in [tags] {
    file {
      path => "/opt/elasticSearch/logstash-1.4.2/output.log"
    }

    elasticsearch {
      cluster => "elasticsearchdev"
    }
  }
}
person Val    schedule 23.12.2015
comment
Пробовал, но не работает. По-прежнему в тегах вывода есть _grokparsefailure. - person Dev Gosain; 24.12.2015
comment
Вы стерли индекс перед попыткой? т.е. вы уверены, что смотрите новые документы, а не старые? - person Val; 24.12.2015
comment
Да. Использовал эту команду для удаления индексов. curl -XDELETE 'localhost: 9200 / logstash-2015.12.24' - person Dev Gosain; 24.12.2015
comment
Хорошо, и можете ли вы убедиться, что не просматриваете документы из прошлых индексов, то есть logstash-2015-12-23, logstash-2015-12-22 и т. Д. - person Val; 24.12.2015
comment
Документы только на текущий день. Я добавил в свой input.log какой-то фиктивный текст, чтобы убедиться, что он собирает последние изменения. Этот фиктивный текст все еще выводится. - person Dev Gosain; 24.12.2015
comment
У меня все еще та же проблема? Кто-нибудь нашел решение для этого? - person Mika Andrianarijaona; 28.06.2016

arrow_upward
1
arrow_downward

Иногда у вас может быть несколько фильтров Grok, и можно ожидать, что некоторые из них не сработают для некоторых событий, но пройдут для отдыха, удаление событий на основе _grokparsefailure не решит задачу.

пример:

input
{
some input
}

filter
{
grok1 {extract ip to my_ip1}

grok2 {extract ip to my_ip2}

grok3 {extract ip to my_ip3}


}

output
{
  if "_grokparsefailure" not in [tags] { <-- This will not write to output if any single grok fails.
  some output
}
}

У меня есть решение отфильтровать на основе какой-то переменной. Любой другой лучший выход здесь ??? пример:

if "10." in ["ip1"] or "10." in ["ip2"] or "10." in ["ip3"]
{
 drop{}
}
person NinjaSolid    schedule 07.08.2017