feat(event_processor): add forwarding event processor and integrate with optimizely (#205)

Author: rashidsp

optimizely/event/event_processor.py

@@ -117,11 +117,11 @@ def _validate_intantiation_props(self, prop, prop_name):
       prop_name: Property name.
 
     Returns:
-      False if property value is None or less than 1 or not a finite number.
+      False if property value is None or less than or equal to 0 or not a finite number.
       False if property name is batch_size and value is a floating point number.
       True otherwise.
     """
-    if (prop_name == 'batch_size' and not isinstance(prop, int)) or prop is None or prop < 1 or \
+    if (prop_name == 'batch_size' and not isinstance(prop, int)) or prop is None or prop <= 0 or \
       not validator.is_finite_number(prop):
       self.logger.info('Using default value for {}.'.format(prop_name))
       return False
@@ -159,11 +159,11 @@ def _run(self):
     """
     try:
       while True:
-        if self._get_time() > self.flushing_interval_deadline:
+        if self._get_time() >= self.flushing_interval_deadline:
           self._flush_queue()
 
         try:
-          item = self.event_queue.get(True, 0.05)
+          item = self.event_queue.get(False)
 
         except queue.Empty:
           time.sleep(0.05)
@@ -283,3 +283,51 @@ def stop(self):
 
     if self.is_running:
       self.logger.error('Timeout exceeded while attempting to close for ' + str(self.timeout_interval) + ' ms.')
+
+
+class ForwardingEventProcessor(BaseEventProcessor):
+  """
+  ForwardingEventProcessor serves as the default EventProcessor.
+
+  The ForwardingEventProcessor sends the LogEvent to EventDispatcher as soon as it is received.
+  """
+
+  def __init__(self, event_dispatcher, logger=None, notification_center=None):
+    """ ForwardingEventProcessor init method to configure event dispatching.
+
+    Args:
+      event_dispatcher: Provides a dispatch_event method which if given a URL and params sends a request to it.
+      logger: Optional component which provides a log method to log messages. By default nothing would be logged.
+      notification_center: Optional instance of notification_center.NotificationCenter.
+    """
+    self.event_dispatcher = event_dispatcher
+    self.logger = _logging.adapt_logger(logger or _logging.NoOpLogger())
+    self.notification_center = notification_center
+
+    if not validator.is_notification_center_valid(self.notification_center):
+      self.logger.error(enums.Errors.INVALID_INPUT.format('notification_center'))
+      self.notification_center = _notification_center.NotificationCenter()
+
+  def process(self, user_event):
+    """ Method to process the user_event by dispatching it.
+    Args:
+      user_event: UserEvent Instance.
+    """
+    if not isinstance(user_event, UserEvent):
+      self.logger.error('Provided event is in an invalid format.')
+      return
+
+    self.logger.debug('Received user_event: ' + str(user_event))
+
+    log_event = EventFactory.create_log_event(user_event, self.logger)
+
+    if self.notification_center is not None:
+      self.notification_center.send_notifications(
+        enums.NotificationTypes.LOG_EVENT,
+        log_event
+      )
+
+    try:
+      self.event_dispatcher.dispatch_event(log_event)
+    except Exception as e:
+      self.logger.exception('Error dispatching event: ' + str(log_event) + ' ' + str(e))

optimizely/exceptions.py

@@ -1,4 +1,4 @@
-# Copyright 2016-2018, Optimizely
+# Copyright 2016-2019, Optimizely
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
@@ -43,7 +43,7 @@ class InvalidGroupException(Exception):
 
 
 class InvalidInputException(Exception):
-  """ Raised when provided datafile, event dispatcher, logger or error handler is invalid. """
+  """ Raised when provided datafile, event dispatcher, logger, event processor or error handler is invalid. """
   pass
 
 

optimizely/helpers/validator.py

@@ -1,4 +1,4 @@
-# Copyright 2016-2018, Optimizely
+# Copyright 2016-2019, Optimizely
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
@@ -72,6 +72,19 @@ def is_config_manager_valid(config_manager):
   return _has_method(config_manager, 'get_config')
 
 
+def is_event_processor_valid(event_processor):
+  """ Given an event_processor, determine if it is valid or not i.e. provides a process method.
+
+  Args:
+    event_processor: Provides a process method to create user events and then send requests.
+
+  Returns:
+    Boolean depending upon whether event_processor is valid or not.
+  """
+
+  return _has_method(event_processor, 'process')
+
+
 def is_error_handler_valid(error_handler):
   """ Given a error_handler determine if it is valid or not i.e. provides a handle_error method.
 

optimizely/optimizely.py

@@ -17,12 +17,13 @@
 from . import event_builder
 from . import exceptions
 from . import logger as _logging
-from .config_manager import StaticConfigManager
 from .config_manager import PollingConfigManager
+from .config_manager import StaticConfigManager
 from .error_handler import NoOpErrorHandler as noop_error_handler
+from .event import event_factory, user_event_factory
+from .event.event_processor import ForwardingEventProcessor
 from .event_dispatcher import EventDispatcher as default_event_dispatcher
-from .helpers import enums
-from .helpers import validator
+from .helpers import enums, validator
 from .notification_center import NotificationCenter
 
 
@@ -38,7 +39,8 @@ def __init__(self,
                user_profile_service=None,
                sdk_key=None,
                config_manager=None,
-               notification_center=None):
+               notification_center=None,
+               event_processor=None):
     """ Optimizely init method for managing Custom projects.
 
     Args:
@@ -56,6 +58,7 @@ def __init__(self,
       notification_center: Optional instance of notification_center.NotificationCenter. Useful when providing own
                            config_manager.BaseConfigManager implementation which can be using the
                            same NotificationCenter instance.
+      event_processor: Processes the given event(s) by creating LogEvent(s) and then dispatching it.
     """
     self.logger_name = '.'.join([__name__, self.__class__.__name__])
     self.is_valid = True
@@ -64,6 +67,9 @@ def __init__(self,
     self.error_handler = error_handler or noop_error_handler
     self.config_manager = config_manager
     self.notification_center = notification_center or NotificationCenter(self.logger)
+    self.event_processor = event_processor or ForwardingEventProcessor(self.event_dispatcher,
+                                                                       self.logger,
+                                                                       self.notification_center)
 
     try:
       self._validate_instantiation_options()
@@ -114,6 +120,9 @@ def _validate_instantiation_options(self):
     if not validator.is_notification_center_valid(self.notification_center):
       raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('notification_center'))
 
+    if not validator.is_event_processor_valid(self.event_processor):
+      raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('event_processor'))
+
   def _validate_user_inputs(self, attributes=None, event_tags=None):
     """ Helper method to validate user inputs.
 
@@ -149,26 +158,23 @@ def _send_impression_event(self, project_config, experiment, variation, user_id,
       attributes: Dict representing user attributes and values which need to be recorded.
     """
 
-    impression_event = self.event_builder.create_impression_event(
+    user_event = user_event_factory.UserEventFactory.create_impression_event(
       project_config,
       experiment,
       variation.id,
       user_id,
       attributes
     )
 
-    self.logger.debug('Dispatching impression event to URL %s with params %s.' % (
-      impression_event.url,
-      impression_event.params
-    ))
-
-    try:
-      self.event_dispatcher.dispatch_event(impression_event)
-    except:
-      self.logger.exception('Unable to dispatch impression event!')
+    self.event_processor.process(user_event)
 
-    self.notification_center.send_notifications(enums.NotificationTypes.ACTIVATE,
-                                                experiment, user_id, attributes, variation, impression_event)
+    # Kept for backward compatibility.
+    # This notification is deprecated and new Decision notifications
+    # are sent via their respective method calls.
+    if len(self.notification_center.notification_listeners[enums.NotificationTypes.ACTIVATE]) > 0:
+      log_event = event_factory.EventFactory.create_log_event(user_event, self.logger)
+      self.notification_center.send_notifications(enums.NotificationTypes.ACTIVATE, experiment,
+                                                  user_id, attributes, variation, log_event.__dict__)
 
   def _get_feature_variable_for_type(self,
                                      project_config,
@@ -359,24 +365,21 @@ def track(self, event_key, user_id, attributes=None, event_tags=None):
       self.logger.info('Not tracking user "%s" for event "%s".' % (user_id, event_key))
       return
 
-    conversion_event = self.event_builder.create_conversion_event(
+    user_event = user_event_factory.UserEventFactory.create_conversion_event(
       project_config,
       event_key,
       user_id,
       attributes,
       event_tags
     )
+
+    self.event_processor.process(user_event)
     self.logger.info('Tracking event "%s" for user "%s".' % (event_key, user_id))
-    self.logger.debug('Dispatching conversion event to URL %s with params %s.' % (
-      conversion_event.url,
-      conversion_event.params
-    ))
-    try:
-      self.event_dispatcher.dispatch_event(conversion_event)
-    except:
-      self.logger.exception('Unable to dispatch conversion event!')
-    self.notification_center.send_notifications(enums.NotificationTypes.TRACK, event_key, user_id,
-                                                attributes, event_tags, conversion_event)
+
+    if len(self.notification_center.notification_listeners[enums.NotificationTypes.TRACK]) > 0:
+      log_event = event_factory.EventFactory.create_log_event(user_event, self.logger)
+      self.notification_center.send_notifications(enums.NotificationTypes.TRACK, event_key, user_id,
+                                                    attributes, event_tags, log_event.__dict__)
 
   def get_variation(self, experiment_key, user_id, attributes=None):
     """ Gets variation where user will be bucketed.

tests/helpers_tests/test_validator.py

@@ -1,4 +1,4 @@
-# Copyright 2016-2018, Optimizely
+# Copyright 2016-2019, Optimizely
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
@@ -20,6 +20,7 @@
 from optimizely import error_handler
 from optimizely import event_dispatcher
 from optimizely import logger
+from optimizely.event import event_processor
 from optimizely.helpers import validator
 
 from tests import base
@@ -42,6 +43,20 @@ def some_other_method(self):
 
     self.assertFalse(validator.is_config_manager_valid(CustomConfigManager()))
 
+  def test_is_event_processor_valid__returns_true(self):
+    """ Test that valid event_processor returns True. """
+
+    self.assertTrue(validator.is_event_processor_valid(event_processor.ForwardingEventProcessor))
+
+  def test_is_event_processor_valid__returns_false(self):
+    """ Test that invalid event_processor returns False. """
+
+    class CustomEventProcessor(object):
+      def some_other_method(self):
+        pass
+
+    self.assertFalse(validator.is_event_processor_valid(CustomEventProcessor))
+
   def test_is_datafile_valid__returns_true(self):
     """ Test that valid datafile returns True. """
 

tests/test_event_processor.py

@@ -18,7 +18,8 @@
 
 from . import base
 from optimizely.event.payload import Decision, Visitor
-from optimizely.event.event_processor import BatchEventProcessor
+from optimizely.event.event_processor import BatchEventProcessor, ForwardingEventProcessor
+from optimizely.event.event_factory import EventFactory
 from optimizely.event.log_event import LogEvent
 from optimizely.event.user_event_factory import UserEventFactory
 from optimizely.helpers import enums
@@ -401,3 +402,81 @@ def on_log_event(log_event):
     self.assertEqual(1, len(self.optimizely.notification_center.notification_listeners[
       enums.NotificationTypes.LOG_EVENT
     ]))
+
+
+class TestForwardingEventDispatcher(object):
+
+  def __init__(self, is_updated=False):
+    self.is_updated = is_updated
+
+  def dispatch_event(self, log_event):
+    if log_event.http_verb == 'POST' and log_event.url == EventFactory.EVENT_ENDPOINT:
+      self.is_updated = True
+    return self.is_updated
+
+
+class ForwardingEventProcessorTest(base.BaseTest):
+
+  def setUp(self, *args, **kwargs):
+    base.BaseTest.setUp(self, 'config_dict_with_multiple_experiments')
+    self.test_user_id = 'test_user'
+    self.event_name = 'test_event'
+    self.optimizely.logger = SimpleLogger()
+    self.notification_center = self.optimizely.notification_center
+    self.event_dispatcher = TestForwardingEventDispatcher(is_updated=False)
+
+    with mock.patch.object(self.optimizely, 'logger') as mock_config_logging:
+      self._event_processor = ForwardingEventProcessor(self.event_dispatcher,
+                                                        mock_config_logging,
+                                                        self.notification_center
+                                                       )
+
+  def _build_conversion_event(self, event_name):
+    return UserEventFactory.create_conversion_event(self.project_config,
+                                                    event_name,
+                                                    self.test_user_id,
+                                                    {},
+                                                    {}
+                                                    )
+
+  def test_event_processor__dispatch_raises_exception(self):
+    """ Test that process logs dispatch failure gracefully. """
+
+    user_event = self._build_conversion_event(self.event_name)
+    log_event = EventFactory.create_log_event(user_event, self.optimizely.logger)
+
+    with mock.patch.object(self.optimizely, 'logger') as mock_client_logging, \
+        mock.patch.object(self.event_dispatcher, 'dispatch_event',
+                          side_effect=Exception('Failed to send.')):
+
+      event_processor = ForwardingEventProcessor(self.event_dispatcher, mock_client_logging, self.notification_center)
+      event_processor.process(user_event)
+
+    mock_client_logging.exception.assert_called_once_with(
+      'Error dispatching event: ' + str(log_event) + ' Failed to send.'
+    )
+
+  def test_event_processor__with_test_event_dispatcher(self):
+    user_event = self._build_conversion_event(self.event_name)
+    self._event_processor.process(user_event)
+    self.assertStrictTrue(self.event_dispatcher.is_updated)
+
+  def test_notification_center(self):
+
+    callback_hit = [False]
+
+    def on_log_event(log_event):
+      self.assertStrictTrue(isinstance(log_event, LogEvent))
+      callback_hit[0] = True
+
+    self.optimizely.notification_center.add_notification_listener(
+      enums.NotificationTypes.LOG_EVENT, on_log_event
+    )
+
+    user_event = self._build_conversion_event(self.event_name)
+    self._event_processor.process(user_event)
+
+    self.assertEqual(True, callback_hit[0])
+    self.assertEqual(1, len(self.optimizely.notification_center.notification_listeners[
+      enums.NotificationTypes.LOG_EVENT
+    ]))