Refactoring to improve maintainability of code (#86)

We want to soon start publishing action requests on Treeherder.
In order to accomodate for that I wanted to improve the code to make it
easier to add this feature.

Some of the improvements are:

treeherder_job_event.py
* Rename treeherder_buildbot to treeherder_job_event as it won't be
  Buildbot only in the future
* Shorten treeherder_job_event's logging to improve readability
* Logging of backfilling action now points to a treeherder link rather
  than a job id (unintelligible info)
* Cleaned up variable naming and reduced code with try/block

worker.py
* We now read treeherder_host from the config files
* Handlers will eventually return an exit code in order to help us
  determine how an action request should show up on Treeherder

Procfile

@@ -1,2 +1,2 @@
-worker2: python pulse_actions/worker.py --config-file configs/production_config.json
+worker2: python pulse_actions/worker.py --config-file configs/worker2.json
 worker3: python pulse_actions/worker.py --config-file configs/talos_config.json --debug

configs/production_config.json → configs/worker2.json

@@ -1,5 +1,5 @@
 {
-    "applabel": "['resultset_actions', 'manual_backfill', 'runnable']",
+    "applabel": "worker2",
     "durable": "true",
     "sources": {
         "resultset_actions": {
@@ -14,5 +14,8 @@
             "exchange": "exchange/treeherder/v1/resultset-runnable-job-actions",
             "topic": "#"
         }
+    },
+    "pulse_actions": {
+        "treeherder_host": "treeherder.mozilla.org"
     }
 }

configs/worker2_stage.json

@@ -0,0 +1,20 @@
+{
+    "durable": "false",
+    "sources": {
+        "resultset_actions": {
+            "exchange": "exchange/treeherder-stage/v1/resultset-actions",
+            "topic": "#.#"
+        },
+        "manual_backfill": {
+            "exchange": "exchange/treeherder-stage/v1/job-actions",
+            "topic": "buildbot.#.backfill"
+        },
+        "runnable": {
+            "exchange": "exchange/treeherder-stage/v1/resultset-runnable-job-actions",
+            "topic": "#"
+        }
+    },
+    "pulse_actions": {
+        "treeherder_host": "treeherder.allizom.org"
+    }
+}

pulse_actions/handlers/treeherder_job_event.py

@@ -0,0 +1,135 @@
+"""
+This module deals with Treeherder's job actions.
+
+Exchange documentation:
+ https://wiki.mozilla.org/Auto-tools/Projects/Pulse/Exchanges#Treeherder:_Job_Actions
+
+- exchange/treeherder/v1/job-actions on buildbot.#.#.
+- exchange/treeherder-stage/v1/job-actions on buildbot.#.#.
+
+
+Sample data from Pulse:
+
+{u'action': u'backfill',
+ u'build_system_type': u'buildbot',
+ u'job_guid': u'cb4c4cf5fdb50bc6b03ec18f50dc19bdf4ce6811',
+ u'job_id': 30210861,
+ u'project': u'mozilla-inbound',
+ u'requester': u'armenzg@mozilla.com',
+ u'version': 1}
+
+Sample job from Treeherder:
+
+[{u'build_architecture': u'x86_64',
+  u'build_os': u'mac',
+  u'build_platform': u'osx-10-10',
+  u'build_platform_id': 69,
+  u'build_system_type': u'buildbot',
+  u'end_timestamp': 1466065682,
+  u'failure_classification_id': 4,
+  u'id': 30210861,
+  u'job_coalesced_to_guid': None,
+  u'job_group_description': u'fill me',
+  u'job_group_id': 9,
+  u'job_group_name': u'Mochitest e10s',
+  u'job_group_symbol': u'M-e10s',
+  u'job_guid': u'cb4c4cf5fdb50bc6b03ec18f50dc19bdf4ce6811',
+  u'job_type_description': u'',
+  u'job_type_id': 178,
+  u'job_type_name': u'Mochitest e10s Browser Chrome',
+  u'job_type_symbol': u'bc2',
+  u'last_modified': u'2016-06-16T08:40:39',
+  u'machine_name': u't-yosemite-r7-0204',
+  u'machine_platform_architecture': u'x86_64',
+  u'machine_platform_os': u'mac',
+  u'option_collection_hash': u'102210fe594ee9b33d82058545b1ed14f4c8206e',
+  u'platform': u'osx-10-10',
+  u'platform_option': u'opt',
+  u'reason': u'scheduler',
+  u'ref_data_name': u'Rev7 MacOSX Yosemite 10.10.5 mozilla-inbound opt test mochitest-e10s-browser-chrome-2',
+  u'result': u'testfailed',
+  u'result_set_id': 33460,
+  u'running_eta': 588,
+  u'signature': u'931d973dbadb4c8e065d10905fc0b83f0eecb7f8',
+  u'start_timestamp': 1466065035,
+  u'state': u'completed',
+  u'submit_timestamp': 1466065032,
+  u'tier': 1,
+  u'who': u'tests-mozilla-inbound-yosemite_r7-opt-unittest'}]
+"""
+import logging
+
+from pulse_actions.utils.misc import filter_invalid_builders
+
+from mozci import query_jobs
+from mozci.mozci import manual_backfill
+from mozci.sources import buildjson
+from thclient import TreeherderClient
+
+LOG = logging.getLogger(__name__.split('.')[-1])
+
+
+def on_event(data, message, dry_run, treeherder_host='treeherder.mozilla.org'):
+    """Act upon Treeherder job events.
+
+    Return if the outcome was successful or not
+    """
+    # Cleaning mozci caches
+    buildjson.BUILDS_CACHE = {}
+    query_jobs.JOBS_CACHE = {}
+
+    treeherder_client = TreeherderClient(host=treeherder_host)
+
+    action = data['action'].capitalize()
+    job_id = data['job_id']
+    repo_name = data['project']
+    status = None
+
+    # We want to know the status of the job we're processing
+    try:
+        job_info = treeherder_client.get_jobs(repo_name, id=job_id)[0]
+    except IndexError:
+        LOG.info("We could not find any job_info for repo_name: %s and "
+                 "job_id: %s" % (repo_name, job_id))
+        return False
+
+    buildername = job_info["ref_data_name"]
+
+    # We want to know the revision associated for this job
+    result_sets = treeherder_client.get_resultsets(repo_name, id=job_info["result_set_id"])
+    revision = result_sets[0]["revision"]
+
+    link_to_job = 'https://{}/#/jobs?repo={}&revision={}&selectedJob={}'.format(
+        treeherder_host,
+        repo_name,
+        revision,
+        job_id
+    )
+
+    LOG.info("{} action requested by {} for '{}'".format(action, data['requester'], buildername))
+
+    buildername = filter_invalid_builders(buildername)
+
+    # Treeherder can send us invalid builder names
+    # https://bugzilla.mozilla.org/show_bug.cgi?id=1242038
+    if buildername is None:
+        status = 'Builder %s was invalid.' % buildername[0]
+
+    else:
+        # There are various actions that can be taken on a job, however, we currently
+        # only process the backfill one
+        if action == "Backfill":
+            manual_backfill(
+                revision=revision,
+                buildername=buildername,
+                dry_run=dry_run,
+            )
+            if not dry_run:
+                status = 'Backfill request sent'
+            else:
+                status = 'Dry-run mode, nothing was backfilled'
+            LOG.debug(status)
+
+    if not dry_run:
+        # We need to ack the message to remove it from our queue
+        message.ack()

pulse_actions/worker.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 import sys
@@ -6,7 +7,7 @@
 from argparse import ArgumentParser
 from timeit import default_timer
 
-import pulse_actions.handlers.treeherder_buildbot as treeherder_buildbot
+import pulse_actions.handlers.treeherder_job_event as treeherder_job_event
 import pulse_actions.handlers.treeherder_resultset as treeherder_resultset
 import pulse_actions.handlers.treeherder_runnable as treeherder_runnable
 import pulse_actions.handlers.talos as talos
@@ -33,41 +34,64 @@ def main():
     else:
         LOG = setup_logging(logging.INFO)
 
+    # Load information only relevant to pulse_actions
+    with open(options.config_file, 'r') as file:
+        pulse_actions_config = json.load(file)['pulse_actions']
+
+    treeherder_host = pulse_actions_config['treeherder_host']
+
     # Disable mozci's validations
     disable_validations()
-    if not options.replay_file:
-        run_listener(options.config_file, options.dry_run)
-    else:
+    if options.replay_file:
         replay_messages(options.replay_file, route, dry_run=True)
+    else:
+        # Normal execution path
+        run_listener(config_file=options.config_file, dry_run=options.dry_run,
+                     treeherder_host=treeherder_host)
+
 
+def route(data, message, dry_run, treeherder_host):
+    ''' We need to map every exchange/topic to a specific handler.
 
-def route(data, message, dry_run):
+    We return if the request was processed succesfully or not
+    '''
+    # XXX: This is not ideal; we should define in the config which exchange uses which handler
+    # XXX: Specify here which treeherder host
     if 'job_id' in data:
-        treeherder_buildbot.on_buildbot_event(data, message, dry_run)
+        exit_code = treeherder_job_event.on_event(data, message, dry_run, treeherder_host)
     elif 'buildernames' in data:
-        treeherder_runnable.on_runnable_job_prod_event(data, message, dry_run)
+        exit_code = treeherder_runnable.on_runnable_job_prod_event(data, message, dry_run)
     elif 'resultset_id' in data:
-        treeherder_resultset.on_resultset_action_event(data, message, dry_run)
+        exit_code = treeherder_resultset.on_resultset_action_event(data, message, dry_run)
     elif data['_meta']['exchange'] == 'exchange/build/normalized':
-        talos.on_event(data, message, dry_run)
+        exit_code = talos.on_event(data, message, dry_run)
     else:
         LOG.error("Exchange not supported by router (%s)." % data)
 
+    return exit_code
+
 
-def run_listener(config_file, dry_run=True):
+def run_listener(config_file, dry_run=True, treeherder_host='treeherder.mozilla.org'):
     # Pulse consumer's callback passes only data and message arguments
-    # to the function, we need to pass dry-run
-    def handler_with_dry_run(data, message):
+    # to the function, we need to pass dry-run to route
+    def message_handler(data, message):
+        # XXX: Each request has to be logged into a unique file
+        # XXX: Upload each logging file into S3
+        # XXX: Report the job to Treeherder as running and then as complete
+        LOG.info('#### New request ####.')
         start_time = default_timer()
-        route(data, message, dry_run)
-        elapsed_time = default_timer() - start_time
+        route(data, message, dry_run, treeherder_host)
+        if not dry_run:
+            message.ack()
+
+        elapsed_time = int(default_timer() - start_time)
         LOG.info('Message {}, took {} seconds to execute'.format(data, str(elapsed_time)))
 
     consumer = create_consumer(
         user=os.environ['PULSE_USER'],
         password=os.environ['PULSE_PW'],
         config_file_path=config_file,
-        process_message=handler_with_dry_run,
+        process_message=message_handler,
     )
 
     while True:
@@ -82,18 +106,15 @@ def handler_with_dry_run(data, message):
 def parse_args(argv=None):
     parser = ArgumentParser()
     parser.add_argument('--config-file', dest="config_file", type=str)
-    parser.add_argument('--replay-file', dest="replay_file", type=str,
-                        help='You can specify a file with saved pulse_messages to process')
 
-    parser.add_argument("--dry-run",
-                        action="store_true",
-                        dest="dry_run",
-                        help="flag to test without actual push.")
+    parser.add_argument('--debug', action="store_true", dest="debug",
+                        help="Record debug messages.")
 
-    parser.add_argument("--debug",
-                        action="store_true",
-                        dest="debug",
-                        help="set debug for logging.")
+    parser.add_argument('--dry-run', action="store_true", dest="dry_run",
+                        help="Test without actual making changes.")
+
+    parser.add_argument('--replay-file', dest="replay_file", type=str,
+                        help='You can specify a file with saved pulse_messages to process')
 
     options = parser.parse_args(argv)
     return options