Pulse messages that are not processed should not show up on Treeherder (

#93)

We have some pulse messages which we filter out (i.e. ignore) and we don't want them
to submit a 'Sch' job to Treeherder.

This change adds to each of the handlers an ignore() method which allows
us to quickly know if a message will be processed or not.

In such cases we don't submit a Treeherder 'Sch' job.

There are also other significant changes in here:
* all handlers now return an exit code
  * the exit code is used to set the colour of the 'Sch' job
* we log more info for each request
* all handlers now have the same function name to process a pulse message
  * this will be important in the future when wanting to normalize all handlers

pulse_actions/handlers/talos_pgo_jobs.py

@@ -18,63 +18,71 @@
 LOG = logging.getLogger(__name__.split('.')[-1])
 
 
-def on_event(data, message, dry_run, acknowledge):
+def ignored(data):
+    '''It determines if the request will be processed or not.'''
+    try:
+        info = get_buildername_metadata(data['payload']['buildername'])
+        if info['build_type'] == "pgo" and \
+           info['repo_name'] in ['mozilla-inbound', 'fx-team'] and \
+           info['platform_name'] != 'win64':
+            return False
+        else:
+            return True
+    except MissingBuilderError, e:
+        LOG.warning(str(e))
+
+        return True
+
+
+def on_event(data, message, dry_run, acknowledge, **kwargs):
     """
     Whenever PGO builds are completed in mozilla-inbound or fx-team,
     we trigger the corresponding talos jobs twice.
     """
+    if ignored(data):
+        if acknowledge:
+            # We need to ack the message to remove it from our queue
+            message.ack()
+        LOG.debug("'%s' with status %i. Nothing to be done.",
+                  buildername, status)
+        return 0  # SUCCESS
+
     # Cleaning mozci caches
     buildjson.BUILDS_CACHE = {}
     query_jobs.JOBS_CACHE = {}
     payload = data["payload"]
     status = payload["status"]
     buildername = payload["buildername"]
-    try:
-        info = get_buildername_metadata(buildername)
-    except MissingBuilderError, e:
-        LOG.warning(str(e))
+    revision = payload["revision"]
+
+    # Treeherder can send us invalid builder names
+    # https://bugzilla.mozilla.org/show_bug.cgi?id=1242038
+    buildername = filter_invalid_builders(buildername)
+
+    if buildername is None:
         if acknowledge:
             # We need to ack the message to remove it from our queue
             message.ack()
+        return -1  # FAILURE
 
-        return
-    revision = payload["revision"]
+    try:
+
+        trigger_talos_jobs_for_build(
+            buildername=buildername,
+            revision=revision,
+            times=2,
+            priority=0,
+            dry_run=dry_run
+        )
 
-    if info['build_type'] == "pgo" and \
-       info['repo_name'] in ['mozilla-inbound', 'fx-team'] and \
-       info['platform_name'] != 'win64':
-        # Treeherder can send us invalid builder names
-        # https://bugzilla.mozilla.org/show_bug.cgi?id=1242038
-        buildername = filter_invalid_builders(buildername)
-
-        if buildername is None:
-            if acknowledge:
-                # We need to ack the message to remove it from our queue
-                message.ack()
-            return
-
-        try:
-
-            trigger_talos_jobs_for_build(
-                buildername=buildername,
-                revision=revision,
-                times=2,
-                priority=0,
-                dry_run=dry_run
-            )
-
-            if acknowledge:
-                # We need to ack the message to remove it from our queue
-                message.ack()
-
-        except Exception, e:
-            # The message has not been acked so we will try again
-            LOG.warning(str(e))
-            raise
-    else:
         if acknowledge:
             # We need to ack the message to remove it from our queue
             message.ack()
 
-        LOG.debug("'%s' with status %i. Nothing to be done.",
-                  buildername, status)
+        LOG.info('We triggered talos jobs for the build.')
+        return 0  # SUCCESS
+
+    except Exception, e:
+        LOG.warning("The message has not been acknowledged so we can retry it.")
+        LOG.warning(str(e))
+        raise

pulse_actions/handlers/treeherder_add_new_jobs.py

@@ -15,7 +15,18 @@
 MEMORY_SAVING_MODE = True
 
 
-def on_runnable_job_event(data, message, dry_run, treeherder_host, acknowledge):
+def ignored(data):
+    '''It determines if the job will be processed or not.'''
+    return False
+
+
+def on_event(data, message, dry_run, treeherder_host, acknowledge, **kwargs):
+    if ignored(data):
+        if acknowledge:
+            # We need to ack the message to remove it from our queue
+            message.ack()
+        return 0  # SUCCESS
+
     # Cleaning mozci caches
     buildjson.BUILDS_CACHE = {}
     query_jobs.JOBS_CACHE = {}
@@ -49,31 +60,6 @@ def on_runnable_job_event(data, message, dry_run, treeherder_host, acknowledge):
         # We want to see this in the alerts
         LOG.error("Notice that we're letting %s schedule jobs for %s." % (requester,
                                                                           treeherder_link))
-    '''
-    # Everyone can press the button, but only authorized users can trigger jobs
-    # TODO: remove this when proper LDAP identication is set up on TH
-    if not (requester.endswith('@mozilla.com') or author == requester or
-            whitelisted_users(requester)):
-
-        if acknowledge:
-            # Remove message from pulse queue
-            message.ack()
-
-        # We publish a message saying we will not trigger the job
-        pulse_message = {
-            'resultset_id': resultset_id,
-            'requester': requester,
-            'status': "Could not determine if the user is authorized, nothing was triggered."}
-        routing_key = '{}.{}'.format(repo_name, 'runnable')
-        try:
-            message_sender.publish_message(pulse_message, routing_key)
-        except:
-            LOG.warning("Failed to publish message over pulse stream.")
-
-        LOG.error("Requester %s is not allowed to trigger jobs on %s." %
-                  (requester, treeherder_link))
-        return  # Raising an exception adds too much noise
-    '''
 
     LOG.info("New jobs requested by %s for %s" % (requester, treeherder_link))
     LOG.info("List of builders:")
@@ -88,7 +74,7 @@ def on_runnable_job_event(data, message, dry_run, treeherder_host, acknowledge):
         if acknowledge:
             # We need to ack the message to remove it from our queue
             message.ack()
-        return
+        return -1  # FAILURE
 
     builders_graph, other_builders_to_schedule = buildbot_bridge.buildbot_graph_builder(
         builders=buildernames,
@@ -127,3 +113,5 @@ def on_runnable_job_event(data, message, dry_run, treeherder_host, acknowledge):
     if acknowledge:
         # We need to ack the message to remove it from our queue
         message.ack()
+
+    return 0  # SUCCESS

pulse_actions/handlers/treeherder_job_action.py

@@ -70,11 +70,25 @@
 LOG = logging.getLogger(__name__.split('.')[-1])
 
 
-def on_event(data, message, dry_run, treeherder_host, acknowledge):
+def ignored(data):
+    '''It determines if the job will be processed or not.'''
+    if data['action'].capitalize() == "Backfill":
+        return False
+    else:
+        return True
+
+
+def on_event(data, message, dry_run, treeherder_host, acknowledge, **kwargs):
     """Act upon Treeherder job events.
 
     Return if the outcome was successful or not
     """
+    if ignored(data):
+        if acknowledge:
+            # We need to ack the message to remove it from our queue
+            message.ack()
+        return 0  # SUCCESS
+
     # Cleaning mozci caches
     buildjson.BUILDS_CACHE = {}
     query_jobs.JOBS_CACHE = {}
@@ -116,26 +130,32 @@ def on_event(data, message, dry_run, treeherder_host, acknowledge):
 
     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]
+        LOG.info('Treeherder can send us invalid builder names.')
+        LOG.info('See https://bugzilla.mozilla.org/show_bug.cgi?id=1242038.')
+        LOG.warning('Builder %s is invalid.' % buildername[0])
+        return -1  # FAILURE
 
-    else:
+    elif action == "Backfill":
         # 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 acknowledge:
+            if not dry_run:
                 status = 'Backfill request sent'
             else:
-                status = 'Dry-run mode, nothing was backfilled'
+                status = 'Dry-run mode, nothing was backfilled.'
             LOG.debug(status)
 
+    else:
+        LOG.error(
+            'We were not aware of the "{}" action. Please file an issue'.format(action)
+        )
+        return -1  # FAILURE
+
     if acknowledge:
         # We need to ack the message to remove it from our queue
         message.ack()

pulse_actions/handlers/treeherder_push_action.py

@@ -9,7 +9,21 @@
 LOG = logging.getLogger(__name__)
 
 
-def on_resultset_action_event(data, message, dry_run, treeherder_host, acknowledge):
+def ignored(data):
+    '''Ite determines if the job will be processed or not.'''
+    # We do not handle 'cancel_all' action right now, so skip it.
+    if data['action'] == "cancel_all":
+        return True
+    else:
+        return False
+
+
+def on_event(data, message, dry_run, treeherder_host, acknowledge, **kwargs):
+    if ignored(data):
+        if acknowledge:
+            message.ack()
+        return 0  # SUCCESS
+
     # Cleaning mozci caches
     buildjson.BUILDS_CACHE = {}
     query_jobs.JOBS_CACHE = {}
@@ -21,11 +35,6 @@ def on_resultset_action_event(data, message, dry_run, treeherder_host, acknowled
 
     treeherder_client = TreeherderClient(host=treeherder_host)
 
-    # We do not handle 'cancel_all' action right now, so skip it.
-    if action == "cancel_all":
-        if acknowledge:
-            message.ack()
-        return
     LOG.info("%s action requested by %s on repo_name %s with resultset_id: %s" % (
         data['action'],
         data["requester"],
@@ -56,9 +65,15 @@ def on_resultset_action_event(data, message, dry_run, treeherder_host, acknowled
                      'lower then normal'.format(times)
         else:
             status = 'Dry-mode, no request sent'
+    else:
+        raise Exception(
+            'We were not aware of the "{}" action. Please address the code.'.format(action)
+        )
 
     LOG.debug(status)
 
     if acknowledge:
         # We need to ack the message to remove it from our queue
         message.ack()
+
+    return 0  # SUCCESS