Merge pull request #417 from splitio/development

Development

Author: chillaq

.github/CODEOWNERS

@@ -0,0 +1 @@
+* @splitio/sdk

CHANGES.txt

@@ -1,3 +1,6 @@
+9.5.0 (Jul 18, 2023)
+- Improved streaming architecture implementation to apply feature flag updates from the notification received which is now enhanced, improving efficiency and reliability of the whole update system.
+
 9.4.2 (May 15, 2023)
 - Updated terminology on the SDKs codebase to be more aligned with current standard without causing a breaking change. The core change is the term split for feature flag on things like logs and code documentation comments.
 - Added detailed debug logging for redis adapter.

setup.py

@@ -21,7 +21,7 @@
     'pyyaml>=5.4',
     'docopt>=0.6.2',
     'enum34;python_version<"3.4"',
-    'bloom-filter2>=2.0.0',
+    'bloom-filter2>=2.0.0'
 ]
 
 with open(path.join(path.abspath(path.dirname(__file__)), 'splitio', 'version.py')) as f:
@@ -44,7 +44,7 @@
         'uwsgi': ['uwsgi>=2.0.0'],
         'cpphash': ['mmh3cffi==0.2.1'],
     },
-    setup_requires=['pytest-runner'],
+    setup_requires=['pytest-runner', 'pluggy==1.0.0;python_version<"3.7"'],
     classifiers=[
         'Environment :: Console',
         'Intended Audience :: Developers',

splitio/engine/telemetry.py

@@ -6,7 +6,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 from  splitio.storage.inmemmory import InMemoryTelemetryStorage
-from splitio.models.telemetry import CounterConstants
+from splitio.models.telemetry import CounterConstants, UpdateFromSSE
 
 class TelemetryStorageProducer(object):
     """Telemetry storage producer class."""
@@ -139,6 +139,10 @@ def record_session_length(self, session):
         """Record session length."""
         self._telemetry_storage.record_session_length(session)
 
+    def record_update_from_sse(self, event):
+        """Record update from sse."""
+        self._telemetry_storage.record_update_from_sse(event)
+
 class TelemetryStorageConsumer(object):
     """Telemetry storage consumer class."""
 
@@ -271,6 +275,10 @@ def pop_streaming_events(self):
         """Get and reset streaming events."""
         return self._telemetry_storage.pop_streaming_events()
 
+    def pop_update_from_sse(self, event):
+        """Get and reset update from sse."""
+        return self._telemetry_storage.pop_update_from_sse(event)
+
     def get_session_length(self):
         """Get session length"""
         return self._telemetry_storage.get_session_length()
@@ -292,6 +300,7 @@ def pop_formatted_stats(self):
             'iDr': self.get_impressions_stats(CounterConstants.IMPRESSIONS_DROPPED),
             'eQ': self.get_events_stats(CounterConstants.EVENTS_QUEUED),
             'eD': self.get_events_stats(CounterConstants.EVENTS_DROPPED),
+            'ufs': {event.value: self.pop_update_from_sse(event) for event in UpdateFromSSE},
             'lS': {'sp': last_synchronization['split'],
                       'se': last_synchronization['segment'],
                       'im': last_synchronization['impression'],

splitio/models/telemetry.py

@@ -131,6 +131,10 @@ class OperationMode(Enum):
     CONSUMER = 'consumer'
     PARTIAL_CONSUMER = 'partial_consumer'
 
+class UpdateFromSSE(Enum):
+    """Update from sse constants"""
+    SPLIT_UPDATE = 'sp'
+
 def get_latency_bucket_index(micros):
     """
     Find the bucket index for a measured latency.
@@ -482,6 +486,7 @@ def _reset_all(self):
             self._auth_rejections = 0
             self._token_refreshes = 0
             self._session_length = 0
+            self._update_from_sse = {}
 
     def record_impressions_value(self, resource, value):
         """
@@ -519,17 +524,27 @@ def record_events_value(self, resource, value):
             else:
                 return
 
+    def record_update_from_sse(self, event):
+        """
+        Increment the update from sse resource by one.
+
+        """
+        with self._lock:
+            if event.value not in self._update_from_sse:
+                self._update_from_sse[event.value] = 0
+            self._update_from_sse[event.value] += 1
+
     def record_auth_rejections(self):
         """
-        Increament the auth rejection resource by one.
+        Increment the auth rejection resource by one.
 
         """
         with self._lock:
             self._auth_rejections += 1
 
     def record_token_refreshes(self):
         """
-        Increament the token refreshes resource by one.
+        Increment the token refreshes resource by one.
 
         """
         with self._lock:
@@ -604,6 +619,18 @@ def pop_token_refreshes(self):
             self._token_refreshes = 0
             return token_refreshes
 
+    def pop_update_from_sse(self, event):
+        """
+        Pop update from sse
+
+        :return: update from sse value
+        :rtype: int
+        """
+        with self._lock:
+            update_from_sse = self._update_from_sse[event.value]
+            self._update_from_sse[event.value] = 0
+            return update_from_sse
+
 class StreamingEvent(object):
     """
     Streaming event class

splitio/push/manager.py

@@ -44,7 +44,7 @@ def __init__(self, auth_api, synchronizer, feedback_loop, sdk_metadata, telemetr
         """
         self._auth_api = auth_api
         self._feedback_loop = feedback_loop
-        self._processor = MessageProcessor(synchronizer)
+        self._processor = MessageProcessor(synchronizer, telemetry_runtime_producer)
         self._status_tracker = PushStatusTracker(telemetry_runtime_producer)
         self._event_handlers = {
             EventType.MESSAGE: self._handle_message,

splitio/push/parser.py

@@ -7,7 +7,6 @@
 from splitio.util.time import utctime_ms
 from splitio.push.sse import SSE_EVENT_ERROR, SSE_EVENT_MESSAGE
 
-
 class EventType(Enum):
     """Event type enumeration."""
 
@@ -277,7 +276,7 @@ def __str__(self):
 
 
 class BaseUpdate(BaseMessage, metaclass=abc.ABCMeta):
-    """Split data update notification."""
+    """Feature flag data update notification."""
 
     def __init__(self, channel, timestamp, change_number):
         """
@@ -324,11 +323,14 @@ def change_number(self):
 
 
 class SplitChangeUpdate(BaseUpdate):
-    """Split Change notification."""
+    """Feature flag Change notification."""
 
-    def __init__(self, channel, timestamp, change_number):
+    def __init__(self, channel, timestamp, change_number, previous_change_number, feature_flag_definition, compression):
         """Class constructor."""
         BaseUpdate.__init__(self, channel, timestamp, change_number)
+        self._previous_change_number = previous_change_number
+        self._feature_flag_definition = feature_flag_definition
+        self._compression = compression
 
     @property
     def update_type(self):  # pylint:disable=no-self-use
@@ -340,18 +342,48 @@ def update_type(self):  # pylint:disable=no-self-use
         """
         return UpdateType.SPLIT_UPDATE
 
+    @property
+    def previous_change_number(self):  # pylint:disable=no-self-use
+        """
+        Return previous change number
+
+        :returns: The previous change number
+        :rtype: int
+        """
+        return self._previous_change_number
+
+    @property
+    def feature_flag_definition(self):  # pylint:disable=no-self-use
+        """
+        Return feature flag definition
+
+        :returns: The new feature flag definition
+        :rtype: str
+        """
+        return self._feature_flag_definition
+
+    @property
+    def compression(self):  # pylint:disable=no-self-use
+        """
+        Return previous compression type
+
+        :returns: The compression type
+        :rtype: int
+        """
+        return self._compression
+
     def __str__(self):
         """Return string representation."""
         return "SplitChange - changeNumber=%d" % (self.change_number)
 
 
 class SplitKillUpdate(BaseUpdate):
-    """Split Kill notification."""
+    """Feature flag Kill notification."""
 
-    def __init__(self, channel, timestamp, change_number, split_name, default_treatment):  # pylint:disable=too-many-arguments
+    def __init__(self, channel, timestamp, change_number, feature_flag_name, default_treatment):  # pylint:disable=too-many-arguments
         """Class constructor."""
         BaseUpdate.__init__(self, channel, timestamp, change_number)
-        self._split_name = split_name
+        self._feature_flag_name = feature_flag_name
         self._default_treatment = default_treatment
 
     @property
@@ -365,14 +397,14 @@ def update_type(self):  # pylint:disable=no-self-use
         return UpdateType.SPLIT_KILL
 
     @property
-    def split_name(self):
+    def feature_flag_name(self):
         """
-        Return the name of the killed split.
+        Return the name of the killed feature flag.
 
-        :returns: name of the killed split
+        :returns: name of the killed feature flag
         :rtype: str
         """
-        return self._split_name
+        return self._feature_flag_name
 
     @property
     def default_treatment(self):
@@ -387,7 +419,7 @@ def default_treatment(self):
     def __str__(self):
         """Return string representation."""
         return "SplitKill - changeNumber=%d, name=%s, defaultTreatment=%s" % \
-            (self.change_number, self.split_name, self.default_treatment)
+            (self.change_number, self.feature_flag_name, self.default_treatment)
 
 
 class SegmentChangeUpdate(BaseUpdate):
@@ -471,9 +503,9 @@ def _parse_update(channel, timestamp, data):
     """
     update_type = UpdateType(data['type'])
     change_number = data['changeNumber']
-    if update_type == UpdateType.SPLIT_UPDATE:
-        return SplitChangeUpdate(channel, timestamp, change_number)
-    elif update_type == UpdateType.SPLIT_KILL:
+    if update_type == UpdateType.SPLIT_UPDATE and change_number is not None:
+        return SplitChangeUpdate(channel, timestamp, change_number, data.get('pcn'), data.get('d'), data.get('c'))
+    elif update_type == UpdateType.SPLIT_KILL and change_number is not None:
         return SplitKillUpdate(channel, timestamp, change_number,
                                data['splitName'], data['defaultTreatment'])
     elif update_type == UpdateType.SEGMENT_UPDATE:

splitio/push/processor.py

@@ -6,47 +6,46 @@
 from splitio.push.splitworker import SplitWorker
 from splitio.push.segmentworker import SegmentWorker
 
-
 class MessageProcessor(object):
     """Message processor class."""
 
-    def __init__(self, synchronizer):
+    def __init__(self, synchronizer, telemetry_runtime_producer):
         """
         Class constructor.
 
         :param synchronizer: synchronizer component
         :type synchronizer: splitio.sync.synchronizer.Synchronizer
         """
-        self._split_queue = Queue()
+        self._feature_flag_queue = Queue()
         self._segments_queue = Queue()
         self._synchronizer = synchronizer
-        self._split_worker = SplitWorker(synchronizer.synchronize_splits, self._split_queue)
+        self._feature_flag_worker = SplitWorker(synchronizer.synchronize_splits, synchronizer.synchronize_segment, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, synchronizer.segment_storage, telemetry_runtime_producer)
         self._segments_worker = SegmentWorker(synchronizer.synchronize_segment, self._segments_queue)
         self._handlers = {
-            UpdateType.SPLIT_UPDATE: self._handle_split_update,
-            UpdateType.SPLIT_KILL: self._handle_split_kill,
+            UpdateType.SPLIT_UPDATE: self._handle_feature_flag_update,
+            UpdateType.SPLIT_KILL: self._handle_feature_flag_kill,
             UpdateType.SEGMENT_UPDATE: self._handle_segment_change
         }
 
-    def _handle_split_update(self, event):
+    def _handle_feature_flag_update(self, event):
         """
-        Handle incoming split update notification.
+        Handle incoming feature flag update notification.
 
-        :param event: Incoming split change event
+        :param event: Incoming feature flag change event
         :type event: splitio.push.parser.SplitChangeUpdate
         """
-        self._split_queue.put(event)
+        self._feature_flag_queue.put(event)
 
-    def _handle_split_kill(self, event):
+    def _handle_feature_flag_kill(self, event):
         """
-        Handle incoming split kill notification.
+        Handle incoming feature flag kill notification.
 
-        :param event: Incoming split kill event
+        :param event: Incoming feature flag kill event
         :type event: splitio.push.parser.SplitKillUpdate
         """
-        self._synchronizer.kill_split(event.split_name, event.default_treatment,
+        self._synchronizer.kill_split(event.feature_flag_name, event.default_treatment,
                                       event.change_number)
-        self._split_queue.put(event)
+        self._feature_flag_queue.put(event)
 
     def _handle_segment_change(self, event):
         """
@@ -65,10 +64,10 @@ def update_workers_status(self, enabled):
         :type enabled: bool
         """
         if enabled:
-            self._split_worker.start()
+            self._feature_flag_worker.start()
             self._segments_worker.start()
         else:
-            self._split_worker.stop()
+            self._feature_flag_worker.stop()
             self._segments_worker.stop()
 
     def handle(self, event):
@@ -86,6 +85,6 @@ def handle(self, event):
         handle(event)
 
     def shutdown(self):
-        """Stop splits & segments workers."""
-        self._split_worker.stop()
+        """Stop feature flags & segments workers."""
+        self._feature_flag_worker.stop()
         self._segments_worker.stop()