Merge pull request #447 from splitio/development

sanzmauro · web-flow · commit bd1473c3fc85 · 2023-03-20T18:24:59.000-03:00
[8.1.1] Development into master
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -1,5 +1,10 @@
 CHANGES
 
+8.1.1 (Mar 17, 2023)
+- Added retries with backoff when the sdk tries to connect to the Streaming service and it is not available.
+- Updated the way that the sdk write mtks in redis.
+- Fixed calculation of timeUntilReady in telemetry.
+
 8.1.0 (Oct 5, 2022)
 - Added a new impressions mode for the SDK called NONE , to be used in factory when there is no desire to capture impressions on an SDK factory to feed Split's analytics engine. Running NONE mode, the SDK will only capture unique keys evaluated for a particular feature flag instead of full blown impressions.
 
diff --git a/LICENSE b/LICENSE
@@ -1,4 +1,4 @@
-Copyright © 2022 Split Software, Inc.
+Copyright © 2023 Split Software, Inc.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
diff --git a/lib/splitclient-rb/cache/adapters/redis_adapter.rb b/lib/splitclient-rb/cache/adapters/redis_adapter.rb
@@ -99,7 +99,7 @@ def bool(key)
         alias find_sets_by_prefix find_strings_by_prefix
 
         def add_to_set(key, val)
-          @redis.sadd(key, val)
+          @redis.sadd?(key, val)
         end
 
         def delete_from_set(key, val)
diff --git a/lib/splitclient-rb/cache/senders/impressions_adapter/redis_sender.rb b/lib/splitclient-rb/cache/senders/impressions_adapter/redis_sender.rb
@@ -12,14 +12,11 @@ def initialize(config)
         end
 
         def record_uniques_key(uniques)
-          return if uniques.nil? || uniques == {}
+          return if uniques.nil? || uniques.empty?
 
-          formatted = uniques_formatter(uniques).to_json
+          size = @adapter.add_to_queue(unique_keys_key, uniques_formatter(uniques))
 
-          unless formatted.nil?
-            size = @adapter.add_to_queue(unique_keys_key, formatted)
-            @adapter.expire(unique_keys_key, EXPIRE_SECONDS) if formatted.size == size
-          end
+          @adapter.expire(unique_keys_key, EXPIRE_SECONDS) if uniques.length == size
         rescue StandardError => e
           @config.log_found_exception(__method__.to_s, e)
         end
@@ -47,29 +44,24 @@ def expire_impressions_count_key(impressions_count, pipeline_result)
           @adapter.expire(impressions_count_key, EXPIRE_SECONDS) if impressions_count.size == hlen && (pipeline_result.sum - hlen) == total_count
         end
 
-        def impressions_count_key
-          "#{@config.redis_namespace}.impressions.count"
-        end
-
-        def unique_keys_key
-          "#{@config.redis_namespace}.uniquekeys"
-        end
-
         def uniques_formatter(uniques)
-          return if uniques.nil? || uniques.empty?
-
           to_return = []
           uniques.each do |key, value|
             to_return << {
               f: key,
-              k: value.to_a
-            }
+              ks: value.to_a
+            }.to_json
           end
 
           to_return
-        rescue StandardError => e
-          @config.log_found_exception(__method__.to_s, e)
-          nil
+        end
+
+        def impressions_count_key
+          "#{@config.redis_namespace}.impressions.count"
+        end
+
+        def unique_keys_key
+          "#{@config.redis_namespace}.uniquekeys"
         end
       end
     end
diff --git a/lib/splitclient-rb/engine/auth_api_client.rb b/lib/splitclient-rb/engine/auth_api_client.rb
@@ -59,11 +59,7 @@ def process_error(response)
 
       def process_success(response, start)
         @config.logger.debug("Success connection to: #{@config.auth_service_url}") if @config.debug_enabled
-
-        bucket = BinarySearchLatencyTracker.get_bucket((Time.now - start) * 1000.0)
-        @telemetry_runtime_producer.record_sync_latency(Telemetry::Domain::Constants::TOKEN_SYNC, bucket)
-        timestamp = (Time.now.to_f * 1000.0).to_i
-        @telemetry_runtime_producer.record_successful_sync(Telemetry::Domain::Constants::TOKEN_SYNC, timestamp)
+        record_telemetry(start)
 
         body_json = JSON.parse(response.body, symbolize_names: true)
         push_enabled = body_json[:pushEnabled]
@@ -77,7 +73,7 @@ def process_success(response, start)
           @telemetry_runtime_producer.record_token_refreshes
         end
 
-        { push_enabled: push_enabled, token: token, channels: channels, exp: exp, retry: false }
+        { push_enabled: push_enabled, token: token, channels: channels, exp: exp, retry: true }
       end
 
       def control_channels(channels_string)
@@ -88,6 +84,13 @@ def control_channels(channels_string)
 
         channels_string.gsub(control_sec, "#{prefix}#{control_sec}")
       end
+
+      def record_telemetry(start)
+        bucket = BinarySearchLatencyTracker.get_bucket((Time.now - start) * 1000.0)
+        @telemetry_runtime_producer.record_sync_latency(Telemetry::Domain::Constants::TOKEN_SYNC, bucket)
+        timestamp = (Time.now.to_f * 1000.0).to_i
+        @telemetry_runtime_producer.record_successful_sync(Telemetry::Domain::Constants::TOKEN_SYNC, timestamp)
+      end
     end
   end
 end
diff --git a/lib/splitclient-rb/engine/back_off.rb b/lib/splitclient-rb/engine/back_off.rb
@@ -2,7 +2,7 @@
 
 module SplitIoClient
   module Engine
-    BACKOFF_MAX_ALLOWED = 1.8
+    BACKOFF_MAX_ALLOWED = 1800
     class BackOff
       def initialize(back_off_base, attempt = 0, max_allowed = BACKOFF_MAX_ALLOWED)
         @attempt = attempt
diff --git a/lib/splitclient-rb/engine/push_manager.rb b/lib/splitclient-rb/engine/push_manager.rb
@@ -14,7 +14,6 @@ def initialize(config, sse_handler, api_key, telemetry_runtime_producer)
 
       def start_sse
         response = @auth_api_client.authenticate(@api_key)
-
         @config.logger.debug("Auth service response push_enabled: #{response[:push_enabled]}") if @config.debug_enabled
 
         if response[:push_enabled] && @sse_handler.start(response[:token], response[:channels])
@@ -26,7 +25,6 @@ def start_sse
         end
 
         stop_sse
-
         schedule_next_token_refresh(@back_off.interval) if response[:retry]
         false
       rescue StandardError => e
@@ -35,7 +33,8 @@ def start_sse
 
       def stop_sse
         @sse_handler.stop
-        SplitIoClient::Helpers::ThreadHelper.stop(:schedule_next_token_refresh, @config)
+      rescue StandardError => e
+        @config.logger.error(e.inspect)
       end
 
       private
diff --git a/lib/splitclient-rb/engine/synchronizer.rb b/lib/splitclient-rb/engine/synchronizer.rb
@@ -6,8 +6,6 @@ class Synchronizer
       include SplitIoClient::Cache::Fetchers
       include SplitIoClient::Cache::Senders
 
-      ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS = 10
-      ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS = 60
       ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES = 10
 
       def initialize(
@@ -27,6 +25,9 @@ def initialize(
         @telemetry_synchronizer = params[:telemetry_synchronizer]
         @impressions_sender_adapter = params[:impressions_sender_adapter]
         @unique_keys_tracker = params[:unique_keys_tracker]
+
+        @splits_sync_backoff = Engine::BackOff.new(10, 0, 60)
+        @segments_sync_backoff = Engine::BackOff.new(10, 0, 60)
       end
 
       def sync_all(asynchronous = true)
@@ -140,7 +141,7 @@ def fetch_segment(name, target_change_number)
 
       def attempt_segment_sync(name, target_cn, fetch_options, max_retries, retry_delay_seconds, with_backoff)
         remaining_attempts = max_retries
-        backoff = Engine::BackOff.new(ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS, 0, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS) if with_backoff
+        @segments_sync_backoff.reset
 
         loop do
           remaining_attempts -= 1
@@ -150,14 +151,14 @@ def attempt_segment_sync(name, target_cn, fetch_options, max_retries, retry_dela
           return sync_result(true, remaining_attempts) if target_cn <= @segments_repository.get_change_number(name).to_i
           return sync_result(false, remaining_attempts) if remaining_attempts <= 0
 
-          delay = with_backoff ? backoff.interval : retry_delay_seconds
+          delay = with_backoff ? @segments_sync_backoff.interval : retry_delay_seconds
           sleep(delay)
         end
       end
 
       def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_seconds, with_backoff)
         remaining_attempts = max_retries
-        backoff = Engine::BackOff.new(ON_DEMAND_FETCH_BACKOFF_BASE_SECONDS, 0, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_SECONDS) if with_backoff
+        @splits_sync_backoff.reset
 
         loop do
           remaining_attempts -= 1
@@ -167,7 +168,7 @@ def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_secon
           return sync_result(true, remaining_attempts, result[:segment_names]) if target_cn <= @splits_repository.get_change_number
           return sync_result(false, remaining_attempts, result[:segment_names]) if remaining_attempts <= 0
 
-          delay = with_backoff ? backoff.interval : retry_delay_seconds
+          delay = with_backoff ? @splits_sync_backoff.interval : retry_delay_seconds
           sleep(delay)
         end
       end
diff --git a/lib/splitclient-rb/split_config.rb b/lib/splitclient-rb/split_config.rb
@@ -117,7 +117,7 @@ def initialize(opts = {})
 
       @counter_refresh_rate = SplitConfig.default_counter_refresh_rate(@cache_adapter)
 
-      @sdk_start_time = Time.now
+      @sdk_start_time = Time.now.to_f
 
       @on_demand_fetch_retry_delay_seconds = SplitConfig.default_on_demand_fetch_retry_delay_seconds
       @on_demand_fetch_max_retries = SplitConfig.default_on_demand_fetch_max_retries
diff --git a/lib/splitclient-rb/telemetry/domain/structs.rb b/lib/splitclient-rb/telemetry/domain/structs.rb
@@ -13,9 +13,10 @@ module Telemetry
     # s: sdkUrl, e: eventsUrl, a: authUrl, st: streamUrl, t: telemetryUrl
     UrlOverrides = Struct.new(:s, :e, :a, :st, :t)
 
-    # om: operationMode, se: streamingEnabled, st: storage, rr: refreshRate, uo: urlOverrides, iq: impressionsQueueSize,
-    # eq: eventsQueueSize, im: impressionsMode, il: impressionListenerEnabled, hp: httpProxyDetected, af: activeFactories,
-    # rf: redundantActiveFactories, tr: timeUntilSdkReady, bt: burTimeouts, nr: sdkNotReadyUsage, t: tags, i: integrations
+    # om: operationMode, st: storage, af: activeFactories, rf: redundantActiveFactories, t: tags, se: streamingEnabled,
+    # rr: refreshRate, uo: urlOverrides, iq: impressionsQueueSize, eq: eventsQueueSize, im: impressionsMode,
+    # il: impressionListenerEnabled, hp: httpProxyDetected, tr: timeUntilSdkReady, bt: burTimeouts,
+    # nr: sdkNotReadyUsage, i: integrations
     ConfigInit = Struct.new(:om, :st, :af, :rf, :t, :se, :rr, :uo, :iq, :eq, :im, :il, :hp, :tr, :bt, :nr, :i)
 
     # ls: lastSynchronization, ml: clientMethodLatencies, me: clientMethodExceptions, he: httpErros, hl: httpLatencies,
diff --git a/lib/splitclient-rb/telemetry/memory/memory_synchronizer.rb b/lib/splitclient-rb/telemetry/memory/memory_synchronizer.rb
@@ -56,6 +56,7 @@ def synchronize_config(active_factories = nil, redundant_active_factories = nil,
 
         active_factories ||= SplitIoClient.split_factory_registry.active_factories
         redundant_active_factories ||= SplitIoClient.split_factory_registry.redundant_active_factories
+        time_until_ready ||= ((Time.now.to_f - @config.sdk_start_time) * 1000.0).to_i
 
         init_config = ConfigInit.new(mode,
                                      'memory',
@@ -70,7 +71,7 @@ def synchronize_config(active_factories = nil, redundant_active_factories = nil,
                                      impressions_mode,
                                      !@config.impression_listener.nil?,
                                      http_proxy_detected?,
-                                     time_until_ready || Time.now.to_i - @config.sdk_start_time.to_i,
+                                     time_until_ready,
                                      @telemetry_init_consumer.bur_timeouts,
                                      @telemetry_init_consumer.non_ready_usages)
 
diff --git a/lib/splitclient-rb/version.rb b/lib/splitclient-rb/version.rb
@@ -1,3 +1,3 @@
 module SplitIoClient
-  VERSION = '8.1.0'
+  VERSION = '8.1.1'
 end
diff --git a/spec/cache/senders/impressions_sender_adapter_spec.rb b/spec/cache/senders/impressions_sender_adapter_spec.rb
@@ -23,17 +23,19 @@
 
       result = config.cache_adapter.get_from_queue(unique_keys_key, 0)
 
-      expect(result.size).to eq(1)
-      data = JSON.parse(result[0], symbolize_names: true)
+      expect(result.size).to eq(3)
 
-      expect(data[0][:f]).to eq('feature-name-1')
-      expect(data[0][:k].to_s).to eq('["key-1", "key-2", "key-3", "key-4"]')
+      data = JSON.parse(result[0], symbolize_names: true)
+      expect(data[:f]).to eq('feature-name-1')
+      expect(data[:ks].to_s).to eq('["key-1", "key-2", "key-3", "key-4"]')
 
-      expect(data[1][:f]).to eq('feature-name-2')
-      expect(data[1][:k].to_s).to eq('["key-1", "key-2", "key-3", "key-4"]')
+      data1 = JSON.parse(result[1], symbolize_names: true)
+      expect(data1[:f]).to eq('feature-name-2')
+      expect(data1[:ks].to_s).to eq('["key-1", "key-2", "key-3", "key-4"]')
 
-      expect(data[2][:f]).to eq('feature-name-3')
-      expect(data[2][:k].to_s).to eq('["key-1", "key-2", "key-3", "key-4"]')
+      data2 = JSON.parse(result[2], symbolize_names: true)
+      expect(data2[:f]).to eq('feature-name-3')
+      expect(data2[:ks].to_s).to eq('["key-1", "key-2", "key-3", "key-4"]')
     end
 
     it 'record_uniques_key when uniques is nil or empty' do
diff --git a/spec/engine/auth_api_client_spec.rb b/spec/engine/auth_api_client_spec.rb
@@ -22,7 +22,7 @@
 
     expect(response[:push_enabled]).to eq(true)
     expect(response[:channels]).to eq('xxxx_xxxx_segments%2Cxxxx_xxxx_splits%2C%5B%3Foccupancy%3Dmetrics.publishers%5Dcontrol_pri%2C%5B%3Foccupancy%3Dmetrics.publishers%5Dcontrol_sec')
-    expect(response[:retry]).to eq(false)
+    expect(response[:retry]).to eq(true)
   end
 
   it 'auth server return 500' do
diff --git a/spec/engine/impressions/redis_unique_keys_tracker_spec.rb b/spec/engine/impressions/redis_unique_keys_tracker_spec.rb
@@ -36,7 +36,7 @@
     end
 
     result = config.cache_adapter.get_from_queue(key, 0)
-    expect(result.size).to eq(10)
+    expect(result.size).to eq(20)
 
     10.times do |i|
       data = JSON.parse(result[i], symbolize_names: true)
@@ -63,11 +63,11 @@
     sleep 1
 
     result = config.cache_adapter.get_from_queue(key, 0)
-    expect(result.size).to eq(1)
+    expect(result.size).to eq(10)
 
     10.times do |i|
-      data = JSON.parse(result[0], symbolize_names: true)
-      expect(data.size).to eq(10)
+      data = JSON.parse(result[i], symbolize_names: true)
+      expect(data[:ks].size).to eq(2)
     end
 
     cache.clear
diff --git a/spec/engine/sync_manager_spec.rb b/spec/engine/sync_manager_spec.rb
@@ -103,7 +103,7 @@
       sleep(2)
       expect(a_request(:get, 'https://sdk.split.io/api/splitChanges?since=-1')).to have_been_made.once
 
-      expect(config.threads.size).to eq(7)
+      expect(config.threads.size).to eq(8)
     end
   end
 
diff --git a/splitclient-rb.gemspec b/splitclient-rb.gemspec
@@ -57,7 +57,7 @@ Gem::Specification.new do |spec|
   spec.add_runtime_dependency 'jwt', '>= 1.0.0', '< 3.0'
   spec.add_runtime_dependency 'lru_redux', '~> 1.1'
   spec.add_runtime_dependency 'net-http-persistent', '>= 2.9', '< 5.0'
-  spec.add_runtime_dependency 'redis', '>= 4.0.0', '< 5.0'
+  spec.add_runtime_dependency 'redis', '>= 4.0.0', '< 6.0'
   spec.add_runtime_dependency 'socketry', '>= 0.4', '< 1.0'
   spec.add_runtime_dependency 'thread_safe', '~> 0.3'
 end

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-Copyright © 2022 Split Software, Inc.`
	`1`	`+Copyright © 2023 Split Software, Inc.`
`2`	`2`
`3`	`3`	`Licensed under the Apache License, Version 2.0 (the "License");`
`4`	`4`	`you may not use this file except in compliance with the License.`