feat: use new event_bus_producer_config

cms/djangoapps/contentstore/signals/handlers.py

@@ -19,6 +19,7 @@
     XBLOCK_DUPLICATED,
     XBLOCK_PUBLISHED,
 )
+from openedx.core.lib.events import determine_producer_config_for_signal_and_topic
 from openedx_events.event_bus import get_producer
 from pytz import UTC
 
@@ -164,6 +165,13 @@ def listen_for_course_catalog_info_changed(sender, signal, **kwargs):
     """
     Publish COURSE_CATALOG_INFO_CHANGED signals onto the event bus.
     """
+    # temporary: defer to EVENT_BUS_PRODUCER_CONFIG if present
+    producer_config_setting = determine_producer_config_for_signal_and_topic(COURSE_CATALOG_INFO_CHANGED,
+                                                                             'course-catalog-info-changed')
+    if producer_config_setting is True:
+        log.info("Producing course-catalog-info-changed event via config")
+        return
+    log.info("Producing course-catalog-info-changed event via manual send")
     get_producer().send(
         signal=COURSE_CATALOG_INFO_CHANGED, topic='course-catalog-info-changed',
         event_key_field='catalog_info.course_key', event_data={'catalog_info': kwargs['catalog_info']},
@@ -176,8 +184,14 @@ def listen_for_xblock_published(sender, signal, **kwargs):
     """
     Publish XBLOCK_PUBLISHED signals onto the event bus.
     """
+    # temporary: defer to EVENT_BUS_PRODUCER_CONFIG if present
+    topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
+    producer_config_setting = determine_producer_config_for_signal_and_topic(XBLOCK_PUBLISHED, topic)
+    if producer_config_setting is True:
+        log.info("Producing xblock-published event via config")
+        return
     if settings.FEATURES.get("ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS"):
-        topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
+        log.info("Producing xblock-published event via manual send")
         get_producer().send(
             signal=XBLOCK_PUBLISHED, topic=topic,
             event_key_field='xblock_info.usage_key', event_data={'xblock_info': kwargs['xblock_info']},
@@ -190,8 +204,14 @@ def listen_for_xblock_deleted(sender, signal, **kwargs):
     """
     Publish XBLOCK_DELETED signals onto the event bus.
     """
+    # temporary: defer to EVENT_BUS_PRODUCER_CONFIG if present
+    topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
+    producer_config_setting = determine_producer_config_for_signal_and_topic(XBLOCK_DELETED, topic)
+    if producer_config_setting is True:
+        log.info("Producing xblock-deleted event via config")
+        return
     if settings.FEATURES.get("ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS"):
-        topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
+        log.info("Producing xblock-deleted event via manual send")
         get_producer().send(
             signal=XBLOCK_DELETED, topic=topic,
             event_key_field='xblock_info.usage_key', event_data={'xblock_info': kwargs['xblock_info']},
@@ -204,8 +224,14 @@ def listen_for_xblock_duplicated(sender, signal, **kwargs):
     """
     Publish XBLOCK_DUPLICATED signals onto the event bus.
     """
+    # temporary: defer to EVENT_BUS_PRODUCER_CONFIG if present
+    topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
+    producer_config_setting = determine_producer_config_for_signal_and_topic(XBLOCK_DUPLICATED, topic)
+    if producer_config_setting is True:
+        log.info("Producing xblock-duplicated event via config")
+        return
     if settings.FEATURES.get("ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS"):
-        topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
+        log.info("Producing xblock-duplicated event via manual send")
         get_producer().send(
             signal=XBLOCK_DUPLICATED, topic=topic,
             event_key_field='xblock_info.usage_key', event_data={'xblock_info': kwargs['xblock_info']},

cms/envs/common.py

@@ -531,6 +531,19 @@
     # .. toggle_creation_date: 2023-03-31
     # .. toggle_tickets: https://github.com/openedx/edx-platform/pull/32015
     'DISABLE_ADVANCED_SETTINGS': False,
+
+    # .. toggle_name: FEATURES['ENABLE_SEND_XBLOCK_LIFECYCLE_EVENTS_OVER_BUS']
+    # .. toggle_implementation: DjangoSetting
+    # .. toggle_default: False
+    # .. toggle_description: Enables sending xblock lifecycle events over the event bus. Used to create the
+    #   EVENT_BUS_PRODUCER_CONFIG setting
+    # .. toggle_use_cases: opt_in
+    # .. toggle_creation_date: 2023-10-10
+    # .. toggle_target_removal_date: 2023-10-12
+    # .. toggle_warning: The default may be changed in a later release. See
+    #   https://github.com/openedx/openedx-events/issues/265
+    # .. toggle_tickets: https://github.com/edx/edx-arch-experiments/issues/381
+    'ENABLE_SEND_XBLOCK_LIFECYCLE_EVENTS_OVER_BUS': False,
 }
 
 # .. toggle_name: ENABLE_COPPA_COMPLIANCE
@@ -1795,6 +1808,7 @@
 
     # alternative swagger generator for CMS API
     'drf_spectacular',
+    'openedx_events',
 ]
 
 
@@ -2789,6 +2803,66 @@
     'PREPROCESSING_HOOKS': ['cms.lib.spectacular.cms_api_filter'],  # restrict spectacular to CMS API endpoints
 }
 
-#### Event bus publishing ####
-## Will be more filled out as part of https://github.com/edx/edx-arch-experiments/issues/381
-EVENT_BUS_PRODUCER_CONFIG = {}
+
+#### Event bus producing ####
+def _should_send_xblock_events(settings):
+    return settings.FEATURES['ENABLE_SEND_XBLOCK_LIFECYCLE_EVENTS_OVER_BUS']
+
+# .. setting_name: EVENT_BUS_PRODUCER_CONFIG
+# .. setting_default: all events disabled
+# .. setting_description: Dictionary of event_types mapped to dictionaries of topic to topic-related configuration.
+#    Each topic configuration dictionary contains
+#    * `enabled`: a toggle denoting whether the event will be published to the topic. These should be annotated
+#       according to
+#       https://edx.readthedocs.io/projects/edx-toggles/en/latest/how_to/documenting_new_feature_toggles.html
+#    * `event_key_field` which is a period-delimited string path to event data field to use as event key.
+#    Note: The topic names should not include environment prefix as it will be dynamically added based on
+#    EVENT_BUS_TOPIC_PREFIX setting.
+
+EVENT_BUS_PRODUCER_CONFIG = {
+    'org.openedx.content_authoring.course.catalog_info.changed.v1': {
+        'course-catalog-info-changed':
+            {'event_key_field': 'catalog_info.course_key',
+             # .. toggle_name: EVENT_BUS_PRODUCER_CONFIG['org.openedx.content_authoring.course.catalog_info.changed.v1']
+             #    ['course-catalog-info-changed']['enabled']
+             # .. toggle_implementation: DjangoSetting
+             # .. toggle_default: False
+             # .. toggle_description: if enabled, will publish COURSE_CATALOG_INFO_CHANGED events to the event bus on
+             #    the course-catalog-info-changed topics
+             # .. toggle_warning: The default may be changed in a later release. See
+             #    https://github.com/openedx/openedx-events/issues/265
+             # .. toggle_use_cases: opt_in
+             # .. toggle_creation_date: 2023-10-10
+             'enabled': False},
+    },
+    'org.openedx.content_authoring.xblock.published.v1': {
+        'course-authoring-xblock-lifecycle':
+            {'event_key_field': 'xblock_info.usage_key', 'enabled': _should_send_xblock_events},
+    },
+    'org.openedx.content_authoring.xblock.deleted.v1': {
+        'course-authoring-xblock-lifecycle':
+            {'event_key_field': 'xblock_info.usage_key', 'enabled': _should_send_xblock_events},
+    },
+    'org.openedx.content_authoring.xblock.duplicated.v1': {
+        'course-authoring-xblock-lifecycle':
+            {'event_key_field': 'xblock_info.usage_key', 'enabled': _should_send_xblock_events},
+    },
+    # LMS events. These have to be copied over here because lms.common adds some derived entries as well,
+    # and the derivation fails if the keys are missing. If we ever remove the import of lms.common, we can remove these.
+    'org.openedx.learning.certificate.created.v1': {
+        'learning-certificate-lifecycle':
+            {'event_key_field': 'certificate.course.course_key', 'enabled': False},
+    },
+    'org.openedx.learning.certificate.revoked.v1': {
+        'learning-certificate-lifecycle':
+            {'event_key_field': 'certificate.course.course_key', 'enabled': False},
+    },
+}
+
+
+derived_collection_entry('EVENT_BUS_PRODUCER_CONFIG', 'org.openedx.content_authoring.xblock.published.v1',
+                         'course-authoring-xblock-lifecycle', 'enabled')
+derived_collection_entry('EVENT_BUS_PRODUCER_CONFIG', 'org.openedx.content_authoring.xblock.duplicated.v1',
+                         'course-authoring-xblock-lifecycle', 'enabled')
+derived_collection_entry('EVENT_BUS_PRODUCER_CONFIG', 'org.openedx.content_authoring.xblock.deleted.v1',
+                         'course-authoring-xblock-lifecycle', 'enabled')

cms/envs/devstack.py

@@ -309,6 +309,7 @@ def should_show_debug_toolbar(request):  # lint-amnesty, pylint: disable=missing
 # .. toggle_creation_date: 2023-02-21
 # .. toggle_warning: For consistency in user experience, keep the value in sync with the setting of the same name
 #   in the LMS and CMS.
+#   This will be deprecated in favor of ENABLE_SEND_XBLOCK_LIFECYCLE_EVENTS_OVER_BUS
 # .. toggle_tickets: 'https://github.com/openedx/edx-platform/pull/31813'
 FEATURES['ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS'] = True
 FEATURES['ENABLE_SEND_ENROLLMENT_EVENTS_OVER_BUS'] = True

cms/envs/production.py

@@ -18,9 +18,9 @@
 from django.core.exceptions import ImproperlyConfigured
 from django.urls import reverse_lazy
 from edx_django_utils.plugins import add_plugins
+from openedx_events.event_bus import merge_producer_configs
 from path import Path as path
 
-
 from openedx.core.djangoapps.plugins.constants import ProjectType, SettingsType
 
 from .common import *
@@ -85,6 +85,7 @@ def get_env_setting(setting):
         'MKTG_URL_LINK_MAP',
         'MKTG_URL_OVERRIDES',
         'REST_FRAMEWORK',
+        'EVENT_BUS_PRODUCER_CONFIG',
     ]
     for key in KEYS_WITH_MERGED_VALUES:
         if key in __config_copy__:
@@ -670,3 +671,7 @@ def get_env_setting(setting):
 }
 
 INACTIVE_USER_URL = f'http{"s" if HTTPS == "on" else ""}://{CMS_BASE}'
+
+############## Event bus producer ##############
+EVENT_BUS_PRODUCER_CONFIG = merge_producer_configs(EVENT_BUS_PRODUCER_CONFIG,
+                                                   ENV_TOKENS.get('EVENT_BUS_PRODUCER_CONFIG', {}))

common/djangoapps/student/handlers.py

@@ -8,17 +8,26 @@
 from openedx_events.learning.signals import (
     COURSE_UNENROLLMENT_COMPLETED,
 )
+from openedx.core.lib.events import determine_producer_config_for_signal_and_topic
+import logging
+log = logging.getLogger(__name__)
 
 
 @receiver(COURSE_UNENROLLMENT_COMPLETED)
 def course_unenrollment_receiver(sender, signal, **kwargs):
     """
     Removes user notification preference when user un-enrolls from the course
     """
+    topic = getattr(settings, "EVENT_BUS_ENROLLMENT_LIFECYCLE_TOPIC", "course-unenrollment-lifecycle")
+    producer_config_setting = determine_producer_config_for_signal_and_topic(COURSE_UNENROLLMENT_COMPLETED, topic)
+    if producer_config_setting is True:
+        log.info("Producing unenrollment-event event via config")
+        return
     if settings.FEATURES.get("ENABLE_SEND_ENROLLMENT_EVENTS_OVER_BUS"):
+        log.info("Producing unenrollment-event event via manual send")
         get_producer().send(
             signal=COURSE_UNENROLLMENT_COMPLETED,
-            topic=getattr(settings, "EVENT_BUS_ENROLLMENT_LIFECYCLE_TOPIC", "course-unenrollment-lifecycle"),
+            topic=topic,
             event_key_field='enrollment.course.course_key',
             event_data={'enrollment': kwargs.get('enrollment')},
             event_metadata=kwargs.get('metadata')

lms/djangoapps/certificates/config.py

@@ -23,6 +23,7 @@
 # .. toggle_description: When True, the system will publish `CERTIFICATE_CREATED` signals to the event bus. The
 #   `CERTIFICATE_CREATED` signal is emit when a certificate has been awarded to a learner and the creation process has
 #   completed.
+# .. toggle_warning: Will be deprecated in favor of SEND_LEARNING_CERTIFICATE_LIFECYCLE_EVENTS_TO_BUS
 # .. toggle_use_cases: temporary
 # .. toggle_creation_date: 2023-04-11
 # .. toggle_target_removal_date: 2023-07-31
@@ -36,6 +37,7 @@
 # .. toggle_description: When True, the system will publish `CERTIFICATE_REVOKED` signals to the event bus. The
 #   `CERTIFICATE_REVOKED` signal is emit when a certificate has been revoked from a learner and the revocation process
 #   has completed.
+# .. toggle_warning: Will be deprecated in favor of SEND_LEARNING_CERTIFICATE_LIFECYCLE_EVENTS_TO_BUS
 # .. toggle_use_cases: temporary
 # .. toggle_creation_date: 2023-09-15
 # .. toggle_target_removal_date: 2024-01-01

lms/djangoapps/certificates/signals.py

@@ -3,10 +3,12 @@
 """
 
 import logging
+from django.conf import settings
 
 from django.db.models.signals import post_save
 from django.dispatch import receiver
 from openedx_events.event_bus import get_producer
+from edx_django_utils.monitoring import set_custom_attribute
 
 from common.djangoapps.course_modes import api as modes_api
 from common.djangoapps.student.models import CourseEnrollment
@@ -27,6 +29,7 @@
 from lms.djangoapps.certificates.api import auto_certificate_generation_enabled
 from lms.djangoapps.verify_student.services import IDVerificationService
 from openedx.core.djangoapps.content.course_overviews.signals import COURSE_PACING_CHANGED
+from openedx.core.lib.events import determine_producer_config_for_signal_and_topic
 from openedx.core.djangoapps.signals.signals import (
     COURSE_GRADE_NOW_FAILED,
     COURSE_GRADE_NOW_PASSED,
@@ -161,12 +164,43 @@ def _listen_for_enrollment_mode_change(sender, user, course_key, mode, **kwargs)
             return False
 
 
+def _determine_producer_config_for_signal_and_topic(signal, topic):
+    """
+    Utility method to determine the setting for the given signal and topic in EVENT_BUS_PRODUCER_CONFIG
+
+    Records to New Relic for later analysis.
+
+    Parameters
+        signal (OpenEdxPublicSignal): The signal being sent to the event bus
+        topic (string): The topic to which the signal is being sent (without environment prefix)
+
+    Returns
+        True if the signal is enabled for that topic in EVENT_BUS_PRODUCER_CONFIG
+        False if the signal is explicitly disabled for that topic in EVENT_BUS_PRODUCER_CONFIG
+        None if the signal/topic pair is not present in EVENT_BUS_PRODUCER_CONFIG
+    """
+    event_type_producer_configs = getattr(settings, "EVENT_BUS_PRODUCER_CONFIG",
+                                          {}).get(signal.event_type, {})
+    topic_config = event_type_producer_configs.get(topic, {})
+    topic_setting = topic_config.get('enabled', None)
+    set_custom_attribute(f'producer_config_setting_{topic}_{signal.event_type}',
+                         topic_setting if topic_setting is not None else 'Unset')
+    return topic_setting
+
+
 @receiver(CERTIFICATE_CREATED)
 def listen_for_certificate_created_event(sender, signal, **kwargs):  # pylint: disable=unused-argument
     """
     Publish `CERTIFICATE_CREATED` events to the event bus.
     """
+    # temporary: defer to EVENT_BUS_PRODUCER_CONFIG if present
+    producer_config_setting = determine_producer_config_for_signal_and_topic(CERTIFICATE_CREATED,
+                                                                             'learning-certificate-lifecycle')
+    if producer_config_setting is True:
+        log.info("Producing certificate-created event via config")
+        return
     if SEND_CERTIFICATE_CREATED_SIGNAL.is_enabled():
+        log.info("Producing certificate-created event via manual send")
         get_producer().send(
             signal=CERTIFICATE_CREATED,
             topic='learning-certificate-lifecycle',
@@ -181,7 +215,14 @@ def listen_for_certificate_revoked_event(sender, signal, **kwargs):  # pylint: d
     """
     Publish `CERTIFICATE_REVOKED` events to the event bus.
     """
+    # temporary: defer to EVENT_BUS_PRODUCER_CONFIG if present
+    producer_config_setting = determine_producer_config_for_signal_and_topic(CERTIFICATE_REVOKED,
+                                                                             'learning-certificate-lifecycle')
+    if producer_config_setting is True:
+        log.info("Producing certificate-revoked event via config")
+        return
     if SEND_CERTIFICATE_REVOKED_SIGNAL.is_enabled():
+        log.info("Producing certificate-revoked event via manual send")
         get_producer().send(
             signal=CERTIFICATE_REVOKED,
             topic='learning-certificate-lifecycle',