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

Question

我有filebeat发送文件到Logstash。 我在logstash里面有一些过滤器。它看起来像这样：

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

它基本上删除用户数据，因为我不希望他们在日志中。因此，它具有以下功能：

用户[[email protected]] ---->用户[]

我将需要使用插入解析用户内部SHA1或类似的校验和，是这样的：

用户[[email protected]] ---->用户[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 [[email protected]] was idle，那么它将被转化为：

user [d709f370e52b57b4eb75f04e2b3422c4d41a05148cad8f81776d94a048fb70af] was idle