Add exposure deduping (#138)

* Add exposure deduping * feedback and cleanup

Author: tore-statsig

statsig/statsig_logger.py

@@ -1,11 +1,12 @@
 import collections
 import threading
-from typing import Optional
+from typing import Optional, Union
 
 from .retryable_logs import RetryableLogs
 from .evaluation_details import EvaluationDetails
 from .evaluator import _ConfigEvaluation
 from .statsig_event import StatsigEvent
+# from .statsig_user import StatsigUser
 from .layer import Layer
 from .utils import logger
 from .thread_util import spawn_background_thread, THREAD_JOIN_TIMEOUT
@@ -14,10 +15,11 @@
 _LAYER_EXPOSURE_EVENT = "statsig::layer_exposure"
 _GATE_EXPOSURE_EVENT = "statsig::gate_exposure"
 
+_IGNORED_METADATA_KEYS = {'serverTime', 'configSyncTime', 'initTime', 'reason'}
 
 def _safe_add_evaluation_to_event(
- evaluation_details: EvaluationDetails, event: StatsigEvent):
- if evaluation_details is None:
+ evaluation_details: Union[EvaluationDetails, None], event: StatsigEvent):
+ if evaluation_details is None or event is None or event.metadata is None:
  return
 
  event.metadata["reason"] = evaluation_details.reason
@@ -29,10 +31,12 @@ def _safe_add_evaluation_to_event(
 class _StatsigLogger:
  _background_flush: Optional[threading.Thread]
  _background_retry: Optional[threading.Thread]
+ _background_deduper: Optional[threading.Thread]
 
  def __init__(self, net, shutdown_event, statsig_metadata, error_boundary, options):
  self._events = []
  self._retry_logs = collections.deque(maxlen=10)
+ self._deduper = set()
  self._net = net
  self._statsig_metadata = statsig_metadata
  self._local_mode = options.local_mode
@@ -44,6 +48,7 @@ def __init__(self, net, shutdown_event, statsig_metadata, error_boundary, option
  self._shutdown_event = shutdown_event
  self._background_flush = None
  self._background_retry = None
+ self._background_deduper = None
  self.spawn_bg_threads_if_needed()
 
  def spawn_bg_threads_if_needed(self):
@@ -58,6 +63,10 @@ def spawn_bg_threads_if_needed(self):
  self._background_retry = spawn_background_thread(
  self._periodic_retry, (self._shutdown_event,), self._error_boundary)
 
+ if self._background_deduper is None or not self._background_deduper.is_alive():
+ self._background_deduper = spawn_background_thread(
+ self._periodic_dedupe_clear, (self._shutdown_event,), self._error_boundary)
+
  def log(self, event):
  if self._local_mode:
  return
@@ -73,6 +82,9 @@ def log_gate_exposure(self, user, gate, value, rule_id, secondary_exposures,
  "gateValue": "true" if value else "false",
  "ruleID": rule_id,
  }
+ if not self._is_unique_exposure(user, _GATE_EXPOSURE_EVENT, event.metadata):
+ return
+
  if is_manual_exposure:
  event.metadata["isManualExposure"] = "true"
 
@@ -90,6 +102,8 @@ def log_config_exposure(self, user, config, rule_id, secondary_exposures,
  "config": config,
  "ruleID": rule_id,
  }
+ if not self._is_unique_exposure(user, _CONFIG_EXPOSURE_EVENT, event.metadata):
+ return
  if is_manual_exposure:
  event.metadata["isManualExposure"] = "true"
 
@@ -111,20 +125,24 @@ def log_layer_exposure(self, user, layer: Layer, parameter_name: str,
  exposures = config_evaluation.secondary_exposures
  allocated_experiment = config_evaluation.allocated_experiment
 
- event.metadata = {
+ metadata = {
  "config": layer.name,
  "ruleID": layer.rule_id,
  "allocatedExperiment": allocated_experiment,
  "parameterName": parameter_name,
  "isExplicitParameter": "true" if is_explicit else "false"
  }
+ if not self._is_unique_exposure(user, _LAYER_EXPOSURE_EVENT, metadata):
+ return
+ event.metadata = metadata
  if is_manual_exposure:
  event.metadata["isManualExposure"] = "true"
 
  event._secondary_exposures = [] if exposures is None else exposures
 
  _safe_add_evaluation_to_event(
  config_evaluation.evaluation_details, event)
+
  self.log(event)
 
  def flush(self):
@@ -157,11 +175,19 @@ def _periodic_flush(self, shutdown_event):
  except Exception as e:
  self._error_boundary.log_exception(e)
 
+ def _periodic_dedupe_clear(self, shutdown_event):
+ while True:
+ try:
+ if shutdown_event.wait(self._logging_interval):
+ break
+ self._deduper = set()
+ except Exception as e:
+ self._error_boundary.log_exception(e)
+
  def _periodic_retry(self, shutdown_event):
  while True:
  if shutdown_event.wait(self._retry_interval):
  break
-
  length = len(self._retry_logs)
  for _i in range(length):
  try:
@@ -178,3 +204,24 @@ def _periodic_retry(self, shutdown_event):
  return
 
  self._retry_logs.append(RetryableLogs(retry_logs.payload, retry_logs.retries))
+
+ def _is_unique_exposure(self, user, eventName: str, metadata: dict or None) -> bool:
+ if user is None:
+ return True
+ if len(self._deduper) > 10000:
+ self._deduper = set()
+ custom_id_key = ''
+ if user.custom_ids and isinstance(user.custom_ids, dict):
+ custom_id_key = ','.join(user.custom_ids.values())
+
+ metadata_key = ''
+ if metadata and isinstance(metadata, dict):
+ metadata_key = ','.join(str(value) for key, value in metadata.items() if key not in _IGNORED_METADATA_KEYS)
+
+ key = ','.join(str(item) for item in [user.user_id, custom_id_key, eventName, metadata_key])
+
+ if key in self._deduper:
+ return False
+
+ self._deduper.add(key)
+ return True

statsig/statsig_server.py

@@ -37,7 +37,8 @@ def __init__(self) -> None:
  self._initialized = False
 
  def initialize_with_timeout(self, sdkKey: str, options=None):
- thread = threading.Thread(target=self.initialize, args=(sdkKey, options))
+ thread = threading.Thread(
+ target=self.initialize, args=(sdkKey, options))
  thread.start()
  thread.join(timeout=options.init_timeout)
  if thread.is_alive():
@@ -62,7 +63,8 @@ def initialize(self, sdkKey: str, options=None):
  self._options = options
  self.__shutdown_event = threading.Event()
  self.__statsig_metadata = _StatsigMetadata.get()
- self._network = _StatsigNetwork(sdkKey, options, self._errorBoundary)
+ self._network = _StatsigNetwork(
+ sdkKey, options, self._errorBoundary)
  self._logger = _StatsigLogger(
  self._network, self.__shutdown_event, self.__statsig_metadata, self._errorBoundary,
  options)
@@ -81,7 +83,8 @@ def task():
  if not self._verify_inputs(user, gate_name):
  return False
 
- result = self.__check_gate_server_fallback(user, gate_name, log_exposure)
+ result = self.__check_gate_server_fallback(
+ user, gate_name, log_exposure)
  return result.boolean_value
 
  return self._errorBoundary.capture(task, lambda: False)
@@ -98,7 +101,8 @@ def task():
  if not self._verify_inputs(user, config_name):
  return DynamicConfig({}, config_name, "")
 
- result = self.__get_config_server_fallback(user, config_name, log_exposure)
+ result = self.__get_config_server_fallback(
+ user, config_name, log_exposure)
  return DynamicConfig(
  result.json_value, config_name, result.rule_id)
 
@@ -173,7 +177,8 @@ def task():
  self._errorBoundary.swallow(task)
 
  def flush(self):
- self._logger.flush()
+ if self._logger is not None:
+ self._logger.flush()
 
  def shutdown(self):
  def task():
@@ -272,8 +277,10 @@ def _verify_inputs(self, user: StatsigUser, variable_name: str):
  return True
 
  def _verify_bg_threads_running(self):
- self._logger.spawn_bg_threads_if_needed()
- self._spec_store.spawn_bg_threads_if_needed()
+ if self._logger is not None:
+ self._logger.spawn_bg_threads_if_needed()
+ if self._spec_store is not None:
+ self._spec_store.spawn_bg_threads_if_needed()
 
  def __check_gate_server_fallback(
  self, user: StatsigUser, gate_name: str, log_exposure=True):

tests/test_concurrency.py

@@ -1,3 +1,4 @@
+import random
 import threading
 import time
 import os
@@ -92,22 +93,25 @@ def test_checking_and_updating_concurrently(self, mock_post, mock_get):
  for t in self.threads:
  t.join()
 
- self.assertEqual(200, len(statsig.get_instance()._logger._events))
+ self.assertEqual(201, len(statsig.get_instance()._logger._events))
  self.assertEqual(1600, self._event_count)
  statsig.shutdown()
 
  self.assertEqual(0, len(statsig.get_instance()._logger._events))
- self.assertEqual(1800, self._event_count)
+ self.assertEqual(1801, self._event_count)
 
  def run_checks(self, interval, times):
  for x in range(times):
+ salt = str(random.randint(1, 10000000000))
  user = StatsigUser(
- f'user_id_{x}', email="testuser@statsig.com", private_attributes={"test": 123})
+ f'user_id_{x}', email="testuser@statsig.com", private_attributes={"test": 123}, custom_ids={'salt': salt})
  statsig.log_event(StatsigEvent(
  user, "test_event", 1, {"key": "value"}))
  self.assertEqual(True, statsig.check_gate(
  user, "on_for_statsig_email"))
  self.assertEqual(True, statsig.check_gate(user, "always_on_gate"))
+ self.assertFalse(statsig.check_gate(
+ StatsigUser(f'user_id_{salt}{x}'), "on_for_id_list"))
  self.assertTrue(statsig.check_gate(
  StatsigUser("regular_user_id"), "on_for_id_list"))
 

tests/test_logger.py

@@ -1,3 +1,4 @@
+import random
 import unittest
 from unittest.mock import patch
 
@@ -36,23 +37,63 @@ def on_log(url: str, data: dict):
  @patch('requests.post', side_effect=_network_stub.mock)
  def test_log_size(self, mock_post):
  self._instance.check_gate(self._user, "a_gate")
- self._instance.check_gate(self._user, "a_gate")
+ self._instance.check_gate(self._user, "b_gate")
 
  self.assertEqual(len(self._events), 0)
 
- self._instance.check_gate(self._user, "a_gate")
+ self._instance.check_gate(self._user, "c_gate")
  self.assertEqual(len(self._events), 3)
 
- self._instance.check_gate(self._user, "a_gate")
+ self._instance.check_gate(self._user, "d_gate")
  self.assertEqual(len(self._events), 3)
+ self._instance.check_gate(self._user, "e_gate")
+ self._instance.check_gate(self._user, "f_gate")
+ self.assertEqual(len(self._events), 6)
+ 
+ @patch('requests.post', side_effect=_network_stub.mock)
+ def test_exposure_dedupe(self, mock_post):
  self._instance.check_gate(self._user, "a_gate")
  self._instance.check_gate(self._user, "a_gate")
- self.assertEqual(len(self._events), 6)
+
+ self.assertEqual(len(self._events), 0)
 
  self._instance.check_gate(self._user, "a_gate")
+ # doesnt flush yet, because they are deduped
+ self.assertEqual(len(self._events), 0)
+
+ self._instance.check_gate(self._user, "b_gate")
+ self.assertEqual(len(self._events), 0)
+ self._instance.check_gate(self._user, "c_gate")
+ self.assertEqual(len(self._events), 3)
+ self._instance.check_gate(self._user, "a_gate")
+ self._instance.check_gate(self._user, "b_gate")
+ self._instance.check_gate(self._user, "c_gate")
+ self.assertEqual(len(self._events), 3)
+
+ self._instance.get_config(self._user, "a_gate")
+ self._instance.get_config(self._user, "b_gate")
+ self._instance.get_config(self._user, "b_gate")
+ self._instance.get_config(self._user, "b_gate")
+ self._instance.get_config(self._user, "a_gate")
+ self._instance.get_config(self._user, "c_gate")
+ self.assertEqual(len(self._events), 6)
+
+ self._instance.check_gate(self._user, "d_gate")
  self._instance.flush()
  self.assertEqual(len(self._events), 7)
 
+ # get layer does not expose
+ self._instance.get_layer(self._user, "a_gate")
+ self._instance.get_layer(self._user, "b_gate")
+ self._instance.get_layer(self._user, "c_gate")
+ self.assertEqual(len(self._events), 7)
+
+ self._instance.get_experiment(StatsigUser(str(random.randint(1, 10000000000))), "a_gate")
+ self._instance.get_experiment(StatsigUser(str(random.randint(1, 10000000000))), "a_gate")
+ self._instance.get_experiment(StatsigUser(str(random.randint(1, 10000000000))), "a_gate")
+ self.assertEqual(len(self._events), 10)
+
+
  @patch('requests.post', side_effect=_network_stub.mock)
  def test_log_content(self, mock_post):
  self._instance.check_gate(self._user, "a_gate")