[Findit] Integrate with Fracas through Pub/Sub.

appengine/findit/crash/fracas_crash_pipeline.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 datetime
+import json
+import logging
+
+from google.appengine.ext import ndb
+
+from common import appengine_util
+from common import constants
+from common import pubsub_util
+from crash import fracas
+from model import analysis_status
+from model.crash.crash_config import CrashConfig
+from model.crash.fracas_crash_analysis import FracasCrashAnalysis
+from pipeline_wrapper import BasePipeline
+from pipeline_wrapper import pipeline
+
+
+class FracasCrashBasePipeline(BasePipeline):
+  def __init__(self, channel, platform, signature):
+    super(FracasCrashBasePipeline, self).__init__(
+        channel, platform, signature)
+    self.channel = channel
+    self.platform = platform
+    self.signature = signature
+
+  def run(self, *args, **kwargs):
+    raise NotImplementedError()
+
+
+class AnalyzeCrashPipeline(FracasCrashBasePipeline):
+  def _SetErrorWhenAborted(self, aborted):

[Martin Barbella, 2016/04/08 18:10:24]

I don't think this necessarily needs to be in its own function, but as-is the
name is a bit confusing. When seems to imply a callback, so maybe IfAborted
would be clearer. It also doesn't seem like the aborted argument is necessary.

Also, minor nit, just return if not aborted.

[stgao, 2016/04/11 20:32:19]

This function is necessary for unittest purpose with 100% code coverage. It's
more of a workaround. But we can't find a good way to simulate an abortion in
appengine pipeline.
Done.

+    if aborted:
+      logging.error('Aborted analysis for %s, %s, %s',
+                    self.channel, self.platform, self.signature)
+      analysis = FracasCrashAnalysis.Get(
+          self.channel, self.platform, self.signature)
+      analysis.status = analysis_status.ERROR
+      analysis.put()
+
+  def finalized(self):
+    self._SetErrorWhenAborted(self.was_aborted)
+
+  # Arguments number differs from overridden method - pylint: disable=W0221
+  def run(self, channel, platform, signature):
+    analysis = FracasCrashAnalysis.Get(channel, platform, signature)
+
+    # Update analysis status.
+    analysis.pipeline_status_path = self.pipeline_status_path()
+    analysis.status = analysis_status.RUNNING
+    analysis.started_time = datetime.datetime.utcnow()
+    analysis.findit_version = appengine_util.GetCurrentVersion()
+    analysis.put()
+
+    # Run the analysis.
+    result, tags = fracas.FindCulpritForChromeCrash(
+        channel, platform, signature, analysis.stack_trace,
+        analysis.crashed_version, analysis.cpms)
+
+    # Update analysis status and save the analysis result.
+    analysis.completed_time = datetime.datetime.utcnow()
+    analysis.result = result
+    for tag_name, tag_value in tags.iteritems():
+      # Later, we might consider adding arbitrary tags.

[Martin Barbella, 2016/04/08 18:10:24]

This seems ideal. At the very least, this should be a TODO.

[stgao, 2016/04/11 20:32:19]

Maybe let's remove for now. The intention was to avoid adding too many fields to
the NDB model.

[Martin Barbella, 2016/04/11 21:46:31]

There are certainly ways to do this without adding new fields to the model,
though.

[stgao, 2016/04/12 18:19:17]

Yes, there should be. But I'd like to defer the implementation for that until we
need much more arbitrary tags for query and monitoring. Leave a TODO here.

+      if hasattr(analysis, tag_name):
+        setattr(analysis, tag_name, tag_value)
+    analysis.status = analysis_status.COMPLETED
+    analysis.put()
+
+
+class PublishResultPipeline(FracasCrashBasePipeline):
+  def finalized(self):
+    if self.was_aborted:  # pragma: no cover.
+      logging.error('Failed to publish analysis result for %s, %s, %s',
+                    self.channel, self.platform, self.signature)
+
+  # Arguments number differs from overridden method - pylint: disable=W0221
+  def run(self, channel, platform, signature):
+    analysis = FracasCrashAnalysis.Get(channel, platform, signature)

[Sharu, 2016/04/12 00:23:13]

Should we set the self.channel, self.platform, self.signature somewhere?

[stgao, 2016/04/12 18:27:35]

self.channel, self.platform, self.signature are set in the constructor. And
channel,platform,signature will always be passed over to the function ``run`` in
a pipeline.

+    result = {
+        'channel': channel,
+        'platform': platform,
+        'signature': signature,
+        'result': analysis.result,
+    }
+    messages_data = [json.dumps(result, sort_keys=True)]
+
+    crash_config = CrashConfig.Get()
+    topic = crash_config.fracas['analysis_result_pubsub_topic']
+    pubsub_util.PublishMessagesToTopic(messages_data, topic)
+    logging.info('Published analysis result for %s, %s, %s',
+                 channel, platform, signature)
+
+
+class FracasCrashWrapperPipeline(BasePipeline):
+  # Arguments number differs from overridden method - pylint: disable=W0221
+  def run(self, channel, platform, signature):
+    run_analysis = yield AnalyzeCrashPipeline(channel, platform, signature)
+    with pipeline.After(run_analysis):
+      yield PublishResultPipeline(channel, platform, signature)
+
+
+@ndb.transactional
+def _NeedANewAnalysis(

[Martin Barbella, 2016/04/08 18:10:24]

Nit: rename to _NeedsNewAnalysis.

[stgao, 2016/04/11 20:32:19]

Done.

+    channel, platform, signature, stack_trace, chrome_version, cpms):

[Martin Barbella, 2016/04/08 18:10:24]

Same comment about cpms from before.

[stgao, 2016/04/11 20:32:19]

Done.

+  analysis = FracasCrashAnalysis.Get(channel, platform, signature)
+  if analysis and not analysis.failed:
+    # A new analysis is not needed if last one didn't complete or succeeded.
+    # TODO(http://crbug.com/600535): re-analyze if stack trace or regression
+    # range changed.
+    return False
+
+  if not analysis:
+    # A new analysis is needed if there is no analysis yet.
+    analysis = FracasCrashAnalysis.Create(channel, platform, signature)
+
+  analysis.Reset()
+  analysis.crashed_version = chrome_version
+  analysis.stack_trace = stack_trace
+  analysis.cpms = cpms
+  analysis.status = analysis_status.PENDING
+  analysis.requested_time = datetime.datetime.utcnow()
+  analysis.put()
+  return True
+
+
+def ScheduleNewAnalysisForCrash(
+    channel, platform, signature, stack_trace, chrome_version, cpms,

[Martin Barbella, 2016/04/08 18:10:24]

Ditto.

[stgao, 2016/04/11 20:32:19]

Done.

+    queue_name=constants.DEFAULT_QUEUE):
+  """Schedules an analysis."""
+  crash_config = CrashConfig.Get()
+  if platform not in crash_config.fracas.get(
+      'supported_platform_list_by_channel', {}).get(channel, []):
+    # Bail out if either the channel or platform is not supported yet.
+    return False
+
+  if _NeedANewAnalysis(
+      channel, platform, signature, stack_trace, chrome_version, cpms):
+    analysis_pipeline = FracasCrashWrapperPipeline(channel, platform, signature)
+    analysis_pipeline.target = appengine_util.GetTargetNameForModule(
+        constants.CRASH_BACKEND_FRACAS)
+    analysis_pipeline.start(queue_name=queue_name)
+    logging.info('New analysis is scheduled for %s, %s, %s',
+                 channel, platform, signature)
+    return True
+
+  return False