feat: add odp flush interval to sdk_settings (#324)

* add odp flush interval to sdk_settings

Author: andrewleap-optimizely

lib/optimizely.rb

@@ -1203,6 +1203,7 @@ def setup_odp!(sdk_key)
         segments_cache: @sdk_settings.odp_segments_cache,
         fetch_segments_timeout: @sdk_settings.fetch_segments_timeout,
         odp_event_timeout: @sdk_settings.odp_event_timeout,
+        odp_flush_interval: @sdk_settings.odp_flush_interval,
         logger: @logger
       )
 

lib/optimizely/helpers/sdk_settings.rb

@@ -21,7 +21,8 @@
 module Optimizely
   module Helpers
     class OptimizelySdkSettings
-      attr_accessor :odp_disabled, :segments_cache_size, :segments_cache_timeout_in_secs, :odp_segments_cache, :odp_segment_manager, :odp_event_manager, :fetch_segments_timeout, :odp_event_timeout
+      attr_accessor :odp_disabled, :segments_cache_size, :segments_cache_timeout_in_secs, :odp_segments_cache, :odp_segment_manager,
+                    :odp_event_manager, :fetch_segments_timeout, :odp_event_timeout, :odp_flush_interval
 
       # Contains configuration used for Optimizely Project initialization.
       #
@@ -31,26 +32,29 @@ class OptimizelySdkSettings
       # @param odp_segments_cache - A custom odp segments cache. Required methods include: `save(key, value)`, `lookup(key) -> value`, and `reset()`
       # @param odp_segment_manager - A custom odp segment manager. Required method is: `fetch_qualified_segments(user_key, user_value, options)`.
       # @param odp_event_manager - A custom odp event manager. Required method is: `send_event(type:, action:, identifiers:, data:)`
-      # @param fetch_segments_timeout - The timeout in seconds of to fetch odp segments (optional. default = 10).
-      # @param odp_event_timeout - The timeout in seconds of to send odp events (optional. default = 10).
+      # @param odp_segment_request_timeout - Time to wait in seconds for fetch_qualified_segments (optional. default = 10).
+      # @param odp_event_request_timeout - Time to wait in seconds for send_odp_events (optional. default = 10).
+      # @param odp_event_flush_interval - Time to wait in seconds for odp events to accumulate before sending (optional. default = 1).
       def initialize(
         disable_odp: false,
         segments_cache_size: Constants::ODP_SEGMENTS_CACHE_CONFIG[:DEFAULT_CAPACITY],
         segments_cache_timeout_in_secs: Constants::ODP_SEGMENTS_CACHE_CONFIG[:DEFAULT_TIMEOUT_SECONDS],
         odp_segments_cache: nil,
         odp_segment_manager: nil,
         odp_event_manager: nil,
-        fetch_segments_timeout: nil,
-        odp_event_timeout: nil
+        odp_segment_request_timeout: nil,
+        odp_event_request_timeout: nil,
+        odp_event_flush_interval: nil
       )
         @odp_disabled = disable_odp
         @segments_cache_size = segments_cache_size
         @segments_cache_timeout_in_secs = segments_cache_timeout_in_secs
         @odp_segments_cache = odp_segments_cache
         @odp_segment_manager = odp_segment_manager
         @odp_event_manager = odp_event_manager
-        @fetch_segments_timeout = fetch_segments_timeout
-        @odp_event_timeout = odp_event_timeout
+        @fetch_segments_timeout = odp_segment_request_timeout
+        @odp_event_timeout = odp_event_request_timeout
+        @odp_flush_interval = odp_event_flush_interval
       end
     end
   end

lib/optimizely/odp/odp_event_manager.rb

@@ -34,7 +34,8 @@ def initialize(
       api_manager: nil,
       logger: NoOpLogger.new,
       proxy_config: nil,
-      timeout: nil
+      request_timeout: nil,
+      flush_interval: nil
     )
       super()
 
@@ -48,9 +49,9 @@ def initialize(
       # received signal should be sent after adding item to event_queue
       @received = ConditionVariable.new
       @logger = logger
-      @api_manager = api_manager || OdpEventApiManager.new(logger: @logger, proxy_config: proxy_config, timeout: timeout)
-      @batch_size = Helpers::Constants::ODP_EVENT_MANAGER[:DEFAULT_BATCH_SIZE]
-      @flush_interval = Helpers::Constants::ODP_EVENT_MANAGER[:DEFAULT_FLUSH_INTERVAL_SECONDS]
+      @api_manager = api_manager || OdpEventApiManager.new(logger: @logger, proxy_config: proxy_config, timeout: request_timeout)
+      @flush_interval = flush_interval || Helpers::Constants::ODP_EVENT_MANAGER[:DEFAULT_FLUSH_INTERVAL_SECONDS]
+      @batch_size = @flush_interval&.zero? ? 1 : Helpers::Constants::ODP_EVENT_MANAGER[:DEFAULT_BATCH_SIZE]
       @flush_deadline = 0
       @retry_count = Helpers::Constants::ODP_EVENT_MANAGER[:DEFAULT_RETRY_COUNT]
       # current_batch should only be accessed by processing thread

lib/optimizely/odp/odp_manager.rb

@@ -32,7 +32,16 @@ class OdpManager
     ODP_CONFIG_STATE = Helpers::Constants::ODP_CONFIG_STATE
 
     # update_odp_config must be called to complete initialization
-    def initialize(disable:, segments_cache: nil, segment_manager: nil, event_manager: nil, fetch_segments_timeout: nil, odp_event_timeout: nil, logger: nil)
+    def initialize(
+      disable:,
+      segments_cache: nil,
+      segment_manager: nil,
+      event_manager: nil,
+      fetch_segments_timeout: nil,
+      odp_event_timeout: nil,
+      odp_flush_interval: nil,
+      logger: nil
+    )
       @enabled = !disable
       @segment_manager = segment_manager
       @event_manager = event_manager
@@ -52,7 +61,7 @@ def initialize(disable:, segments_cache: nil, segment_manager: nil, event_manage
         @segment_manager = Optimizely::OdpSegmentManager.new(segments_cache, nil, @logger, timeout: fetch_segments_timeout)
       end
 
-      @event_manager ||= Optimizely::OdpEventManager.new(logger: @logger, timeout: odp_event_timeout)
+      @event_manager ||= Optimizely::OdpEventManager.new(logger: @logger, request_timeout: odp_event_timeout, flush_interval: odp_flush_interval)
 
       @segment_manager.odp_config = @odp_config
     end

spec/odp/odp_event_manager_spec.rb

@@ -368,7 +368,7 @@
       event_manager.stop!
     end
 
-    it 'should flush when timeout is reached' do
+    it 'should flush when flush interval is reached' do
       allow(SecureRandom).to receive(:uuid).and_return(test_uuid)
       event_manager = Optimizely::OdpEventManager.new(logger: spy_logger)
       allow(event_manager.api_manager).to receive(:send_odp_events).once.with(api_key, api_host, odp_events).and_return(false)
@@ -385,6 +385,22 @@
       event_manager.stop!
     end
 
+    it 'should flush when flush interval is zero' do
+      allow(SecureRandom).to receive(:uuid).and_return(test_uuid)
+      event_manager = Optimizely::OdpEventManager.new(logger: spy_logger)
+      allow(event_manager.api_manager).to receive(:send_odp_events).once.with(api_key, api_host, odp_events).and_return(false)
+      event_manager.instance_variable_set('@flush_interval', 0.0)
+      event_manager.start!(odp_config)
+
+      event_manager.send_event(**events[0])
+      event_manager.send_event(**events[1])
+      sleep(0.1) until event_manager.instance_variable_get('@event_queue').empty?
+
+      expect(spy_logger).not_to have_received(:log).with(Logger::ERROR, anything)
+      expect(spy_logger).to have_received(:log).once.with(Logger::DEBUG, 'ODP event queue: flushing on interval.')
+      event_manager.stop!
+    end
+
     it 'should discard events received before datafile is ready and process normally' do
       allow(SecureRandom).to receive(:uuid).and_return(test_uuid)
       odp_config = Optimizely::OdpConfig.new

spec/project_spec.rb

@@ -4463,6 +4463,30 @@ def callback(_args); end
       expect(spy_logger).to have_received(:log).once.with(Logger::INFO, 'ODP is not enabled.')
     end
 
+    it 'should accept zero for flush interval' do
+      stub_request(:get, "https://cdn.optimizely.com/datafiles/#{sdk_key}.json")
+        .to_return(status: 200, body: config_body_integrations_JSON)
+      sdk_settings = Optimizely::Helpers::OptimizelySdkSettings.new(odp_event_flush_interval: 0)
+      project = Optimizely::Project.new(nil, nil, spy_logger, error_handler, false, nil, sdk_key, nil, nil, nil, [], sdk_settings)
+      event_manager = project.odp_manager.instance_variable_get('@event_manager')
+      expect(event_manager.instance_variable_get('@flush_interval')).to eq 0
+      project.close
+
+      expect(spy_logger).not_to have_received(:log).with(Logger::ERROR, anything)
+    end
+
+    it 'should use default for flush interval when nil' do
+      stub_request(:get, "https://cdn.optimizely.com/datafiles/#{sdk_key}.json")
+        .to_return(status: 200, body: config_body_integrations_JSON)
+      sdk_settings = Optimizely::Helpers::OptimizelySdkSettings.new(odp_event_flush_interval: nil)
+      project = Optimizely::Project.new(nil, nil, spy_logger, error_handler, false, nil, sdk_key, nil, nil, nil, [], sdk_settings)
+      event_manager = project.odp_manager.instance_variable_get('@event_manager')
+      expect(event_manager.instance_variable_get('@flush_interval')).to eq 1
+      project.close
+
+      expect(spy_logger).not_to have_received(:log).with(Logger::ERROR, anything)
+    end
+
     it 'should accept cache_size' do
       stub_request(:get, "https://cdn.optimizely.com/datafiles/#{sdk_key}.json")
         .to_return(status: 200, body: config_body_integrations_JSON)
@@ -4502,6 +4526,34 @@ def callback(_args); end
       expect(spy_logger).not_to have_received(:log).with(Logger::ERROR, anything)
     end
 
+    it 'should use default cache_size and cache_timeout when not provided' do
+      stub_request(:get, "https://cdn.optimizely.com/datafiles/#{sdk_key}.json")
+        .to_return(status: 200, body: config_body_integrations_JSON)
+      sdk_settings = Optimizely::Helpers::OptimizelySdkSettings.new
+      project = Optimizely::Project.new(nil, nil, spy_logger, error_handler, false, nil, sdk_key, nil, nil, nil, [], sdk_settings)
+      segment_manager = project.odp_manager.instance_variable_get('@segment_manager')
+      segments_cache = segment_manager.instance_variable_get('@segments_cache')
+      expect(segments_cache.capacity).to eq 10_000
+      expect(segments_cache.timeout).to eq 600
+      project.close
+
+      expect(spy_logger).not_to have_received(:log).with(Logger::ERROR, anything)
+    end
+
+    it 'should accept zero cache_size and cache_timeout' do
+      stub_request(:get, "https://cdn.optimizely.com/datafiles/#{sdk_key}.json")
+        .to_return(status: 200, body: config_body_integrations_JSON)
+      sdk_settings = Optimizely::Helpers::OptimizelySdkSettings.new(segments_cache_size: 0, segments_cache_timeout_in_secs: 0)
+      project = Optimizely::Project.new(nil, nil, spy_logger, error_handler, false, nil, sdk_key, nil, nil, nil, [], sdk_settings)
+      segment_manager = project.odp_manager.instance_variable_get('@segment_manager')
+      segments_cache = segment_manager.instance_variable_get('@segments_cache')
+      expect(segments_cache.capacity).to eq 0
+      expect(segments_cache.timeout).to eq 0
+      project.close
+
+      expect(spy_logger).not_to have_received(:log).with(Logger::ERROR, anything)
+    end
+
     it 'should accept valid custom cache' do
       class CustomCache # rubocop:disable Lint/ConstantDefinitionInBlock
         def reset; end