Merge pull request #13995 from edx/efischer/override_celery_queues

Additional Celery Queues

cms/envs/aws.py

@@ -93,13 +93,6 @@ ALTERNATE_QUEUES = [
     DEFAULT_PRIORITY_QUEUE.replace(QUEUE_VARIANT, alternate + '.')
     for alternate in ALTERNATE_QUEUE_ENVS
 ]
-CELERY_QUEUES.update(
-    {
-        alternate: {}
-        for alternate in ALTERNATE_QUEUES
-        if alternate not in CELERY_QUEUES.keys()
-    }
-)
 CELERY_ROUTES = "{}celery.Router".format(QUEUE_VARIANT)
 
 ############# NON-SECURE ENV CONFIG ##############################
@@ -372,6 +365,18 @@ BROKER_URL = "{0}://{1}:{2}@{3}/{4}".format(CELERY_BROKER_TRANSPORT,
                                             CELERY_BROKER_HOSTNAME,
                                             CELERY_BROKER_VHOST)
 
+# Allow CELERY_QUEUES to be overwritten before adding alternates
+ENV_CELERY_QUEUES = ENV_TOKENS.get('CELERY_QUEUES', None)
+if ENV_CELERY_QUEUES:
+    CELERY_QUEUES = {queue: {} for queue in ENV_CELERY_QUEUES}
+CELERY_QUEUES.update(
+    {
+        alternate: {}
+        for alternate in ALTERNATE_QUEUES
+        if alternate not in CELERY_QUEUES.keys()
+    }
+)
+
 # Event tracking
 TRACKING_BACKENDS.update(AUTH_TOKENS.get("TRACKING_BACKENDS", {}))
 EVENT_TRACKING_BACKENDS['tracking_logs']['OPTIONS']['backends'].update(AUTH_TOKENS.get("EVENT_TRACKING_BACKENDS", {}))

cms/envs/yaml_config.py

@@ -65,6 +65,34 @@ EMAIL_BACKEND = 'django_ses.SESBackend'
 SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
 DEFAULT_FILE_STORAGE = 'storages.backends.s3boto.S3BotoStorage'
 
+GIT_REPO_EXPORT_DIR = '/edx/var/edxapp/export_course_repos'
+SESSION_INACTIVITY_TIMEOUT_IN_SECONDS = None
+EMAIL_FILE_PATH = None
+STATIC_URL_BASE = None
+STATIC_ROOT_BASE = None
+SESSION_COOKIE_NAME = None
+ADDL_INSTALLED_APPS = []
+AUTH_USE_CAS = False
+CAS_ATTRIBUTE_CALLBACK = None
+MICROSITE_ROOT_DIR = ''
+CMS_SEGMENT_KEY = None
+DATADOG = {}
+ADDL_INSTALLED_APPS = []
+LOCAL_LOGLEVEL = 'INFO'
+##############################################################
+#
+# ENV TOKEN IMPORT
+#
+# Currently non-secure and secure settings are managed
+# in two yaml files. This section imports the non-secure
+# settings and modifies them in code if necessary.
+#
+
+with open(CONFIG_ROOT / CONFIG_PREFIX + "env.yaml") as env_file:
+    ENV_TOKENS = yaml.safe_load(env_file)
+
+ENV_TOKENS = convert_tokens(ENV_TOKENS)
+
 ##############################################################
 #
 # DEFAULT SETTINGS FOR CELERY
@@ -98,40 +126,17 @@ LOW_PRIORITY_QUEUE = 'edx.{0}core.low'.format(QUEUE_VARIANT)
 CELERY_DEFAULT_QUEUE = DEFAULT_PRIORITY_QUEUE
 CELERY_DEFAULT_ROUTING_KEY = DEFAULT_PRIORITY_QUEUE
 
-CELERY_QUEUES = {
-    HIGH_PRIORITY_QUEUE: {},
-    LOW_PRIORITY_QUEUE: {},
-    DEFAULT_PRIORITY_QUEUE: {}
-}
+ENV_CELERY_QUEUES = ENV_TOKENS.get('CELERY_QUEUES', None)
+if ENV_CELERY_QUEUES:
+    CELERY_QUEUES = {queue: {} for queue in ENV_CELERY_QUEUES}
+else:
+    CELERY_QUEUES = {
+        HIGH_PRIORITY_QUEUE: {},
+        LOW_PRIORITY_QUEUE: {},
+        DEFAULT_PRIORITY_QUEUE: {}
+    }
 
 CELERY_ALWAYS_EAGER = False
-GIT_REPO_EXPORT_DIR = '/edx/var/edxapp/export_course_repos'
-SESSION_INACTIVITY_TIMEOUT_IN_SECONDS = None
-EMAIL_FILE_PATH = None
-STATIC_URL_BASE = None
-STATIC_ROOT_BASE = None
-SESSION_COOKIE_NAME = None
-ADDL_INSTALLED_APPS = []
-AUTH_USE_CAS = False
-CAS_ATTRIBUTE_CALLBACK = None
-MICROSITE_ROOT_DIR = ''
-CMS_SEGMENT_KEY = None
-DATADOG = {}
-ADDL_INSTALLED_APPS = []
-LOCAL_LOGLEVEL = 'INFO'
-##############################################################
-#
-# ENV TOKEN IMPORT
-#
-# Currently non-secure and secure settings are managed
-# in two yaml files. This section imports the non-secure
-# settings and modifies them in code if necessary.
-#
-
-with open(CONFIG_ROOT / CONFIG_PREFIX + "env.yaml") as env_file:
-    ENV_TOKENS = yaml.safe_load(env_file)
-
-ENV_TOKENS = convert_tokens(ENV_TOKENS)
 
 ##########################################
 # Merge settings from common.py

lms/envs/aws.py

@@ -104,13 +104,6 @@ ALTERNATE_QUEUES = [
     DEFAULT_PRIORITY_QUEUE.replace(QUEUE_VARIANT, alternate + '.')
     for alternate in ALTERNATE_QUEUE_ENVS
 ]
-CELERY_QUEUES.update(
-    {
-        alternate: {}
-        for alternate in ALTERNATE_QUEUES
-        if alternate not in CELERY_QUEUES.keys()
-    }
-)
 CELERY_ROUTES = "{}celery.Router".format(QUEUE_VARIANT)
 
 # If we're a worker on the high_mem queue, set ourselves to die after processing
@@ -275,6 +268,18 @@ BULK_EMAIL_ROUTING_KEY_SMALL_JOBS = ENV_TOKENS.get('BULK_EMAIL_ROUTING_KEY_SMALL
 # Queue to use for updating persistent grades
 RECALCULATE_GRADES_ROUTING_KEY = ENV_TOKENS.get('RECALCULATE_GRADES_ROUTING_KEY', LOW_PRIORITY_QUEUE)
 
+# Allow CELERY_QUEUES to be overwritten before adding alternates
+ENV_CELERY_QUEUES = ENV_TOKENS.get('CELERY_QUEUES', None)
+if ENV_CELERY_QUEUES:
+    CELERY_QUEUES = {queue: {} for queue in ENV_CELERY_QUEUES}
+CELERY_QUEUES.update(
+    {
+        alternate: {}
+        for alternate in ALTERNATE_QUEUES
+        if alternate not in CELERY_QUEUES.keys()
+    }
+)
+
 # following setting is for backward compatibility
 if ENV_TOKENS.get('COMPREHENSIVE_THEME_DIR', None):
     COMPREHENSIVE_THEME_DIR = ENV_TOKENS.get('COMPREHENSIVE_THEME_DIR')

lms/envs/yaml_config.py

@@ -102,6 +102,21 @@ LOCAL_LOGLEVEL = 'INFO'
 
 ##############################################################
 #
+# ENV TOKEN IMPORT
+#
+# Currently non-secure and secure settings are managed
+# in two yaml files. This section imports the non-secure
+# settings and modifies them in code if necessary.
+#
+
+with open(CONFIG_ROOT / CONFIG_PREFIX + "env.yaml") as env_file:
+    ENV_TOKENS = yaml.safe_load(env_file)
+
+# Works around an Ansible bug
+ENV_TOKENS = convert_tokens(ENV_TOKENS)
+
+##############################################################
+#
 # DEFAULT SETTINGS FOR CELERY
 #
 
@@ -135,12 +150,16 @@ HIGH_MEM_QUEUE = 'edx.{0}core.high_mem'.format(QUEUE_VARIANT)
 CELERY_DEFAULT_QUEUE = DEFAULT_PRIORITY_QUEUE
 CELERY_DEFAULT_ROUTING_KEY = DEFAULT_PRIORITY_QUEUE
 
-CELERY_QUEUES = {
-    HIGH_PRIORITY_QUEUE: {},
-    LOW_PRIORITY_QUEUE: {},
-    DEFAULT_PRIORITY_QUEUE: {},
-    HIGH_MEM_QUEUE: {},
-}
+ENV_CELERY_QUEUES = ENV_TOKENS.get('CELERY_QUEUES', None)
+if ENV_CELERY_QUEUES:
+    CELERY_QUEUES = {queue: {} for queue in ENV_CELERY_QUEUES}
+else:
+    CELERY_QUEUES = {
+        HIGH_PRIORITY_QUEUE: {},
+        LOW_PRIORITY_QUEUE: {},
+        DEFAULT_PRIORITY_QUEUE: {},
+        HIGH_MEM_QUEUE: {},
+    }
 
 # If we're a worker on the high_mem queue, set ourselves to die after processing
 # one request to avoid having memory leaks take down the worker server. This env
@@ -150,22 +169,6 @@ CELERY_QUEUES = {
 if os.environ.get('QUEUE') == 'high_mem':
     CELERYD_MAX_TASKS_PER_CHILD = 1
 
-
-##############################################################
-#
-# ENV TOKEN IMPORT
-#
-# Currently non-secure and secure settings are managed
-# in two yaml files. This section imports the non-secure
-# settings and modifies them in code if necessary.
-#
-
-with open(CONFIG_ROOT / CONFIG_PREFIX + "env.yaml") as env_file:
-    ENV_TOKENS = yaml.safe_load(env_file)
-
-# Works around an Ansible bug
-ENV_TOKENS = convert_tokens(ENV_TOKENS)
-
 ##########################################
 # Merge settings from common.py
 #