require 'redis/ick/version' require 'redis/script_manager' class Redis # Binds Lua code to provide the Ick operations in Redis. # class Ick # TODO: rdoc # Creates an Ick accessor. # # @param redis Redis # # @param statsd a stats proxy. May be nil, else expected to respond # to :increment and :timing. # def initialize(redis, statsd: nil) if !redis.is_a?(Redis) raise ArgumentError, "not a Redis: #{redis}" end if statsd if !statsd.respond_to?(:increment) raise ArgumentError, 'no statsd.increment' end if !statsd.respond_to?(:timing) raise ArgumentError, 'no statsd.timeing' end if !statsd.respond_to?(:time) raise ArgumentError, 'no statsd.time' end end @redis = redis @statsd = statsd end attr_accessor :redis attr_accessor :statsd # Reports a single count on the requested metric to statsd (if any). # # @param metric String # def _statsd_increment(metric) statsd.increment(metric) if statsd end # Reports the specified timing on the requested metric to statsd (if # any). # # @param metric String # def _statsd_timing(metric,time) statsd.timing(metric,time) if statsd end # Executes the block (if any) and reports its timing in milliseconds # on the requested metric to statsd (if any). # # @param metric String # # @return the value of the block, or nil if none # def _statsd_time(metric) if statsd statsd.time(metric) do block_given? ? yield : nil end else block_given? ? yield : nil end end # Removes all data associated with the Ick in Redis at key. # # Similar to DEL key, http://redis.io/commands/del, but may # delete multiple keys which together implement the Ick data # structure. # # @param ick_key String the base key for the Ick # # @return an integer, the number of Redis keys deleted, which will # be >= 1 if an Ick existed at key. # def ickdel(ick_key) if !ick_key.is_a?(String) raise ArgumentError, "bogus non-String ick_key #{ick_key}" end _statsd_increment('profile.ick.ickdel.calls') _statsd_time('profile.ick.ickdel.time') do _eval(LUA_ICKDEL,ick_key) end end # Fetches stats. # # @param ick_key String the base key for the Ick # # @return If called outside a Redis pipeline, a Hash with stats # about the Ick at ick_key, if any, else nil. If called within a # pipeline, returns a redis::Future whose value is a Hash or nil as # before. # def ickstats(ick_key) if !ick_key.is_a?(String) raise ArgumentError, "bogus non-String ick_key #{ick_key}" end _statsd_increment('profile.ick.ickstats.calls') raw_ickstats_results = nil _statsd_time('profile.ick.time.ickstats') do raw_ickstats_results = _eval(LUA_ICKSTATS,ick_key) end if raw_ickstats_results.is_a?(Redis::Future) # # We extend the Redis::Future with a continuation so we can add # our own post-processing. # class << raw_ickstats_results alias_method :original_value, :value def value ::Redis::Ick._postprocess_ickstats_results(original_value) end end raw_ickstats_results else ::Redis::Ick._postprocess_ickstats_results(raw_ickstats_results) end end def self._postprocess_ickstats_results(raw_ickstats_results) return nil if !raw_ickstats_results # # LUA_ICKSTATS returned bulk data response [k,v,k,v,...] # stats = Hash[*raw_ickstats_results] # # From http://redis.io/commands/eval, the "Lua to Redis conversion # table" states that: # # Lua number -> Redis integer reply (the number is converted # into an integer) # # ...If you want to return a float from Lua you should return # it as a string. # # LUA_ICKSTATS works around this by converting certain stats to # strings. We reverse that conversion here. # stats.keys.select{|k|/_min$/ =~ k || /_max$/ =~ k}.each do |k| next if !stats[k] stats[k] = (/^\d+$/ =~ stats[k]) ? stats[k].to_i : stats[k].to_f end stats end # Adds all the specified members with the specified scores to the # Ick stored at key. # # Entries are stored in order by score. Lower-scored entries will # pop out in reserve before higher-scored entries. Re-adding an # entry which already exists in the producer set with a new score # results in the entry having the lowest of the old and new scores. # # Similar to http://redis.io/commands/zadd with a modified NX # option, operating on the producer set. # # Usage: # # ick.ickadd(ick_key,score,member[,score,member]*) # # Suggested usage is for scores to be a Unix timestamp indicating # when something became dirty. # # @param ick_key String the base key for the Ick # # @param score_member_pairs Array of Arrays of [score,message] # # @return a pair, the number of new values followed by the numer of # changed scores. # def ickadd(ick_key,*score_member_pairs) if !ick_key.is_a?(String) raise ArgumentError, "bogus non-String ick_key #{ick_key}" end if score_member_pairs.size.odd? raise ArgumentError, "bogus odd-numbered #{score_member_pairs}" end score_member_pairs.each_slice(2) do |slice| score, member = slice if ! score.is_a? Numeric raise ArgumentError, "bogus non-Numeric score #{score}" end if ! member.is_a? String raise ArgumentError, "bogus non-String member #{member}" end end _statsd_increment('profile.ick.ickadd.calls') _statsd_timing('profile.ick.ickadd.pairs',score_member_pairs.size / 2) _statsd_time('profile.ick.time.ickadd') do _eval(LUA_ICKADD,ick_key,*score_member_pairs) end end # Tops up the consumer set up to max_size by shifting the # lowest-scored elements from the producer set into the consumer set # until the consumer set cardinality reaches max_size or the # producer set is exhausted. # # The reserved elements are meant to represent consumer # work-in-progress. If they are not committed, they will be # returned again in future calls to ickreserve. # # Note that the Lua for ick is irritating like so: # # - you add in the pattern [ score_number, member_string, ... ] # - you retrieve in the pattern [ member_string, score_string, ... ] # # Native ZADD and ZRANGE WITHSCORES exhibit this same irritating # inconsistency: Ick is annoyance-compatible with Redis sorted sets. # # However, by analogy with redis-rb's Redis.current.zrange(), this # Ruby wrapper method pairs up the results for you, and converts the # string scores to floats. # # - you get from this method [[ member_string, score_number] , ... ] # # @param ick_key String the base key for the Ick # # @param max_size max number of messages to reserve # # @return a list of up to max_size pairs, similar to # Redis.current.zrange() withscores: [ member_string, score_number ] # representing the lowest-scored elements from the producer set. # def ickreserve(ick_key,max_size=0) if !ick_key.is_a?(String) raise ArgumentError, "bogus non-String ick_key #{ick_key}" end if !max_size.is_a?(Integer) raise ArgumentError, "bogus non-Integer max_size #{max_size}" end if max_size < 0 raise ArgumentError, "bogus negative #{max_size}" end _statsd_increment('profile.ick.ickreserve.calls') _statsd_timing('profile.ick.ickreserve.max_size',max_size) raw_ickreserve_results = nil _statsd_time('profile.ick.time.ickreserve') do raw_ickreserve_results = _eval( LUA_ICKRESERVE, ick_key, max_size ) end if raw_ickreserve_results.is_a?(Redis::Future) # # We extend the Redis::Future with a continuation so we can # add our own post-processing. # class << raw_ickreserve_results alias_method :original_value, :value def value original_value.each_slice(2).map do |p| [ p[0], ::Redis::Ick._floatify(p[1]) ] end end end raw_ickreserve_results else results = raw_ickreserve_results.each_slice(2).map do |p| [ p[0], ::Redis::Ick._floatify(p[1]) ] end _statsd_timing('profile.ick.ickreserve.num_results',results.size) results end end # Removes the indicated members from the producer set, if present. # # Similar to ZREM ick_key [member]*, per # http://redis.io/commands/zrem, operating on the consumer set only. # # Usage: # # ick.ickcommit(ick_key,memberA,memberB,...) # # Committed elements are meant to represent consumer work-completed. # # @param ick_key String the base key for the Ick # # @param members members to be committed out pf the pset # # @return an integer, the number of members removed from the # producer set, not including non existing members. # def ickcommit(ick_key,*members) if !ick_key.is_a?(String) raise ArgumentError, "bogus non-String ick_key #{ick_key}" end _statsd_increment('profile.ick.ickcommit.calls') _statsd_timing('profile.ick.ickcommit.members',members.size) _statsd_time('profile.ick.time.ickcommit') do _eval(LUA_ICKCOMMIT,ick_key,*members) end end # Converts a string str into a Float, and recognizes 'inf', '-inf', # etc. # # So we can be certain of compatibility, this was stolen with tweaks # from https://github.com/redis/redis-rb/blob/master/lib/redis.rb. # def self._floatify(str) raise ArgumentError, "not String: #{str}" if !str.is_a?(String) if (inf = str.match(/^(-)?inf/i)) (inf[1] ? -1.0 : 1.0) / 0.0 else Float(str) end end # Runs the specified lua in the redis against the specifified Ick. # def _eval(lua,ick_key,*args) if !lua.is_a?(String) raise ArgumentError, "bogus non-String lua #{lua}" end if !ick_key.is_a?(String) raise ArgumentError, "bogus non-String ick_key #{ick_key}" end Redis::ScriptManager.eval_gently(redis,lua,[ick_key],args) end ####################################################################### # # The Ick Data Model in Redis # # - At ick_key, we keep a simple manifest string. Currently, only # 'ick.v1' is expected or supported. This is for future-proofing # # - At "#{ick_key}/ick/{#{ick_key}}/pset" we keep a sorted set. # Ick, the "producer set", into which new messages are pushed by # ickadd. # # - At "#{ick_key}/ick/{#{ick_key}}/cset" we keep another sorted # set, the "consumer set", where messages are held between # ickreserve and ickcommit. # # These name patterns were chosen carefully, so that under the Redis # Cluster Specification, http://redis.io/topics/cluster-spec, all # three keys will always be hashed to the same HASH_SLOT. # # Note that if ick_key contain a user-specified prescriptive hashing # subsequence like "{foo}", then that {}-sequence appears at the # front of the key, hence will serve as the prescriptive hash key # for all the derived keys which use ick_keys a prefix. # # If ick_key does not contain a {}-sequence, then the portion of all # derived keys ".../{#{ick_key}}/..." provides one which hashes to # the same slot as ick_key. # # Thus, we know all keys will be present on these same shard at # run-time. # # The ickadd op adds entries to the pset, but only if they do not # already exist. # # The ickreserve op moves entries from the pset into the cset. # # The ickcommit op removes entries from the pset. # # WARNING: If ick_key itself contains an {}-expr, this hashslot # matching algorithm will break in RedisLabs Enterprise Cluster due # to the newly-discovered Yikes Curly Brace Surprise. See # https://github.com/ProsperWorks/ALI/pull/1132 for details. # ####################################################################### ####################################################################### # LUA_ICK_PREFIX ####################################################################### # # A snippet of Lua code which is common to all the Ick scripts. # # For convenience and to avoid repeating code, we set up # some computed key names. # # For safety, we check that the ick_ver, ick_pset, and ick_cset # either do not exist or exit with the correct types and values to # be identifiable as an Ick. # # All scripts in the LUA_ICK series expect only one KEYS, the root # key of the Ick data structure. We expect a version flag as a # string at this key. Keys for other data are computed from KEYS[1] # in such a way as to guarantee they all hash to the same slot. # LUA_ICK_PREFIX = %{ local ick_key = KEYS[1] local ick_ver = redis.call('GET',ick_key) local ick_pset_key = ick_key .. '/ick/{' .. ick_key .. '}/pset' local ick_cset_key = ick_key .. '/ick/{' .. ick_key .. '}/cset' local ick_ver_type = redis.call('TYPE',ick_key).ok local ick_pset_type = redis.call('TYPE',ick_pset_key).ok local ick_cset_type = redis.call('TYPE',ick_cset_key).ok if (false ~= ick_ver and 'ick.v1' ~= ick_ver) then return redis.error_reply('unrecognized ick version ' .. ick_ver) end if ('none' ~= ick_ver_type and 'string' ~= ick_ver_type) then return redis.error_reply('ick defense: expected string at ' .. ick_ver_key .. ', found ' .. ick_ver_type) end if ('none' ~= ick_pset_type and 'zset' ~= ick_pset_type) then return redis.error_reply('ick defense: expected string at ' .. ick_pset_key .. ', found ' .. ick_pset_type) end if ('none' ~= ick_cset_type and 'zset' ~= ick_cset_type) then return redis.error_reply('ick defense: expected string at ' .. ick_cset_key .. ', found ' .. ick_cset_type) end if ('none' == ick_ver_type) then if ('none' ~= ick_pset_type) then return redis.error_reply('ick defense: no ver at ' .. ick_ver_key .. ', but found pset at ' .. ick_pset_key) end if ('none' ~= ick_cset_type) then return redis.error_reply('ick defense: no ver at ' .. ick_ver_key .. ', but found cset at ' .. ick_cset_key) end end }.freeze ####################################################################### # LUA_ICKDEL ####################################################################### # # Removes all keys associated with the Ick at KEYS[1]. # # @param uses no ARGV # # @return the number of Redis keys deleted, which will be 0 if and # only if no Ick existed at KEYS[1] # LUA_ICKDEL = (LUA_ICK_PREFIX + %{ return redis.call('DEL',ick_key,ick_pset_key,ick_cset_key) }).freeze ####################################################################### # LUA_ICKSTATS ####################################################################### # # @param uses no ARGV # # @return a bulk data response with statistics about the Ick at # KEYS[1], or nil if none. # # Note: At http://redis.io/commands/eval, the "Lua to Redis # conversion table" stats: # # Lua number -> Redis integer reply (the number is converted # into an integer) # # ...If you want to return a float from Lua you should return # it as a string. # # We follow this recommendation in our Lua below where we convert # our numeric responses to strings with "tostring(tonumber(n))". # LUA_ICKSTATS = (LUA_ICK_PREFIX + %{ if (false == ick_ver) then return nil end local ick_pset_size = redis.call('ZCARD',ick_pset_key) local ick_cset_size = redis.call('ZCARD',ick_cset_key) local ick_stats = { 'key', ick_key, 'keys', { ick_key, ick_pset_key, ick_cset_key }, 'ver', ick_ver, 'cset_size', ick_cset_size, 'pset_size', ick_pset_size, 'total_size', ick_cset_size + ick_pset_size, } local pset_min = nil local pset_max = nil if ick_pset_size > 0 then pset_min = redis.call('ZRANGE',ick_pset_key, 0, 0,'WITHSCORES')[2] table.insert(ick_stats, 'pset_min') table.insert(ick_stats, tostring(tonumber(pset_min))) pset_max = redis.call('ZRANGE',ick_pset_key,-1,-1,'WITHSCORES')[2] table.insert(ick_stats, 'pset_max') table.insert(ick_stats, tostring(tonumber(pset_max))) end local cset_min = nil local cset_max = nil if ick_cset_size > 0 then cset_min = redis.call('ZRANGE',ick_cset_key, 0, 0,'WITHSCORES')[2] table.insert(ick_stats, 'cset_min') table.insert(ick_stats, tostring(tonumber(cset_min))) cset_max = redis.call('ZRANGE',ick_cset_key,-1,-1,'WITHSCORES')[2] table.insert(ick_stats, 'cset_max') table.insert(ick_stats, tostring(tonumber(cset_max))) end local total_min = nil if pset_min and cset_min then total_min = math.min(cset_min,pset_min) elseif pset_min then total_min = pset_min elseif cset_min then total_min = cset_min end if total_min then table.insert(ick_stats, 'total_min') table.insert(ick_stats, tostring(tonumber(total_min))) end local total_max = nil if pset_max and cset_max then total_max = math.max(cset_max,pset_max) elseif pset_max then total_max = pset_max elseif cset_max then total_max = cset_max end if total_max then table.insert(ick_stats, 'total_max') table.insert(ick_stats, tostring(tonumber(total_max))) end return ick_stats }).freeze ####################################################################### # LUA_ICKADD ####################################################################### # # Adds members to the cset as per ZADD. Where a member is # re-written, we always take the lowest score. # # Thus, scores are only allowed to move downward. changes to score. # # Creates the Ick if necessary. # # @param ARGV a sequence of score,member pairs as per Redis ZADD. # # @return a pair of numbers [num_new, num_changed] # LUA_ICKADD = (LUA_ICK_PREFIX + %{ local num_args = table.getn(ARGV) if 1 == (num_args % 2) then return redis.error_reply("odd number of arguments for 'ickadd' command") end local num_new = 0 local num_changed = 0 for i = 1,num_args,2 do local score = tonumber(ARGV[i]) local member = ARGV[i+1] local old_score = redis.call('ZSCORE',ick_pset_key,member) if false == old_score then redis.call('ZADD',ick_pset_key,score,member) num_new = num_new + 1 elseif score < tonumber(old_score) then redis.call('ZADD',ick_pset_key,score,member) num_changed = num_changed + 1 end end redis.call('SETNX', ick_key, 'ick.v1') return { num_new, num_changed } }).freeze ####################################################################### # LUA_ICKRESERVE ####################################################################### # # Tops up the cset to up to size ARGV[1] by shifting the # lowest-scored members over from the pset. # # The cset might already be full, in which case we may shift fewer # than ARGV[1] elements. # # The same score-folding happens as per ICKADD. Thus where there # are duplicate messages, we may remove more members from the pset # than we add to the cset. # # @param ARGV a single number, batch_size, the desired # size for cset and to be returned # # @return a bulk response, up to ARGV[1] pairs [member,score,...] # LUA_ICKRESERVE = (LUA_ICK_PREFIX + %{ local target_cset_size = tonumber(ARGV[1]) while true do local ick_cset_size = redis.call('ZCARD',ick_cset_key) if ick_cset_size and target_cset_size <= ick_cset_size then break end local first_in_pset = redis.call('ZRANGE',ick_pset_key,0,0,'WITHSCORES') if 0 == table.getn(first_in_pset) then break end local first_member = first_in_pset[1] local first_score = tonumber(first_in_pset[2]) redis.call('ZREM',ick_pset_key,first_member) local old_score = redis.call('ZSCORE',ick_cset_key,first_member) if false == old_score or first_score < tonumber(old_score) then redis.call('ZADD',ick_cset_key,first_score,first_member) end end redis.call('SETNX', ick_key, 'ick.v1') if target_cset_size <= 0 then return {} else local max = target_cset_size - 1 return redis.call('ZRANGE',ick_cset_key,0,max,'WITHSCORES') end }).freeze ####################################################################### # LUA_ICKCOMMIT ####################################################################### # # Removes specified members from the pset. # # @param ARGV a list of members to be removed from the cset # # @return the number of members removed # # Note: This this Lua unpacks ARGV with the iterator ipairs() # instead of unpack() to avoid a "too many results to unpack" # failure at 8000 args. However, the loop over many redis.call is # regrettably heavy-weight. From a performance standpoint it # would be preferable to call ZREM in larger batches. # LUA_ICKCOMMIT = (LUA_ICK_PREFIX + %{ redis.call('SETNX', ick_key, 'ick.v1') if 0 == table.getn(ARGV) then return 0 end local num_removed = 0 for i,v in ipairs(ARGV) do num_removed = num_removed + redis.call('ZREM',ick_cset_key,v) end return num_removed }).freeze end end