Refactor pulse_actions (#74)

It makes pulse_actions a bit easier to comprehend by removing a lot of boiler plate.
It also gives the ability of replaying messages.

.gitignore

@@ -1,11 +1,8 @@
 *.pyc
 *.egg-info/*
-*.swp
-*.txt
 # Packages
 dist
-.idea/
-venv/
 .tox/
 .cache/
 .coverage
+requirements.txt

Procfile

@@ -1,2 +1 @@
-worker1: python pulse_actions/worker.py --topic-base backfilling --dry-run
-worker2: python pulse_actions/worker.py --topic-base resultset_actions,manual_backfill,runnable --debug
+worker2: python pulse_actions/worker.py --config-file configs/production_config.json --dry-run False

README.rst

@@ -44,6 +44,12 @@ From Pypi::
 
     pip install pulse-actions
 
+Replaying sample data
+=====================
+You can have re-process sample data from real requests to re-test any new changes:
+
+    python pulse_actions/worker.py --replay-file data/sample_queue.json
+
 Running
 =======
 
@@ -75,13 +81,6 @@ Adding more functionality
 Pulse Actions can be expanded to add more functionality. Here_ is a step-by-step guide for creating a "Hello World" client with pulse_actions.
 
 
-Requirements
-------------
-
-* mozci
-* mozillapulse
-
-
 .. _Pulse: https://wiki.mozilla.org/Auto-tools/Projects/Pulse
 .. _1180732: https://bugzilla.mozilla.org/show_bug.cgi?id=1180732
 .. _wiki: https://wiki.mozilla.org/Auto-tools/Projects/Pulse_actions

configs/production_config.json

@@ -0,0 +1,18 @@
+{
+    "applabel": "['resultset_actions', 'manual_backfill', 'runnable']",
+    "durable": "true",
+    "sources": {
+        "resultset_actions": {
+            "exchange": "exchange/treeherder/v1/resultset-actions",
+            "topic": "#.#"
+        },
+        "manual_backfill": {
+            "exchange": "exchange/treeherder/v1/job-actions",
+            "topic": "buildbot.#.backfill"
+        },
+        "runnable": {
+            "exchange": "exchange/treeherder/v1/resultset-runnable-job-actions",
+            "topic": "#"
+        }
+    }
+}

data/failing_requests.json

@@ -0,0 +1 @@
+{u'project': u'try', u'buildernames': [u'Android 4.3 armv7 API 15+ try opt test crashtest-1', u'Android 4.3 armv7 API 15+ try opt test crashtest-2', u'Android 4.3 armv7 API 15+ try debug test crashtest-1', u'Android 4.3 armv7 API 15+ try debug test crashtest-2', u'Android 4.3 armv7 API 15+ try debug test crashtest-3', u'Android 4.3 armv7 API 15+ try debug test crashtest-4'], u'version': 1, u'requester': u'cmanchester@mozilla.com', u'resultset_id': u'87913'}

pulse_actions/handlers/treeherder_buildbot.py

@@ -8,7 +8,6 @@
 """
 import logging
 
-from pulse_actions.publisher import MessageHandler
 from pulse_actions.utils.misc import filter_invalid_builders
 
 from mozci import query_jobs
@@ -82,19 +81,7 @@ def on_buildbot_event(data, message, dry_run, stage=False):
             status = 'Backfill request sent'
         else:
             status = 'Dry-run mode, nothing was backfilled'
-
-    # Send a pulse message showing what we did
-    message_sender = MessageHandler()
-    pulse_message = {
-        'job_id': job_id,
-        'action': action,
-        'requester': data['requester'],
-        'status': status}
-    routing_key = '{}.{}'.format(repo_name, action)
-    try:
-        message_sender.publish_message(pulse_message, routing_key)
-    except:
-        LOG.warning("Failed to publish message over pulse stream.")
+        LOG.debug(status)
 
     if not dry_run:
         # We need to ack the message to remove it from our queue

pulse_actions/handlers/treeherder_resultset.py

@@ -5,7 +5,6 @@
 from mozci.ci_manager import BuildAPIManager
 from mozci.sources import buildjson
 from thclient import TreeherderClient
-from pulse_actions.publisher import MessageHandler
 
 LOG = logging.getLogger(__name__)
 
@@ -54,6 +53,7 @@ def on_resultset_action_event(data, message, dry_run, stage=False):
             status = 'trigger_missing_jobs request sent'
         else:
             status = 'Dry-mode, no request sent'
+
     elif action == "trigger_all_talos_jobs":
         trigger_all_talos_jobs(
             repo_name=repo_name,
@@ -68,18 +68,7 @@ def on_resultset_action_event(data, message, dry_run, stage=False):
         else:
             status = 'Dry-mode, no request sent'
 
-    # Send a pulse message showing what we did
-    message_sender = MessageHandler()
-    pulse_message = {
-        'resultset_id': resultset_id,
-        'action': action,
-        'requester': data['requester'],
-        'status': status}
-    routing_key = '{}.{}'.format(repo_name, action)
-    try:
-        message_sender.publish_message(pulse_message, routing_key)
-    except:
-        LOG.warning("Failed to publish message over pulse stream.")
+    LOG.debug(status)
 
     if not dry_run:
         # We need to ack the message to remove it from our queue

pulse_actions/handlers/treeherder_runnable.py

@@ -1,6 +1,5 @@
 import logging
 
-from pulse_actions.publisher import MessageHandler
 from pulse_actions.utils.misc import whitelisted_users, filter_invalid_builders
 
 from mozci import query_jobs
@@ -41,11 +40,9 @@ def on_runnable_job_event(data, message, dry_run, stage):
     resultset = treeherder_client.get_resultsets(repo_name, id=resultset_id)[0]
     revision = resultset["revision"]
     author = resultset["author"]
-    status = None
 
     treeherder_link = TREEHERDER % {'repo': repo_name, 'revision': resultset['revision']}
 
-    message_sender = MessageHandler()
     if not (requester.endswith('@mozilla.com') or author == requester or
             whitelisted_users(requester)):
         # We want to see this in the alerts
@@ -126,19 +123,6 @@ def on_runnable_job_event(data, message, dry_run, stage):
     else:
         LOG.info("We don't have anything to schedule through Buildapi")
 
-    # Send a pulse message showing what we did
-    message_sender = MessageHandler()
-    pulse_message = {
-        'resultset_id': resultset_id,
-        'graph': builders_graph,
-        'requester': requester,
-        'status': status}
-    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.")
-
     if not dry_run:
         # We need to ack the message to remove it from our queue
         message.ack()

pulse_actions/utils/log_util.py

@@ -0,0 +1,41 @@
+import logging
+
+LOG = None
+
+
+def setup_logging(logging_level):
+    global LOG
+    if LOG:
+        return LOG
+    # Let's use the root logger
+    LOG = logging.getLogger()
+    LOG.setLevel(logging.DEBUG)
+
+    format = '%(asctime)s %(name)s\t %(levelname)s:\t %(message)s'
+    formatter = logging.Formatter(format, datefmt='%H:%M:%S')
+    # Handler 1 - Store all INFO messages in a specific file
+    # This logging.txt file will *only* show messages for that same day as Heroku
+    # dynos restart every day - Use papertrail for more details.
+    fh = logging.FileHandler('logging.txt')
+    fh.setLevel(logging.INFO)
+    fh.setFormatter(formatter)
+    LOG.addHandler(fh)
+
+    # Handler 2 - Store all DEBUG messages in a specific file
+    fh = logging.FileHandler('logging_debug.txt')
+    fh.setLevel(logging.DEBUG)
+    fh.setFormatter(formatter)
+    LOG.addHandler(fh)
+
+    # Handler 3 - Output to console (this is the output for Papertrail)
+    console = logging.StreamHandler()
+    console.setLevel(logging_level)
+    formatter = logging.Formatter('%(name)s\t %(levelname)s:\t %(message)s')
+    console.setFormatter(formatter)
+    LOG.addHandler(console)
+
+    LOG.info("Console output logs %s level messages." % logging.getLevelName(logging_level))
+
+    # Reduce logging for other noisy modules
+    logging.getLogger("requests").setLevel(logging.WARNING)
+    logging.getLogger("amqp").setLevel(logging.WARNING)