Merge pull request #19 from edx/yro/add_onsite_workers

Reintegrate onsite GPU Workers / Update Celery

instance_config.yaml

 ---
+## NOTE!
+# NEVER COMMIT A SECRET HERE!
+#
+# This file is meant as a guide, but should not be assumed to deploy to production.
+# To deploy changes to this file to production,
+# change in @edx/veda-secure/shared/instance_config.yaml
+#
+#
 
-# ---
-# AWS
-# ---
-veda_s3_hotstore_bucket:
-veda_deliverable_bucket:
-edx_s3_endpoint_bucket:
-aws_video_images_bucket:
-aws_video_images_prefix: ''
-
-# ---
-# Celery
-# ---
-celery_threads: 1
-
-# ---
-# VAL user creds
-# ---
-val_api_url:
-val_video_images_url:
-val_client_id:
-val_password:
-val_secret_key:
-val_token_url:
-val_username:
-
-## VEDA API Auth
-veda_api_url:
-veda_auth_url:
-veda_client_id:
-veda_secret_key:
-veda_token_url:
-
-## CLOUD
-rabbitmq_broker: ""
-rabbitmq_pass: ""
-rabbitmq_user: ""
-
-instance_prefix: ""
+# ----------
+onsite_worker: False
+veda_s3_hotstore_bucket: ""
+##---
+# This is a list of encodes and their respective course
+# boolean matches
+encode_dict:
+    review_proc:
+        - review
+    mobile_override:
+        - override
+    s3_proc:
+        - mobile_low
+        - audio_mp3
+        - 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
+
+    audio_mp3:
+        - audio_mp3
+
+    youtube:
+        - youtube
+    review:
+    hls:
+        - hls
+
+
+#--
+# Global settings
+heal_start: 3
+heal_end: 144
+global_timeout: 40
+ffmpeg_compiled: 'ffmpeg'
+ffprobe_compiled: 'ffprobe'
+
+...

requirements.txt

@@ -2,8 +2,8 @@ PyYAML==3.11
 boto==2.39.0
 nose==1.3.3
 requests==2.10.0
-celery==3.1.18
+celery==4.1.0
 newrelic
 # TODO! https://openedx.atlassian.net/browse/EDUCATOR-2279
 # this way is required when installing using `pip install requirements.txt` so that `encode_profiles.json` is accessible
--e git+https://github.com/yro/chunkey.git@14f9c330d4ba5a4290dd0d17429b142618bc2c9f#egg=chunkey
+-e git+https://github.com/yro/chunkey.git@f659c95552413d340ccab7bd27e8d63a874fd316#egg=chunkey

video_worker/__init__.py

@@ -18,7 +18,14 @@ from video_worker.api_communicate import UpdateAPIStatus
 from celeryapp import deliverable_route
 from video_worker.generate_encode import CommandGenerate
 from video_worker.generate_delivery import Deliverable
-from video_worker.global_vars import ENCODE_WORK_DIR, VAL_TRANSCODE_STATUS, NODE_TRANSCODE_STATUS
+
+from video_worker.global_vars import (
+    HOME_DIR,
+    ENCODE_WORK_DIR,
+    VAL_TRANSCODE_STATUS,
+    NODE_TRANSCODE_STATUS,
+    BOTO_TIMEOUT
+)
 from video_worker.reporting import Output
 from video_worker.validate import ValidateVideo
 from video_worker.video_images import VideoImages
@@ -29,9 +36,8 @@ try:
 except:
     pass
 
-boto.config.set('Boto', 'http_socket_timeout', '10')
-
-logging.basicConfig()
+boto.config.set('Boto', 'http_socket_timeout', BOTO_TIMEOUT)
+logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 
@@ -45,17 +51,14 @@ class VideoWorker(object):
         self.encode_profile = kwargs.get('encode_profile', None)
         self.VideoObject = kwargs.get('VideoObject', None)
 
-        # Working Dir Config
-        self.workdir = kwargs.get('workdir', None)
-        if self.workdir is None:
-            if self.jobid is None:
-                self.workdir = ENCODE_WORK_DIR
-            else:
-                self.workdir = os.path.join(ENCODE_WORK_DIR, self.jobid)
-
-            if not os.path.exists(ENCODE_WORK_DIR):
-                os.mkdir(ENCODE_WORK_DIR)
-
+        self.instance_yaml = kwargs.get(
+            'instance_yaml',
+            os.path.join(
+                os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+                'instance_config.yaml'
+            )
+        )
+        self.workdir = kwargs.get('workdir', self.determine_workdir())
         self.ffcommand = None
         self.source_file = kwargs.get('source_file', None)
         self.output_file = None
@@ -64,6 +67,14 @@ class VideoWorker(object):
         self.encoded = False
         self.delivered = False
 
+    def determine_workdir(self):
+        if not os.path.exists(ENCODE_WORK_DIR):
+            os.mkdir(ENCODE_WORK_DIR)
+        if self.jobid is None:
+            return ENCODE_WORK_DIR
+        else:
+            return os.path.join(ENCODE_WORK_DIR, self.jobid)
+
     def test(self):
         """
         Run tests
@@ -128,6 +139,9 @@ class VideoWorker(object):
         # generate video images command and update S3 and edxval
         # run against 'hls' encode only
         if self.encode_profile == 'hls':
+            # Run HLS encode
+            self._hls_pipeline()
+            # Auto-video Images
             VideoImages(
                 video_object=self.VideoObject,
                 work_dir=self.workdir,
@@ -135,8 +149,7 @@ class VideoWorker(object):
                 jobid=self.jobid,
                 settings=self.settings
             ).create_and_update()
-            # Run HLS encode
-            self._hls_pipeline()
+
         else:
             self._static_pipeline()
 
@@ -179,14 +192,23 @@ class VideoWorker(object):
 
         os.chdir(self.workdir)
 
-        V1 = Chunkey(
-            mezz_file=os.path.join(self.workdir, self.source_file),
-            DELIVER_BUCKET=self.settings['edx_s3_endpoint_bucket'],
-            clean=False
-        )
+        if self.settings['onsite_worker'] is True:
+            hls_chunk_instance = Chunkey(
+                mezz_file=os.path.join(self.workdir, self.source_file),
+                DELIVER_BUCKET=self.settings['edx_s3_endpoint_bucket'],
+                clean=False,
+                ACCESS_KEY_ID=self.settings['edx_access_key_id'],
+                SECRET_ACCESS_KEY=self.settings['edx_secret_access_key']
+            )
+        else:
+            hls_chunk_instance = Chunkey(
+                mezz_file=os.path.join(self.workdir, self.source_file),
+                DELIVER_BUCKET=self.settings['edx_s3_endpoint_bucket'],
+                clean=False,
+            )
 
-        if V1.complete:
-            self.endpoint_url = V1.manifest_url
+        if hls_chunk_instance.complete:
+            self.endpoint_url = hls_chunk_instance.manifest_url
 
     def _engine_intake(self):
         """
@@ -197,7 +219,13 @@ class VideoWorker(object):
             return
 
         if self.source_file is None:
-            conn = S3Connection()
+            if self.settings['onsite_worker'] is True:
+                conn = S3Connection(
+                    self.settings['veda_access_key_id'],
+                    self.settings['veda_secret_access_key']
+                )
+            else:
+                conn = S3Connection()
             try:
                 bucket = conn.get_bucket(self.settings['veda_s3_hotstore_bucket'])
             except S3ResponseError:
@@ -266,6 +294,9 @@ class VideoWorker(object):
         if not os.path.exists(os.path.join(self.workdir, self.source_file)):
             logger.error('[VIDEO_WORKER] Source File (local) NOT FOUND - Input')
             return
+        print '%s : %s' % (self.VideoObject.veda_id, self.encode_profile)
+        # to be polite
+        print
         process = subprocess.Popen(
             self.ffcommand,
             stdout=subprocess.PIPE,
@@ -273,7 +304,6 @@ class VideoWorker(object):
             shell=True,
             universal_newlines=True
         )
-        print '%s : %s' % (self.VideoObject.veda_id, self.encode_profile)
         Output.status_bar(process=process)
         # to be polite
         print
@@ -288,11 +318,15 @@ class VideoWorker(object):
         Validate encode by matching (w/in 5 sec) encode duration,
         as well as standard validation tests
         """
-        self.encoded = ValidateVideo(
-            filepath=os.path.join(self.workdir, self.output_file),
-            product_file=True,
-            VideoObject=self.VideoObject
-        ).valid
+        if self.output_file is None:
+            self.encoded = False
+            return
+        else:
+            self.encoded = ValidateVideo(
+                filepath=os.path.join(self.workdir, self.output_file),
+                product_file=True,
+                VideoObject=self.VideoObject
+            ).valid
 
     def _deliver_file(self):
         """

video_worker/celeryapp.py

@@ -7,7 +7,6 @@ Start Celery Worker (if VEDA-attached node)
 from celery import Celery
 import os
 import shutil
-import sys
 
 from video_worker.utils import get_config
 from video_worker.global_vars import ENCODE_WORK_DIR
@@ -16,16 +15,12 @@ from video_worker.global_vars import ENCODE_WORK_DIR
 settings = get_config()
 
 
-
-def cel_Start():
+def cel_start():
     app = Celery(
         settings.setdefault('celery_app_name', ''),
         broker='amqp://' + settings.setdefault('rabbitmq_user', '') +
                ':' + settings.setdefault('rabbitmq_pass', '') +
                '@' + settings.setdefault('rabbitmq_broker', '') + ':5672//',
-        backend='amqp://' + settings.setdefault('rabbitmq_user', '') +
-                ':' + settings.setdefault('rabbitmq_pass', '') +
-                '@' + settings.setdefault('rabbitmq_broker', '') + ':5672//',
         include=['celeryapp']
     )
 
@@ -34,13 +29,20 @@ def cel_Start():
         CELERY_IGNORE_RESULT=True,
         CELERY_TASK_RESULT_EXPIRES=10,
         CELERYD_PREFETCH_MULTIPLIER=1,
-        CELERY_ACCEPT_CONTENT=['pickle', 'json', 'msgpack', 'yaml']
+        CELERY_ACCEPT_CONTENT=['json'],
+        CELERY_TASK_PUBLISH_RETRY=True,
+        CELERY_TASK_PUBLISH_RETRY_POLICY={
+            "max_retries": 3,
+            "interval_start": 0,
+            "interval_step": 1,
+            "interval_max": 5
+        }
     )
 
     return app
 
 
-app = cel_Start()
+app = cel_start()
 
 
 @app.task(name='worker_encode')
@@ -98,5 +100,5 @@ def test_command(message):
 
 
 if __name__ == '__main__':
-    app = cel_Start()
+    app = cel_start()
     app.start()

video_worker/generate_delivery.py

@@ -83,15 +83,14 @@ class Deliverable():
         Upload single part (under threshold in node_config)
         node_config MULTI_UPLOAD_BARRIER
         """
-        try:
-            conn = boto.connect_s3()
-            delv_bucket = conn.get_bucket(settings['veda_deliverable_bucket'])
-
-        except S3ResponseError:
-            ErrorObject().print_error(
-                message='Deliverable Fail: s3 Connection Error - Singleton'
+        if settings['onsite_worker'] is True:
+            conn = boto.connect_s3(
+                settings['veda_access_key_id'],
+                settings['veda_secret_access_key']
             )
-            return False
+        else:
+            conn = boto.connect_s3()
+        delv_bucket = conn.get_bucket(settings['veda_deliverable_bucket'])
 
         upload_key = Key(delv_bucket)
         upload_key.key = self.output_file
@@ -129,14 +128,14 @@ class Deliverable():
         sys.stdout.flush()
 
         # Connect to s3
-        try:
-            c = boto.connect_s3()
-            b = c.lookup(settings['veda_deliverable_bucket'])
-        except S3ResponseError:
-            ErrorObject().print_error(
-                message='Deliverable Fail: s3 Connection Error - Multipart'
+        if settings['onsite_worker'] is True:
+            conn = boto.connect_s3(
+                settings['veda_access_key_id'],
+                settings['veda_secret_access_key']
             )
-            return False
+        else:
+            conn = boto.connect_s3()
+        b = conn.lookup(settings['veda_deliverable_bucket'])
 
         if b is None:
             ErrorObject().print_error(

video_worker/global_vars.py

@@ -5,7 +5,7 @@ Globals
 """
 
 import os
-
+from os.path import expanduser
 from video_worker.utils import get_config, ROOT_DIR
 
 DEFAULT_ENCODE_WORK_DIR = os.path.join(ROOT_DIR, 'ENCODE_WORKDIR')
@@ -28,7 +28,7 @@ HLS_SUBSTITUTE = 'mobile_low'
 
 # For BOTO Multipart uploader
 MULTI_UPLOAD_BARRIER = 2000000000
-BOTO_TIMEOUT = 60
+BOTO_TIMEOUT = '60'
 
 # Settings for testing
 TEST_VIDEO_DIR = os.path.join(
@@ -42,3 +42,5 @@ NODE_COLORS_BLUE = '\033[94m'
 NODE_COLORS_GREEN = '\033[92m'
 NODE_COLORS_RED = '\033[91m'
 NODE_COLORS_END = '\033[0m'
+
+HOME_DIR = expanduser("~")

video_worker/tests/data/test_instance_config.yaml

 #---#
 # Test instance_config for edx-video-worker
 #---#
+# Onsite
+onsite_worker: False
+
 # VEDA credentials
 veda_api_url: 'http://dummy-veda-api-url'
 veda_token_url: 'http://dummy-veda-token-url'

video_worker/tests/test_video_worker.py

@@ -14,9 +14,6 @@ from video_worker.abstractions import Video
 from video_worker.global_vars import ENCODE_WORK_DIR
 from video_worker.utils import get_config
 
-from utils import TEST_INSTANCE_YAML
-
-
 worker_settings = get_config()
 
 

video_worker/video_images.py

@@ -122,7 +122,13 @@ class VideoImages(object):
         """
         Upload auto generated images to S3.
         """
-        s3_connection = S3Connection()
+        if self.settings['onsite_worker'] is True:
+            s3_connection = S3Connection(
+                self.settings['edx_access_key_id'],
+                self.settings['edx_secret_access_key']
+            )
+        else:
+            s3_connection = S3Connection()
 
         try:
             bucket = s3_connection.get_bucket(self.settings['aws_video_images_bucket'])