[Findit] Refactor Findit pipeline.

appengine/findit/waterfall/start_try_job_on_demand_pipeline.py

Index: appengine/findit/waterfall/start_try_job_on_demand_pipeline.py
diff --git a/appengine/findit/waterfall/start_try_job_on_demand_pipeline.py b/appengine/findit/waterfall/start_try_job_on_demand_pipeline.py
index 750000a2d4c06372fae8462ef683b30b9ed60160..d84cbb793222b46a9a673a18b783b3e9f3dd6c54 100644
--- a/appengine/findit/waterfall/start_try_job_on_demand_pipeline.py
+++ b/appengine/findit/waterfall/start_try_job_on_demand_pipeline.py
@@ -2,29 +2,122 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+import logging
+
 from common.pipeline_wrapper import BasePipeline
-from model.wf_analysis import WfAnalysis
+from common.waterfall import failure_type
 from waterfall import try_job_util
+from waterfall.identify_try_job_culprit_pipeline import (
+    IdentifyTryJobCulpritPipeline)
+from waterfall.monitor_try_job_pipeline import MonitorTryJobPipeline
+from waterfall.process_swarming_task_result_pipeline import (
+    ProcessSwarmingTaskResultPipeline)
+from waterfall.schedule_compile_try_job_pipeline import (
+    ScheduleCompileTryJobPipeline)
+from waterfall.schedule_test_try_job_pipeline import (
+    ScheduleTestTryJobPipeline)
+
+
+def _GetLastPassCompile(build_number, failed_steps):
+  if (failed_steps.get('compile') and
+      failed_steps['compile']['first_failure'] == build_number and
+      failed_steps['compile'].get('last_pass') is not None):
+    return failed_steps['compile']['last_pass']
+  return None
+
+
+def _GetLastPassTest(build_number, failed_steps):
+  for step_failure in failed_steps.itervalues():
+    for test_failure in step_failure['tests'].itervalues():
+      if (test_failure['first_failure'] == build_number and
+          test_failure.get('last_pass') is not None):
+        return test_failure['last_pass']
+  return None
+
+
+def _GetLastPass(build_number, failure_info, try_job_type):
+  if try_job_type == failure_type.COMPILE:
+    return _GetLastPassCompile(build_number, failure_info['failed_steps'])
+  elif try_job_type == failure_type.TEST:
+    return _GetLastPassTest(build_number, failure_info['failed_steps'])
+  else:
+    return None
+
+
+def _GetSuspectsFromHeuristicResult(heuristic_result):
+  if not heuristic_result:
+    return []
+
+  suspected_revisions = set()
+  for failure in heuristic_result.get('failures', []):
+    for cl in failure['suspected_cls']:
+      suspected_revisions.add(cl['revision'])
+  return list(suspected_revisions)
 
 
 class StartTryJobOnDemandPipeline(BasePipeline):
 
   # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self, failure_info, signals, build_completed, force_try_job,
-          heuristic_result):
+  def run(self, master_name, builder_name, build_number, failure_info,
+          signals, heuristic_result, build_completed, force_try_job):
     """Starts a try job if one is needed for the given failure."""
+
     if not build_completed:  # Only start try-jobs for completed builds.
-      return False
-
-    failure_result_map = try_job_util.ScheduleTryJobIfNeeded(
-        failure_info, signals=signals, heuristic_result=heuristic_result,
-        force_try_job=force_try_job)
-
-    # Save reference to the try-jobs if any was scheduled.
-    master_name = failure_info['master_name']
-    builder_name = failure_info['builder_name']
-    build_number = failure_info['build_number']
-    analysis = WfAnalysis.Get(master_name, builder_name, build_number)
-    analysis.failure_result_map = failure_result_map
-    analysis.put()
-    return True
+      return
+
+    need_try_job = try_job_util.NeedANewTryJob(
+        master_name, builder_name, build_number, failure_info, signals,
+        heuristic_result, force_try_job)
+
+    if not need_try_job:
+      return
+
+    try_job_type = failure_info['failure_type']
+    last_pass = _GetLastPass(build_number, failure_info, try_job_type)
+    if last_pass is None:  # pragma: no cover
+      logging.warning('Couldn"t start try job because last_pass is not found.')

[lijeffrey, 2016/08/09 04:07:34]

is this a normal scenario that last pass is unavailable? If it's not supposed to
happen, should we change this to logging.error and include master_name,
builder_name, build_number?

[chanli, 2016/08/09 17:08:43]

There are some cases:
1. If a failure has happened on more than 20 consecutive builds
2. If the builder is newly added 
3. There is something wrong

I'm not sure if we should escalate this to error. But you're write we should add
build info to the message.

+      return
+
+    blame_list = failure_info['builds'][str(build_number)]['blame_list']
+    good_revision = failure_info['builds'][str(last_pass)]['chromium_revision']
+    bad_revision = failure_info['builds'][str(build_number)][
+        'chromium_revision']
+    suspected_revisions = _GetSuspectsFromHeuristicResult(heuristic_result)
+
+    if try_job_type == failure_type.COMPILE:
+      compile_targets = try_job_util.GetFailedTargetsFromSignals(
+          signals, master_name, builder_name)
+      try_job_id = yield ScheduleCompileTryJobPipeline(
+          master_name, builder_name, build_number, good_revision, bad_revision,
+          try_job_type, compile_targets, suspected_revisions)
+      try_job_result = yield MonitorTryJobPipeline(
+          master_name, builder_name, build_number, try_job_type, try_job_id)
+      yield IdentifyTryJobCulpritPipeline(
+          master_name, builder_name, build_number, blame_list, try_job_type,
+          try_job_id, try_job_result)
+    else:
+      # If try_job_type is other type, the pipeline has returned.
+      # So here the try_job_type is failure_type.TEST.
+
+      # Waits and gets the swarming tasks' results.
+      reliable_tests = []
+      for step_name, step_failure in failure_info['failed_steps'].iteritems():
+        step_has_first_time_failure = False
+        for test_failure in step_failure['tests'].itervalues():

[lijeffrey, 2016/08/09 04:07:34]

I would break this part into a separate helper method to determine whether or
not a first time failure can be found

def _HasFirstTimeFailure(tests, build_number):
  ...


step_has_first_time_failure = _HasFirstTimeFailure(step_failure['tests'],
build_number)

This would make the code cleaner and more testable

[chanli, 2016/08/09 17:08:42]

Done.

+          if test_failure['first_failure'] == build_number:
+            step_has_first_time_failure = True
+            break
+        if not step_has_first_time_failure:
+          continue
+        task_result = yield ProcessSwarmingTaskResultPipeline(
+            master_name, builder_name, build_number, step_name)
+        reliable_tests.append(task_result)
+
+      try_job_id = yield ScheduleTestTryJobPipeline(
+          master_name, builder_name, build_number, good_revision, bad_revision,
+          try_job_type, suspected_revisions, *reliable_tests)
+      try_job_result = yield MonitorTryJobPipeline(
+          master_name, builder_name, build_number, try_job_type, try_job_id)
+      yield IdentifyTryJobCulpritPipeline(
+          master_name, builder_name, build_number, blame_list, try_job_type,
+          try_job_id, try_job_result)