Update config.yaml to define jobs rather than plugins

Pay off some technical debt and make config.yaml define jobs rather than
plugins.

Change-Id: Ib4aa17649c02ab246d31515c8073230e8259cc4e

4 files changed, 78 insertions, 53 deletions

diff --git a/turbo_hipster/lib/models.py b/turbo_hipster/lib/models.py
index 3599766..403a7f1 100644
--- a/turbo_hipster/lib/models.py
+++ b/turbo_hipster/lib/models.py
@@ -19,6 +19,7 @@ import logging
 import os
 import pkg_resources
 import socket
+import uuid
 
 from turbo_hipster.lib import common
 from turbo_hipster.lib import utils
@@ -28,9 +29,13 @@ class Task(object):
     """ A base object for running a job (aka Task) """
     log = logging.getLogger("task")
 
-    def __init__(self, worker_server, plugin_config, job_name):
+    def __init__(self, worker_server, job_name, job_config):
+        # TODO(jhesketh): remove the need for worker_server here
         self.worker_server = worker_server
-        self.plugin_config = plugin_config
+        # NOTE(jhesketh): job_config may be in the old format where name
+        # refers to the plugin and function is the job name. Thus these should
+        # never be used in a job, instead use the provided job_name.
+        self.job_config = job_config
         self.job_name = job_name
         self._reset()
 
@@ -52,16 +57,16 @@ class Task(object):
         self.messages = []
         self.current_step = 0
         self.log_handler = None
+        self.th_uuid = str(uuid.uuid4())[-12:]
 
     def _prep_working_dir(self):
-        self.job_identifier = utils.determine_job_identifier(
-            self.job_arguments,
-            self.plugin_config['function'],
-            self.job.unique
-        )
+        # Use the th_uuid so that if the same job is somehow taken twice from
+        # zuul we won't re-use zuul's uuid. This shouldn't happen but if it
+        # does it prevents overwriting previous results
         self.job_working_dir = os.path.join(
             self.worker_server.config['jobs_working_dir'],
-            self.job_identifier
+            self.th_uuid,
+            self.job_arguments['LOG_PATH']
         )
         self.job_results_dir = os.path.join(
             self.job_working_dir,
@@ -221,7 +226,7 @@ class Task(object):
 
         if 'publish_logs' in self.worker_server.config:
             index_url = utils.push_file(
-                self.job_identifier, self.job_results_dir,
+                self.job_arguments['LOG_PATH'], self.job_results_dir,
                 self.worker_server.config['publish_logs'])
             self.log.debug("Index URL found at %s" % index_url)
             self.work_data['url'] = index_url
@@ -229,14 +234,14 @@ class Task(object):
         if 'ZUUL_EXTRA_SWIFT_URL' in self.job_arguments:
             # Upload to zuul's url as instructed
             utils.zuul_swift_upload(self.job_working_dir, self.job_arguments)
-            self.work_data['url'] = self.job_identifier
+            self.work_data['url'] = self.job_arguments['LOG_PATH']
 
 
 class ShellTask(Task):
     log = logging.getLogger("task.shell_task")
 
-    def __init__(self, worker_server, plugin_config, job_name):
-        super(ShellTask, self).__init__(worker_server, plugin_config, job_name)
+    def __init__(self, worker_server, job_name, job_config):
+        super(ShellTask, self).__init__(worker_server, job_name, job_config)
         # Define the number of steps we will do to determine our progress.
         self.total_steps = 5
 
@@ -285,7 +290,7 @@ class ShellTask(Task):
 
         self.log.debug("Grab the patchset we want to test against")
         local_path = os.path.join(self.worker_server.config['git_working_dir'],
-                                  self.job_name, job_args['ZUUL_PROJECT'])
+                                  self.th_uuid, job_args['ZUUL_PROJECT'])
         if not os.path.exists(local_path):
             os.makedirs(local_path)
 
@@ -305,7 +310,7 @@ class ShellTask(Task):
     @common.task_step
     def _execute_script(self):
         # Run script
-        cmd = self.plugin_config['shell_script']
+        cmd = self.job_config['shell_script']
         cmd += (
             (' %(git_path)s %(job_working_dir)s %(unique_id)s')
             % {
@@ -339,8 +344,8 @@ class ShellTask(Task):
     def _handle_cleanup(self):
         """Handle and cleanup functions. Shutdown if requested to so that no
         further jobs are ran if the environment is dirty."""
-        if ('shutdown-th' in self.plugin_config and
-            self.plugin_config['shutdown-th']):
+        if ('shutdown-th' in self.job_config and
+            self.job_config['shutdown-th']):
             self.worker_server.shutdown_gracefully()
 
     @common.task_step
diff --git a/turbo_hipster/lib/utils.py b/turbo_hipster/lib/utils.py
index aee31e7..3a8fc2c 100644
--- a/turbo_hipster/lib/utils.py
+++ b/turbo_hipster/lib/utils.py
@@ -263,12 +263,6 @@ def scp_push_file(results_set_name, file_path, local_config):
     pass
 
 
-def determine_job_identifier(zuul_arguments, job, unique):
-    # use new determined path from zuul
-    path = zuul_arguments['LOG_PATH']
-    return path
-
-
 def zuul_swift_upload(file_path, job_arguments):
     """Upload working_dir to swift as per zuul's instructions"""
     # NOTE(jhesketh): Zuul specifies an object prefix in the destination so
diff --git a/turbo_hipster/task_plugins/real_db_upgrade/task.py b/turbo_hipster/task_plugins/real_db_upgrade/task.py
index 635566c..25bd5ba 100644
--- a/turbo_hipster/task_plugins/real_db_upgrade/task.py
+++ b/turbo_hipster/task_plugins/real_db_upgrade/task.py
@@ -40,8 +40,8 @@ class Runner(models.ShellTask):
 
     log = logging.getLogger("task.real_db_upgrade")
 
-    def __init__(self, worker_server, plugin_config, job_name):
-        super(Runner, self).__init__(worker_server, plugin_config, job_name)
+    def __init__(self, worker_server, job_name, job_config):
+        super(Runner, self).__init__(worker_server, job_name, job_config)
 
         # Set up the runner worker
         self.datasets = []
@@ -69,10 +69,7 @@ class Runner(models.ShellTask):
             if (self.job_arguments['ZUUL_PROJECT'] ==
                     dataset['config']['project'] and
                     self._get_project_command(dataset['config']['type'])):
-                dataset['determined_path'] = utils.determine_job_identifier(
-                    self.job_arguments, self.plugin_config['function'],
-                    self.job.unique
-                )
+                dataset['determined_path'] = self.job_arguments['LOG_PATH']
                 dataset['job_log_file_path'] = os.path.join(
                     self.worker_server.config['jobs_working_dir'],
                     dataset['determined_path'],
@@ -129,7 +126,7 @@ class Runner(models.ShellTask):
         if len(self.datasets) > 0:
             return self.datasets
 
-        datasets_path = self.plugin_config['datasets_dir']
+        datasets_path = self.job_config['datasets_dir']
         for ent in os.listdir(datasets_path):
             dataset_dir = os.path.join(datasets_path, ent)
             if (os.path.isdir(dataset_dir) and os.path.isfile(
diff --git a/turbo_hipster/worker_server.py b/turbo_hipster/worker_server.py
index fae193d..5a2ff00 100644
--- a/turbo_hipster/worker_server.py
+++ b/turbo_hipster/worker_server.py
@@ -47,15 +47,14 @@ class Server(threading.Thread):
         # Config init
         self.zuul_manager = None
         self.zuul_client = None
-        self.plugins = []
         self.services_started = False
 
         # TODO: Make me unique (random?) and we should be able to run multiple
         # instances of turbo-hipster on the one host
         self.worker_name = os.uname()[1]
 
-        self.tasks = {}
-        self.load_plugins()
+        self.jobs = {}
+        self.load_jobs()
 
     def load_extra_configuration(self):
         if isdir(self.config["conf_d"]):
@@ -84,41 +83,71 @@ class Server(threading.Thread):
                             filename=log_file,
                             level=logging.DEBUG)
 
+    def load_jobs(self):
+        # Legacy, load the plugins first
+        self.load_plugins()
+
+        self.log.debug("Loading jobs")
+        if 'jobs' in self.config:
+            for job in self.config['jobs']:
+                try:
+                    plugin = 'shell_script'
+                    if 'plugin' in job:
+                        plugin = job['plugin']
+
+                    module = __import__('turbo_hipster.task_plugins.' +
+                                        plugin + '.task',
+                                        fromlist='turbo_hipster.task_plugins' +
+                                        plugin)
+
+                    self.jobs[job['name']] = {
+                        'name': job['name'],
+                        'plugin': plugin,
+                        'job_config': job,
+                        'runner': module.Runner(self, job['name'], job),
+                    }
+                    self.log.debug('Job %s loaded' % job['name'])
+                except Exception as e:
+                    self.log.exception("Failure loading job")
+                    self.log.exception(e)
+
     def load_plugins(self):
         """ Load the available plugins from task_plugins """
         self.log.debug('Loading plugins')
         # Load plugins
-        for plugin in self.config['plugins']:
-            self.plugins.append({
-                'module': __import__('turbo_hipster.task_plugins.' +
-                                     plugin['name'] + '.task',
-                                     fromlist='turbo_hipster.task_plugins' +
-                                     plugin['name']),
-                'plugin_config': plugin
-            })
-            self.log.debug('Plugin %s loaded' % plugin['name'])
+        if 'plugins' in self.config:
+            for plugin in self.config['plugins']:
+                try:
+                    module = __import__('turbo_hipster.task_plugins.' +
+                                        plugin['name'] + '.task',
+                                        fromlist='turbo_hipster.task_plugins' +
+                                        plugin['name'])
+
+                    self.jobs[plugin['function']] = {
+                        'name': plugin['function'],
+                        'plugin': plugin['name'],
+                        'plugin_config': plugin,
+                        'runner': module.Runner(
+                            self, plugin['function'], plugin
+                        ),
+                    }
+                    self.log.debug('Job %s loaded' % plugin['function'])
+                except Exception as e:
+                    self.log.exception("Failure loading plugin")
+                    self.log.exception(e)
 
     def start_zuul_client(self):
         """ Run the tasks """
         self.log.debug('Starting zuul client')
         self.zuul_client = worker_manager.ZuulClient(self)
 
-        for task_number, plugin in enumerate(self.plugins):
-            module = plugin['module']
-            job_name = '%s-%s-%s' % (plugin['plugin_config']['name'],
-                                     self.worker_name, task_number)
-            self.tasks[job_name] = module.Runner(
-                self,
-                plugin['plugin_config'],
-                job_name
-            )
-            self.zuul_client.add_function(plugin['plugin_config']['function'],
-                                          self.tasks[job_name])
+        for job in self.jobs.values():
+            self.zuul_client.add_function(job['name'], job['runner'])
 
         self.zuul_client.start()
 
     def start_zuul_manager(self):
-        self.zuul_manager = worker_manager.ZuulManager(self, self.tasks)
+        self.zuul_manager = worker_manager.ZuulManager(self, self.jobs)
         self.zuul_manager.start()
 
     def shutdown_gracefully(self):