class LogStash::Filters::ZeroMQ
ZeroMQ
filter. This is the best way to send an event externally for filtering It works much like an exec filter would by sending the event “offsite” for processing and waiting for a response
The protocol here is:
* REQ sent with JSON-serialized logstash event * REP read expected to be the full JSON 'filtered' event * - if reply read is an empty string, it will cancel the event.
Note that this is a limited subset of the zeromq functionality in inputs and outputs. The only topology that makes sense here is: REQ/REP.
Public Class Methods
new(params)
click to toggle source
Calls superclass method
LogStash::Filters::Base::new
# File lib/logstash/filters/zeromq.rb, line 55 def initialize(params) super(params) @threadsafe = false end
Public Instance Methods
filter(event)
click to toggle source
# File lib/logstash/filters/zeromq.rb, line 80 def filter(event) return unless filter?(event) # TODO (lusis) # Need to set a timeout on the socket # If it never gets a reply, filtering stops cold begin if @field @logger.debug("0mq: sending", :request => event[@field]) error_check(@zsocket.send_string(event[@field]), "in send_string") else @logger.debug("0mq: sending", :request => event.to_json) error_check(@zsocket.send_string(event.to_json), "in send_string") end reply = '' rc = @zsocket.recv_string(reply) error_check(rc, "in recv_string") # If we receive an empty reply, this is an indication that the filter # wishes to cancel this event. if reply.empty? event.cancel return end @logger.debug("0mq: receiving", :reply => reply) if @field event[@field] = event.sprintf(reply) filter_matched(event) else reply = JSON.parse(reply) event.overwrite(reply) end filter_matched(event) rescue => e @logger.warn("0mq filter exception", :address => @address, :exception => e, :backtrace => e.backtrace) end end
register()
click to toggle source
# File lib/logstash/filters/zeromq.rb, line 62 def register require "ffi-rzmq" require "logstash/util/zeromq" self.class.send(:include, LogStash::Util::ZeroMQ) @zsocket = context.socket(ZMQ::REQ) error_check(@zsocket.setsockopt(ZMQ::LINGER, 1), "while setting ZMQ::LINGER == 1)") if @sockopt setopts(@zsocket, @sockopt) end setup(@zsocket, @address) end
Private Instance Methods
server?()
click to toggle source
# File lib/logstash/filters/zeromq.rb, line 119 def server? @mode == "server" end