Skip to content

Commit da58a64

Browse files
sanzmaurosanzmauro
authored
Merge pull request #478 from splitio/SDKS-7190
Telemetry and fetch segments
2 parents 4d5a865 + 7a1f920 commit da58a64

File tree

19 files changed

+143
-79
lines changed

19 files changed

+143
-79
lines changed

‎lib/splitclient-rb.rb‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
require 'splitclient-rb/managers/split_manager'
4343
require 'splitclient-rb/helpers/thread_helper'
4444
require 'splitclient-rb/helpers/decryption_helper'
45+
require 'splitclient-rb/helpers/util'
4546
require 'splitclient-rb/split_factory'
4647
require 'splitclient-rb/split_factory_builder'
4748
require 'splitclient-rb/split_config'

‎lib/splitclient-rb/engine/api/splits.rb‎

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -46,21 +46,11 @@ def splits_with_segment_names(splits_json)
4646

4747
parsed_splits[:segment_names] =
4848
parsed_splits[:splits].each_with_object(Set.new) do |split, splits|
49-
splits << segment_names(split)
49+
splits << Helpers::Util.segment_names_by_feature_flag(split)
5050
end.flatten
5151

5252
parsed_splits
5353
end
54-
55-
def segment_names(split)
56-
split[:conditions].each_with_object(Set.new) do |condition, names|
57-
condition[:matcherGroup][:matchers].each do |matcher|
58-
next if matcher[:userDefinedSegmentMatcherData].nil?
59-
60-
names << matcher[:userDefinedSegmentMatcherData][:segmentName]
61-
end
62-
end
63-
end
6454
end
6555
end
6656
end

‎lib/splitclient-rb/helpers/util.rb‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
# frozen_string_literal: true
2+
3+
module SplitIoClient
4+
module Helpers
5+
class Util
6+
def self.segment_names_by_feature_flag(feature_flag)
7+
feature_flag[:conditions].each_with_object(Set.new) do |condition, names|
8+
condition[:matcherGroup][:matchers].each do |matcher|
9+
next if matcher[:userDefinedSegmentMatcherData].nil?
10+
11+
names << matcher[:userDefinedSegmentMatcherData][:segmentName]
12+
end
13+
end
14+
end
15+
end
16+
end
17+
end

‎lib/splitclient-rb/split_factory.rb‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -184,7 +184,7 @@ def build_synchronizer
184184

185185
def build_streaming_components
186186
@push_status_queue = Queue.new
187-
splits_worker = SSE::Workers::SplitsWorker.new(@synchronizer, @config, @splits_repository)
187+
splits_worker = SSE::Workers::SplitsWorker.new(@synchronizer, @config, @splits_repository, @runtime_producer, @segment_fetcher)
188188
segments_worker = SSE::Workers::SegmentsWorker.new(@synchronizer, @config, @segments_repository)
189189
notification_manager_keeper = SSE::NotificationManagerKeeper.new(@config, @runtime_producer, @push_status_queue)
190190
notification_processor = SSE::NotificationProcessor.new(@config, splits_worker, segments_worker)

‎lib/splitclient-rb/sse/workers/splits_worker.rb‎

Lines changed: 48 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,14 @@ module SplitIoClient
44
module SSE
55
module Workers
66
class SplitsWorker
7-
def initialize(synchronizer, config, feature_flags_repository)
7+
def initialize(synchronizer, config, feature_flags_repository, telemetry_runtime_producer, segment_fetcher)
88
@synchronizer = synchronizer
99
@config = config
1010
@feature_flags_repository = feature_flags_repository
1111
@queue = Queue.new
1212
@running = Concurrent::AtomicBoolean.new(false)
13+
@telemetry_runtime_producer = telemetry_runtime_producer
14+
@segment_fetcher = segment_fetcher
1315
end
1416

1517
def start
@@ -29,7 +31,7 @@ def stop
2931
end
3032

3133
@running.make_false
32-
SplitIoClient::Helpers::ThreadHelper.stop(:split_update_worker, @config)
34+
Helpers::ThreadHelper.stop(:split_update_worker, @config)
3335
end
3436

3537
def add_to_queue(notification)
@@ -39,38 +41,47 @@ def add_to_queue(notification)
3941

4042
private
4143

42-
def return_split_from_json(notification)
43-
JSON.parse(
44-
SplitIoClient::Helpers::DecryptionHelper.get_encoded_definition(
45-
notification.data['c'],
46-
notification.data['d']
47-
),
48-
symbolize_names: true
49-
)
44+
def perform_thread
45+
@config.threads[:split_update_worker] = Thread.new do
46+
@config.logger.debug('starting feature_flags_worker ...') if @config.debug_enabled
47+
perform
48+
end
5049
end
5150

52-
def check_update(notification)
53-
@feature_flags_repository.get_change_number == notification.data['pcn'] && !notification.data['d'].nil?
51+
def perform
52+
while (notification = @queue.pop)
53+
@config.logger.debug("feature_flags_worker change_number dequeue #{notification.data['changeNumber']}")
54+
case notification.data['type']
55+
when SSE::EventSource::EventTypes::SPLIT_UPDATE
56+
success = update_feature_flag(notification)
57+
@synchronizer.fetch_splits(notification.data['changeNumber']) unless success
58+
when SSE::EventSource::EventTypes::SPLIT_KILL
59+
kill_feature_flag(notification)
60+
end
61+
end
5462
end
5563

5664
def update_feature_flag(notification)
57-
return if @feature_flags_repository.get_change_number.to_i > notification.data['changeNumber']
65+
return true if @feature_flags_repository.get_change_number.to_i >= notification.data['changeNumber']
66+
return false unless !notification.data['d'].nil? && @feature_flags_repository.get_change_number == notification.data['pcn']
5867

59-
if check_update(notification)
60-
begin
61-
new_split = return_split_from_json(notification)
62-
if SplitIoClient::Engine::Models::Split.archived?(new_split)
63-
@feature_flags_repository.remove_split(new_split)
64-
else
65-
@feature_flags_repository.add_split(new_split)
66-
end
67-
@feature_flags_repository.set_change_number(notification.data['changeNumber'])
68-
return
69-
rescue StandardError => e
70-
@config.logger.debug("Failed to update Split: #{e.inspect}") if @config.debug_enabled
71-
end
68+
new_split = return_split_from_json(notification)
69+
if Engine::Models::Split.archived?(new_split)
70+
@feature_flags_repository.remove_split(new_split)
71+
else
72+
@feature_flags_repository.add_split(new_split)
73+
74+
fetch_segments_if_not_exists(new_split)
7275
end
73-
@synchronizer.fetch_splits(notification.data['changeNumber'])
76+
77+
@feature_flags_repository.set_change_number(notification.data['changeNumber'])
78+
@telemetry_runtime_producer.record_updates_from_sse(Telemetry::Domain::Constants::SPLITS)
79+
80+
true
81+
rescue StandardError => e
82+
@config.logger.debug("Failed to update Split: #{e.inspect}") if @config.debug_enabled
83+
84+
false
7485
end
7586

7687
def kill_feature_flag(notification)
@@ -85,23 +96,18 @@ def kill_feature_flag(notification)
8596
@synchronizer.fetch_splits(notification.data['changeNumber'])
8697
end
8798

88-
def perform
89-
while (notification = @queue.pop)
90-
@config.logger.debug("feature_flags_worker change_number dequeue #{notification.data['changeNumber']}")
91-
case notification.data['type']
92-
when SSE::EventSource::EventTypes::SPLIT_UPDATE
93-
update_feature_flag(notification)
94-
when SSE::EventSource::EventTypes::SPLIT_KILL
95-
kill_feature_flag(notification)
96-
end
97-
end
99+
def return_split_from_json(notification)
100+
split_json = Helpers::DecryptionHelper.get_encoded_definition(notification.data['c'], notification.data['d'])
101+
102+
JSON.parse(split_json, symbolize_names: true)
98103
end
99104

100-
def perform_thread
101-
@config.threads[:split_update_worker] = Thread.new do
102-
@config.logger.debug('starting feature_flags_worker ...') if @config.debug_enabled
103-
perform
104-
end
105+
def fetch_segments_if_not_exists(feature_flag)
106+
segment_names = Helpers::Util.segment_names_by_feature_flag(feature_flag)
107+
return if segment_names.nil?
108+
109+
@feature_flags_repository.set_segment_names(segment_names)
110+
@segment_fetcher.fetch_segments_if_not_exists(segment_names)
105111
end
106112
end
107113
end

‎lib/splitclient-rb/telemetry/domain/constants.rb‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@ class Constants
3636
TREATMENT_WITH_CONFIG = 'treatmentWithConfig'
3737
TREATMENTS_WITH_CONFIG = 'treatmentsWithConfig'
3838
TRACK = 'track'
39+
40+
SPLITS = 'splits'
3941
end
4042
end
4143
end

‎lib/splitclient-rb/telemetry/domain/structs.rb‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,11 +22,14 @@ module Telemetry
2222
# ls: lastSynchronization, ml: clientMethodLatencies, me: clientMethodExceptions, he: httpErros, hl: httpLatencies,
2323
# tr: tokenRefreshes, ar: authRejections, iq: impressionsQueued, ide: impressionsDeduped, idr: impressionsDropped,
2424
# spc: splitsCount, sec: segmentCount, skc: segmentKeyCount, sl: sessionLengthMs, eq: eventsQueued, ed: eventsDropped,
25-
# se: streamingEvents, t: tags
26-
Usage = Struct.new(:ls, :ml, :me, :he, :hl, :tr, :ar, :iq, :ide, :idr, :spc, :sec, :skc, :sl, :eq, :ed, :se, :t)
25+
# se: streamingEvents, t: tags, ufs: updates from sse
26+
Usage = Struct.new(:ls, :ml, :me, :he, :hl, :tr, :ar, :iq, :ide, :idr, :spc, :sec, :skc, :sl, :eq, :ed, :se, :t, :ufs)
2727

2828
# t: treatment, ts: treatments, tc: treatmentWithConfig, tcs: treatmentsWithConfig, tr: track
2929
ClientMethodLatencies = Struct.new(:t, :ts, :tc, :tcs, :tr)
3030
ClientMethodExceptions = Struct.new(:t, :ts, :tc, :tcs, :tr)
31+
32+
# sp: splits
33+
UpdatesFromSSE = Struct.new(:sp)
3134
end
3235
end

‎lib/splitclient-rb/telemetry/memory/memory_runtime_consumer.rb‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,13 @@ def session_length
9494
@adapter.session_length.value
9595
end
9696

97+
def pop_updates_from_sse
98+
splits = @adapter.updates_from_sse[Domain::Constants::SPLITS]
99+
@adapter.updates_from_sse[Domain::Constants::SPLITS] = 0
100+
101+
UpdatesFromSSE.new(splits)
102+
end
103+
97104
private
98105

99106
def find_last_synchronization(type)

‎lib/splitclient-rb/telemetry/memory/memory_runtime_producer.rb‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,12 @@ def record_session_length(session)
7676
rescue StandardError => e
7777
@config.log_found_exception(__method__.to_s, e)
7878
end
79+
80+
def record_updates_from_sse(event)
81+
@adapter.updates_from_sse[event] += 1
82+
rescue StandardError => e
83+
@config.log_found_exception(__method__.to_s, e)
84+
end
7985
end
8086
end
8187
end

‎lib/splitclient-rb/telemetry/memory/memory_synchronizer.rb‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,8 @@ def synchronize_stats
3434
@telemetry_runtime_consumer.events_stats(Domain::Constants::EVENTS_QUEUED),
3535
@telemetry_runtime_consumer.events_stats(Domain::Constants::EVENTS_DROPPED),
3636
@telemetry_runtime_consumer.pop_streaming_events,
37-
@telemetry_runtime_consumer.pop_tags)
37+
@telemetry_runtime_consumer.pop_tags,
38+
@telemetry_runtime_consumer.pop_updates_from_sse)
3839

3940
@telemetry_api.record_stats(format_stats(usage))
4041
rescue StandardError => e
@@ -163,7 +164,8 @@ def format_stats(usage)
163164
eQ: usage.eq,
164165
eD: usage.ed,
165166
sE: usage.se,
166-
t: usage.t
167+
t: usage.t,
168+
ufs: usage.ufs.to_h
167169
}
168170
end
169171

0 commit comments

Comments
 (0)