Update rb segment matcher

Author: chillaq

splitio/engine/evaluator.py

@@ -11,7 +11,7 @@
 from splitio.optional.loaders import asyncio
 
 CONTROL = 'control'
-EvaluationContext = namedtuple('EvaluationContext', ['flags', 'segment_memberships', 'segment_rbs_memberships', 'segment_rbs_conditions', 'excluded_rbs_segments'])
+EvaluationContext = namedtuple('EvaluationContext', ['flags', 'segment_memberships', 'rbs_segments'])
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -115,59 +115,24 @@ def context_for(self, key, feature_names):
         :rtype: EvaluationContext
         """
         pending = set(feature_names)
+        pending_rbs = set()
         splits = {}
+        rb_segments = {}
         pending_memberships = set()
-        pending_rbs_memberships = set()
-        while pending:
+        while pending or pending_rbs:
             fetched = self._flag_storage.fetch_many(list(pending))
-            features = filter_missing(fetched)
-            splits.update(features)
-            pending = set()
-            for feature in features.values():
-                cf, cs, crbs = get_dependencies(feature)
-                for rbs in crbs:
-                    rbs_cf, rbs_cs, rbs_crbs = get_dependencies(self._rbs_segment_storage.get(rbs))                
-                    cf.extend(rbs_cf)
-                    cs.extend(rbs_cs)
-                    crbs.extend(rbs_crbs)
-
-                pending.update(filter(lambda f: f not in splits, cf))
-                pending_memberships.update(cs)
-                pending_rbs_memberships.update(crbs)
-        
-        rbs_segment_memberships = {}
-        rbs_segment_conditions = {}
-        excluded_rbs_segments = set()
-        key_membership = False
-        segment_memberhsip = False
-        for rbs_segment in pending_rbs_memberships:
-            rbs_segment_obj = self._rbs_segment_storage.get(rbs_segment)
-            pending_memberships.update(rbs_segment_obj.get_condition_segment_names())
-            
-            key_membership = key in rbs_segment_obj.excluded.get_excluded_keys()
-            segment_memberhsip = False
-            for excluded_segment in rbs_segment_obj.excluded.get_excluded_segments():
-                if excluded_segment.type == SegmentType.STANDARD and self._segment_storage.segment_contains(excluded_segment.name, key):
-                    segment_memberhsip = True
-
-                if excluded_segment.type == SegmentType.RULE_BASED:
-                    rbs_segment = self._rbs_segment_storage.get(excluded_segment.name)
-                    if rbs_segment is not None:
-                        excluded_rbs_segments.add(rbs_segment)
-                
-            rbs_segment_memberships.update({rbs_segment: segment_memberhsip or key_membership})
-            if not (segment_memberhsip or key_membership):
-                rbs_segment_conditions.update({rbs_segment: [condition for condition in rbs_segment_obj.conditions]})
-            
+            fetched_rbs = self._rbs_segment_storage.fetch_many(list(pending_rbs))
+            features, rbsegments, splits, rb_segments = update_objects(fetched, fetched_rbs, splits, rb_segments)
+            pending, pending_memberships, pending_rbs = get_pending_objects(features, splits, rbsegments, rb_segments, pending_memberships)
+                        
         return EvaluationContext(
             splits, 
             { segment: self._segment_storage.segment_contains(segment, key)
                 for segment in pending_memberships
             },
-            rbs_segment_memberships,
-            rbs_segment_conditions,
-            excluded_rbs_segments        
+            rb_segments
         )
+        
 
 class AsyncEvaluationDataFactory:
 
@@ -186,72 +151,36 @@ async def context_for(self, key, feature_names):
         :rtype: EvaluationContext
         """
         pending = set(feature_names)
+        pending_rbs = set()
         splits = {}
+        rb_segments = {}
         pending_memberships = set()
-        pending_rbs_memberships = set()
-        while pending:
+        while pending or pending_rbs:
             fetched = await self._flag_storage.fetch_many(list(pending))
-            features = filter_missing(fetched)
-            splits.update(features)
-            pending = set()
-            for feature in features.values():
-                cf, cs, crbs = get_dependencies(feature)     
-                for rbs in crbs:
-                    rbs_cf, rbs_cs, rbs_crbs = get_dependencies(await self._rbs_segment_storage.get(rbs))                
-                    cf.extend(rbs_cf)
-                    cs.extend(rbs_cs)
-                    crbs.extend(rbs_crbs)
-               
-                pending.update(filter(lambda f: f not in splits, cf))
-                pending_memberships.update(cs)
-                pending_rbs_memberships.update(crbs)
-                
-        rbs_segment_memberships = {}
-        rbs_segment_conditions = {}
-        excluded_rbs_segments = set()
-        key_membership = False
-        segment_memberhsip = False
-        for rbs_segment in pending_rbs_memberships:
-            rbs_segment_obj = await self._rbs_segment_storage.get(rbs_segment)
-            pending_memberships.update(rbs_segment_obj.get_condition_segment_names())
-            
-            key_membership = key in rbs_segment_obj.excluded.get_excluded_keys()
-            segment_memberhsip = False
-            for excluded_segment in rbs_segment_obj.excluded.get_excluded_segments():
-                if excluded_segment.type == SegmentType.STANDARD and await self._segment_storage.segment_contains(excluded_segment.name, key):
-                    segment_memberhsip = True
-
-                if excluded_segment.type == SegmentType.RULE_BASED:
-                    rbs_segment = await self._rbs_segment_storage.get(excluded_segment.name)
-                    if rbs_segment is not None:
-                        excluded_rbs_segments.add(rbs_segment)
-                            
-            rbs_segment_memberships.update({rbs_segment: segment_memberhsip or key_membership})
-            if not (segment_memberhsip or key_membership):
-                rbs_segment_conditions.update({rbs_segment: [condition for condition in rbs_segment_obj.conditions]})
+            fetched_rbs = await self._rbs_segment_storage.fetch_many(list(pending_rbs))
+            features, rbsegments, splits, rb_segments = update_objects(fetched, fetched_rbs, splits, rb_segments)
+            pending, pending_memberships, pending_rbs = get_pending_objects(features, splits, rbsegments, rb_segments, pending_memberships)
 
         segment_names = list(pending_memberships)
         segment_memberships = await asyncio.gather(*[
             self._segment_storage.segment_contains(segment, key)
             for segment in segment_names
         ])
+        
         return EvaluationContext(
             splits, 
             dict(zip(segment_names, segment_memberships)),
-            rbs_segment_memberships,
-            rbs_segment_conditions,
-            excluded_rbs_segments        
+            rb_segments
         )
 
-
-def get_dependencies(feature):
+def get_dependencies(object):
     """
     :rtype: tuple(list, list)
     """
     feature_names = []
     segment_names = []
     rbs_segment_names = []
-    for condition in feature.conditions:
+    for condition in object.conditions:
         for matcher in condition.matchers:
             if isinstance(matcher,RuleBasedSegmentMatcher):
                 rbs_segment_names.append(matcher._rbs_segment_name)
@@ -264,3 +193,34 @@ def get_dependencies(feature):
 
 def filter_missing(features):
     return {k: v for (k, v) in features.items() if v is not None}
+
+def get_pending_objects(features, splits, rbsegments, rb_segments, pending_memberships):
+    pending = set()
+    pending_rbs = set()
+    for feature in features.values():
+        cf, cs, crbs = get_dependencies(feature)
+        pending.update(filter(lambda f: f not in splits, cf))
+        pending_memberships.update(cs)
+        pending_rbs.update(filter(lambda f: f not in rb_segments, crbs))
+
+    for rb_segment in rbsegments.values():
+        cf, cs, crbs = get_dependencies(rb_segment)
+        pending.update(filter(lambda f: f not in splits, cf))
+        pending_memberships.update(cs)
+        for excluded_segment in rb_segment.excluded.get_excluded_segments():
+            if excluded_segment.type == SegmentType.STANDARD:
+                pending_memberships.add(excluded_segment.name)
+            else:
+                pending_rbs.update(filter(lambda f: f not in rb_segments, [excluded_segment.name]))
+        pending_rbs.update(filter(lambda f: f not in rb_segments, crbs))
+    
+    return pending, pending_memberships, pending_rbs
+    
+def update_objects(fetched, fetched_rbs, splits, rb_segments):
+    features = filter_missing(fetched)
+    rbsegments = filter_missing(fetched_rbs)
+    splits.update(features)
+    rb_segments.update(rbsegments)
+    
+    return features, rbsegments, splits, rb_segments
+    

splitio/models/grammar/matchers/rule_based_segment.py

@@ -1,5 +1,6 @@
 """Rule based segment matcher classes."""
 from splitio.models.grammar.matchers.base import Matcher
+from splitio.models.rule_based_segments import SegmentType
 
 class RuleBasedSegmentMatcher(Matcher):
 
@@ -29,15 +30,15 @@ def _match(self, key, attributes=None, context=None):
         if self._rbs_segment_name == None:
             return False
 
-        # Check if rbs segment has exclusions
-        if context['ec'].segment_rbs_memberships.get(self._rbs_segment_name):
-            return False
-        
-        for rbs_segment in context['ec'].excluded_rbs_segments:
-            if self._match_conditions(rbs_segment.conditions, key, attributes, context):
-                return True
+        rb_segment = context['ec'].rbs_segments.get(self._rbs_segment_name)
 
-        return self._match_conditions(context['ec'].segment_rbs_conditions.get(self._rbs_segment_name), key, attributes, context):
+        if key in rb_segment.excluded.get_excluded_keys():
+            return False 
+
+        if self._match_dep_rb_segments(rb_segment.excluded.get_excluded_segments(), key, attributes, context):
+            return False
+                
+        return self._match_conditions(rb_segment.conditions, key, attributes, context)
 
     def _add_matcher_specific_properties_to_json(self):
         """Return UserDefinedSegment specific properties."""
@@ -51,3 +52,23 @@ def _match_conditions(self, rbs_segment_conditions, key, attributes, context):
         for parsed_condition in rbs_segment_conditions:
             if parsed_condition.matches(key, attributes, context):
                 return True
+            
+        return False
+            
+    def _match_dep_rb_segments(self, excluded_rb_segments, key, attributes, context):
+        for excluded_rb_segment in excluded_rb_segments:
+            if excluded_rb_segment.type == SegmentType.STANDARD:
+                if context['ec'].segment_memberships[excluded_rb_segment.name]:
+                    return True
+            else:
+                excluded_segment = context['ec'].rbs_segments.get(excluded_rb_segment.name)
+                if key in excluded_segment.excluded.get_excluded_keys():
+                    return True
+
+                if self._match_dep_rb_segments(excluded_segment.excluded.get_excluded_segments(), key, attributes, context):
+                    return True
+                    
+                if self._match_conditions(excluded_segment.conditions, key, attributes, context):
+                    return True 
+
+        return False

splitio/storage/inmemmory.py

@@ -230,6 +230,9 @@ def contains(self, segment_names):
         """
         with self._lock:
             return set(segment_names).issubset(self._rule_based_segments.keys())
+        
+    def fetch_many(self, segment_names):
+        return {rb_segment_name: self.get(rb_segment_name) for rb_segment_name in segment_names}
 
 class InMemoryRuleBasedSegmentStorageAsync(RuleBasedSegmentsStorage):
     """InMemory implementation of a feature flag storage base."""    
@@ -354,6 +357,9 @@ async def contains(self, segment_names):
         async with self._lock:
             return set(segment_names).issubset(self._rule_based_segments.keys())
 
+    async def fetch_many(self, segment_names):
+        return {rb_segment_name: await self.get(rb_segment_name) for rb_segment_name in segment_names}
+
 class InMemorySplitStorageBase(SplitStorage):
     """InMemory implementation of a feature flag storage base."""
 

splitio/storage/pluggable.py

@@ -177,6 +177,25 @@ def get_segment_names(self):
             _LOGGER.error('Error getting rule based segments names from storage')
             _LOGGER.debug('Error: ', exc_info=True)
             return None
+        
+    def fetch_many(self, rb_segment_names):
+        """
+        Retrieve rule based segments.
+
+        :param rb_segment_names: Names of the rule based segments to fetch.
+        :type rb_segment_names: list(str)
+
+        :return: A dict with rule based segment objects parsed from queue.
+        :rtype: dict(rb_segment_names, splitio.models.rile_based_segment.RuleBasedSegment)
+        """
+        try:
+            prefix_added = [self._prefix.format(segment_name=rb_segment_name) for rb_segment_name in rb_segment_names]
+            return {rb_segment['name']: rule_based_segments.from_raw(rb_segment) for rb_segment in self._pluggable_adapter.get_many(prefix_added)}
+
+        except Exception:
+            _LOGGER.error('Error getting rule based segments from storage')
+            _LOGGER.debug('Error: ', exc_info=True)
+            return None        
 
 class PluggableRuleBasedSegmentsStorageAsync(PluggableRuleBasedSegmentsStorageBase):
     """Pluggable storage for rule based segments."""
@@ -256,6 +275,25 @@ async def get_segment_names(self):
             _LOGGER.debug('Error: ', exc_info=True)
             return None
 
+    async def fetch_many(self, rb_segment_names):
+        """
+        Retrieve rule based segments.
+
+        :param rb_segment_names: Names of the rule based segments to fetch.
+        :type rb_segment_names: list(str)
+
+        :return: A dict with rule based segment objects parsed from queue.
+        :rtype: dict(rb_segment_names, splitio.models.rile_based_segment.RuleBasedSegment)
+        """
+        try:
+            prefix_added = [self._prefix.format(segment_name=rb_segment_name) for rb_segment_name in rb_segment_names]
+            return {rb_segment['name']: rule_based_segments.from_raw(rb_segment) for rb_segment in await self._pluggable_adapter.get_many(prefix_added)}
+
+        except Exception:
+            _LOGGER.error('Error getting rule based segments from storage')
+            _LOGGER.debug('Error: ', exc_info=True)
+            return None        
+
 class PluggableSplitStorageBase(SplitStorage):
     """InMemory implementation of a feature flag storage."""
 

splitio/storage/redis.py

@@ -131,6 +131,35 @@ def get_large_segment_names(self):
         """
         pass    
 
+    def fetch_many(self, segment_names):
+        """
+        Retrieve rule based segment.
+
+        :param segment_names: Names of the rule based segments to fetch.
+        :type segment_names: list(str)
+
+        :return: A dict with rule based segment objects parsed from redis.
+        :rtype: dict(segment_name, splitio.models.rule_based_segment.RuleBasedSegment)
+        """
+        to_return = dict()
+        try:
+            keys = [self._get_key(segment_name) for segment_name in segment_names]
+            raw_rbs_segments = self._redis.mget(keys)
+            _LOGGER.debug("Fetchting rule based segment [%s] from redis" % segment_names)
+            _LOGGER.debug(raw_rbs_segments)
+            for i in range(len(raw_rbs_segments)):
+                rbs_segment = None
+                try:
+                    rbs_segment = rule_based_segments.from_raw(json.loads(raw_rbs_segments[i]))
+                except (ValueError, TypeError):
+                    _LOGGER.error('Could not parse rule based segment.')
+                    _LOGGER.debug("Raw rule based segment that failed parsing attempt: %s", raw_rbs_segments[i])
+                to_return[segment_names[i]] = rbs_segment
+        except RedisAdapterException:
+            _LOGGER.error('Error fetching rule based segments from storage')
+            _LOGGER.debug('Error: ', exc_info=True)
+        return to_return
+
 class RedisRuleBasedSegmentsStorageAsync(RuleBasedSegmentsStorage):
     """Redis-based storage for rule based segments."""
 
@@ -246,6 +275,35 @@ async def get_large_segment_names(self):
         """
         pass    
 
+    async def fetch_many(self, segment_names):
+        """
+        Retrieve rule based segment.
+
+        :param segment_names: Names of the rule based segments to fetch.
+        :type segment_names: list(str)
+
+        :return: A dict with rule based segment objects parsed from redis.
+        :rtype: dict(segment_name, splitio.models.rule_based_segment.RuleBasedSegment)
+        """
+        to_return = dict()
+        try:
+            keys = [self._get_key(segment_name) for segment_name in segment_names]
+            raw_rbs_segments = await self._redis.mget(keys)
+            _LOGGER.debug("Fetchting rule based segment [%s] from redis" % segment_names)
+            _LOGGER.debug(raw_rbs_segments)
+            for i in range(len(raw_rbs_segments)):
+                rbs_segment = None
+                try:
+                    rbs_segment = rule_based_segments.from_raw(json.loads(raw_rbs_segments[i]))
+                except (ValueError, TypeError):
+                    _LOGGER.error('Could not parse rule based segment.')
+                    _LOGGER.debug("Raw rule based segment that failed parsing attempt: %s", raw_rbs_segments[i])
+                to_return[segment_names[i]] = rbs_segment
+        except RedisAdapterException:
+            _LOGGER.error('Error fetching rule based segments from storage')
+            _LOGGER.debug('Error: ', exc_info=True)
+        return to_return
+
 class RedisSplitStorageBase(SplitStorage):
     """Redis-based storage base for feature flags."""