From 805bf6d0dca1341ad2e1ea7e52dbd46516f7a5bc Mon Sep 17 00:00:00 2001 From: Bilal Al-Shahwany Date: Tue, 11 Jul 2023 11:17:16 -0700 Subject: [PATCH 1/4] added memory imps async storage --- splitio/storage/inmemmory.py | 115 ++++++++++++++++++++++--- tests/storage/test_inmemory_storage.py | 90 ++++++++++++++++++- 2 files changed, 193 insertions(+), 12 deletions(-) diff --git a/splitio/storage/inmemmory.py b/splitio/storage/inmemmory.py index 8dd35cef..ef9b7670 100644 --- a/splitio/storage/inmemmory.py +++ b/splitio/storage/inmemmory.py @@ -3,10 +3,12 @@ import threading import queue from collections import Counter +import pytest from splitio.models.segments import Segment from splitio.models.telemetry import HTTPErrors, HTTPLatencies, MethodExceptions, MethodLatencies, LastSynchronization, StreamingEvents, TelemetryConfig, TelemetryCounters, CounterConstants from splitio.storage import SplitStorage, SegmentStorage, ImpressionStorage, EventStorage, TelemetryStorage +from splitio.optional.loaders import asyncio MAX_SIZE_BYTES = 5 * 1024 * 1024 MAX_TAGS = 10 @@ -310,7 +312,43 @@ def get_segments_keys_count(self): return total_count -class InMemoryImpressionStorage(ImpressionStorage): +class InMemoryImpressionStorageBase(ImpressionStorage): + """In memory implementation of an impressions base storage.""" + + def set_queue_full_hook(self, hook): + """ + Set a hook to be called when the queue is full. + + :param h: Hook to be called when the queue is full + """ + if callable(hook): + self._queue_full_hook = hook + + def put(self, impressions): + """ + Put one or more impressions in storage. + + :param impressions: List of one or more impressions to store. + :type impressions: list + """ + pass + + def pop_many(self, count): + """ + Pop the oldest N impressions from storage. + + :param count: Number of impressions to pop. + :type count: int + """ + pass + + def clear(self): + """ + Clear data. + """ + pass + +class InMemoryImpressionStorage(InMemoryImpressionStorageBase): """In memory implementation of an impressions storage.""" def __init__(self, queue_size, telemetry_runtime_producer): @@ -325,15 +363,6 @@ def __init__(self, queue_size, telemetry_runtime_producer): self._queue_full_hook = None self._telemetry_runtime_producer = telemetry_runtime_producer - def set_queue_full_hook(self, hook): - """ - Set a hook to be called when the queue is full. - - :param h: Hook to be called when the queue is full - """ - if callable(hook): - self._queue_full_hook = hook - def put(self, impressions): """ Put one or more impressions in storage. @@ -382,6 +411,72 @@ def clear(self): self._impressions = queue.Queue(maxsize=self._queue_size) +class InMemoryImpressionStorageAsync(InMemoryImpressionStorageBase): + """In memory implementation of an impressions async storage.""" + + def __init__(self, queue_size, telemetry_runtime_producer): + """ + Construct an instance. + + :param eventsQueueSize: How many events to queue before forcing a submission + """ + self._queue_size = queue_size + self._impressions = asyncio.Queue(maxsize=queue_size) + self._lock = asyncio.Lock() + self._queue_full_hook = None + self._telemetry_runtime_producer = telemetry_runtime_producer + + async def put(self, impressions): + """ + Put one or more impressions in storage. + + :param impressions: List of one or more impressions to store. + :type impressions: list + """ + impressions_stored = 0 + try: + async with self._lock: + for impression in impressions: + if self._impressions.qsize() == self._queue_size: + raise asyncio.QueueFull + await self._impressions.put(impression) + impressions_stored += 1 + _LOGGER.error(impressions_stored) + self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, len(impressions)) + return True + except asyncio.QueueFull: + self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_DROPPED, len(impressions) - impressions_stored) + self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, impressions_stored) + if self._queue_full_hook is not None and callable(self._queue_full_hook): + await self._queue_full_hook() + _LOGGER.warning( + 'Impression queue is full, failing to add more impressions. \n' + 'Consider increasing parameter `impressionsQueueSize` in configuration' + ) + return False + + async def pop_many(self, count): + """ + Pop the oldest N impressions from storage. + + :param count: Number of impressions to pop. + :type count: int + """ + impressions = [] + async with self._lock: + while not self._impressions.empty() and count > 0: + impressions.append(await self._impressions.get()) + count -= 1 + return impressions + + async def clear(self): + """ + Clear data. + """ + async with self._lock: + self._impressions = asyncio.Queue(maxsize=self._queue_size) + + class InMemoryEventStorage(EventStorage): """ In memory storage for events. diff --git a/tests/storage/test_inmemory_storage.py b/tests/storage/test_inmemory_storage.py index 7319548d..785241ab 100644 --- a/tests/storage/test_inmemory_storage.py +++ b/tests/storage/test_inmemory_storage.py @@ -11,7 +11,7 @@ from splitio.engine.telemetry import TelemetryStorageProducer from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \ - InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage + InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage, InMemoryImpressionStorageAsync class InMemorySplitStorageTests(object): @@ -325,7 +325,7 @@ def test_clear(self, mocker): storage.clear() assert storage._impressions.qsize() == 0 - def test_push_pop_impressions(self, mocker): + def test_impressions_dropped(self, mocker): """Test pushing and retrieving impressions.""" telemetry_storage = InMemoryTelemetryStorage() telemetry_producer = TelemetryStorageProducer(telemetry_storage) @@ -338,6 +338,92 @@ def test_push_pop_impressions(self, mocker): assert(telemetry_storage._counters._impressions_dropped == 1) assert(telemetry_storage._counters._impressions_queued == 2) + +class InMemoryImpressionsStorageAsyncTests(object): + """InMemory impressions async storage test cases.""" + + @pytest.mark.asyncio + async def test_push_pop_impressions(self, mocker): + """Test pushing and retrieving impressions.""" + telemetry_storage = InMemoryTelemetryStorage() + telemetry_producer = TelemetryStorageProducer(telemetry_storage) + telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() + storage = InMemoryImpressionStorageAsync(100, telemetry_runtime_producer) + await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) + await storage.put([Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) + await storage.put([Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) + assert(telemetry_storage._counters._impressions_queued == 3) + + # Assert impressions are retrieved in the same order they are inserted. + assert await storage.pop_many(1) == [ + Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654) + ] + assert await storage.pop_many(1) == [ + Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654) + ] + assert await storage.pop_many(1) == [ + Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654) + ] + + # Assert inserting multiple impressions at once works and maintains order. + impressions = [ + Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654), + Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654), + Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654) + ] + assert await storage.put(impressions) + + # Assert impressions are retrieved in the same order they are inserted. + assert await storage.pop_many(1) == [ + Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654) + ] + assert await storage.pop_many(1) == [ + Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654) + ] + assert await storage.pop_many(1) == [ + Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654) + ] + + @pytest.mark.asyncio + async def test_queue_full_hook(self, mocker): + """Test queue_full_hook is executed when the queue is full.""" + storage = InMemoryImpressionStorageAsync(100, mocker.Mock()) + self.hook_called = False + async def queue_full_hook(): + self.hook_called = True + + storage.set_queue_full_hook(queue_full_hook) + impressions = [ + Impression('key%d' % i, 'feature1', 'on', 'l1', 123456, 'b1', 321654) + for i in range(0, 101) + ] + await storage.put(impressions) + await queue_full_hook() + assert self.hook_called == True + + @pytest.mark.asyncio + async def test_clear(self, mocker): + """Test clear method.""" + storage = InMemoryImpressionStorageAsync(100, mocker.Mock()) + await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) + assert storage._impressions.qsize() == 1 + await storage.clear() + assert storage._impressions.qsize() == 0 + + @pytest.mark.asyncio + async def test_impressions_dropped(self, mocker): + """Test pushing and retrieving impressions.""" + telemetry_storage = InMemoryTelemetryStorage() + telemetry_producer = TelemetryStorageProducer(telemetry_storage) + telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() + storage = InMemoryImpressionStorageAsync(2, telemetry_runtime_producer) + await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) + await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) + await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) + assert(telemetry_storage._counters._impressions_dropped == 1) + assert(telemetry_storage._counters._impressions_queued == 2) + + class InMemoryEventsStorageTests(object): """InMemory events storage test cases.""" From b22a606a122cc49e8bedd68a0b03677265522271 Mon Sep 17 00:00:00 2001 From: Bilal Al-Shahwany Date: Tue, 11 Jul 2023 13:29:34 -0700 Subject: [PATCH 2/4] polish --- splitio/storage/inmemmory.py | 1 - 1 file changed, 1 deletion(-) diff --git a/splitio/storage/inmemmory.py b/splitio/storage/inmemmory.py index ef9b7670..93646aed 100644 --- a/splitio/storage/inmemmory.py +++ b/splitio/storage/inmemmory.py @@ -3,7 +3,6 @@ import threading import queue from collections import Counter -import pytest from splitio.models.segments import Segment from splitio.models.telemetry import HTTPErrors, HTTPLatencies, MethodExceptions, MethodLatencies, LastSynchronization, StreamingEvents, TelemetryConfig, TelemetryCounters, CounterConstants From b605593875aebf870f2ad9238a72b7220cddc1f1 Mon Sep 17 00:00:00 2001 From: Bilal Al-Shahwany Date: Fri, 14 Jul 2023 09:28:27 -0700 Subject: [PATCH 3/4] added async telemetry classes --- splitio/storage/inmemmory.py | 336 +++++++++++++++++++++++-- tests/storage/test_inmemory_storage.py | 23 +- 2 files changed, 337 insertions(+), 22 deletions(-) diff --git a/splitio/storage/inmemmory.py b/splitio/storage/inmemmory.py index 93646aed..972cbf8c 100644 --- a/splitio/storage/inmemmory.py +++ b/splitio/storage/inmemmory.py @@ -5,7 +5,9 @@ from collections import Counter from splitio.models.segments import Segment -from splitio.models.telemetry import HTTPErrors, HTTPLatencies, MethodExceptions, MethodLatencies, LastSynchronization, StreamingEvents, TelemetryConfig, TelemetryCounters, CounterConstants +from splitio.models.telemetry import HTTPErrors, HTTPLatencies, MethodExceptions, MethodLatencies, LastSynchronization, StreamingEvents, TelemetryConfig, TelemetryCounters, CounterConstants, \ + HTTPErrorsAsync, HTTPLatenciesAsync, MethodExceptionsAsync, MethodLatenciesAsync, LastSynchronizationAsync, StreamingEventsAsync, TelemetryConfigAsync, TelemetryCountersAsync + from splitio.storage import SplitStorage, SegmentStorage, ImpressionStorage, EventStorage, TelemetryStorage from splitio.optional.loaders import asyncio @@ -441,11 +443,11 @@ async def put(self, impressions): await self._impressions.put(impression) impressions_stored += 1 _LOGGER.error(impressions_stored) - self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, len(impressions)) + await self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, len(impressions)) return True except asyncio.QueueFull: - self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_DROPPED, len(impressions) - impressions_stored) - self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, impressions_stored) + await self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_DROPPED, len(impressions) - impressions_stored) + await self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, impressions_stored) if self._queue_full_hook is not None and callable(self._queue_full_hook): await self._queue_full_hook() _LOGGER.warning( @@ -556,14 +558,158 @@ def clear(self): with self._lock: self._events = queue.Queue(maxsize=self._queue_size) -class InMemoryTelemetryStorage(TelemetryStorage): +class InMemoryTelemetryStorageBase(TelemetryStorage): + """In-memory telemetry storage base.""" + + def _reset_tags(self): + self._tags = [] + + def _reset_config_tags(self): + self._config_tags = [] + + def record_config(self, config, extra_config): + """Record configurations.""" + pass + + def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count): + """Record active and redundant factories.""" + pass + + def record_ready_time(self, ready_time): + """Record ready time.""" + pass + + def add_tag(self, tag): + """Record tag string.""" + pass + + def add_config_tag(self, tag): + """Record tag string.""" + pass + + def record_bur_time_out(self): + """Record block until ready timeout.""" + pass + + def record_not_ready_usage(self): + """record non-ready usage.""" + pass + + def record_latency(self, method, latency): + """Record method latency time.""" + pass + + def record_exception(self, method): + """Record method exception.""" + pass + + def record_impression_stats(self, data_type, count): + """Record impressions stats.""" + pass + + def record_event_stats(self, data_type, count): + """Record events stats.""" + pass + + def record_successful_sync(self, resource, time): + """Record successful sync.""" + pass + + def record_sync_error(self, resource, status): + """Record sync http error.""" + pass + + def record_sync_latency(self, resource, latency): + """Record latency time.""" + pass + + def record_auth_rejections(self): + """Record auth rejection.""" + pass + + def record_token_refreshes(self): + """Record sse token refresh.""" + pass + + def record_streaming_event(self, streaming_event): + """Record incoming streaming event.""" + pass + + def record_session_length(self, session): + """Record session length.""" + pass + + def get_bur_time_outs(self): + """Get block until ready timeout.""" + pass + + def get_non_ready_usage(self): + """Get non-ready usage.""" + pass + + def get_config_stats(self): + """Get all config info.""" + pass + + def pop_exceptions(self): + """Get and reset method exceptions.""" + pass + + def pop_tags(self): + """Get and reset tags.""" + pass + + def pop_config_tags(self): + """Get and reset tags.""" + pass + + def pop_latencies(self): + """Get and reset eval latencies.""" + pass + + def get_impressions_stats(self, type): + """Get impressions stats""" + pass + + def get_events_stats(self, type): + """Get events stats""" + pass + + def get_last_synchronization(self): + """Get last sync""" + pass + + def pop_http_errors(self): + """Get and reset http errors.""" + pass + + def pop_http_latencies(self): + """Get and reset http latencies.""" + pass + + def pop_auth_rejections(self): + """Get and reset auth rejections.""" + pass + + def pop_token_refreshes(self): + """Get and reset token refreshes.""" + pass + + def pop_streaming_events(self): + """Get and reset streaming events""" + pass + + def get_session_length(self): + """Get session length""" + pass + + +class InMemoryTelemetryStorage(InMemoryTelemetryStorageBase): """In-memory telemetry storage.""" def __init__(self): """Constructor""" self._lock = threading.RLock() - self._reset_tags() - self._reset_config_tags() self._method_exceptions = MethodExceptions() self._last_synchronization = LastSynchronization() self._counters = TelemetryCounters() @@ -572,14 +718,9 @@ def __init__(self): self._http_latencies = HTTPLatencies() self._streaming_events = StreamingEvents() self._tel_config = TelemetryConfig() - - def _reset_tags(self): with self._lock: - self._tags = [] - - def _reset_config_tags(self): - with self._lock: - self._config_tags = [] + self._reset_tags() + self._reset_config_tags() def record_config(self, config, extra_config): """Record configurations.""" @@ -726,6 +867,173 @@ def get_session_length(self): """Get session length""" return self._counters.get_session_length() + +class InMemoryTelemetryStorageAsync(InMemoryTelemetryStorageBase): + """In-memory telemetry async storage.""" + + async def create(): + """Constructor""" + self = InMemoryTelemetryStorageAsync() + self._lock = asyncio.Lock() + self._method_exceptions = await MethodExceptionsAsync.create() + self._last_synchronization = await LastSynchronizationAsync.create() + self._counters = await TelemetryCountersAsync.create() + self._http_sync_errors = await HTTPErrorsAsync.create() + self._method_latencies = await MethodLatenciesAsync.create() + self._http_latencies = await HTTPLatenciesAsync.create() + self._streaming_events = await StreamingEventsAsync.create() + self._tel_config = await TelemetryConfigAsync.create() + async with self._lock: + self._reset_tags() + self._reset_config_tags() + return self + + async def record_config(self, config, extra_config): + """Record configurations.""" + await self._tel_config.record_config(config, extra_config) + + async def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count): + """Record active and redundant factories.""" + await self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count) + + async def record_ready_time(self, ready_time): + """Record ready time.""" + await self._tel_config.record_ready_time(ready_time) + + async def add_tag(self, tag): + """Record tag string.""" + async with self._lock: + if len(self._tags) < MAX_TAGS: + self._tags.append(tag) + + async def add_config_tag(self, tag): + """Record tag string.""" + async with self._lock: + if len(self._config_tags) < MAX_TAGS: + self._config_tags.append(tag) + + async def record_bur_time_out(self): + """Record block until ready timeout.""" + await self._tel_config.record_bur_time_out() + + async def record_not_ready_usage(self): + """record non-ready usage.""" + await self._tel_config.record_not_ready_usage() + + async def record_latency(self, method, latency): + """Record method latency time.""" + await self._method_latencies.add_latency(method,latency) + + async def record_exception(self, method): + """Record method exception.""" + await self._method_exceptions.add_exception(method) + + async def record_impression_stats(self, data_type, count): + """Record impressions stats.""" + await self._counters.record_impressions_value(data_type, count) + + async def record_event_stats(self, data_type, count): + """Record events stats.""" + await self._counters.record_events_value(data_type, count) + + async def record_successful_sync(self, resource, time): + """Record successful sync.""" + await self._last_synchronization.add_latency(resource, time) + + async def record_sync_error(self, resource, status): + """Record sync http error.""" + await self._http_sync_errors.add_error(resource, status) + + async def record_sync_latency(self, resource, latency): + """Record latency time.""" + await self._http_latencies.add_latency(resource, latency) + + async def record_auth_rejections(self): + """Record auth rejection.""" + await self._counters.record_auth_rejections() + + async def record_token_refreshes(self): + """Record sse token refresh.""" + await self._counters.record_token_refreshes() + + async def record_streaming_event(self, streaming_event): + """Record incoming streaming event.""" + await self._streaming_events.record_streaming_event(streaming_event) + + async def record_session_length(self, session): + """Record session length.""" + await self._counters.record_session_length(session) + + async def get_bur_time_outs(self): + """Get block until ready timeout.""" + return await self._tel_config.get_bur_time_outs() + + async def get_non_ready_usage(self): + """Get non-ready usage.""" + return await self._tel_config.get_non_ready_usage() + + async def get_config_stats(self): + """Get all config info.""" + return await self._tel_config.get_stats() + + async def pop_exceptions(self): + """Get and reset method exceptions.""" + return await self._method_exceptions.pop_all() + + async def pop_tags(self): + """Get and reset tags.""" + async with self._lock: + tags = self._tags + self._reset_tags() + return tags + + async def pop_config_tags(self): + """Get and reset tags.""" + async with self._lock: + tags = self._config_tags + self._reset_config_tags() + return tags + + async def pop_latencies(self): + """Get and reset eval latencies.""" + return await self._method_latencies.pop_all() + + async def get_impressions_stats(self, type): + """Get impressions stats""" + return await self._counters.get_counter_stats(type) + + async def get_events_stats(self, type): + """Get events stats""" + return await self._counters.get_counter_stats(type) + + async def get_last_synchronization(self): + """Get last sync""" + return await self._last_synchronization.get_all() + + async def pop_http_errors(self): + """Get and reset http errors.""" + return await self._http_sync_errors.pop_all() + + async def pop_http_latencies(self): + """Get and reset http latencies.""" + return await self._http_latencies.pop_all() + + async def pop_auth_rejections(self): + """Get and reset auth rejections.""" + return await self._counters.pop_auth_rejections() + + async def pop_token_refreshes(self): + """Get and reset token refreshes.""" + return await self._counters.pop_token_refreshes() + + async def pop_streaming_events(self): + return await self._streaming_events.pop_streaming_events() + + async def get_session_length(self): + """Get session length""" + return await self._counters.get_session_length() + + class LocalhostTelemetryStorage(): """Localhost telemetry storage.""" def do_nothing(*_, **__): diff --git a/tests/storage/test_inmemory_storage.py b/tests/storage/test_inmemory_storage.py index 785241ab..7d3b7f6b 100644 --- a/tests/storage/test_inmemory_storage.py +++ b/tests/storage/test_inmemory_storage.py @@ -8,10 +8,11 @@ from splitio.models.impressions import Impression from splitio.models.events import Event, EventWrapper import splitio.models.telemetry as ModelTelemetry -from splitio.engine.telemetry import TelemetryStorageProducer +from splitio.engine.telemetry import TelemetryStorageProducer, TelemetryStorageProducerAsync from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \ - InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage, InMemoryImpressionStorageAsync + InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage, InMemoryImpressionStorageAsync, InMemoryTelemetryStorageAsync + class InMemorySplitStorageTests(object): @@ -345,8 +346,8 @@ class InMemoryImpressionsStorageAsyncTests(object): @pytest.mark.asyncio async def test_push_pop_impressions(self, mocker): """Test pushing and retrieving impressions.""" - telemetry_storage = InMemoryTelemetryStorage() - telemetry_producer = TelemetryStorageProducer(telemetry_storage) + telemetry_storage = await InMemoryTelemetryStorageAsync.create() + telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() storage = InMemoryImpressionStorageAsync(100, telemetry_runtime_producer) await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) @@ -387,7 +388,10 @@ async def test_push_pop_impressions(self, mocker): @pytest.mark.asyncio async def test_queue_full_hook(self, mocker): """Test queue_full_hook is executed when the queue is full.""" - storage = InMemoryImpressionStorageAsync(100, mocker.Mock()) + telemetry_storage = await InMemoryTelemetryStorageAsync.create() + telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) + telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() + storage = InMemoryImpressionStorageAsync(100, telemetry_runtime_producer) self.hook_called = False async def queue_full_hook(): self.hook_called = True @@ -404,7 +408,10 @@ async def queue_full_hook(): @pytest.mark.asyncio async def test_clear(self, mocker): """Test clear method.""" - storage = InMemoryImpressionStorageAsync(100, mocker.Mock()) + telemetry_storage = await InMemoryTelemetryStorageAsync.create() + telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) + telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() + storage = InMemoryImpressionStorageAsync(100, telemetry_runtime_producer) await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) assert storage._impressions.qsize() == 1 await storage.clear() @@ -413,8 +420,8 @@ async def test_clear(self, mocker): @pytest.mark.asyncio async def test_impressions_dropped(self, mocker): """Test pushing and retrieving impressions.""" - telemetry_storage = InMemoryTelemetryStorage() - telemetry_producer = TelemetryStorageProducer(telemetry_storage) + telemetry_storage = await InMemoryTelemetryStorageAsync.create() + telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage) telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer() storage = InMemoryImpressionStorageAsync(2, telemetry_runtime_producer) await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)]) From d143770e1186bfdc78cd6b7de5bea49b56ce791d Mon Sep 17 00:00:00 2001 From: Bilal Al-Shahwany Date: Wed, 19 Jul 2023 11:36:03 -0700 Subject: [PATCH 4/4] clean up --- splitio/storage/inmemmory.py | 1 - 1 file changed, 1 deletion(-) diff --git a/splitio/storage/inmemmory.py b/splitio/storage/inmemmory.py index 972cbf8c..edcfe36c 100644 --- a/splitio/storage/inmemmory.py +++ b/splitio/storage/inmemmory.py @@ -442,7 +442,6 @@ async def put(self, impressions): raise asyncio.QueueFull await self._impressions.put(impression) impressions_stored += 1 - _LOGGER.error(impressions_stored) await self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, len(impressions)) return True except asyncio.QueueFull: