Skip to content

updated engine classes and repo helper #539

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Jan 10, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 18 additions & 20 deletions lib/splitclient-rb/clients/split_client.rb
Original file line number Diff line number Diff line change
Expand Up @@ -267,7 +267,7 @@ def treatments(key, feature_flag_names, attributes = {}, calling_method = 'get_t
to_return = Hash.new
sanitized_feature_flag_names.each {|name|
to_return[name.to_sym] = control_treatment_with_config
impressions << @impressions_manager.build_impression(matching_key, bucketing_key, name.to_sym, control_treatment_with_config.merge({ label: Engine::Models::Label::NOT_READY }), { attributes: attributes, time: nil })
impressions << { :impression => @impressions_manager.build_impression(matching_key, bucketing_key, name.to_sym, control_treatment_with_config.merge({ :label => Engine::Models::Label::NOT_READY }), false, { attributes: attributes, time: nil }), :disabled => false }
}
@impressions_manager.track(impressions)
return to_return
Expand All @@ -278,7 +278,6 @@ def treatments(key, feature_flag_names, attributes = {}, calling_method = 'get_t
valid_feature_flag_names << feature_flag_name unless feature_flag_name.nil?
}
start = Time.now
impressions_total = []

feature_flags = @splits_repository.splits(valid_feature_flag_names)
treatments = Hash.new
Expand All @@ -291,15 +290,14 @@ def treatments(key, feature_flag_names, attributes = {}, calling_method = 'get_t
next
end
treatments_labels_change_numbers, impressions = evaluate_treatment(feature_flag, key, bucketing_key, matching_key, attributes, calling_method)
impressions_total.concat(impressions) unless impressions.nil?
treatments[key] =
{
treatment: treatments_labels_change_numbers[:treatment],
config: treatments_labels_change_numbers[:config]
}
@impressions_manager.track(impressions) unless impressions.empty?
end
record_latency(calling_method, start)
@impressions_manager.track(impressions_total) unless impressions_total.empty?

treatments.merge(invalid_treatments)
end
Expand Down Expand Up @@ -332,50 +330,50 @@ def treatment(key, feature_flag_name, attributes = {}, split_data = nil, store_i
end

feature_flag = @splits_repository.get_split(feature_flag_name)
treatments, impressions = evaluate_treatment(feature_flag, feature_flag_name, bucketing_key, matching_key, attributes, calling_method, multiple)
treatments, impressions_decorator = evaluate_treatment(feature_flag, feature_flag_name, bucketing_key, matching_key, attributes, calling_method, multiple)

@impressions_manager.track(impressions) unless impressions.nil?
@impressions_manager.track(impressions_decorator) unless impressions_decorator.nil?
treatments
end

def evaluate_treatment(feature_flag, feature_flag_name, bucketing_key, matching_key, attributes, calling_method, multiple = false)
impressions = []
impressions_decorator = []
begin
start = Time.now
if feature_flag.nil? && ready?
@config.logger.warn("#{calling_method}: you passed #{feature_flag_name} that " \
'does not exist in this environment, please double check what feature flags exist in the Split user interface')
return parsed_treatment(control_treatment.merge({ label: Engine::Models::Label::NOT_FOUND }), multiple), nil
return parsed_treatment(control_treatment.merge({ :label => Engine::Models::Label::NOT_FOUND }), multiple), nil
end
treatment_data =

if !feature_flag.nil? && ready?
@evaluator.evaluate_feature_flag(
treatment_data = @evaluator.evaluate_feature_flag(
{ bucketing_key: bucketing_key, matching_key: matching_key }, feature_flag, attributes
)
impressions_disabled = feature_flag[:impressionsDisabled]
else
@config.logger.error("#{calling_method}: the SDK is not ready, results may be incorrect for feature flag #{feature_flag_name}. Make sure to wait for SDK readiness before using this method.")
control_treatment.merge({ label: Engine::Models::Label::NOT_READY })
treatment_data = control_treatment.merge({ :label => Engine::Models::Label::NOT_READY })
impressions_disabled = false
end

record_latency(calling_method, start)
impression = @impressions_manager.build_impression(matching_key, bucketing_key, feature_flag_name, treatment_data, { attributes: attributes, time: nil })
impressions << impression unless impression.nil?
impression_decorator = { :impression => @impressions_manager.build_impression(matching_key, bucketing_key, feature_flag_name, treatment_data, impressions_disabled, params={ attributes: attributes, time: nil }), :disabled => impressions_disabled }
impressions_decorator << impression_decorator unless impression_decorator.nil?
rescue StandardError => e
@config.log_found_exception(__method__.to_s, e)

record_exception(calling_method)
impression_decorator = { :impression => @impressions_manager.build_impression(matching_key, bucketing_key, feature_flag_name, control_treatment, false, { attributes: attributes, time: nil }), :disabled => false }
impressions_decorator << impression_decorator unless impression_decorator.nil?

impression = @impressions_manager.build_impression(matching_key, bucketing_key, feature_flag_name, control_treatment, { attributes: attributes, time: nil })
impressions << impression unless impression.nil?

return parsed_treatment(control_treatment.merge({ label: Engine::Models::Label::EXCEPTION }), multiple), impressions
return parsed_treatment(control_treatment.merge({ :label => Engine::Models::Label::EXCEPTION }), multiple), impressions_decorator
end

return parsed_treatment(treatment_data, multiple), impressions
return parsed_treatment(treatment_data, multiple), impressions_decorator
end

def control_treatment
{ treatment: Engine::Models::Treatment::CONTROL }
{ :treatment => Engine::Models::Treatment::CONTROL }
end

def control_treatment_with_config
Expand Down
50 changes: 23 additions & 27 deletions lib/splitclient-rb/engine/common/impressions_manager.rb
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,16 @@ def initialize(config,
@unique_keys_tracker = unique_keys_tracker
end

def build_impression(matching_key, bucketing_key, split_name, treatment, params = {})
impression_data = impression_data(matching_key, bucketing_key, split_name, treatment, params[:time])

def build_impression(matching_key, bucketing_key, split_name, treatment_data, impressions_disabled, params = {})
impression_data = impression_data(matching_key, bucketing_key, split_name, treatment_data, params[:time])
begin
case @config.impressions_mode
when :debug # In DEBUG mode we should calculate the pt only.
impression_data[:pt] = @impression_observer.test_and_set(impression_data)
when :none # In NONE mode we should track the total amount of evaluations and the unique keys.
if @config.impressions_mode == :none || impressions_disabled
@impression_counter.inc(split_name, impression_data[:m])
@unique_keys_tracker.track(split_name, matching_key)
elsif @config.impressions_mode == :debug # In DEBUG mode we should calculate the pt only.
impression_data[:pt] = @impression_observer.test_and_set(impression_data)
else # In OPTIMIZED mode we should track the total amount of evaluations and deduplicate the impressions.
impression_data[:pt] = @impression_observer.test_and_set(impression_data)

@impression_counter.inc(split_name, impression_data[:m]) unless impression_data[:pt].nil?
end
rescue StandardError => e
Expand All @@ -40,24 +37,24 @@ def build_impression(matching_key, bucketing_key, split_name, treatment, params
impression(impression_data, params[:attributes])
end

def track(impressions)
return if impressions.empty?

stats = { dropped: 0, queued: 0, dedupe: 0 }
begin
case @config.impressions_mode
when :none
return
when :debug
track_debug_mode(impressions, stats)
when :optimized
track_optimized_mode(impressions, stats)
def track(impressions_decorator)
return if impressions_decorator.empty?

impressions_decorator.each do |impression_decorator|
impression_router.add_bulk([impression_decorator[:impression]])
stats = { dropped: 0, queued: 0, dedupe: 0 }
begin
next if @config.impressions_mode == :none || impression_decorator[:disabled]
if @config.impressions_mode == :debug
track_debug_mode([impression_decorator[:impression]], stats)
else
track_optimized_mode([impression_decorator[:impression]], stats)
end
rescue StandardError => e
@config.log_found_exception(__method__.to_s, e)
ensure
record_stats(stats)
end
rescue StandardError => e
@config.log_found_exception(__method__.to_s, e)
ensure
record_stats(stats)
impression_router.add_bulk(impressions)
end
end

Expand Down Expand Up @@ -126,11 +123,10 @@ def track_debug_mode(impressions, stats)

def track_optimized_mode(impressions, stats)
optimized_impressions = impressions.select { |imp| should_queue_impression?(imp[:i]) }

stats[:dedupe] = impressions.length - optimized_impressions.length
return if optimized_impressions.empty?

stats[:dropped] = @impressions_repository.add_bulk(optimized_impressions)
stats[:dedupe] = impressions.length - optimized_impressions.length
stats[:queued] = optimized_impressions.length - stats[:dropped]
end
end
Expand Down
8 changes: 4 additions & 4 deletions lib/splitclient-rb/engine/synchronizer.rb
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ def start_periodic_data_recording
events_sender
start_telemetry_sync_task
end

impressions_count_sender
start_unique_keys_tracker_task
end
Expand Down Expand Up @@ -175,7 +175,7 @@ def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_secon

# Starts thread which loops constantly and sends impressions to the Split API
def impressions_sender
ImpressionsSender.new(@impressions_repository, @config, @impressions_api).call unless @config.impressions_mode == :none
ImpressionsSender.new(@impressions_repository, @config, @impressions_api).call
end

# Starts thread which loops constantly and sends events to the Split API
Expand All @@ -185,7 +185,7 @@ def events_sender

# Starts thread which loops constantly and sends impressions count to the Split API
def impressions_count_sender
ImpressionsCountSender.new(@config, @impression_counter, @impressions_sender_adapter).call unless @config.impressions_mode == :debug
ImpressionsCountSender.new(@config, @impression_counter, @impressions_sender_adapter).call
end

def start_telemetry_sync_task
Expand All @@ -203,7 +203,7 @@ def sync_result(success, remaining_attempts, segment_names = nil)
def sync_splits_and_segments
@config.logger.debug('Synchronizing feature flags and segments ...') if @config.debug_enabled
splits_result = @split_fetcher.fetch_splits

splits_result[:success] && @segment_fetcher.fetch_segments
end
end
Expand Down
7 changes: 7 additions & 0 deletions lib/splitclient-rb/helpers/repository_helper.rb
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,13 @@ def self.update_feature_flag_repository(feature_flag_repository, feature_flags,
next
end

unless feature_flag.key?(:impressionsDisabled)
feature_flag[:impressionsDisabled] = false
if config.debug_enabled
config.logger.debug("feature flag (#{feature_flag[:name]}) does not have impressionsDisabled field, setting it to false")
end
end

config.logger.debug("storing feature flag (#{feature_flag[:name]})") if config.debug_enabled
to_add.push(feature_flag)
end
Expand Down
15 changes: 2 additions & 13 deletions lib/splitclient-rb/split_factory.rb
Original file line number Diff line number Diff line change
Expand Up @@ -230,33 +230,22 @@ def build_telemetry_synchronizer(flag_sets, flag_sets_invalid)
end

def build_unique_keys_tracker
if @config.impressions_mode != :none
@unique_keys_tracker = Engine::Impressions::NoopUniqueKeysTracker.new
return
end

bf = Cache::Filter::BloomFilter.new(30_000_000)
filter_adapter = Cache::Filter::FilterAdapter.new(@config, bf)
cache = Concurrent::Hash.new
@unique_keys_tracker = Engine::Impressions::UniqueKeysTracker.new(@config, filter_adapter, @impressions_sender_adapter, cache)
end

def build_impressions_observer
if (@config.cache_adapter == :redis && @config.impressions_mode != :optimized) ||
(@config.cache_adapter == :memory && @config.impressions_mode == :none)
if (@config.cache_adapter == :redis && @config.impressions_mode != :optimized)
@impression_observer = Observers::NoopImpressionObserver.new
else
@impression_observer = Observers::ImpressionObserver.new
end
end

def build_impression_counter
case @config.impressions_mode
when :debug
@impression_counter = Engine::Common::NoopImpressionCounter.new
else
@impression_counter = Engine::Common::ImpressionCounter.new
end
@impression_counter = Engine::Common::ImpressionCounter.new
end

def build_impressions_sender_adapter
Expand Down
32 changes: 16 additions & 16 deletions spec/cache/repositories/impressions_repository_spec.rb
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,8 @@
it 'adds impressions' do
params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, params)
impressions << impressions_manager.build_impression('matching_key1', nil, :bar, treatment2, params)
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, false, params), :disabled => false }
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :bar, treatment2, false, params), :disabled => false }
impressions_manager.track(impressions)

expect(repository.batch).to match_array(result)
Expand All @@ -67,8 +67,8 @@
it 'adds impressions in bulk' do
params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, params)
impressions << impressions_manager.build_impression('matching_key1', nil, :bar, treatment2, params)
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, false, params), :disabled => false }
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :bar, treatment2, false, params), :disabled => false }
impressions_manager.track(impressions)

expect(repository.batch).to match_array(result)
Expand All @@ -80,7 +80,7 @@
config.labels_enabled = false
params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, params)
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, false, params), :disabled => false }
impressions_manager.track(impressions)

expect(repository.batch.first[:i][:r]).to be_nil
Expand All @@ -89,8 +89,8 @@
it 'bulk size less than the actual queue' do
params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, params)
impressions << impressions_manager.build_impression('matching_key1', nil, :foo, treatment2, params)
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :foo, treatment1, false, params), :disabled => false }
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :foo, treatment2, false, params), :disabled => false }
impressions_manager.track(impressions)

config.impressions_bulk_size = 1
Expand Down Expand Up @@ -142,8 +142,8 @@
treatment = { treatment: 'on', label: 'sample_rule', change_number: 1_533_177_602_748 }
params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key1', nil, :foo1, treatment, params)
impressions << impressions_manager.build_impression('matching_key2', nil, :foo1, treatment, params)
impressions << { :impression => impressions_manager.build_impression('matching_key1', nil, :foo1, treatment, false, params), :disabled => false }
impressions << { :impression => impressions_manager.build_impression('matching_key2', nil, :foo1, treatment, false, params), :disabled => false }
impressions_manager.track(impressions)

expect(repository.batch.size).to eq(1)
Expand Down Expand Up @@ -200,8 +200,8 @@
expect(config.impressions_adapter).to receive(:expire).once.with(anything, 3600)
params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key', nil, :foo1, treatment, params)
impressions << impressions_manager.build_impression('matching_key', nil, :foo1, treatment, params)
impressions << { :impression => impressions_manager.build_impression('matching_key', nil, :foo1, treatment, false, params), :disabled => false }
impressions << { :impression => impressions_manager.build_impression('matching_key', nil, :foo1, treatment, false, params), :disabled => false }
impressions_manager.track(impressions)
end

Expand All @@ -211,7 +211,7 @@

params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key', nil, :foo1, treatment, params)
impressions << { :impression => impressions_manager.build_impression('matching_key', nil, :foo1, treatment, false, params), :disabled => false }
impressions_manager.track(impressions)

expect(repository.batch).to eq([])
Expand All @@ -221,8 +221,8 @@
other_treatment = { treatment: 'on', label: 'sample_rule_2', change_number: 1_533_177_602_748 }
params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << impressions_manager.build_impression('matching_key', nil, :foo1, treatment, params)
impressions << impressions_manager.build_impression('matching_key', nil, :foo2, other_treatment, params)
impressions << { :impression => impressions_manager.build_impression('matching_key', nil, :foo1, treatment, false, params), :disabled => false }
impressions << { :impression => impressions_manager.build_impression('matching_key', nil, :foo2, other_treatment, false, params), :disabled => false }
impressions_manager.track(impressions)

adapter.get_from_queue('SPLITIO.impressions', 0).map do |e|
Expand Down Expand Up @@ -252,8 +252,8 @@

params = { attributes: {}, time: 1_478_113_516_002 }
impressions = []
impressions << custom_impressions_manager.build_impression('matching_key', nil, :foo1, treatment, params)
impressions << custom_impressions_manager.build_impression('matching_key', nil, :foo2, other_treatment, params)
impressions << { :impression => custom_impressions_manager.build_impression('matching_key', nil, :foo1, treatment, false, params), :disabled => false }
impressions << { :impression => custom_impressions_manager.build_impression('matching_key', nil, :foo2, other_treatment, false, params), :disabled => false }
custom_impressions_manager.track(impressions)

custom_adapter.get_from_queue('SPLITIO.impressions', 0).map do |e|
Expand Down
Loading