Logstash: переименование вложенных полей на основе некоторых условий

Question

Я пытаюсь переименовать вложенные поля из Elasticsearch при миграции на Amazonelasticsearch

В документе я хочу изменить

1.Если поле value имеет тип JSON. Измените значение поля значение на ключевое слово-значение и удалите «пробел-значение» и «стандарт-значение», если они имеются

2.Если размер поля имеет размербольше 15. Измените значение поля значение на значение-значение

 "_source": {
          "applicationid" : "appid",
          "interactionId": "716bf006-7280-44ea-a52f-c79da36af1c5",
          "interactionInfo": [
            {
              "value": """{"edited":false}""",
              "value-standard": """{"edited":false}""",
              "value-whitespace" :  """{"edited":false}"""
              "title": "msgMeta"
            },
            {
              "title": "msg",
              "value": "hello testing",
            },
            {
              "title": "testing",
              "value": "I have a text that can be done and changed only the size exist more than 20 so we applied value-standard ",
            }
          ],
          "uniqueIdentifier": "a21ed89c-b634-4c7f-ca2c-8be6f31ae7b3",
        }
      }

конечный результат должен быть

 "_source": {
          "applicationid" : "appid",
          "interactionId": "716bf006-7280-44ea-a52f-c79da36af1c5",
          "interactionInfo": [
            {
              "value-keyword": """{"edited":false}""",
              "title": "msgMeta"
            },
            {
              "title": "msg",
              "value": "hello testing",
            },
            {
              "title": "testing",
              "value-standard": "I have a text that can be done and changed only the size exist more than 20 and so we applied value-standard  ",
            }
          ],
          "uniqueIdentifier": "a21ed89c-b634-4c7f-ca2c-8be6f31ae7b3",
        }
      }

Answer 1

Нашел решение для этого. Я использовал фильтр ruby ​​в Logstash для проверки каждого документа, а также вложенного документа. Вот код рубина

require 'json'

def register(param)
end

def filter(event)
  infoarray = event.get("interactionInfo")
  infoarray.each {  |x|
      if x.include?"value"
         value = x["value"]
         if value.length > 15
           apply_only_keyword(x)
         end
       end
      if x.include?"value"
        value = x["value"]
         if validate_json(value)
           apply_only_keyword(x)
         end
       end
  }
event.set("interactionInfo",infoarray)
return [event]
end


def validate_json(value)
  if value.nil?
    return false
  end
  JSON.parse(value)
  return true
rescue JSON::ParserError => e
  return false
end

def apply_only_keyword(x)
  x["value-keyword"] = x["value"]
  x.delete("value")
  if x.include?"value-standard"
    x.delete("value-standard")
  end
  if x.include?"value-whitespace"
    x.delete("value-whitespace")
  end
end

Answer 2

Для 2) вы можете сделать это следующим образом:

filter {
    if [_source][interactionInfo][2][value] =~ /.{15,15}/ {

        mutate {
            rename => ["[_source][interactionInfo][2][value]","[_source][interactionInfo][2][value-standard]"]
        }
    }
}

Регулярное выражение .{15,15} соответствует любой строке длиной 15 символов. Если поле короче 15 символов, регулярное выражение не совпадает и mutate#rename не применяется.

для 1), одним из возможных решений будет попытка проанализировать поле с помощью фильтра jsonи если тега _jsonparsefailure нет, переименуйте поле.