temp: Possible repro branch for DD trace concatenation

- Add `/celery_repro` URL to run a sample task
- Send Celery tasks to a broker, rather than running in-process
- Hardcode a broker URL
- Log all celery signals

See edx/edx-arch-experiments#692

lms/envs/devstack.py

@@ -29,12 +29,6 @@
 
 SESSION_COOKIE_NAME = 'lms_sessionid'
 
-# By default don't use a worker, execute tasks as if they were local functions
-CELERY_ALWAYS_EAGER = True
-# When the celery task is run eagerly, it is executed locally while sharing the
-# thread and its request cache with the active Django Request. In that case,
-# do not clear the cache.
-CLEAR_REQUEST_CACHE_ON_TASK_COMPLETION = False
 HTTPS = 'off'
 
 LMS_ROOT_URL = f'http://{LMS_BASE}'

lms/urls.py

@@ -108,6 +108,7 @@
     path('', include('lms.djangoapps.static_template_view.urls')),
 
     path('heartbeat', include('openedx.core.djangoapps.heartbeat.urls')),
+    path('celery_repro', include('openedx.core.djangoapps.celery_repro.urls')),
 
     path('i18n/', include('django.conf.urls.i18n')),
 

openedx/core/djangoapps/celery_repro/urls.py

@@ -0,0 +1,6 @@
+from django.urls import path
+from openedx.core.djangoapps.celery_repro.views import run_celery_task
+
+urlpatterns = [
+    path('', run_celery_task, name='celery_repro'),
+]

openedx/core/djangoapps/celery_repro/views.py

@@ -0,0 +1,14 @@
+import ddtrace.tracer
+from celery import shared_task
+from common.djangoapps.util.json_request import JsonResponse
+
+
+@shared_task
+def sample_task():
+    print("📋📋📋 Task is running in this process.")
+
+
+def run_celery_task(request):
+    print(f"ℹ️ Current root span: {ddtrace.tracer.current_root_span()._pprint()}")
+    sample_task.apply_async()
+    return JsonResponse({}, status=200)

openedx/core/lib/celery/__init__.py

@@ -21,10 +21,34 @@
 
 # WARNING: Do not refer to this unless you are cms.celery or
 # lms.celery. See module docstring!
-APP = Celery('proj')
+APP = Celery('proj', broker="redis://:[email protected]:6379/10")
 
 APP.conf.task_protocol = 1
 # Using a string here means the worker will not have to
 # pickle the object when using Windows.
 APP.config_from_object('django.conf:settings')
 APP.autodiscover_tasks()
+
+
+import celery.signals
+
+CELERY_SIGNAL_NAMES = [
+    'before_task_publish',
+    'after_task_publish',
+    'task_prerun',
+    'task_postrun',
+    'task_retry',
+    'task_success',
+    'task_failure',
+    'task_internal_error',
+    'task_received',
+    'task_revoked',
+    'task_unknown',
+    'task_rejected',
+]
+
+def log_celery_signal(*args, **kwargs):
+    print(f"⚠️⚠️⚠️ Celery signal received: {args=} {kwargs=}")
+
+for signal_name in CELERY_SIGNAL_NAMES:
+    getattr(celery.signals, signal_name).connect(log_celery_signal, weak=False)