Refactor batching of exposure events

Author: Strech

lib/datadog/open_feature/component.rb

@@ -33,7 +33,7 @@ def initialize(settings, agent_settings, logger:, telemetry:)
         @telemetry = telemetry
 
         transport = Transport::HTTP.exposures(agent_settings: agent_settings, logger: logger)
-        @worker = Exposures::Worker.new(transport: transport, logger: logger)
+        @worker = Exposures::Worker.new(settings: settings, transport: transport, logger: logger)
         @reporter = Exposures::Reporter.new(@worker, telemetry: telemetry, logger: logger)
         @engine = EvaluationEngine.new(@reporter, telemetry: telemetry, logger: logger)
       end

lib/datadog/open_feature/exposures.rb

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

lib/datadog/open_feature/exposures/batch.rb

@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+module Datadog
+  module OpenFeature
+    module Exposures
+      # This class builds a batch of exposures and context to be sent to the Agent
+      class BatchBuilder
+        def initialize(settings)
+          @context = build_context(settings)
+        end
+
+        def payload_for(events)
+          {
+            context: @context,
+            exposures: events.map(&:to_h)
+          }
+        end
+
+        private
+
+        def build_context(settings)
+          env = extract_env(settings)
+          service = extract_service(settings)
+          version = extract_version(settings)
+
+          context = {}
+          context[:env] = env if env
+          context[:service] = service if service
+          context[:version] = version if version
+
+          context
+        end
+
+        def extract_env(settings)
+          return settings.env if settings.respond_to?(:env)
+          return settings.tags['env'] if settings.respond_to?(:tags)
+
+          nil
+        end
+
+        def extract_service(settings)
+          return settings.service if settings.respond_to?(:service)
+          return settings.tags['service'] if settings.respond_to?(:tags)
+
+          nil
+        end
+
+        def extract_version(settings)
+          return settings.version if settings.respond_to?(:version)
+          return settings.tags['version'] if settings.respond_to?(:tags)
+
+          nil
+        end
+      end
+    end
+  end
+end

lib/datadog/open_feature/exposures/batch_builder.rb

@@ -6,7 +6,7 @@ module Datadog
   module OpenFeature
     module Exposures
       module Models
-        # A data model for an exposure event.
+        # A data model for an exposure event
         class Event
           TARGETING_KEY_FIELD = 'targeting_key'
           ALLOWED_FIELD_TYPES = [
@@ -95,8 +95,6 @@ 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

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

@@ -1,14 +1,15 @@
 # frozen_string_literal: true
 
+require_relative '../../core/workers/queue'
 require_relative '../../core/workers/polling'
 
-require_relative 'batch'
 require_relative 'buffer'
-require_relative 'context'
+require_relative 'batch_builder'
 
 module Datadog
   module OpenFeature
     module Exposures
+      # This class is responsible for sending exposures to the Agent
       class Worker
         include Datadog::Core::Workers::Queue
         include Datadog::Core::Workers::Polling
@@ -18,17 +19,16 @@ class Worker
 
         attr_reader :logger
 
-        # NOTE: Context builder and the data model is not finished
         def initialize(
+          settings:,
           transport:,
           logger: Datadog.logger,
           flush_interval_seconds: DEFAULT_FLUSH_INTERVAL_SECONDS,
-          buffer_limit: DEFAULT_BUFFER_LIMIT,
-          context_builder: nil
+          buffer_limit: DEFAULT_BUFFER_LIMIT
         )
           @logger = logger
           @transport = transport
-          @context_builder = context_builder || -> { Context.build }
+          @batch_builder = BatchBuilder.new(settings)
           @buffer_limit = buffer_limit
           @flush_mutex = Mutex.new
 
@@ -67,12 +67,12 @@ def dequeue
 
         def flush
           events, dropped = dequeue
-          send_events(events || [], dropped || 0)
+          send_events(events || [], dropped.to_i)
         end
 
         def perform(*args)
           events, dropped = args
-          send_events(events || [], dropped || 0)
+          send_events(events || [], dropped.to_i)
         end
 
         private
@@ -84,7 +84,7 @@ def send_events(events, dropped)
             logger.debug { "OpenFeature: Exposure worker dropped #{dropped} event(s) due to full buffer" }
           end
 
-          payload = Batch.new(context: @context_builder.call, exposures: events).to_h
+          payload = @batch_builder.payload_for(events)
           send_payload(payload)
         end
 

lib/datadog/open_feature/exposures/worker.rb

@@ -0,0 +1,24 @@
+module Datadog
+  module OpenFeature
+    module Exposures
+      class BatchBuilder
+        @context: ::Hash[::Symbol, ::String]
+
+        def initialize: (Core::Configuration::Settings settings) -> void
+
+        def payload_for: (::Array[Models::Event] events) -> ::Hash[::Symbol, untyped]
+
+        private
+
+        def build_context: (Core::Configuration::Settings settings) -> ::Hash[::Symbol, ::String]
+
+        def extract_env: (Core::Configuration::Settings settings) -> ::String?
+
+        def extract_service: (Core::Configuration::Settings settings) -> ::String?
+
+        def extract_version: (Core::Configuration::Settings settings) -> ::String?
+      end
+    end
+  end
+end
+

sig/datadog/open_feature/exposures/batch.rbs

@@ -8,14 +8,14 @@ module Datadog
         DEFAULT_FLUSH_INTERVAL_SECONDS: ::Integer
         DEFAULT_BUFFER_LIMIT: ::Integer
 
-        attr_reader logger: ::Datadog::Core::Logger
+        attr_reader logger: Core::Logger
 
         def initialize: (
+          settings: Core::Configuration::Settings,
           transport: Transport::Exposures::Transport,
           logger: Core::Logger,
           ?flush_interval_seconds: ::Integer,
-          ?buffer_limit: ::Integer,
-          ?context_builder: (^() -> ::Hash[::Symbol, untyped])
+          ?buffer_limit: ::Integer
         ) -> void
 
         def start: () -> void