Merge pull request #85 from launchdarkly/eb/ch28328/feature-store-sup…

…port

factor common logic out of RedisFeatureStore, add integrations module

lib/ldclient-rb.rb

@@ -1,4 +1,5 @@
 require "ldclient-rb/version"
+require "ldclient-rb/interfaces"
 require "ldclient-rb/util"
 require "ldclient-rb/evaluation"
 require "ldclient-rb/flags_state"
@@ -16,6 +17,6 @@
 require "ldclient-rb/non_blocking_thread_pool"
 require "ldclient-rb/event_summarizer"
 require "ldclient-rb/events"
-require "ldclient-rb/redis_store"
 require "ldclient-rb/requestor"
 require "ldclient-rb/file_data_source"
+require "ldclient-rb/integrations"

lib/ldclient-rb/impl.rb

@@ -0,0 +1,10 @@
+
+module LaunchDarkly
+  #
+  # Low-level implementation classes. Everything in this module should be considered non-public
+  # and subject to change with any release.
+  #
+  module Impl
+    # code is in ldclient-rb/impl/
+  end
+end

lib/ldclient-rb/impl/integrations/redis_impl.rb

@@ -0,0 +1,150 @@
+require "concurrent/atomics"
+require "json"
+
+module LaunchDarkly
+  module Impl
+    module Integrations
+      module Redis
+        #
+        # Internal implementation of the Redis feature store, intended to be used with CachingStoreWrapper.
+        #
+        class RedisFeatureStoreCore
+          begin
+            require "redis"
+            require "connection_pool"
+            REDIS_ENABLED = true
+          rescue ScriptError, StandardError
+            REDIS_ENABLED = false
+          end
+
+          def initialize(opts)
+            if !REDIS_ENABLED
+              raise RuntimeError.new("can't use Redis feature store because one of these gems is missing: redis, connection_pool")
+            end
+
+            @redis_opts = opts[:redis_opts] || Hash.new
+            if opts[:redis_url]
+              @redis_opts[:url] = opts[:redis_url]
+            end
+            if !@redis_opts.include?(:url)
+              @redis_opts[:url] = LaunchDarkly::Integrations::Redis::default_redis_url
+            end
+            max_connections = opts[:max_connections] || 16
+            @pool = opts[:pool] || ConnectionPool.new(size: max_connections) do
+              ::Redis.new(@redis_opts)
+            end
+            @prefix = opts[:prefix] || LaunchDarkly::Integrations::Redis::default_prefix
+            @logger = opts[:logger] || Config.default_logger
+            @test_hook = opts[:test_hook]  # used for unit tests, deliberately undocumented
+
+            @stopped = Concurrent::AtomicBoolean.new(false)
+
+            with_connection do |redis|
+              @logger.info("RedisFeatureStore: using Redis instance at #{redis.connection[:host]}:#{redis.connection[:port]} \
+      and prefix: #{@prefix}")
+            end
+          end
+
+          def init_internal(all_data)
+            count = 0
+            with_connection do |redis|
+              all_data.each do |kind, items|
+                redis.multi do |multi|
+                  multi.del(items_key(kind))
+                  count = count + items.count
+                  items.each { |key, item|
+                    redis.hset(items_key(kind), key, item.to_json)
+                  }
+                end
+              end
+            end
+            @logger.info { "RedisFeatureStore: initialized with #{count} items" }
+          end
+
+          def get_internal(kind, key)
+            with_connection do |redis|
+              get_redis(redis, kind, key)
+            end
+          end
+
+          def get_all_internal(kind)
+            fs = {}
+            with_connection do |redis|
+              hashfs = redis.hgetall(items_key(kind))
+              hashfs.each do |k, json_item|
+                f = JSON.parse(json_item, symbolize_names: true)
+                fs[k.to_sym] = f
+              end
+            end
+            fs
+          end
+
+          def upsert_internal(kind, new_item)
+            base_key = items_key(kind)
+            key = new_item[:key]
+            try_again = true
+            final_item = new_item
+            while try_again
+              try_again = false
+              with_connection do |redis|
+                redis.watch(base_key) do
+                  old_item = get_redis(redis, kind, key)
+                  before_update_transaction(base_key, key)
+                  if old_item.nil? || old_item[:version] < new_item[:version]
+                    result = redis.multi do |multi|
+                      multi.hset(base_key, key, new_item.to_json)
+                    end
+                    if result.nil?
+                      @logger.debug { "RedisFeatureStore: concurrent modification detected, retrying" }
+                      try_again = true
+                    end
+                  else
+                    final_item = old_item
+                    action = new_item[:deleted] ? "delete" : "update"
+                    @logger.warn { "RedisFeatureStore: attempted to #{action} #{key} version: #{old_item[:version]} \
+    in '#{kind[:namespace]}' with a version that is the same or older: #{new_item[:version]}" }
+                  end
+                  redis.unwatch
+                end
+              end
+            end
+            final_item
+          end
+
+          def initialized_internal?
+            with_connection { |redis| redis.exists(items_key(FEATURES)) }
+          end
+
+          def stop
+            if @stopped.make_true
+              @pool.shutdown { |redis| redis.close }
+            end
+          end
+
+          private
+
+          def before_update_transaction(base_key, key)
+            @test_hook.before_update_transaction(base_key, key) if !@test_hook.nil?
+          end
+
+          def items_key(kind)
+            @prefix + ":" + kind[:namespace]
+          end
+
+          def cache_key(kind, key)
+            kind[:namespace] + ":" + key.to_s
+          end
+
+          def with_connection
+            @pool.with { |redis| yield(redis) }
+          end
+
+          def get_redis(redis, kind, key)
+            json_item = redis.hget(items_key(kind), key)
+            json_item.nil? ? nil : JSON.parse(json_item, symbolize_names: true)
+          end
+        end
+      end
+    end
+  end
+end

lib/ldclient-rb/in_memory_store.rb

@@ -20,6 +20,8 @@ module LaunchDarkly
   # streaming API.
   #
   class InMemoryFeatureStore
+    include LaunchDarkly::Interfaces::FeatureStore
+
     def initialize
       @items = Hash.new
       @lock = Concurrent::ReadWriteLock.new

lib/ldclient-rb/integrations.rb

@@ -0,0 +1,27 @@
+require "ldclient-rb/integrations/redis"
+require "ldclient-rb/integrations/util/store_wrapper"
+
+module LaunchDarkly
+  #
+  # Tools for connecting the LaunchDarkly client to other software.
+  #
+  module Integrations
+    #
+    # Integration with [Redis](https://redis.io/).
+    #
+    # @since 5.5.0
+    #
+    module Redis
+      # code is in ldclient-rb/impl/integrations/redis_impl
+    end
+
+    #
+    # Support code that may be helpful in creating integrations.
+    #
+    # @since 5.5.0
+    #
+    module Util
+      # code is in ldclient-rb/integrations/util/
+    end
+  end
+end

lib/ldclient-rb/integrations/redis.rb

@@ -0,0 +1,48 @@
+require "ldclient-rb/redis_store"  # eventually we will just refer to impl/integrations/redis_impl directly
+
+module LaunchDarkly
+  module Integrations
+    module Redis
+      #
+      # Default value for the `redis_url` option for {new_feature_store}. This points to an instance of
+      # Redis running at `localhost` with its default port.
+      #
+      # @return [String]  the default Redis URL
+      #
+      def self.default_redis_url
+        'redis://localhost:6379/0'
+      end
+
+      #
+      # Default value for the `prefix` option for {new_feature_store}.
+      #
+      # @return [String]  the default key prefix
+      #
+      def self.default_prefix
+        'launchdarkly'
+      end
+
+      #
+      # Creates a Redis-backed persistent feature store.
+      #
+      # To use this method, you must first have the `redis` and `connection-pool` gems installed. Then,
+      # put the object returned by this method into the `feature_store` property of your
+      # client configuration ({LaunchDarkly::Config}).
+      #
+      # @param opts [Hash] the configuration options
+      # @option opts [String] :redis_url (default_redis_url)  URL of the Redis instance (shortcut for omitting `redis_opts`)
+      # @option opts [Hash] :redis_opts  options to pass to the Redis constructor (if you want to specify more than just `redis_url`)
+      # @option opts [String] :prefix (default_prefix)  namespace prefix to add to all hash keys used by LaunchDarkly
+      # @option opts [Logger] :logger  a `Logger` instance; defaults to `Config.default_logger`
+      # @option opts [Integer] :max_connections  size of the Redis connection pool
+      # @option opts [Integer] :expiration_seconds (15)  expiration time for the in-memory cache, in seconds; 0 for no local caching
+      # @option opts [Integer] :capacity (1000)  maximum number of items in the cache
+      # @option opts [Object] :pool  custom connection pool, if desired
+      # @return [LaunchDarkly::Interfaces::FeatureStore]  a feature store object
+      #
+      def self.new_feature_store(opts)
+        return RedisFeatureStore.new(opts)
+      end
+    end
+  end
+end