Merge pull request #246 from eduNEXT/cag/add-openedx-event-signal

feat: add event bus backend

README.rst

@@ -149,6 +149,61 @@ An example configuration for ``AsyncRoutingBackend`` is provided below::
     }
 
 
+Event Bus Routing
+-----------------
+
+``event-tracking`` provides a solution for routing events to the Event Bus
+using the ``EventBusBackend``. It extends ``RoutingBackend`` but sends events
+to the Event Bus.
+
+It can:
+
+* Process event through the configured processors.
+* If the event is allowed via `EVENT_BUS_TRACKING_LOGS`, send it to the Event Bus.
+
+Make sure to enable the setting: ``SEND_TRACKING_EVENT_EMITTED_SIGNAL`` to allow the
+``EventBusBackend`` to send events to the Event Bus.
+
+An example configuration for ``EventBusBackend`` is provided below::
+
+    EVENT_TRACKING_BACKENDS = {
+        'xapi': {
+            'ENGINE':  'eventtracking.backends.event_bus.EventBusBackend',
+            'OPTIONS': {
+                'backend_name': 'xapi',
+                'processors': [
+                    {
+                        'ENGINE': 'eventtracking.processors.regex_filter.RegexFilter',
+                        'OPTIONS':{
+                            'filter_type': 'allowlist',
+                            'regular_expressions': [
+                                'edx.course.enrollment.activated',
+                                'edx.course.enrollment.deactivated',
+                            ]
+                        }
+                    }
+                ],
+                'backends': {
+                    'xapi': {
+                        'ENGINE': 'dummy.backend.engine',
+                        'OPTIONS': {
+                            ...
+                        }
+                    }
+                },
+            },
+        },
+        'tracking_logs': {
+            ...
+        }
+        ...
+    }
+
+    EVENT_BUS_TRACKING_LOGS = [
+        'edx.course.enrollment.activated',
+        'edx.course.enrollment.deactivated',
+    ]
+
 Roadmap
 -------
 

doc/user_guide/api/eventtracking.backends.rst

@@ -42,3 +42,11 @@ eventtracking.backends.segment
     :undoc-members:
     :show-inheritance:
 
+
+eventtracking.backends.event_bus
+------------------------------
+
+.. automodule:: eventtracking.backends.event_bus
+    :members:
+    :undoc-members:
+    :show-inheritance:

eventtracking/__init__.py

@@ -1,3 +1,3 @@
 """A simple event tracking library"""
 
-__version__ = '2.2.0'
+__version__ = '2.3.0'

eventtracking/backends/event_bus.py

@@ -0,0 +1,52 @@
+"""Event tracker backend that emits events to the event-bus."""
+
+import json
+import logging
+from datetime import datetime
+
+from django.conf import settings
+from openedx_events.analytics.data import TrackingLogData
+from openedx_events.analytics.signals import TRACKING_EVENT_EMITTED
+
+from eventtracking.backends.logger import DateTimeJSONEncoder
+from eventtracking.backends.routing import RoutingBackend
+from eventtracking.config import SEND_TRACKING_EVENT_EMITTED_SIGNAL
+
+logger = logging.getLogger(__name__)
+
+
+class EventBusRoutingBackend(RoutingBackend):
+    """
+    Event tracker backend for the event bus.
+    """
+
+    def send(self, event):
+        """
+        Send the tracking log event to the event bus by emitting the
+        TRACKING_EVENT_EMITTED signal using custom metadata.
+        """
+        if not SEND_TRACKING_EVENT_EMITTED_SIGNAL.is_enabled():
+            return
+
+        name = event.get("name")
+
+        if name not in getattr(settings, "EVENT_BUS_TRACKING_LOGS", []):
+            return
+
+        data = json.dumps(event.get("data"), cls=DateTimeJSONEncoder)
+        context = json.dumps(event.get("context"), cls=DateTimeJSONEncoder)
+
+        timestamp = event.get("timestamp")
+
+        if isinstance(timestamp, str):
+            timestamp = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%f%z")
+
+        tracking_log = TrackingLogData(
+            name=event.get("name"),
+            timestamp=timestamp,
+            data=data,
+            context=context,
+        )
+        TRACKING_EVENT_EMITTED.send_event(tracking_log=tracking_log)
+
+        logger.info(f"Tracking log {tracking_log.name} emitted to the event bus.")

eventtracking/backends/routing.py

@@ -118,7 +118,7 @@ def process_event(self, event):
                     processed_event = modified_event
             except EventEmissionExit:
                 raise
-            except Exception:  # pylint: disable=broad-except
+            except Exception:
                 LOG.exception(
                     'Failed to execute processor: %s', str(processor)
                 )
@@ -142,7 +142,7 @@ def send_to_backends(self, event):
                 LOG.info('[send_to_backends] Failed to send edx event "%s" to "%s" backend. "%s" backend has'
                          ' not been enabled, [%s]', event["name"], name, name, repr(exc)
                          )
-            except Exception:  # pylint: disable=broad-except
+            except Exception:
                 LOG.exception(
                     'Unable to send edx event "%s" to backend: %s', event["name"], name
                 )

eventtracking/backends/tests/test_event_bus.py

@@ -0,0 +1,69 @@
+"""
+Test the async routing backend.
+"""
+
+import json
+from datetime import datetime
+from unittest import TestCase
+from unittest.mock import patch
+
+from django.test import override_settings
+from openedx_events.analytics.data import TrackingLogData
+
+from eventtracking.backends.event_bus import EventBusRoutingBackend
+
+
+class TestAsyncRoutingBackend(TestCase):
+    """
+    Test the async routing backend.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.sample_event = {
+            "name": "sample_event",
+            "data": {"foo": "bar"},
+            "timestamp": "2020-01-01T12:12:12.000000+00:00",
+            "context": {"baz": "qux"},
+        }
+
+    @patch("eventtracking.backends.event_bus.EventBusRoutingBackend.send")
+    def test_successful_send(self, mocked_send_event):
+        backend = EventBusRoutingBackend()
+        backend.send(self.sample_event)
+        mocked_send_event.assert_called_once_with(self.sample_event)
+
+    @override_settings(
+        SEND_TRACKING_EVENT_EMITTED_SIGNAL=True,
+        EVENT_BUS_TRACKING_LOGS=["sample_event"],
+    )
+    @patch("eventtracking.backends.event_bus.TRACKING_EVENT_EMITTED.send_event")
+    def test_successful_send_event(self, mock_send_event):
+        backend = EventBusRoutingBackend()
+        backend.send(self.sample_event)
+
+        mock_send_event.assert_called()
+        self.assertDictContainsSubset(
+            {
+                "tracking_log": TrackingLogData(
+                    name=self.sample_event["name"],
+                    timestamp=datetime.strptime(
+                        self.sample_event["timestamp"], "%Y-%m-%dT%H:%M:%S.%f%z"
+                    ),
+                    data=json.dumps(self.sample_event["data"]),
+                    context=json.dumps(self.sample_event["context"]),
+                )
+            },
+            mock_send_event.call_args.kwargs,
+        )
+
+    @patch(
+        "eventtracking.backends.event_bus.SEND_TRACKING_EVENT_EMITTED_SIGNAL.is_enabled"
+    )
+    @patch("eventtracking.backends.event_bus.TRACKING_EVENT_EMITTED.send_event")
+    def test_event_is_disabled(self, mock_send_event, mock_is_enabled):
+        mock_is_enabled.return_value = False
+        backend = EventBusRoutingBackend()
+        backend.send(self.sample_event)
+        mock_is_enabled.assert_called_once()
+        mock_send_event.assert_not_called()

eventtracking/config.py

@@ -0,0 +1,19 @@
+"""
+This module contains various configuration settings via
+waffle switches for the event-tracking app.
+"""
+
+from edx_toggles.toggles import SettingToggle
+
+# .. toggle_name: SEND_TRACKING_EVENT_EMITTED_SIGNAL
+# .. toggle_implementation: SettingToggle
+# .. toggle_default: False
+# .. toggle_description: When True, the system will publish `TRACKING_EVENT_EMITTED` signals to the event bus. The
+#   `TRACKING_EVENT_EMITTED` signal is emit when a tracking log is emitted.
+# .. toggle_use_cases: opt_in
+# .. toggle_creation_date: 2023-10-26
+SEND_TRACKING_EVENT_EMITTED_SIGNAL = SettingToggle(
+    'SEND_TRACKING_EVENT_EMITTED_SIGNAL',
+    default=False,
+    module_name=__name__
+)

eventtracking/django/apps.py

@@ -21,3 +21,5 @@ def ready(self):
         # pylint: disable=import-outside-toplevel
         from eventtracking.django.django_tracker import override_default_tracker
         override_default_tracker()
+
+        from eventtracking import handlers  # pylint: disable=import-outside-toplevel, unused-import

eventtracking/handlers.py

@@ -0,0 +1,61 @@
+"""
+This module contains the handlers for signals emitted by the analytics app.
+"""
+import json
+import logging
+
+from django.dispatch import receiver
+from openedx_events.analytics.signals import TRACKING_EVENT_EMITTED
+from openedx_events.tooling import SIGNAL_PROCESSED_FROM_EVENT_BUS
+
+from eventtracking.backends.event_bus import EventBusRoutingBackend
+from eventtracking.processors.exceptions import EventEmissionExit
+from eventtracking.tracker import get_tracker
+
+logger = logging.getLogger(__name__)
+
+
+@receiver(TRACKING_EVENT_EMITTED)
+def send_tracking_log_to_backends(
+    sender, signal, **kwargs
+):  # pylint: disable=unused-argument
+    """
+    Listen for the TRACKING_EVENT_EMITTED signal and send the event to the enabled backends.
+
+    The process is the following:
+
+    1. Unserialize the tracking log from the signal.
+    2. Get the tracker instance to get the enabled backends (mongo, event_bus, logger, etc).
+    3. Get the event bus backends that are the interested in the signals (multiple can be configured).
+    4. Transform the event with the configured processors.
+    5. Send the transformed event to the different event bus backends.
+
+    This allows us to only send the tracking log to the event bus once and the event bus will send
+    the transformed event to the different configured backends.
+    """
+    if not kwargs.get(SIGNAL_PROCESSED_FROM_EVENT_BUS, False):
+        logger.debug("Event received from a non-event bus backend, skipping...")
+        return
+    tracking_log = kwargs.get("tracking_log")
+
+    event = {
+        "name": tracking_log.name,
+        "timestamp": tracking_log.timestamp,
+        "data": json.loads(tracking_log.data),
+        "context": json.loads(tracking_log.context),
+    }
+
+    tracker = get_tracker()
+
+    engines = {
+        name: engine
+        for name, engine in tracker.backends.items()
+        if isinstance(engine, EventBusRoutingBackend)
+    }
+    for name, engine in engines.items():
+        try:
+            processed_event = engine.process_event(event)
+            logger.info('Successfully processed event "{}"'.format(event["name"]))
+            engine.send_to_backends(processed_event.copy())
+        except EventEmissionExit:
+            logger.info("[EventEmissionExit] skipping event {}".format(event["name"]))