Индексирование JSON массивов Logstash

Логсташ потрясающий. Я могу отправить JSON таким образом (многослойный для удобочитаемости):

{
  "a": "one"
  "b": {
    "alpha":"awesome"
  }
}

И затем запрос для этой строки в кибане, используя поисковый запрос b.alpha:awesome. Ницца.

Однако теперь у меня есть строка журнала JSON:

{
  "different":[
    {
      "this": "one",
      "that": "uno"
    },
    {
      "this": "two"
    }
  ]
}

И я хотел бы найти эту строку с поиском, например different.this:two (или different.this:one или different.that:uno)

Если бы я напрямую использовал Lucene, я бы перебирал массив different и создавал новый индекс поиска для каждого хэша внутри него, но Logstash в настоящее время, похоже, глотает эту строку следующим образом:

different: {this: one, that: uno}, {this: two}

Что не поможет мне искать строки журнала, используя different.this или different.that.

Есть ли какие-либо мысли относительно кодека, фильтра или изменения кода, которые я могу сделать, чтобы включить это?

Ответ 1

Вы можете написать свой собственный filter (скопируйте и вставьте, переименуйте имя класса, config_name и перепишите filter(event) метод) или изменить текущий JSON фильтр (источник на Github)

Вы можете найти исходный код фильтра JSON (Ruby class) в следующем пути logstash-1.x.x\lib\logstash\filters, названном как json.rb. Фильтр JSON анализирует содержимое как JSON следующим образом

begin
  # TODO(sissel): Note, this will not successfully handle json lists
  # like your text is '[ 1,2,3 ]' JSON.parse gives you an array (correctly)
  # which won't merge into a hash. If someone needs this, we can fix it
  # later.
  dest.merge!(JSON.parse(source))

  # If no target, we target the root of the event object. This can allow
  # you to overwrite @timestamp. If so, let parse it as a timestamp!
  if [email protected] && event[TIMESTAMP].is_a?(String)
    # This is a hack to help folks who are mucking with @timestamp during
    # their json filter. You aren't supposed to do anything with
    # "@timestamp" outside of the date filter, but nobody listens... ;)
    event[TIMESTAMP] = Time.parse(event[TIMESTAMP]).utc
  end

  filter_matched(event)
rescue => e
  event.tag("_jsonparsefailure")
  @logger.warn("Trouble parsing json", :source => @source,
               :raw => event[@source], :exception => e)
  return
end

Вы можете изменить процедуру синтаксического анализа, чтобы изменить исходный JSON

  json  = JSON.parse(source)
  if json.is_a?(Hash)
    json.each do |key, value| 
        if value.is_a?(Array)
            value.each_with_index do |object, index|
                #modify as you need
                object["index"]=index
            end
        end
    end
  end
  #save modified json
  ......
  dest.merge!(json)

то вы можете изменить свой файл конфигурации, чтобы использовать/ваш новый/модифицированный фильтр JSON и поместить его в \logstash-1.x.x\lib\logstash\config

Это мой elastic_with_json.conf с модифицированным фильтром json.rb

input{
    stdin{

    }
}filter{
    json{
        source => "message"
    }
}output{
    elasticsearch{
        host=>localhost
    }stdout{

    }
}

если вы хотите использовать свой новый фильтр, вы можете настроить его с помощью config_name

class LogStash::Filters::Json_index < LogStash::Filters::Base

  config_name "json_index"
  milestone 2
  ....
end

и настройте его

input{
    stdin{

    }
}filter{
    json_index{
        source => "message"
    }
}output{
    elasticsearch{
        host=>localhost
    }stdout{

    }
}

Надеюсь, что это поможет.

Ответ 2

Для быстрого и грязного взлома я использовал фильтр Ruby и код ниже, не нужно больше использовать фильтр "json"

input {
  stdin{}
}

filter {
  grok {
    match => ["message","(?<json_raw>.*)"]
  }
  ruby {
    init => "
      def parse_json obj, pname=nil, event
         obj = JSON.parse(obj) unless obj.is_a? Hash
         obj = obj.to_hash unless obj.is_a? Hash

         obj.each {|k,v|
         p = pname.nil?? k : pname
         if v.is_a? Array
           v.each_with_index {|oo,ii|               
             parse_json_array(oo,ii,p,event)
           }
           elsif v.is_a? Hash
             parse_json(v,p,event)
           else
             p = pname.nil?? k : [pname,k].join('.')
             event[p] = v
           end
         }
        end

        def parse_json_array obj, i,pname, event
          obj = JSON.parse(obj) unless obj.is_a? Hash
          pname_ = pname
          if obj.is_a? Hash
            obj.each {|k,v|
              p=[pname_,i,k].join('.')
              if v.is_a? Array
                v.each_with_index {|oo,ii|
                  parse_json_array(oo,ii,p,event)
                }
              elsif v.is_a? Hash
                parse_json(v,p, event)
              else
                event[p] = v
              end
            }
          else
            n = [pname_, i].join('.')
            event[n] = obj
          end
        end
      "
      code => "parse_json(event['json_raw'].to_s,nil,event) if event['json_raw'].to_s.include? ':'"
    }


  }

output {
  stdout{codec => rubydebug}
}

Тестирование структуры json

{"id":123, "members":[{"i":1, "arr":[{"ii":11},{"ii":22}]},{"i":2}], "im_json":{"id":234, "members":[{"i":3},{"i":4}]}}

и это то, что выводит

      {
           "message" => "{\"id\":123, \"members\":[{\"i\":1, \"arr\":[{\"ii\":11},{\"ii\":22}]},{\"i\":2}], \"im_json\":{\"id\":234, \"members\":[{\"i\":3},{\"i\":4}]}}",
          "@version" => "1",
        "@timestamp" => "2014-07-25T00:06:00.814Z",
              "host" => "Leis-MacBook-Pro.local",
          "json_raw" => "{\"id\":123, \"members\":[{\"i\":1, \"arr\":[{\"ii\":11},{\"ii\":22}]},{\"i\":2}], \"im_json\":{\"id\":234, \"members\":[{\"i\":3},{\"i\":4}]}}",
                "id" => 123,
       "members.0.i" => 1,
"members.0.arr.0.ii" => 11,
"members.0.arr.1.ii" => 22,
       "members.1.i" => 2,
           "im_json" => 234,
       "im_json.0.i" => 3,
       "im_json.1.i" => 4
      }

Ответ 3

Решением, которое мне понравилось, является рубиновый фильтр, потому что это требует от нас не писать другой фильтр. Однако это решение создает поля, которые находятся в "корне" JSON, и трудно отслеживать, как выглядел оригинальный документ.

Я придумал что-то похожее, которое легче следовать и является рекурсивным решением, поэтому оно чище.

ruby {
    init => "
        def arrays_to_hash(h)
          h.each do |k,v|
            # If v is nil, an array is being iterated and the value is k.
            # If v is not nil, a hash is being iterated and the value is v.
            value = v || k
            if value.is_a?(Array)
                # "value" is replaced with "value_hash" later.
                value_hash = {}
                value.each_with_index do |v, i|
                    value_hash[i.to_s] = v
                end
                h[k] = value_hash
            end

            if value.is_a?(Hash) || value.is_a?(Array)
              arrays_to_hash(value)
            end
          end
        end
      "
      code => "arrays_to_hash(event.to_hash)"
}

Он преобразует массивы в каждый номер с номером индекса. Подробнее: - http://blog.abhijeetr.com/2016/11/logstashelasticsearch-best-way-to.html