FFM-10837 Harden Metrics / Support anonymous targets (#88)

Author: erdirowlands

docs/further_reading.md

@@ -25,6 +25,15 @@ You can pass the configuration in as options when the SDK client is created.
 | pollInterval    | with_poll_interval(120)                                  | When running in stream mode, the interval in seconds that we poll for changes.                                                                   | 60                                   |
 | maxAuthRetries  | with_max_auth_retries(10)                                | The number of retry attempts to make if client authentication fails on a retryable HTTP error                                                    | 10                                   |
 
+# Anonymous Target
+
+If you do not want a `Target` to be sent to Harness servers, you can use the `anonymous` attribute. 
+
+```python
+    target = Target(identifier='my_identifier', name='my_name', anonymous=True)
+```            
+
+
 ## Logging Configuration
 The SDK provides a logger that wraps the standard python logging package.  You can import and use it with:
 ```python

featureflags/__init__.py

@@ -2,4 +2,4 @@
 
 __author__ = """Enver Bisevac"""
 __email__ = "[email protected]"
-__version__ = '1.4.0'
+__version__ = '1.5.0'

featureflags/analytics.py

@@ -1,6 +1,6 @@
 import time
 from threading import Lock, Thread
-from typing import Dict, List, Union
+from typing import Dict, List, Union, Set
 import concurrent.futures
 
 import attr
@@ -23,7 +23,8 @@
 from .sdk_logging_codes import info_metrics_thread_started, \
     info_metrics_success, warn_post_metrics_failed, \
     info_metrics_thread_existed, info_metrics_target_exceeded, \
-    warn_post_metrics_target_batch_failed, info_metrics_target_batch_success
+    warn_post_metrics_target_batch_failed, info_metrics_target_batch_success, \
+    info_evaluation_metrics_exceeded
 from .util import log
 
 FF_METRIC_TYPE = 'FFMETRICS'
@@ -33,7 +34,7 @@
 VARIATION_VALUE_ATTRIBUTE = 'variationValue'
 TARGET_ATTRIBUTE = 'target'
 SDK_VERSION_ATTRIBUTE = 'SDK_VERSION'
-SDK_VERSION = '1.4.0'
+SDK_VERSION = '1.5.0'
 SDK_TYPE_ATTRIBUTE = 'SDK_TYPE'
 SDK_TYPE = 'server'
 SDK_LANGUAGE_ATTRIBUTE = 'SDK_LANGUAGE'
@@ -64,13 +65,28 @@ def __init__(self, config: Config, client: AuthenticatedClient,
         self._config = config
         self._client = client
         self._environment = environment
+
+        # Evaluation metrics
         self._data: Dict[str, AnalyticsEvent] = {}
+        # This allows for up to 2K flags with 5 variations each per interval
+        self._max_evaluation_metrics = 10000
+        self._max_evaluation_metrics_exceeded = False
+
+        # Target metrics - batch based
         self._target_data_batches: List[Dict[str, MetricTargetData]] = [{}]
-        self._max_number_of_batches = 200
-        self._max_batch_size = 1000
-        self._current_batch_index = 0
+
+        # This allows for 100k unique targets per interval
+        self._max_number_of_target_batches = 100
+        self._max_target_batch_size = 1000
+        self._current_target_batch_index = 0
         self.max_target_data_exceeded = False
 
+        # Allow us to track targets for the life of the SDK instance, to
+        # prevent sending the same target more than once. This also helps
+        # unique targets get processed each interval, because duplicate
+        # targets are not making up the 100K limit.
+        self._seen_targets: Set[str] = set()
+
         self._running = False
         self._runner = Thread(target=self._sync)
         self._runner.daemon = True
@@ -86,28 +102,49 @@ def enqueue(self, target: Target, identifier: str,
 
         self._lock.acquire()
         try:
-            # Store unique evaluation events. We map a unique evaluation
-            # event to its count.
-            unique_evaluation_key = self.get_key(event)
-            if unique_evaluation_key in self._data:
-                self._data[unique_evaluation_key].count += 1
+            # Check if adding a new metric would exceed the 10,000 limit
+            if len(self._data) < self._max_evaluation_metrics:
+
+                # Store unique evaluation events. We map a unique evaluation
+                # event to its count.
+                unique_evaluation_key = self.get_key(event)
+                if unique_evaluation_key in self._data:
+                    self._data[unique_evaluation_key].count += 1
+                else:
+                    event.count = 1
+                    self._data[unique_evaluation_key] = event
             else:
-                event.count = 1
-                self._data[unique_evaluation_key] = event
+                if not self._max_evaluation_metrics_exceeded:
+                    self._max_evaluation_metrics_exceeded = True
+                    info_evaluation_metrics_exceeded()
+
+            # Don't store this target if it is anonymous. Note, we currently
+            # don't do this check above for evaluation metrics, because we use
+            # the global target.
+            if target.anonymous:
+                return
+
+            unique_target_key = self.get_target_key(event)
 
-            # Check if we're on our final batch - if we are, and we've
+            # If we've seen this target before, don't process it
+            if unique_target_key in self._seen_targets:
+                return
+
+            self._seen_targets.add(unique_target_key)
+
+            # Check if we're on our final target batch - if we are, and we've
             # exceeded the max batch size just return early.
-            if len(self._target_data_batches) >= self._max_number_of_batches:
+            if len(self._target_data_batches) >= \
+                    self._max_number_of_target_batches:
                 if len(self._target_data_batches[
-                           self._current_batch_index]) >= \
-                        self._max_batch_size:
+                           self._current_target_batch_index]) >= \
+                        self._max_target_batch_size:
                     if not self.max_target_data_exceeded:
                         self.max_target_data_exceeded = True
                         info_metrics_target_exceeded()
                     return
 
             if event.target is not None and not event.target.anonymous:
-                unique_target_key = self.get_target_key(event)
 
                 # Store unique targets. If the target already exists
                 # in any of the batches, don't continue processing it
@@ -118,16 +155,17 @@ def enqueue(self, target: Target, identifier: str,
                 # If we've exceeded the max batch size for the current
                 # batch, then create a new batch and start using it.
                 if len(self._target_data_batches[
-                           self._current_batch_index]) >= self._max_batch_size:
+                           self._current_target_batch_index]) >= \
+                        self._max_target_batch_size:
                     self._target_data_batches.append({})
-                    self._current_batch_index += 1
+                    self._current_target_batch_index += 1
 
                 target_name = event.target.name
                 # If the target has no name use the identifier
                 if not target_name:
                     target_name = event.target.identifier
                 self._target_data_batches[
-                    self._current_batch_index][unique_target_key] = \
+                    self._current_target_batch_index][unique_target_key] = \
                     MetricTargetData(
                         identifier=event.target.identifier,
                         name=target_name,
@@ -209,8 +247,9 @@ def _send_data(self) -> None:
         finally:
             self._data = {}
             self._target_data_batches = [{}]
-            self._current_batch_index = 0
+            self._current_target_batch_index = 0
             self.max_target_data_exceeded = False
+            self._max_evaluation_metrics_exceeded = False
             self._lock.release()
 
         body: Metrics = Metrics(target_data=target_data,
@@ -290,6 +329,13 @@ def process_target(self, target_data, unique_target):
         )
         target_data.append(td)
 
+    def is_target_seen(self, target: AnalyticsEvent) -> bool:
+        unique_target_key = self.get_target_key(target)
+
+        with self._lock:
+            seen = unique_target_key in self._seen_targets
+        return seen
+
     def close(self) -> None:
         self._running = False
         if len(self._data) > 0:

featureflags/client.py

@@ -316,8 +316,8 @@ def number_variation(self, identifier: str, target: Target,
             return default
 
     def int_or_float_variation(self, identifier: str, target: Target,
-                                default: Union[float, int]) -> Union[
-        float, int]:
+                               default: Union[float, int]) -> \
+            Union[float, int]:
 
         # If initialization has failed, then return the default variation
         # immediately

featureflags/evaluations/evaluator.py

@@ -14,7 +14,7 @@
                                                 STARTS_WITH_OPERATOR)
 from featureflags.evaluations.distribution import Distribution
 from featureflags.evaluations.enum import FeatureState
-from featureflags.evaluations.feature import FeatureConfig
+from featureflags.evaluations.feature import FeatureConfig, FeatureConfigKind
 from featureflags.evaluations.serving_rule import ServingRule, ServingRules
 from featureflags.evaluations.variation import Variation
 from featureflags.evaluations.variation_map import VariationMap
@@ -38,7 +38,7 @@ class Evaluator(object):
     def __init__(self, provider: QueryInterface):
         self.provider = provider
 
-    def get_kind(self, identifier) -> Optional[str]:
+    def get_kind(self, identifier) -> Optional[FeatureConfigKind]:
         fc = self.provider.get_flag(identifier)
         if not fc:
             return None

featureflags/sdk_logging_codes.py

@@ -41,6 +41,8 @@ def get_sdk_code_message(key):
               "analytics interval will not be sent",
         7005: "Target metrics batches succeeded:",
         7006: "Target metrics batch/batches failed:",
+        7007: "Evaluation metrics exceeded max size, remaining unique "
+              "evaluations for this analytics interval will not be sent",
         # SDK_CACHE_8xxx
         8005: "Fetching feature by identifier attempt",
         8006: "Fetching segment by identifier attempt",
@@ -127,6 +129,10 @@ def info_metrics_target_batch_success(message):
     log.info(sdk_err_msg(7005, message))
 
 
+def info_evaluation_metrics_exceeded():
+    log.info(sdk_err_msg(7007))
+
+
 def info_metrics_thread_existed():
     log.info(sdk_err_msg(7001))
 

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 1.4.0
+current_version = 1.5.0
 commit = True
 tag = True
 

setup.py

@@ -57,6 +57,6 @@
     test_suite="tests",
     tests_require=test_requirements,
     url="https://github.com/harness/ff-python-server-sdk",
-    version='1.4.0',
+    version='1.5.0',
     zip_safe=False,
 )

tests/unit/test_sdk_logging_codes.py

@@ -16,6 +16,7 @@ def test_logs_dont_raise_exception():
     sdk_codes.info_metrics_thread_started(10)
     sdk_codes.info_metrics_thread_existed()
     sdk_codes.info_metrics_success()
+    sdk_codes.info_evaluation_metrics_exceeded()
     sdk_codes.info_metrics_target_exceeded()
     sdk_codes.warn_auth_failed_srv_defaults()
     sdk_codes.warn_failed_init_auth_error()