diff --git a/lib/sidekiq_alive.rb b/lib/sidekiq_alive.rb index ed08853..6a5e274 100644 --- a/lib/sidekiq_alive.rb +++ b/lib/sidekiq_alive.rb @@ -8,6 +8,8 @@ require "sidekiq_alive/helpers" require "sidekiq_alive/redis" +SIDEKIQ_ALIVE_INDEX_NAME = "sidekiq-alive-index" + module SidekiqAlive class << self def start @@ -28,6 +30,7 @@ def start logger.info(startup_info) register_current_instance + store_alive_key # Passing the hostname argument it's only for debugging enqueued jobs SidekiqAlive::Worker.perform_async(hostname) @@ -63,11 +66,13 @@ def unregister_current_instance # Delete any pending jobs for this instance logger.info(shutdown_info) purge_pending_jobs - redis.delete(current_instance_register_key) + redis.zrem(SIDEKIQ_ALIVE_INDEX_NAME, current_instance_register_key) end def registered_instances - redis.match("#{config.registered_instance_key}::*") + # before we return we make sure we expire old keys + expire_old_keys + redis.zrange(SIDEKIQ_ALIVE_INDEX_NAME, 0, -1) end def purge_pending_jobs @@ -136,6 +141,7 @@ def startup_info port: config.port, ttl: config.time_to_live, queue: current_queue, + register_set: SIDEKIQ_ALIVE_INDEX_NAME, liveness_key: current_lifeness_key, register_key: current_instance_register_key, } @@ -144,11 +150,21 @@ def startup_info end def successful_startup_text - "Successfully started sidekiq-alive, registered with key: #{current_instance_register_key}" + "Successfully started sidekiq-alive, registered with key: "\ + "#{current_instance_register_key} on set #{SIDEKIQ_ALIVE_INDEX_NAME}" + end + + def expire_old_keys + # we get every key that should be expired by now + keys_to_expire = redis.zrangebyscore(SIDEKIQ_ALIVE_INDEX_NAME, 0, Time.now.to_i) + # then we remove it + keys_to_expire.each { |key| redis.zrem(SIDEKIQ_ALIVE_INDEX_NAME, key) } end def register_instance(instance_name) - redis.set(instance_name, time: Time.now.to_i, ex: config.registration_ttl.to_i) + expiration = Time.now.to_i + config.registration_ttl.to_i + redis.zadd(SIDEKIQ_ALIVE_INDEX_NAME, expiration, instance_name) + expire_old_keys end end end diff --git a/lib/sidekiq_alive/redis/base.rb b/lib/sidekiq_alive/redis/base.rb index 3627083..0c9f853 100644 --- a/lib/sidekiq_alive/redis/base.rb +++ b/lib/sidekiq_alive/redis/base.rb @@ -7,7 +7,19 @@ def set(...) raise(NotImplementedError) end - def match(key) + def zadd(set_key, ex, key) + raise(NotImplementedError) + end + + def zrange(set_key, start, stop) + raise(NotImplementedError) + end + + def zrangebyscore(set_key, min, max) + raise(NotImplementedError) + end + + def zrem(set_key, key) raise(NotImplementedError) end diff --git a/lib/sidekiq_alive/redis/redis_client_gem.rb b/lib/sidekiq_alive/redis/redis_client_gem.rb index 55c4094..08415fa 100644 --- a/lib/sidekiq_alive/redis/redis_client_gem.rb +++ b/lib/sidekiq_alive/redis/redis_client_gem.rb @@ -15,8 +15,20 @@ def get(key) Sidekiq.redis { |redis| redis.call("GET", key) } end - def match(key) - Sidekiq.redis { |redis| redis.scan("MATCH", key).map { |key| key } } + def zadd(set_key, ex, key) + Sidekiq.redis { |redis| redis.call("ZADD", set_key, ex, key) } + end + + def zrange(set_key, start, stop) + Sidekiq.redis { |redis| redis.call("ZRANGE", set_key, start, stop) } + end + + def zrangebyscore(set_key, min, max) + Sidekiq.redis { |redis| redis.call("ZRANGEBYSCORE", set_key, min, max) } + end + + def zrem(set_key, key) + Sidekiq.redis { |redis| redis.call("ZREM", set_key, key) } end def delete(key) diff --git a/lib/sidekiq_alive/redis/redis_gem.rb b/lib/sidekiq_alive/redis/redis_gem.rb index 0e0bc8a..48840d7 100644 --- a/lib/sidekiq_alive/redis/redis_gem.rb +++ b/lib/sidekiq_alive/redis/redis_gem.rb @@ -15,16 +15,20 @@ def get(key) redis.get(key) end - def match(key) - keys = [] - cursor = 0 - - loop do - cursor, found_keys = redis.scan(cursor, match: key, count: 1000) - keys += found_keys if found_keys - break if cursor.to_i == 0 - end - keys + def zadd(set_key, ex, key) + redis.zadd(set_key, ex, key) + end + + def zrange(set_key, start, stop) + redis.zrange(set_key, start, stop) + end + + def zrangebyscore(set_key, min, max) + redis.zrangebyscore(set_key, min, max) + end + + def zrem(set_key, key) + redis.zrem(set_key, key) end def delete(key) diff --git a/spec/redis_spec.rb b/spec/redis_spec.rb index 0b58cae..7c0ca73 100644 --- a/spec/redis_spec.rb +++ b/spec/redis_spec.rb @@ -8,8 +8,10 @@ redis.set("hello", time: time, ex: 60) expect(redis.ttl("hello") > 1).to(be(true)) expect(redis.get("hello")).to(eq(time)) - expect(redis.match("hello")).to(eq(["hello"])) - expect(redis.delete("hello")).to(eq(1)) - expect(redis.get("hello")).to(be(nil)) + redis.zadd("test_set", Time.now.to_i, "test-key-1") + redis.zadd("test_set", Time.now.to_i, "test-key-2") + expect(redis.zrange("test_set", 0, -1)).to(eq(["test-key-1", "test-key-2"])) + expect(redis.zrem("test_set", "test-key-1")) + expect(redis.zrange("test_set", 0, -1)).to(eq(["test-key-2"])) end end diff --git a/spec/sidekiq_alive_spec.rb b/spec/sidekiq_alive_spec.rb index 0030fd4..c5c70ff 100644 --- a/spec/sidekiq_alive_spec.rb +++ b/spec/sidekiq_alive_spec.rb @@ -133,7 +133,6 @@ it "::registered_instances" do SidekiqAlive.start - expect(SidekiqAlive.registered_instances.count).to(eq(1)) expect(SidekiqAlive.registered_instances.first).to(include("test-hostname")) end