Update Celery Queuenames

default_config.yaml

----
-# ---
-# Database information
-# ---
-
-# SANDBOX
-#DATABASES:
-#    default:
-#        ENGINE: django.db.backends.sqlite3
-#        NAME: sandbox.db
-
-## PRODUCTION
-DATABASES:
-    default:
-        ENGINE: 'django.db.backends.mysql'
-        NAME:
-        USER:
-        PASSWORD:
-        HOST:
-        PORT: '3306'
-
-django_secret_key: ""
-
-# ---
-# AWS Buckets, Prefixes
-# ---
-# Studio/Platform
-edx_s3_ingest_prefix:
-edx_s3_ingest_bucket:
-edx_s3_endpoint_bucket:
-# CF
-edx_cloudfront_prefix:
-# Images
-aws_video_images_bucket:
-aws_video_images_prefix: "video-images/"
-# VEDA Internal
-veda_s3_upload_bucket:
-veda_s3_hotstore_bucket:
-veda_deliverable_bucket:
-# Settings
-multi_upload_barrier: 2000000000
-
-
-# ---
-# email vars
-# ---
-veda_noreply_email:
-admin_email:
-
-# ---
-# VEDA API
-# ---
-## VEDA API Auth
-veda_api_url:
-veda_auth_url:
-veda_client_id:
-veda_secret_key:
-veda_token_url:
-
-# ---
-# VAL
-# ---
-val_api_url:
-val_token_url:
-val_video_images_url:
-# Credentials
-val_client_id:
-val_secret_key:
-val_password:
-val_username:
-
-# ---
-# Celery Info
-# ---
-celery_app_name: veda_production
-# can do multiple queues like so: foo,bar,baz
-main_celery_queue: encode_worker
-celery_receiver_queue: encode_worker
-largefile_celery_queue: large_encode_worker
-celery_stat_queue: transcode_stat
-largefile_queue_barrier: 1000000000
-celery_threads: 1
-
-rabbitmq_broker:
-rabbitmq_pass:
-rabbitmq_user:
-
-# ---
-# Shotgun Variables (internal mediateam)
-# ---
-sg_server_path:
-sg_script_name:
-sg_script_key:
-
-# ---
-# Endpoints
-# ---
-threeplay_ftphost:
-xuetang_api_url:
-xuetang_api_shared_secret:
-
-## Encoding Config
-ffmpeg_compiled: "ffmpeg"
-ffprobe_compiled: "ffprobe"
-target_aspect_ratio: 1.7777778
-
-# ----------
-##---
-# This is a list of encodes and their respective course
-# boolean matches
-encode_dict:
-    review_proc:
-        - review
-    mobile_override:
-        - override
-    s3_proc:
-        - mobile_high
-        - mobile_low
-        - audio_mp3
-        - desktop_webm
-        - desktop_mp4
-        - hls
-
-    yt_proc:
-        - youtube
-
-##---
-# This is a list of encode profiles and their val profile matches
-# boolean matches
-val_profile_dict:
-    mobile_low:
-        - mobile_low
-
-    desktop_mp4:
-        - desktop_mp4
-
-    override:
-        - desktop_mp4
-        - mobile_low
-        - mobile_high
-
-    mobile_high:
-        - mobile_high
-
-    audio_mp3:
-        - audio_mp3
-
-    desktop_webm:
-        - desktop_webm
-
-    youtube:
-        - youtube
-    review:
-    hls:
-        - hls
-
-#--
-# Heal settings
-heal_start: 1
-heal_end: 144
-
-...

tests/test_node_connect.py

-
-import os
-import sys
-import unittest
-
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from video_worker.config import WorkerSetup
-"""
-test connection to celery cluster
-
-"""
-
-
-class TestCeleryConnect(unittest.TestCase):
-
-    def setUp(self):
-        self.WS = WorkerSetup()
-        if os.path.exists(self.WS.instance_yaml):
-            self.WS.run()
-        self.settings = self.WS.settings_dict
-
-    def test_celery_setup(self):
-        if not os.path.exists(self.WS.instance_yaml):
-            self.assertTrue(True)
-            return None
-
-        salient_variables = [
-            'celery_app_name',
-            'celery_receiver_queue',
-            'rabbitmq_user',
-            'rabbitmq_pass',
-            'rabbitmq_broker'
-        ]
-
-        for s in salient_variables:
-            self.assertFalse(len(self.settings[s]) == 0)
-
-    @unittest.skip("not implemented")
-    def test_celery_credentials(self):
-        if not os.path.exists(self.WS.instance_yaml):
-            self.assertTrue(True)
-            return None
-
-        """
-        This is yuck, but I am in a hurry
-        """
-        os.chdir(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-        worker_call = 'python celeryapp.py worker --loglevel=info --concurrency=1 -Q ' \
-            + str(self.settings['celery_receiver_queue'])
-        a1 = subprocess.Popen(
-            worker_call, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True
-        )
-
-        print '** 10 sec of sleep while node connects to cluster **'
-        time.sleep(10)
-        a1.kill()  # Otherwise it's forever
-
-        test_command = 'Connected to amqp://' + self.settings['rabbitmq_user'] + \
-            ':**@' + self.settings['rabbitmq_broker'] + ':5672//'
-
-        for line in iter(a1.stdout.readline, b''):
-            print line
-            if test_command in line:
-                self.assertTrue(True)
-                return None
-
-        self.assertFalse(True)
-
-
-def main():
-    unittest.main()
-
-if __name__ == '__main__':
-    sys.exit(main())

worker.sh

@@ -9,7 +9,7 @@ ROOTDIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 cd ${ROOTDIR}
 
 # Get vars from yaml
-QUEUE=$(cat ${ROOTDIR}/instance_config.yaml | grep celery_receiver_queue)
+QUEUE=$(cat ${ROOTDIR}/instance_config.yaml | grep celery_worker_queue)
 QUEUE=${QUEUE#*: }
 CONCUR=$(cat ${ROOTDIR}/instance_config.yaml | grep celery_threads)
 CONCUR=${CONCUR#*: }