[Findit] Add three sub-pipelines to analyze build failure.

appengine/findit/waterfall/build_failure_analysis_pipelines.py

 # Copyright 2014 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.
 
 from datetime import datetime
 import logging
 
 from google.appengine.ext import ndb
 
 from model.build_analysis import BuildAnalysis
 from model.build_analysis_status import BuildAnalysisStatus
 from waterfall.base_pipeline import BasePipeline
 from waterfall.detect_first_failure_pipeline import DetectFirstFailurePipeline
+from waterfall.extract_signal_pipeline import ExtractSignalPipeline
+from waterfall.identify_culprit_pipeline import IdentifyCulpritPipeline
+from waterfall.pull_changelog_pipeline import PullChangelogPipeline
 
 
 class BuildFailurePipeline(BasePipeline):
 
   def __init__(self, master_name, builder_name, build_number):
     super(BuildFailurePipeline, self).__init__(
         master_name, builder_name, build_number)
     self.master_name = master_name
     self.builder_name = builder_name
     self.build_number = build_number
 
   def finalized(self):
-    analysis = BuildAnalysis.GetBuildAnalysis(
-        self.master_name, self.builder_name, self.build_number)
+    # When this root pipeline or its sub-pipelines still run into any error
+    # after auto-retries, this root pipeline will be aborted. So, mark the
+    # analysis as ERROR. The analysis is created before the pipeline starts.
     if self.was_aborted:  # pragma: no cover
-      analysis.status = BuildAnalysisStatus.ERROR
-    else:
-      analysis.status = BuildAnalysisStatus.ANALYZED
-    analysis.put()
+      analysis = BuildAnalysis.GetBuildAnalysis(
+          self.master_name, self.builder_name, self.build_number)
+      if analysis:  # In case the analysis is deleted manually.
+        analysis.status = BuildAnalysisStatus.ERROR
+        analysis.put()
 
   # Arguments number differs from overridden method - pylint: disable=W0221
   def run(self, master_name, builder_name, build_number):
     analysis = BuildAnalysis.GetBuildAnalysis(
         master_name, builder_name, build_number)
     analysis.pipeline_url = self.pipeline_status_url()
     analysis.status = BuildAnalysisStatus.ANALYZING
     analysis.start_time = datetime.utcnow()
     analysis.put()
 
-    yield DetectFirstFailurePipeline(master_name, builder_name, build_number)
+    # The yield statements below return PipelineFutures, which allow subsequent
+    # pipelines to refer to previous output values.
+    # https://github.com/GoogleCloudPlatform/appengine-pipelines/wiki/Python
+    failure_info = yield DetectFirstFailurePipeline(
+        master_name, builder_name, build_number)
+    change_logs = yield PullChangelogPipeline(failure_info)
+    signals = yield ExtractSignalPipeline(failure_info)
+    yield IdentifyCulpritPipeline(failure_info, change_logs, signals)
 
 
 @ndb.transactional
 def NeedANewAnalysis(master_name, builder_name, build_number, force):
   """Checks status of analysis for the build and decides if a new one is needed.
 
   A BuildAnalysis entity for the given build will be created if none exists.
 
   Returns:
     True if an analysis is needed, otherwise False.
@@ skipping 28 matching lines @@
     pipeline_job = BuildFailurePipeline(master_name, builder_name, build_number)
     pipeline_job.start(queue_name=queue_name)
 
     logging.info('An analysis triggered on build %s, %s, %s: %s',
                  master_name, builder_name, build_number,
                  pipeline_job.pipeline_status_url())
   else:  # pragma: no cover
     logging.info('Analysis was already triggered or the result is recent.')
 
   return BuildAnalysis.GetBuildAnalysis(master_name, builder_name, build_number)