Rework Reporter and the data model

Author: Strech

lib/datadog/open_feature/component.rb

@@ -34,16 +34,13 @@ def initialize(settings, agent_settings, logger:, telemetry:)
 
         transport = Transport::HTTP.exposures(agent_settings: agent_settings, logger: logger)
         @worker = Exposures::Worker.new(transport: transport, logger: logger)
-        @reporter = Exposures::Reporter.new(worker: @worker, logger: logger)
+        @reporter = Exposures::Reporter.new(@worker, telemetry: telemetry, logger: logger)
         @engine = EvaluationEngine.new(@reporter, telemetry: telemetry, logger: logger)
       end
 
       def shutdown!
-        @reporter&.flush
-        return unless defined?(@worker) && @worker
-
-        @worker.flush
-        @worker.stop(true)
+        @worker&.flush
+        @worker&.stop(true)
       end
     end
   end

lib/datadog/open_feature/exposures.rb

@@ -7,7 +7,6 @@ module Exposures
   end
 end
 
-require_relative 'exposures/event'
 require_relative 'exposures/context'
 require_relative 'exposures/batch'
 require_relative 'exposures/buffer'

lib/datadog/open_feature/exposures/deduplicator.rb

@@ -1,8 +1,7 @@
 # frozen_string_literal: true
 
 require 'zlib'
-
-require 'datadog/core/utils/lru_cache'
+require_relative '../../core/utils/lru_cache'
 
 module Datadog
   module OpenFeature
@@ -15,9 +14,9 @@ def initialize(limit: DEFAULT_CACHE_LIMIT)
           @mutex = Mutex.new
         end
 
-        def duplicate?(flag_key, targeting_key, allocation_key:, variation_key:)
-          cache_key = digest(flag_key, targeting_key)
-          cache_digest = digest(allocation_key, variation_key)
+        def duplicate?(event)
+          cache_key = digest(event.flag_key, event.targeting_key)
+          cache_digest = digest(event.allocation_key, event.variation_key)
 
           stored = @cache[cache_key]
           return true if stored == cache_digest

lib/datadog/open_feature/exposures/event.rb

@@ -0,0 +1,12 @@
+# frozen_string_literal: true
+
+module Datadog
+  module OpenFeature
+    module Exposures
+      module Models
+      end
+    end
+  end
+end
+
+require_relative 'models/event'

lib/datadog/open_feature/exposures/models.rb

@@ -0,0 +1,107 @@
+# frozen_string_literal: true
+
+require_relative '../../../core/utils/time'
+
+module Datadog
+  module OpenFeature
+    module Exposures
+      module Models
+        # A data model for an exposure event.
+        class Event
+          TARGETING_KEY_FIELD = 'targeting_key'
+          ALLOWED_FIELD_TYPES = [
+            String,
+            Integer,
+            Float,
+            TrueClass,
+            FalseClass
+          ].freeze
+
+          # NOTE: The result is a Hash-like structure like this
+          #
+          # {
+          #     "flag": "boolean-one-of-matches",
+          #     "variationType": "INTEGER",
+          #     "defaultValue": 0,
+          #     "targetingKey": "haley",
+          #     "attributes": {
+          #       "not_matches_flag": "False"
+          #     },
+          #     "result": {
+          #       "value": 4,
+          #       "variant": "4",
+          #       "flagMetadata": {
+          #         "allocationKey": "4-for-not-matches",
+          #         "variationType": "number",
+          #         "doLog": true
+          #       }
+          #     }
+          # }
+          class << self
+            def build(result, context:)
+              payload = {
+                timestamp: current_timestamp_ms,
+                allocation: {
+                  key: result.dig('result', 'flagMetadata', 'allocationKey').to_s
+                },
+                flag: {
+                  key: result['flag'].to_s
+                },
+                variant: {
+                  key: result.dig('result', 'variant').to_s
+                },
+                subject: {
+                  id: result['targetingKey'].to_s,
+                  attributes: extract_attributes(context)
+                }
+              }
+
+              new(payload)
+            end
+
+            private
+
+            def extract_attributes(context)
+              context.fields.select do |key, value|
+                next false if key == TARGETING_KEY_FIELD
+                next true if ALLOWED_FIELD_TYPES.include?(value.class)
+
+                false
+              end
+            end
+
+            def current_timestamp_ms
+              (Datadog::Core::Utils::Time.now.to_f * 1000).to_i
+            end
+          end
+
+          def initialize(payload)
+            @payload = payload
+          end
+
+          def flag_key
+            @payload.dig(:flag, :key).to_s
+          end
+
+          def targeting_key
+            @payload.dig(:subject, :id).to_s
+          end
+
+          def allocation_key
+            @payload.dig(:allocation, :key).to_s
+          end
+
+          def variation_key
+            @payload.dig(:variant, :key).to_s
+          end
+
+          # NOTE: The schema is this
+          #       https://github.com/DataDog/dd-source/blob/c10946901aaa103db960883c20161833a664e093/domains/evp-workers/apps/exposures-worker/schemas/exposure.json
+          def to_h
+            @payload
+          end
+        end
+      end
+    end
+  end
+end

lib/datadog/open_feature/exposures/models/event.rb

@@ -1,133 +1,30 @@
 # frozen_string_literal: true
 
-require 'zlib'
-require 'thread'
-
-require 'datadog/appsec/api_security/lru_cache'
-
-require_relative 'event'
+require_relative 'models/event'
+require_relative 'deduplicator'
 
 module Datadog
   module OpenFeature
     module Exposures
       class Reporter
-        DEFAULT_CACHE_LIMIT = 1_000
-
-        def initialize(worker:, cache: nil, logger: Datadog.logger, time_provider: Time)
+        def initialize(worker, telemetry:, logger: Datadog.logger)
           @worker = worker
           @logger = logger
-          @time_provider = time_provider
-          @cache = cache || Datadog::AppSec::APISecurity::LRUCache.new(DEFAULT_CACHE_LIMIT)
-          @cache_mutex = Mutex.new
+          @telemetry = telemetry
+          @deduplicator = Deduplicator.new
         end
 
-        def report(result:, context: nil)
-          payload = normalize(result, context)
-          return false if payload.nil?
-
-          cache_key = payload[:cache_key]
-          digest = payload[:digest]
-
-          return false if cache_key && digest && duplicate?(cache_key, digest)
+        def report(result, context:)
+          return false unless result.dig('result', 'flagMetadata', 'doLog')
 
-          event = build_event(payload)
-          return false if event.nil?
+          event = Models::Event.build(result, context: context)
+          return false if @deduplicator.duplicate?(event)
 
           @worker.enqueue(event)
         rescue => e
           @logger.debug { "OpenFeature: Reporter failed to enqueue exposure: #{e.class}: #{e.message}" }
-          false
-        end
-
-        def flush
-          @cache_mutex.synchronize { @cache.clear }
-        end
-
-        private
-
-        def duplicate?(cache_key, digest)
-          @cache_mutex.synchronize do
-            stored = @cache[cache_key]
-            return true if stored == digest
-
-            @cache.store(cache_key, digest)
-            false
-          end
-        end
-
-        def normalize(result, context)
-          data = ensure_hash(result)
-          result_data = ensure_hash(data[:result] || data['result'])
-          flag_metadata = ensure_hash(result_data[:flagMetadata] || result_data['flagMetadata'])
-
-          flag_key = data[:flag] || data['flag']
-          subject_id = extract_subject_id(data, context)
-          allocation_key = flag_metadata[:allocationKey] || flag_metadata['allocationKey']
-          evaluation_key = result_data[:variant] || result_data['variant']
-
-          return nil if flag_key.nil? || subject_id.nil?
-
-          variant = evaluation_key || result_data[:value] || result_data['value']
-          variant_key = variant.nil? ? nil : variant.to_s
-          return nil if variant_key.nil?
 
-          {
-            flag_key: flag_key,
-            subject_id: subject_id,
-            subject_type: nil,
-            subject_attributes: extract_attributes(data, context),
-            allocation_key: allocation_key,
-            variant_key: variant_key,
-            cache_key: cache_key(flag_key, subject_id),
-            digest: allocation_key.nil? ? nil : digest(allocation_key, variant_key)
-          }
-        end
-
-        def build_event(payload)
-          Event.new(
-            timestamp: @time_provider.now,
-            allocation_key: payload[:allocation_key],
-            flag_key: payload[:flag_key],
-            variant_key: payload[:variant_key],
-            subject_id: payload[:subject_id],
-            subject_type: payload[:subject_type],
-            subject_attributes: payload[:subject_attributes]
-          )
-        end
-
-        def extract_subject_id(data, context)
-          data[:targetingKey] || data['targetingKey'] || context_value(context, :targeting_key) ||
-            context_value(context, :targetingKey) || context_value(context, :targetingkey)
-        end
-
-        def extract_attributes(data, context)
-          attributes = data[:attributes] || data['attributes']
-          return attributes if attributes.is_a?(Hash)
-
-          context_value(context, :attributes).is_a?(Hash) ? context_value(context, :attributes) : nil
-        end
-
-        def context_value(context, key)
-          case context
-          when Hash
-            context[key] || context[key.to_s]
-          else
-            context.respond_to?(key) ? context.public_send(key) : nil
-          end
-        end
-
-        def ensure_hash(value)
-          value.is_a?(Hash) ? value : {}
-        end
-
-        def cache_key(flag_key, subject_id)
-          return nil if flag_key.nil? || subject_id.nil?
-
-          "#{flag_key}:#{subject_id}"
-        end
-
-        def digest(allocation_key, evaluation_key)
-          Zlib.crc32("#{allocation_key}:#{evaluation_key}")
+          false
         end
       end
     end