Moving ScheduleNewAnalysis to break the cycle

appengine/findit/handlers/crash/crash_handler.py

 # Copyright 2016 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import base64
 import json
 import logging
 
 from common import constants
+from common import appengine_util
 from common.base_handler import BaseHandler
 from common.base_handler import Permission
-from crash.crash_pipeline import FinditForClientID
+from crash import crash_pipeline
+from crash.crash_pipeline import CrashWrapperPipeline
 from crash.crash_report import CrashReport
 
 
 class CrashHandler(BaseHandler):
   PERMISSION_LEVEL = Permission.ANYONE
 
   def HandlePost(self):
     """Handles push delivery from Pub/Sub for crash data.
 
     The crash data should be in the following json format:
@@ skipping 70 matching lines @@
     }
     """
     try:
       received_message = json.loads(self.request.body)
       pubsub_message = received_message['message']
       crash_data = json.loads(base64.b64decode(pubsub_message['data']))
 
       logging.info('Processing message %s from subscription %s.',
                    pubsub_message['message_id'],
                    received_message['subscription'])
+      logging.info('Crash data is %s', json.dumps(crash_data))
+      ScheduleNewAnalysis(crash_data)
 
-      logging.info('Crash data is %s', json.dumps(crash_data))
-
-      client_id = crash_data['client_id']
-      FinditForClientID(client_id).ScheduleNewAnalysis(crash_data,
-          queue_name=constants.CRASH_ANALYSIS_QUEUE[client_id])
     except (KeyError, ValueError):  # pragma: no cover.
       # TODO: save exception in datastore and create a page to show them.
       logging.exception('Failed to process crash message')
       logging.info(self.request.body)
+
+
+# TODO(http://crbug.com/659345): try to break the cycle re pipeline_cls.

[stgao, 2016/11/02 01:28:27]

This seems invalid now.

[wrengr, 2016/11/02 23:52:26]

Done.

+# TODO(http://crbug.com/659346): we don't cover anything after the
+# call to _NeedsNewAnalysis.
+def ScheduleNewAnalysis(crash_data):
+  """Create a pipeline object to perform the analysis, and start it.

[stgao, 2016/11/02 01:28:27]

nit: Create -> Creates.
This is the convention in the infra repo, although we might still have some
violations in the code.

[wrengr, 2016/11/02 23:52:26]

Done.

+
+  If we can detect that the analysis doesn't need to be performed
+  (e.g., it was already performed, or the ``crash_data`` is empty so
+  there's nothig we can do), then we will skip creating the pipeline
+  at all.
+
+  This method is only ever called from ``handlers.crash.crash_handler``
+  and so should surely be moved there instead of living here.

[stgao, 2016/11/02 01:28:27]

invalid now.

[wrengr, 2016/11/02 23:52:26]

Done.

+
+  Args:
+    crash_data (JSON): ??
+
+  Returns:
+    True if we started a new pipeline; False otherwise.
+  """
+  client_id = crash_data['client_id']
+  # N.B., must call FinditForClientID indirectly, for mock testing.
+  findit_client = crash_pipeline.FinditForClientID(client_id)
+
+  # Check policy and modify the crash_data as needed.
+  crash_data = findit_client.CheckPolicy(crash_data)
+  if crash_data is None:
+    return False
+
+  # Detect the regression range, and decide if we actually need to
+  # run a new anlaysis or not.
+  if not findit_client._NeedsNewAnalysis(crash_data):
+    return False
+
+  crash_identifiers = crash_data['crash_identifiers']
+  # N.B., we cannot pass ``self`` directly to the _pipeline_cls, because
+  # it is not JSON-serializable (and there's no way to make it such,
+  # since JSON-serializability is defined by JSON-encoders rather than
+  # as methods on the objects being encoded).
+  pipeline = CrashWrapperPipeline(client_id, crash_identifiers)
+  # Attribute defined outside __init__ - pylint: disable=W0201
+  pipeline.target = appengine_util.GetTargetNameForModule(
+      constants.CRASH_BACKEND[client_id])
+  queue_name = constants.CRASH_ANALYSIS_QUEUE[client_id]
+  pipeline.start(queue_name=queue_name)
+  logging.info('New %s analysis is scheduled for %s', client_id,
+               repr(crash_identifiers))
+  return True