Merge pull request #43 from edx/yro/update_celery_queues

Yro/update celery queues

bin/deliver

@@ -66,7 +66,7 @@ class DeliverCli:
                     'worker',
                     '--loglevel=info',
                     '--concurrency=' + str(auth_dict['celery_threads']),
-                    '-Q ' + auth_dict['celery_stat_queue'],
+                    '-Q ' + auth_dict['celery_deliver_queue'],
                     '-n deliver.%h'
                 ))
             )

control/veda_file_ingest.py

@@ -163,22 +163,14 @@ class VedaIngest:
         if len(self.encode_list) == 0:
             return None
 
-        """
-        send job to queue
-        """
-        if self.video_proto.filesize > self.auth_dict['largefile_queue_barrier']:
-            cel_queue = self.auth_dict['largefile_celery_queue']
-        else:
-            cel_queue = self.auth_dict['main_celery_queue']
-
+        # Enqueue
         for e in self.encode_list:
-            # print e
             veda_id = self.video_proto.veda_id
             encode_profile = e
             jobid = uuid.uuid1().hex[0:10]
             celeryapp.worker_task_fire.apply_async(
                 (veda_id, encode_profile, jobid),
-                queue=cel_queue
+                queue=self.auth_dict['celery_worker_queue']
             )
 
         """

control/veda_heal.py

@@ -81,23 +81,15 @@ class VedaHeal(object):
                 VAC.call()
             self.val_status = None
 
-            if len(encode_list) > 0:
-                """
-                send job to queue
-                """
-                if v.video_orig_filesize > self.auth_dict['largefile_queue_barrier']:
-                    cel_queue = self.auth_dict['largefile_celery_queue']
-                else:
-                    cel_queue = self.auth_dict['main_celery_queue']
-
-                for e in encode_list:
-                    veda_id = v.edx_id
-                    encode_profile = e
-                    jobid = uuid.uuid1().hex[0:10]
-                    celeryapp.worker_task_fire.apply_async(
-                        (veda_id, encode_profile, jobid),
-                        queue=cel_queue
-                    )
+            # Enqueue
+            for e in encode_list:
+                veda_id = v.edx_id
+                encode_profile = e
+                jobid = uuid.uuid1().hex[0:10]
+                celeryapp.worker_task_fire.apply_async(
+                    (veda_id, encode_profile, jobid),
+                    queue=self.auth_dict['celery_worker_queue']
+                )
 
     def determine_fault(self, video_object):
         """

frontend/abvid_reporting.py

@@ -39,7 +39,10 @@ with open(auth_yaml, 'r') as stream:
 
 
 def report_status(status, abvid_serial, youtube_id):
-    v1 = VedaUpload.objects.filter(video_serial=abvid_serial).latest()
+    try:
+        v1 = VedaUpload.objects.filter(video_serial=abvid_serial).latest()
+    except ObjectDoesNotExist:
+        return
 
     if len(youtube_id) > 0:
         excuse = ''

instance_config.yaml

@@ -82,13 +82,11 @@ val_username:
 # ---
 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_worker_queue: encode_worker
+celery_deliver_queue: deliver_worker
 celery_threads: 1
 
+
 rabbitmq_broker:
 rabbitmq_pass:
 rabbitmq_user:

youtube_callback/sftp_id_retrieve.py

@@ -144,7 +144,7 @@ def domxml_parser(file):
     """
 
     if 'status-' not in file:
-        return None
+        return
 
     upload_data = {
         'datetime': None,
@@ -157,7 +157,9 @@ def domxml_parser(file):
     try:
         tree = ET.parse(os.path.join(workdir, file))
     except ET.ParseError:
-        return None
+        return
+    except IOError:
+        return
     root = tree.getroot()
     for child in root: