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/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__
+)

requirements/base.in

@@ -8,3 +8,5 @@ pytz
 six
 celery
 edx-django-utils
+openedx-events>=9.5.1
+edx-toggles

requirements/base.txt

@@ -8,6 +8,8 @@ amqp==5.2.0
     # via kombu
 asgiref==3.7.2
     # via django
+attrs==23.2.0
+    # via openedx-events
 backports-zoneinfo[tzdata]==0.2.1
     # via
     #   celery
@@ -26,30 +28,54 @@ click==8.1.7
     #   click-didyoumean
     #   click-plugins
     #   click-repl
+    #   code-annotations
     #   edx-django-utils
 click-didyoumean==0.3.0
     # via celery
 click-plugins==1.1.1
     # via celery
 click-repl==0.3.0
     # via celery
+code-annotations==1.6.0
+    # via edx-toggles
 django==3.2.24
     # via
     #   -c https://raw.githubusercontent.com/edx/edx-lint/master/edx_lint/files/common_constraints.txt
     #   -r requirements/base.in
     #   django-crum
     #   django-waffle
     #   edx-django-utils
+    #   edx-toggles
+    #   openedx-events
 django-crum==0.7.9
-    # via edx-django-utils
+    # via
+    #   edx-django-utils
+    #   edx-toggles
 django-waffle==4.1.0
-    # via edx-django-utils
+    # via
+    #   edx-django-utils
+    #   edx-toggles
 edx-django-utils==5.10.1
+    # via
+    #   -r requirements/base.in
+    #   edx-toggles
+    #   openedx-events
+edx-opaque-keys[django]==2.5.1
+    # via openedx-events
+edx-toggles==5.1.1
     # via -r requirements/base.in
+fastavro==1.9.3
+    # via openedx-events
+jinja2==3.1.3
+    # via code-annotations
 kombu==5.3.5
     # via celery
+markupsafe==2.1.5
+    # via jinja2
 newrelic==9.6.0
     # via edx-django-utils
+openedx-events==9.5.1
+    # via -r requirements/base.in
 pbr==6.0.0
     # via stevedore
 prompt-toolkit==3.0.43
@@ -59,26 +85,38 @@ psutil==5.9.8
 pycparser==2.21
     # via cffi
 pymongo==3.13.0
-    # via -r requirements/base.in
+    # via
+    #   -r requirements/base.in
+    #   edx-opaque-keys
 pynacl==1.5.0
     # via edx-django-utils
 python-dateutil==2.8.2
     # via celery
+python-slugify==8.0.4
+    # via code-annotations
 pytz==2024.1
     # via
     #   -r requirements/base.in
     #   django
+pyyaml==6.0.1
+    # via code-annotations
 six==1.16.0
     # via
     #   -r requirements/base.in
     #   python-dateutil
 sqlparse==0.4.4
     # via django
 stevedore==5.1.0
-    # via edx-django-utils
+    # via
+    #   code-annotations
+    #   edx-django-utils
+    #   edx-opaque-keys
+text-unidecode==1.3
+    # via python-slugify
 typing-extensions==4.9.0
     # via
     #   asgiref
+    #   edx-opaque-keys
     #   kombu
 tzdata==2024.1
     # via