elasticsearch - Logstash-如何在字段内动态创建校验和

Question

我有文件拍子发送文件到Logstash。
我在logstash中有一些过滤器。看起来像这样:

filter {
 if [message] =~ /user/{
  mutate {
        gsub => ["message", "user \[(.*?)] was", "user [] was"]              
      }
  }
}

它基本上删除了用户数据，因为我不希望它们出现在日志中。因此，它执行以下操作:

user [myemail@email.com] ----> user []

我将需要使用SHA1或类似的校验和插入到已分析的用户内部，如下所示:

user [myemail@email.com] ----> user [CHECKSUMISHERE]

详细错误在这里。

[2017-04-26T13:13:53,153][ERROR][logstash.pipeline        ] A plugin had an unrecoverable error. Will restart this plugin.
  Plugin: <LogStash::Inputs::Beats port=>5043, codec=><LogStash::Codecs::JSON id=>"json_bf758128-700d-4332-a0c0-c958a6c9dc09", enable_metric=>true, charset=>"UTF-8">, id=>"8d67450b6c5fcad922dd223d89206b7b8d5c884d-1", enable_metric=>true, host=>"0.0.0.0", ssl=>false, ssl_verify_mode=>"none", include_codec_tag=>true, ssl_handshake_timeout=>10000, congestion_threshold=>5, target_field_for_codec=>"message", tls_min_version=>1, tls_max_version=>1.2, cipher_suites=>["TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384", "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384", "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256", "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", "TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA384", "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384", "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256"], client_inactivity_timeout=>60>
  Error: event executor terminated
  Exception: Java::JavaUtilConcurrent::RejectedExecutionException
  Stack: io.netty.util.concurrent.SingleThreadEventExecutor.reject(io/netty/util/concurrent/SingleThreadEventExecutor.java:840)
io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(io/netty/util/concurrent/SingleThreadEventExecutor.java:342)
io.netty.util.concurrent.SingleThreadEventExecutor.addTask(io/netty/util/concurrent/SingleThreadEventExecutor.java:335)
io.netty.util.concurrent.SingleThreadEventExecutor.execute(io/netty/util/concurrent/SingleThreadEventExecutor.java:765)
io.netty.channel.AbstractChannel$AbstractUnsafe.register(io/netty/channel/AbstractChannel.java:475)
io.netty.channel.SingleThreadEventLoop.register(io/netty/channel/SingleThreadEventLoop.java:80)
io.netty.channel.SingleThreadEventLoop.register(io/netty/channel/SingleThreadEventLoop.java:74)
io.netty.channel.MultithreadEventLoopGroup.register(io/netty/channel/MultithreadEventLoopGroup.java:85)
io.netty.bootstrap.AbstractBootstrap.initAndRegister(io/netty/bootstrap/AbstractBootstrap.java:330)
io.netty.bootstrap.AbstractBootstrap.doBind(io/netty/bootstrap/AbstractBootstrap.java:281)
io.netty.bootstrap.AbstractBootstrap.bind(io/netty/bootstrap/AbstractBootstrap.java:277)
io.netty.bootstrap.AbstractBootstrap.bind(io/netty/bootstrap/AbstractBootstrap.java:259)
org.logstash.beats.Server.listen(org/logstash/beats/Server.java:68)
java.lang.reflect.Method.invoke(java/lang/reflect/Method.java:498)
RUBY.run(/Users/xxx/Downloads/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.12-java/lib/logstash/inputs/beats.rb:213)
RUBY.inputworker(/Users/xxx/Downloads/elk/logstash/logstash-core/lib/logstash/pipeline.rb:425)
RUBY.start_input(/Users/xxx/Downloads/elk/logstash/logstash-core/lib/logstash/pipeline.rb:419)
java.lang.Thread.run(java/lang/Thread.java:745)
[2017-04-26T13:13:54,979][DEBUG][logstash.agent           ] Reading config file {:config_file=>"/Users/xxx/Downloads/elk/logstash/config/first-pipeline.conf"}
[2017-04-26T13:13:54,980][DEBUG][logstash.agent           ] no configuration change for pipeline {:pipeline=>"main"}

Answer 1

您可以使用 ruby filter执行该转换。

filter {
  if [message] =~ /user/{
    ruby {
      init => "require 'digest'"
      code => "
        message = event.get('message')
        email = message.match(/user \[(.*?)\] was/)
        if email.present?
          sha = Digest::SHA256.hexdigest email[0].captures
          event.set('message', message.gsub(email[0].captures, sha))
        end
      "
    }
  }
}

如果 message看起来像 user [john@doe.com] was idle，那么它将被转换为:

 user [d709f370e52b57b4eb75f04e2b3422c4d41a05148cad8f81776d94a048fb70af] was idle