Impression dedupe support

CHANGES.txt

@@ -1,5 +1,10 @@
 CHANGES
 
+7.2.0 (Sep 25, 2020)
+- Added deduplication logic for impressions data.
+  - Now there are two modes for Impressions when the SDK is in standalone mode, OPTIMIZED (default) that only ships unique impressions and DEBUG for times where you need to send ALL impressions to debug an integration.
+  - Impression listener remains unchanged and will still get all impressions.
+
 7.1.3 (Jul 31, 2020)
 - Updated rake development dependency to ~> 12.3.3.
 

ext/murmurhash/MurmurHash3.java

@@ -191,8 +191,9 @@ private static long getLittleEndianLong(final byte[] data, final int index) {
                 (((long) data[index + 7] & 0xff) << 56);
     }
 
-    public static long[] hash128x64(final byte[] data) {
-        return hash128x64(data, 0, data.length, 0);
+    public static long[] hash128x64(final String data, final long seed) {
+        final byte[] dataBytes = data.getBytes();
+        return hash128x64(dataBytes, 0, dataBytes.length, seed);
     }
 
     /**

lib/splitclient-rb.rb

@@ -13,6 +13,7 @@
 require 'splitclient-rb/cache/fetchers/segment_fetcher'
 require 'splitclient-rb/cache/fetchers/split_fetcher'
 require 'splitclient-rb/cache/hashers/impression_hasher'
+require 'splitclient-rb/cache/observers/impression_observer'
 require 'splitclient-rb/cache/repositories/repository'
 require 'splitclient-rb/cache/repositories/segments_repository'
 require 'splitclient-rb/cache/repositories/splits_repository'
@@ -29,6 +30,7 @@
 require 'splitclient-rb/cache/senders/impressions_sender'
 require 'splitclient-rb/cache/senders/metrics_sender'
 require 'splitclient-rb/cache/senders/events_sender'
+require 'splitclient-rb/cache/senders/impressions_count_sender'
 require 'splitclient-rb/cache/senders/localhost_repo_cleaner'
 require 'splitclient-rb/cache/stores/store_utils'
 require 'splitclient-rb/cache/stores/localhost_split_builder'
@@ -53,6 +55,8 @@
 require 'splitclient-rb/engine/api/segments'
 require 'splitclient-rb/engine/api/splits'
 require 'splitclient-rb/engine/api/events'
+require 'splitclient-rb/engine/common/impressions_counter'
+require 'splitclient-rb/engine/common/impressions_manager'
 require 'splitclient-rb/engine/parser/condition'
 require 'splitclient-rb/engine/parser/partition'
 require 'splitclient-rb/engine/parser/evaluator'

lib/splitclient-rb/cache/hashers/impression_hasher.rb

@@ -10,7 +10,7 @@ def initialize
         end
       end
 
-      def process(impression)        
+      def process(impression)
         impression_data = "#{unknown_if_null(impression[:k])}"
         impression_data << ":#{unknown_if_null(impression[:f])}"
         impression_data << ":#{unknown_if_null(impression[:t])}"

lib/splitclient-rb/cache/observers/impression_observer.rb

@@ -0,0 +1,22 @@
+module SplitIoClient
+  module Observers
+    class ImpressionObserver
+      LAST_SEEN_CACHE_SIZE = 500000
+
+      def initialize
+        @cache = LruRedux::TTL::ThreadSafeCache.new(LAST_SEEN_CACHE_SIZE)
+        @impression_hasher = Hashers::ImpressionHasher.new
+      end
+
+      def test_and_set(impression)
+        return if impression.nil?
+
+        hash = @impression_hasher.process(impression)
+        previous = @cache[hash]
+        @cache[hash] = impression[:m]
+
+        previous.nil? ? nil : [previous, impression[:m]].min
+      end
+    end
+  end
+end

lib/splitclient-rb/cache/repositories/impressions/memory_repository.rb

@@ -10,18 +10,10 @@ def initialize(config)
             @adapter = @config.impressions_adapter
           end
 
-          # Store impression data in the selected adapter
-          def add(matching_key, bucketing_key, split_name, treatment, time)
-            @adapter.add_to_queue(
-              m: metadata,
-              i: impression_data(
-                matching_key,
-                bucketing_key,
-                split_name,
-                treatment,
-                time
-              )
-            )
+          def add_bulk(impressions)
+            impressions.each do |impression|
+              @adapter.add_to_queue(impression)
+            end
           rescue ThreadError # queue is full
             if random_sampler.rand(1..1000) <= 2 # log only 0.2 % of the time
               @config.logger.warn("Dropping impressions. Current size is \
@@ -30,12 +22,6 @@ def add(matching_key, bucketing_key, split_name, treatment, time)
             end
           end
 
-          def add_bulk(key, bucketing_key, treatments, time)
-            treatments.each do |split_name, treatment|
-              add(key, bucketing_key, split_name, treatment, time)
-            end
-          end
-
           def batch
             return [] if @config.impressions_bulk_size.zero?
 

lib/splitclient-rb/cache/repositories/impressions/redis_repository.rb

@@ -12,28 +12,17 @@ def initialize(config)
             @adapter = @config.impressions_adapter
           end
 
-          def add(matching_key, bucketing_key, split_name, treatment, time)
-            add_bulk(matching_key, bucketing_key, { split_name => treatment }, time)
-          end
-
-          def add_bulk(matching_key, bucketing_key, treatments, time)
-            impressions = treatments.map do |split_name, treatment|
-              {
-                m: metadata,
-                i: impression_data(
-                  matching_key,
-                  bucketing_key,
-                  split_name,
-                  treatment,
-                  time
-                )
-              }.to_json
+          def add_bulk(impressions)
+            impressions_json = impressions.map do |impression|
+              impression.to_json
             end
 
-            impressions_list_size = @adapter.add_to_queue(key, impressions)
+            impressions_list_size = @adapter.add_to_queue(key, impressions_json)
 
             # Synchronizer might not be running
-            @adapter.expire(key, EXPIRE_SECONDS) if impressions.size == impressions_list_size
+            @adapter.expire(key, EXPIRE_SECONDS) if impressions_json.size == impressions_list_size
+          rescue StandardError => e
+            @config.logger.error("Exception while add_bulk: #{e}")
           end
 
           def get_impressions(number_of_impressions = 0)

lib/splitclient-rb/cache/repositories/impressions_repository.rb

@@ -6,7 +6,7 @@ module Repositories
       # Repository which forwards impressions interface to the selected adapter
       class ImpressionsRepository < Repository
         extend Forwardable
-        def_delegators :@repository, :add, :add_bulk, :batch, :clear, :empty?
+        def_delegators :@repository, :add_bulk, :batch, :clear, :empty?
 
         def initialize(config)
           super(config)
@@ -17,32 +17,6 @@ def initialize(config)
                        Repositories::Impressions::RedisRepository.new(@config)
                      end
         end
-
-        protected
-
-        def impression_data(matching_key, bucketing_key, split_name, treatment, timestamp)
-          {
-            k: matching_key,
-            b: bucketing_key,
-            f: split_name,
-            t: treatment[:treatment],
-            r: applied_rule(treatment[:label]),
-            c: treatment[:change_number],
-            m: timestamp
-          }
-        end
-
-        def metadata
-          {
-            s: "#{@config.language}-#{@config.version}",
-            i: @config.machine_ip,
-            n: @config.machine_name
-          }
-        end
-
-        def applied_rule(label)
-          @config.labels_enabled ? label : nil
-        end
       end
     end
   end

lib/splitclient-rb/cache/routers/impression_router.rb

@@ -18,31 +18,31 @@ def initialize(config)
       end
     end
 
-    def add(impression)
-      enqueue(impression)
-    end
-
     def add_bulk(impressions)
-      impressions[:split_names].each do |split_name|
-        enqueue(
-          split_name: split_name.to_s,
-          matching_key: impressions[:matching_key],
-          bucketing_key: impressions[:bucketing_key],
-          time: impressions[:time],
-          treatment: {
-            label: impressions[:treatments_labels_change_numbers][split_name.to_sym][:label],
-            treatment: impressions[:treatments_labels_change_numbers][split_name.to_sym][:treatment],
-            change_number: impressions[:treatments_labels_change_numbers][split_name.to_sym][:change_number]
-          },
-          attributes: impressions[:attributes]
-        ) unless impressions[:treatments_labels_change_numbers][split_name.to_sym].nil?
+      impressions.each do |impression|
+        enqueue(impression)
       end
     end
 
     private
 
     def enqueue(impression)
-      @queue.push(impression) if @listener
+      imp = {
+        split_name: impression[:i][:f],
+        matching_key: impression[:i][:k],
+        bucketing_key: impression[:i][:b],
+        time: impression[:i][:m],
+        treatment: {
+          label: impression[:i][:r],
+          treatment: impression[:i][:t],
+          change_number: impression[:i][:c]
+        },
+        previous_time: impression[:i][:pt],
+        attributes: impression[:attributes]
+      }
+      @queue.push(imp) if @listener
+    rescue StandardError => error
+      @config.log_found_exception(__method__.to_s, error)
     end
 
     def router_thread

lib/splitclient-rb/cache/senders/impressions_count_sender.rb

@@ -0,0 +1,73 @@
+# frozen_string_literal: true
+
+module SplitIoClient
+  module Cache
+    module Senders
+      class ImpressionsCountSender
+        COUNTER_REFRESH_RATE_SECONDS = 1800
+
+        def initialize(config, impression_counter, impressions_api)
+          @config = config
+          @impression_counter = impression_counter
+          @impressions_api = impressions_api
+        end
+
+        def call
+          impressions_count_thread
+
+          if defined?(PhusionPassenger)
+            PhusionPassenger.on_event(:starting_worker_process) do |forked|
+              impressions_count_thread if forked
+            end
+          end
+        end
+
+        private
+
+        def impressions_count_thread
+          @config.threads[:impressions_count_sender] = Thread.new do
+            begin
+              @config.logger.info('Starting impressions count service')
+
+              loop do
+                post_impressions_count
+
+                sleep(COUNTER_REFRESH_RATE_SECONDS)
+              end
+            rescue SplitIoClient::SDKShutdownException
+              post_impressions_count
+
+              @config.logger.info('Posting impressions count due to shutdown')
+            end
+          end
+
+          def post_impressions_count
+            @impressions_api.post_count(formatter(@impression_counter.pop_all))
+          rescue StandardError => error
+            @config.log_found_exception(__method__.to_s, error)
+          end
+
+          def formatter(counts)
+            return if counts.empty?
+
+            formated_counts = {pf: []}
+
+            counts.each do |key, value|              
+              key_splited = key.split('::')
+
+              formated_counts[:pf] << {
+                f: key_splited[0].to_s, # feature name
+                m: key_splited[1].to_i, # time frame
+                rc: value # count
+              }
+            end
+
+            formated_counts
+          rescue StandardError => error
+            @config.log_found_exception(__method__.to_s, error)
+          end
+        end
+      end
+    end
+  end
+end

lib/splitclient-rb/cache/senders/impressions_formatter.rb

@@ -17,12 +17,10 @@ def call(fetch_all_impressions, raw_impressions = nil)
 
           formatted_impressions = unique_features(filtered_impressions).each_with_object([]) do |feature, memo|
             feature_impressions = feature_impressions(filtered_impressions, feature)
-            ip = feature_impressions.first[:m][:i]
             current_impressions = current_impressions(feature_impressions)
             memo << {
-              testName: feature.to_sym,
-              keyImpressions: current_impressions,
-              ip: ip
+              f: feature.to_sym,
+              i: current_impressions
             }
           end
 
@@ -40,12 +38,13 @@ def feature_impressions(filtered_impressions, feature)
         def current_impressions(feature_impressions)
           feature_impressions.map do |impression|
             {
-              keyName: impression[:i][:k],
-              treatment: impression[:i][:t],
-              time: impression[:i][:m],
-              bucketingKey: impression[:i][:b],
-              label: impression[:i][:r],
-              changeNumber: impression[:i][:c]
+              k: impression[:i][:k],
+              t: impression[:i][:t],
+              m: impression[:i][:m],
+              b: impression[:i][:b],
+              r: impression[:i][:r],
+              c: impression[:i][:c],
+              pt: impression[:i][:pt]
             }
           end
         end
@@ -73,7 +72,8 @@ def impression_hash(impression)
           "#{impression[:i][:k]}:" \
           "#{impression[:i][:b]}:" \
           "#{impression[:i][:c]}:" \
-          "#{impression[:i][:t]}"
+          "#{impression[:i][:t]}:" \
+          "#{impression[:i][:pt]}"
         end
       end
     end

lib/splitclient-rb/cache/senders/impressions_sender.rb

@@ -4,10 +4,10 @@ module SplitIoClient
   module Cache
     module Senders
       class ImpressionsSender
-        def initialize(impressions_repository, api_key, config)
+        def initialize(impressions_repository, config, impressions_api)
           @impressions_repository = impressions_repository
-          @api_key = api_key
           @config = config
+          @impressions_api = impressions_api
         end
 
         def call
@@ -50,7 +50,7 @@ def post_impressions(fetch_all_impressions = true)
         end
 
         def impressions_api
-          @impressions_api ||= SplitIoClient::Api::Impressions.new(@api_key, @config)
+          @impressions_api
         end
       end
     end