added clear methods

Author: Matias Melograno

splitio/client/client.py

@@ -90,6 +90,9 @@ def _make_evaluation(self, key, feature, attributes, method_name, metric_name):
             if self.destroyed:
                 _LOGGER.error("Client has already been destroyed - no calls possible")
                 return CONTROL, None
+            if self._factory._waiting_fork():
+                _LOGGER.error("Client is not ready - no calls possible")
+                return CONTROL, None
 
             start = int(round(time.time() * 1000))
 
@@ -143,6 +146,9 @@ def _make_evaluations(self, key, features, attributes, method_name, metric_name)
         if self.destroyed:
             _LOGGER.error("Client has already been destroyed - no calls possible")
             return input_validator.generate_control_treatments(features, method_name)
+        if self._factory._waiting_fork():
+            _LOGGER.error("Client is not ready - no calls possible")
+            return input_validator.generate_control_treatments(features, method_name)
 
         start = int(round(time.time() * 1000))
 
@@ -363,6 +369,9 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
         if self.destroyed:
             _LOGGER.error("Client has already been destroyed - no calls possible")
             return False
+        if self._factory._waiting_fork():
+            _LOGGER.error("Client is not ready - no calls possible")
+            return False
 
         key = input_validator.validate_track_key(key)
         event_type = input_validator.validate_event_type(event_type)

splitio/client/config.py

@@ -56,7 +56,7 @@
     'machineName': None,
     'machineIp': None,
     'splitFile': os.path.join(os.path.expanduser('~'), '.split'),
-    'shouldHandlePostFork': False,
+    'preforkedInitialization': False,
 }
 
 

splitio/client/factory.py

@@ -91,7 +91,7 @@ def __init__(  # pylint: disable=too-many-arguments
             recorder,
             sync_manager=None,
             sdk_ready_flag=None,
-            should_handle_post_fork=False,
+            preforked_initialization=False,
     ):
         """
         Class constructor.
@@ -108,25 +108,27 @@ def __init__(  # pylint: disable=too-many-arguments
         :type sdk_ready_flag: threading.Event
         :param recorder: StatsRecorder instance
         :type recorder: StatsRecorder
+        :param preforked_initialization: Whether should be instantiated as preforked or not.
+        :type preforked_initialization: bool
         """
         self._apikey = apikey
         self._storages = storages
         self._labels_enabled = labels_enabled
         self._sync_manager = sync_manager
         self._sdk_internal_ready_flag = sdk_ready_flag
         self._recorder = recorder
-        self._should_handle_post_fork = should_handle_post_fork
-        self._start_ready_updater()
+        self._preforked_initialization = preforked_initialization
+        self._start_status_updater()
 
-    def _start_ready_updater(self):
+    def _start_status_updater(self):
         """
-        Perform ready updater
+        Perform status updater
         """
-        # If we have a ready flag, it means we have sync tasks that need to finish
-        # before the SDK client becomes ready.
-        if self._should_handle_post_fork:
+        if self._preforked_initialization:
             self._status = Status.WAITING_FORK
             return
+        # If we have a ready flag, it means we have sync tasks that need to finish
+        # before the SDK client becomes ready.
         if self._sdk_internal_ready_flag is not None:
             self._sdk_ready_flag = threading.Event()
             self._status = Status.NOT_INITIALIZED
@@ -243,35 +245,37 @@ def destroyed(self):
         """
         return self._status == Status.DESTROYED
 
-    @property
-    def waiting_fork(self):
+    def _waiting_fork(self):
         """
-        Return whether the factory is waiting for fork recreation or not.
+        Return whether the factory is waiting to be recreated by forking or not.
 
-        :return: True if the factory is waiting for fork recreation. False otherwise.
+        :return: True if the factory is waiting to be recreated by forking. False otherwise.
         :rtype: bool
         """
         return self._status == Status.WAITING_FORK
 
     def handle_post_fork(self):
         """
-        Function capable to re-synchronize splits data on fork processes.
+        Function in charge of starting periodic/realtime synchronization after a fork.
         """
-        if not self.waiting_fork:
+        if not self._waiting_fork():
             _LOGGER.warning('Cannot call handle_post_fork')
             return
-        self._should_handle_post_fork = False  # Reset for updater
         self._sync_manager.recreate()
         sdk_ready_flag = threading.Event()
         self._sdk_internal_ready_flag = sdk_ready_flag
         self._sync_manager._ready_flag = sdk_ready_flag
+        self._get_storage('telemetry').clear()
+        self._get_storage('impressions').clear()
+        self._get_storage('events').clear()
         initialization_thread = threading.Thread(
             target=self._sync_manager.start,
             name="SDKInitializer",
         )
         initialization_thread.setDaemon(True)
         initialization_thread.start()
-        self._start_ready_updater()
+        self._preforked_initialization = False  # reset for status updater
+        self._start_status_updater()
 
 
 def _wrap_impression_listener(listener, metadata):
@@ -360,9 +364,9 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
 
     synchronizer = Synchronizer(synchronizers, tasks)
 
-    should_handle_post_fork = cfg.get('shouldHandlePostFork', False)
+    preforked_initialization = cfg.get('preforkedInitialization', False)
 
-    sdk_ready_flag = threading.Event() if not should_handle_post_fork else None
+    sdk_ready_flag = threading.Event() if not preforked_initialization else None
     manager = Manager(sdk_ready_flag, synchronizer, apis['auth'], cfg['streamingEnabled'],
                       streaming_api_base_url)
 
@@ -376,10 +380,11 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         storages['impressions'],
     )
 
-    if should_handle_post_fork:
+    if preforked_initialization:
         synchronizer.sync_all()
+        synchronizer._split_synchronizers._segment_sync.shutdown()
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
-                            recorder, manager, should_handle_post_fork=should_handle_post_fork)
+                            recorder, manager, preforked_initialization=preforked_initialization)
 
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer")
     initialization_thread.setDaemon(True)

splitio/client/localhost.py

@@ -30,6 +30,10 @@ def pop_many(self, *_, **__):  # pylint: disable=arguments-differ
         """Accept any arguments and do nothing."""
         pass
 
+    def clear(self, *_, **__):  # pylint: disable=arguments-differ
+        """Accept any arguments and do nothing."""
+        pass
+
 
 class LocalhostEventsStorage(EventStorage):
     """Impression storage that doesn't cache anything."""
@@ -42,6 +46,10 @@ def pop_many(self, *_, **__):  # pylint: disable=arguments-differ
         """Accept any arguments and do nothing."""
         pass
 
+    def clear(self, *_, **__):  # pylint: disable=arguments-differ
+        """Accept any arguments and do nothing."""
+        pass
+
 
 class LocalhostTelemetryStorage(TelemetryStorage):
     """Impression storage that doesn't cache anything."""
@@ -69,3 +77,7 @@ def pop_counters(self, *_, **__):  # pylint: disable=arguments-differ
     def pop_gauges(self, *_, **__):  # pylint: disable=arguments-differ
         """Accept any arguments and do nothing."""
         pass
+
+    def clear(self, *_, **__):  # pylint: disable=arguments-differ
+        """Accept any arguments and do nothing."""
+        pass

splitio/client/manager.py

@@ -32,6 +32,9 @@ def split_names(self):
         if self._factory.destroyed:
             _LOGGER.error("Client has already been destroyed - no calls possible.")
             return []
+        if self._factory._waiting_fork():
+            _LOGGER.error("Client is not ready - no calls possible")
+            return []
 
         if not self._factory.ready:
             _LOGGER.warning(
@@ -51,6 +54,9 @@ def splits(self):
         if self._factory.destroyed:
             _LOGGER.error("Client has already been destroyed - no calls possible.")
             return []
+        if self._factory._waiting_fork():
+            _LOGGER.error("Client is not ready - no calls possible")
+            return []
 
         if not self._factory.ready:
             _LOGGER.warning(
@@ -72,7 +78,10 @@ def split(self, feature_name):
         """
         if self._factory.destroyed:
             _LOGGER.error("Client has already been destroyed - no calls possible.")
-            return []
+            return None
+        if self._factory._waiting_fork():
+            _LOGGER.error("Client is not ready - no calls possible")
+            return None
 
         feature_name = input_validator.validate_manager_feature_name(
             feature_name,

splitio/storage/__init__.py

@@ -237,6 +237,13 @@ def pop_many(self, count):
         """
         pass
 
+    @abc.abstractmethod
+    def clear(self):
+        """
+        Clear data.
+        """
+        pass
+
 
 @add_metaclass(abc.ABCMeta)
 class ImpressionPipelinedStorage(object):
@@ -279,6 +286,13 @@ def pop_many(self, count):
         """
         pass
 
+    @abc.abstractmethod
+    def clear(self):
+        """
+        Clear data.
+        """
+        pass
+
 
 @add_metaclass(abc.ABCMeta)
 class TelemetryStorage(object):
@@ -346,6 +360,13 @@ def pop_latencies(self):
         """
         pass
 
+    @abc.abstractmethod
+    def clear(self):
+        """
+        Clear data.
+        """
+        pass
+
 
 @add_metaclass(abc.ABCMeta)
 class TelemetryPipelinedStorage(object):

splitio/storage/inmemmory.py

@@ -291,6 +291,7 @@ def __init__(self, queue_size):
 
         :param eventsQueueSize: How many events to queue before forcing a submission
         """
+        self._queue_size = queue_size
         self._impressions = queue.Queue(maxsize=queue_size)
         self._lock = threading.Lock()
         self._queue_full_hook = None
@@ -339,6 +340,13 @@ def pop_many(self, count):
                 count -= 1
         return impressions
 
+    def clear(self):
+        """
+        Clear data.
+        """
+        with self._lock:
+            self._impressions = queue.Queue(maxsize=self._queue_size)
+
 
 class InMemoryEventStorage(EventStorage):
     """
@@ -353,6 +361,7 @@ def __init__(self, eventsQueueSize):
 
         :param eventsQueueSize: How many events to queue before forcing a submission
         """
+        self._queue_size = eventsQueueSize
         self._lock = threading.Lock()
         self._events = queue.Queue(maxsize=eventsQueueSize)
         self._queue_full_hook = None
@@ -407,6 +416,13 @@ def pop_many(self, count):
         self._size = 0
         return events
 
+    def clear(self):
+        """
+        Clear data.
+        """
+        with self._lock:
+            self._events = queue.Queue(maxsize=self._queue_size)
+
 
 class InMemoryTelemetryStorage(TelemetryStorage):
     """In-Memory implementation of telemetry storage interface."""
@@ -498,3 +514,14 @@ def pop_latencies(self):
                 return self._latencies
             finally:
                 self._latencies = {}
+
+    def clear(self):
+        """
+        Clear data.
+        """
+        with self._latencies_lock:
+            self._latencies = {}
+        with self._gauges_lock:
+            self._gauges = {}
+        with self._counters_lock:
+            self._counters = {}

splitio/storage/redis.py

@@ -453,6 +453,12 @@ def pop_many(self, count):
         """
         raise NotImplementedError('Only redis-consumer mode is supported.')
 
+    def clear(self):
+        """
+        Clear data.
+        """
+        raise NotImplementedError('Not supported for redis.')
+
 
 class RedisEventsStorage(EventStorage):
     """Redis based event storage class."""
@@ -517,6 +523,12 @@ def pop_many(self, count):
         """
         raise NotImplementedError('Only redis-consumer mode is supported.')
 
+    def clear(self):
+        """
+        Clear data.
+        """
+        raise NotImplementedError('Not supported for redis.')
+
 
 class RedisTelemetryStorage(TelemetryStorage, TelemetryPipelinedStorage):
     """Redis-based Telemetry storage."""
@@ -694,3 +706,9 @@ def pop_latencies(self):
         :rtype: list
         """
         raise NotImplementedError('Only redis-consumer mode is supported.')
+
+    def clear(self):
+        """
+        Clear data.
+        """
+        raise NotImplementedError('Not supported for redis.')

splitio/storage/uwsgi.py

@@ -511,6 +511,12 @@ def acknowledge_flush(self):
         """Acknowledge that a flush has been requested."""
         self._uwsgi.cache_del(self._IMPRESSIONS_FLUSH, _SPLITIO_LOCK_CACHE_NAMESPACE)
 
+    def clear(self):
+        """
+        Clear data.
+        """
+        raise NotImplementedError('Not supported for uwsgi.')
+
 
 class UWSGIEventStorage(EventStorage):
     """Events storage interface."""
@@ -602,6 +608,12 @@ def acknowledge_flush(self):
         """Acknowledge that a flush has been requested."""
         self._uwsgi.cache_del(self._EVENTS_FLUSH, _SPLITIO_LOCK_CACHE_NAMESPACE)
 
+    def clear(self):
+        """
+        Clear data.
+        """
+        raise NotImplementedError('Not supported for uwsgi.')
+
 
 class UWSGITelemetryStorage(TelemetryStorage):
     """Telemetry storage interface."""
@@ -726,3 +738,9 @@ def pop_latencies(self):
                 self._LATENCIES_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             self._uwsgi.cache_del(self._LATENCIES_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             return json.loads(latencies_raw) if latencies_raw else {}
+
+    def clear(self):
+        """
+        Clear data.
+        """
+        raise NotImplementedError('Not supported for uwsgi.')

splitio/sync/segment.py

@@ -35,7 +35,6 @@ def recreate(self):
         Create worker_pool on forked processes.
 
         """
-        self.shutdown()
         self._worker_pool = workerpool.WorkerPool(10, self.synchronize_segment)
         self._worker_pool.start()