Class: LogStash::Filters::CacheRedis

Inherits:

Base

• Object
• Base
• LogStash::Filters::CacheRedis

Defined in:

lib/logstash/filters/cache_redis.rb

Instance Method Summary

• #filter(event) ⇒ Object
• #register ⇒ Object

Instance Method Details

#filter(event) ⇒ Object

# File 'lib/logstash/filters/cache_redis.rb', line 124

def filter(event)

    # TODO: Maybe refactor the interface into a more flexible one with two
    #       main configs 'cmd' & 'args'. Then it would be possible to eliminate
    #       all if clauses and replace it through one hashmap call, where
    #       the hashmap would be a mapping from 'cmd' -> <cmd_function_ref>
    #       E.q.: cmds.fetch(event.get(@llen), &method(:cmd_not_found_err))
    max_retries = @max_retries
    begin
        @redis ||= connect

        if @get
            event.set(@target, @redis.get(event.get(@get)))
        end

        if @set
            @redis.set(event.get(@set), event.get(@source))
        end

        if @exists
            @redis.exists(event.get(@exists))
        end

        if @del
            @redis.del(event.get(@del))
        end

        if @hget
            event.set(@target, @redis.hget(event.get(@hget), event.get(@source)))
        end

        if @hset
            @redis.hset(event.get(@hset), event.get(@field), event.get(@source))
        end

        if @sadd
            @redis.sadd(event.get(@sadd), event.get(@source))
        end

        if @sismember
            event.set(@target, @redis.sismember(event.get(@sismember), event.get(@source)))
        end

        if @smembers
            @redis.smembers(event.get(@smembers))
        end

        if @scard
            event.set(@target, @redis.scard(event.get(@scard)))
        end

        if @llen
            key = @cmd_key_is_formatted ? event.sprintf(@llen) : event.get(@llen)
            event.set(@target, @redis.llen(key))
        end

        if @rpush
            @redis.rpush(event.get(@rpush), event.get(@source))
        end

        if @rpushnx
            key = @cmd_key_is_formatted ? event.sprintf(@rpushnx) : event.get(@rpushnx)
            max_lock_retries = @max_lock_retries
            begin
                @lock_manager ||= connect_lockmanager
                @lock_manager.lock!("lock_#{key}", @lock_timeout) do
                    @redis.rpush(key, event.get(@source)) unless @redis.exists(key)
                end
            rescue Redlock::LockError => e
                @logger.warn("Failed to lock section 'rpushnx' for key: #{key}",
                             :event => event, :exception => e)
                sleep @lock_retry_interval
                max_lock_retries -= 1
                unless max_lock_retries < 0
                    retry
                else
                    @logger.error("Max retries reached for trying to lock section 'rpushnx' for key: #{key}",
                                  :event => event, :exception => e)
                end
            end
        end

        if @rpop
            event.set(@target, @redis.rpop(event.get(@rpop)))
        end

        if @lget
            event.set(@target, @redis.lrange(event.get(@lget), 0, -1))
        end

    rescue => e
        @logger.warn("Failed to send event to Redis, retrying after #{@reconnect_interval} seconds...", :event => event,
                     :exception => e, :backtrace => e.backtrace)
        sleep @reconnect_interval
        @redis = nil
        @lock_manager = nil
        max_retries -= 1
        unless max_retries < 0
            retry
        else
            @logger.error("Max retries reached for trying to execute a command",
                          :event => event, :exception => e)
        end
    end

    # filter_matched should go in the last line of our successful code
    filter_matched(event)
end

#register ⇒ Object

# File 'lib/logstash/filters/cache_redis.rb', line 114

def register
    @redis = nil
    @lock_manager = nil
    if @shuffle_hosts
        @host.shuffle!
    end
    @host_idx = 0
end