logstash 使用 ruby​​ 过滤删除嵌套字段

Question

我有一个包含大量垃圾邮件字段的索引（超过 300 个）。它们都是嵌套的，看起来像这样：

kv.amp-1-234
kv.amp-1-abc
kv.amp-1-efg

所以我想做过滤并使用 remove_fields 来摆脱它们。 想使用修剪过滤器，但我不能 - 他们不支持嵌套键删除。 我不能用 筛选 { 变异{ 删除字段

因为它不支持正则表达式。

我看到唯一的方法是通过 ruby​​ 过滤：

  ruby {
    code => "
    event.to_hash.keys.each { |k|
    if k.start_with?('[kv.amp-1][k]')
      event.remove(k)
    end
    }
   "
   }

但它不起作用。我只需要一个使用 ruby​​ 过滤器删除嵌套键的示例（不需要正则表达式，因为 start_with? 已经足够好了）

使用logstash 5.4.2

整个配置文件：

input {

  kafka {
    topics      => "gelfEvents"
    bootstrap_servers => "elk.sonic-dev.us-east-1:9092"
    group_id    => "gelfEventsCG-elk"
    # client_id   => ""
    tags        => [ "gelf2kafka" ]
    codec       => json { charset => "UTF-8" }
  }

}

filter {
  if "gelf2kafka" in [tags] {
    mutate {
      rename => { "short_message" => "message" }
      rename => { "host" => "[host][name]" }

    }
    ruby {
      code => "event.to_hash.keys.each do |key| next unless key[0,1] == '_'; if key == '_' then event.remove(key); next; end; event.set(key[1..-1], event.remove(key)) end"
    }

#trying stuff
#   ruby {
#    code => "
#    event.to_hash.keys.each { |k|
#    if k.start_with?('[tags][k]')
#      event.remove(k)
#    end
#    }
#   "
#   }


    date {
      match => [ "timestamp", "UNIX" ]
      remove_field => [ "timestamp" ]
      # target => "timestamp_new"
    }

    translate {
      field       => "level"
      destination => "level"
      exact       => true
      regex       => false
      override    => true
      fallback    => "no match for %{level}"
      dictionary  => [ "6", "INFO",
                       "4", "WARN",
                       "3", "ERROR",
                       "7", "DEBUG" ]
    }

  }

}

output {
  if "gelf2kafka" in [tags] {
#    stdout { codec => rubydebug }

    elasticsearch {
      # protocol        => "http"
      hosts            => "localhost"
      index           => "logstash-gelf2kafka-%{+YYYY.MM.dd}"
      flush_size      => 100
      idle_flush_time => 1
    }



    statsd {
      host => "localhost"
      namespace => "logstash"
      sender => "_all" # "%{host}"
      increment => "LogsKafka2ElasticSearchAll.gelfEvents.LinesSent"
    }

  }
}

Answer 1

我将假设方括号是您要检查的字符串的一部分。如果是这样，您的问题是您没有将 k 插入到字符串中。

这会有点棘手，因为您的代码已经在双引号字符串中，但这应该可以：

k.start_with?(%{[kv.amp-1][#{k}]})

我还想问你是否需要 to_hash 调用，但这取决于 event 到底是什么类型的对象。