[Findit] Refactor Findit pipeline.

appengine/findit/waterfall/try_job_util.py

Index: appengine/findit/waterfall/try_job_util.py
diff --git a/appengine/findit/waterfall/try_job_util.py b/appengine/findit/waterfall/try_job_util.py
index eaa4209bf663a6f600d080d825e16ffdbfdb6733..fc02e99750e014fe3bd0a30e070142ccdd5112af 100644
--- a/appengine/findit/waterfall/try_job_util.py
+++ b/appengine/findit/waterfall/try_job_util.py
@@ -2,6 +2,7 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+from collections import defaultdict
 from datetime import datetime
 import logging
 
@@ -15,70 +16,14 @@ from model.wf_analysis import WfAnalysis
 from model.wf_build import WfBuild
 from model.wf_failure_group import WfFailureGroup
 from model.wf_try_job import WfTryJob
-from waterfall import swarming_tasks_to_try_job_pipeline
 from waterfall import waterfall_config
-from waterfall.try_job_type import TryJobType
 
 
-def _CheckFailureForTryJobKey(
-    master_name, builder_name, build_number,
-    failure_result_map, failed_step_or_test, failure):
-  """Compares the current_failure and first_failure for each failed_step/test.
-
-  If equal, a new try_job needs to start;
-  If not, apply the key of the first_failure's try_job to this failure.
-  """
-  # TODO(chanli): Need to compare failures across builders
-  # after the grouping of failures is implemented.
-  # TODO(chanli): Need to handle cases where first failure is actually
-  # more than 20 builds back. The implementation should not be here,
-  # but need to be taken care of.
-  if not failure.get('last_pass'):
-    # Bail out since cannot figure out the good_revision.
-    return False, None
-
-  if failure['current_failure'] == failure['first_failure']:
-    failure_result_map[failed_step_or_test] = '%s/%s/%s' % (
-        master_name, builder_name, build_number)
-    return True, failure['last_pass']  # A new try_job is needed.
-  else:
-    failure_result_map[failed_step_or_test] = '%s/%s/%s' % (
-        master_name, builder_name, failure['first_failure'])
-    return False, None
-
-
-def _CheckIfNeedNewTryJobForTestFailure(
-    failure_level, master_name, builder_name, build_number,
-    failure_result_map, failures):
-  """Traverses failed steps or tests to check if a new try job is needed."""
-  need_new_try_job = False
-  last_pass = build_number
-  targeted_tests = {} if failure_level == 'step' else []
-
-  for failure_name, failure in failures.iteritems():
-    if 'tests' in failure:
-      failure_result_map[failure_name] = {}
-      failure_targeted_tests, failure_need_try_job, failure_last_pass = (
-          _CheckIfNeedNewTryJobForTestFailure(
-              'test', master_name, builder_name, build_number,
-              failure_result_map[failure_name], failure['tests']))
-      if failure_need_try_job:
-        targeted_tests[failure_name] = failure_targeted_tests
-    else:
-      failure_need_try_job, failure_last_pass = _CheckFailureForTryJobKey(
-          master_name, builder_name, build_number,
-          failure_result_map, failure_name, failure)
-      if failure_need_try_job:
-        if failure_level == 'step':
-          targeted_tests[failure_name] = []
-        else:
-          targeted_tests.append(failure.get('base_test_name', failure_name))
-
-    need_new_try_job = need_new_try_job or failure_need_try_job
-    last_pass = (failure_last_pass if failure_last_pass and
-                 failure_last_pass < last_pass else last_pass)
+def _ShouldBailOutForOutdatedBuild(build):
+  return (datetime.utcnow() - build.start_time).days > 0
 
-  return targeted_tests, need_new_try_job, last_pass
+def _CurrentBuildKey(master_name, builder_name, build_number):
+  return '%s/%s/%d' % (master_name, builder_name, build_number)
 
 
 def _BlameListsIntersection(blame_list_1, blame_list_2):
@@ -298,43 +243,81 @@ def ReviveOrCreateTryJobEntity(
   return try_job_entity_revived_or_created
 
 
-def _NeedANewTryJob(
-    master_name, builder_name, build_number, build_failure_type, failed_steps,
-    failure_result_map, builds, signals, heuristic_result, force_try_job=False):
-  """Checks if a new try_job is needed."""
-  need_new_try_job = False
-  last_pass = build_number
+def _NeedANewTryJobForCompile(

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

nit: how about _NeedANewCompileTryJob/_NeedANewTestTryJob?

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

Done.

+    master_name, builder_name, build_number, failure_info):
 
-  if 'compile' in failed_steps:
-    try_job_type = TryJobType.COMPILE
-    targeted_tests = None
-    need_new_try_job, last_pass = _CheckFailureForTryJobKey(
-        master_name, builder_name, build_number,
-        failure_result_map, TryJobType.COMPILE, failed_steps['compile'])
-  else:
-    try_job_type = TryJobType.TEST
-    targeted_tests, need_new_try_job, last_pass = (
-        _CheckIfNeedNewTryJobForTestFailure(
-            'step', master_name, builder_name, build_number, failure_result_map,
-            failed_steps))
+  compile_failure = failure_info['failed_steps'].get('compile', {})
+  if compile_failure:
+    analysis = WfAnalysis.Get(master_name, builder_name, build_number)
+    analysis.failure_result_map['compile'] = '%s/%s/%d' % (
+        master_name, builder_name, compile_failure['first_failure'])
+    analysis.put()
+
+    if compile_failure['first_failure'] == compile_failure['current_failure']:
+      return True
+
+  return False
+
+def _NeedANewTryJobForTest(
+    master_name, builder_name, build_number, failure_info, force_try_job):
+
+  if failure_info['failure_type'] != failure_type.TEST:
+    return False
+
+  if not force_try_job:
+    if waterfall_config.ShouldSkipTestTryJobs(master_name, builder_name):

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

nit: It looks like these if statements can be combined into an and

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

Done.

+      logging.info('Test try jobs on %s, %s are not supported yet.',
+                   master_name, builder_name)
+      return False
+
+  analysis = WfAnalysis.Get(master_name, builder_name, build_number)
+  failure_result_map = analysis.failure_result_map

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

nit: break this part of the code into a separate function like
_CurrentBuildKeyInFailureResultMap(analysis) (or master_name, builder_name,
build_number) or something to make the functions easier to test

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

Done.

+  current_build_key = _CurrentBuildKey(master_name, builder_name, build_number)
+  for step_keys in failure_result_map.itervalues():
+    for test_key in step_keys.itervalues():
+      if test_key == current_build_key:
+        return True
+  return False
+
+def NeedANewTryJob(
+    master_name, builder_name, build_number, failure_info, signals,
+    heuristic_result, force_try_job=False):
+
+  tryserver_mastername, tryserver_buildername = (
+      waterfall_config.GetTrybotForWaterfallBuilder(master_name, builder_name))
 
+  try_job_type = failure_info['failure_type']
+  if not tryserver_mastername or not tryserver_buildername:
+    logging.info('%s, %s is not supported yet.', master_name, builder_name)
+    return False
+
+  if not force_try_job:
+    build = WfBuild.Get(master_name, builder_name, build_number)
+
+    if _ShouldBailOutForOutdatedBuild(build):
+      logging.error('Build time %s is more than 24 hours old. '
+                    'Try job will not be triggered.' % build.start_time)
+      return False
 
-  need_new_try_job = (
-      need_new_try_job and ReviveOrCreateTryJobEntity(
-          master_name, builder_name, build_number, force_try_job))
+  need_new_try_job = (_NeedANewTryJobForCompile(
+      master_name, builder_name, build_number, failure_info)
+      if try_job_type == failure_type.COMPILE else
+      _NeedANewTryJobForTest(
+          master_name, builder_name, build_number, failure_info, force_try_job))
 
-  # TODO(josiahk): Integrate _IsBuildFailureUniqueAcrossPlatforms() into
-  # need_new_try_job boolean
   if need_new_try_job:
+    # TODO(josiahk): Integrate this into need_new_try_job boolean
     _IsBuildFailureUniqueAcrossPlatforms(
-        master_name, builder_name, build_number, build_failure_type,
-        builds[str(build_number)]['blame_list'], failed_steps, signals,
-        heuristic_result)
+        master_name, builder_name, build_number, try_job_type,
+        failure_info['builds'][str(build_number)]['blame_list'],
+        failure_info['failed_steps'], signals, heuristic_result)
 
-  return need_new_try_job, last_pass, try_job_type, targeted_tests
+  need_new_try_job = need_new_try_job and ReviveOrCreateTryJobEntity(
+      master_name, builder_name, build_number, force_try_job)
+  return need_new_try_job
 
 
-def _GetFailedTargetsFromSignals(signals, master_name, builder_name):
+def GetFailedTargetsFromSignals(signals, master_name, builder_name):
   compile_targets = []
 
   if not signals or 'compile' not in signals:
@@ -354,90 +337,3 @@ def _GetFailedTargetsFromSignals(signals, master_name, builder_name):
       compile_targets.append(source_target.get('target'))
 
   return compile_targets
-
-
-def _GetSuspectsFromHeuristicResult(heuristic_result):
-  suspected_revisions = set()
-  if not heuristic_result:
-    return list(suspected_revisions)
-  for failure in heuristic_result.get('failures', []):
-    for cl in failure['suspected_cls']:
-      suspected_revisions.add(cl['revision'])
-  return list(suspected_revisions)
-
-
-def _ShouldBailOutForOutdatedBuild(build):
-  return (datetime.utcnow() - build.start_time).days > 0
-
-
-def ScheduleTryJobIfNeeded(failure_info, signals, heuristic_result,
-                           force_try_job=False):
-  master_name = failure_info['master_name']
-  builder_name = failure_info['builder_name']
-  build_number = failure_info['build_number']
-  failed_steps = failure_info.get('failed_steps', [])
-  builds = failure_info.get('builds', {})
-
-  tryserver_mastername, tryserver_buildername = (
-      waterfall_config.GetTrybotForWaterfallBuilder(master_name, builder_name))
-
-  if not tryserver_mastername or not tryserver_buildername:
-    logging.info('%s, %s is not supported yet.', master_name, builder_name)
-    return {}
-
-  if not force_try_job:
-    build = WfBuild.Get(master_name, builder_name, build_number)
-
-    if _ShouldBailOutForOutdatedBuild(build):
-      logging.error('Build time %s is more than 24 hours old. '
-                    'Try job will not be triggered.' % build.start_time)
-      return {}
-
-    if (failure_info['failure_type'] == failure_type.TEST and
-        waterfall_config.ShouldSkipTestTryJobs(master_name, builder_name)):
-      logging.info('Test try jobs on %s, %s are not supported yet.',
-                   master_name, builder_name)
-      return {}
-
-  failure_result_map = {}
-  need_new_try_job, last_pass, try_job_type, targeted_tests = (
-      _NeedANewTryJob(master_name, builder_name, build_number,
-                      failure_info['failure_type'], failed_steps,
-                      failure_result_map, builds, signals, heuristic_result,
-                      force_try_job))
-
-  if need_new_try_job:
-    compile_targets = (_GetFailedTargetsFromSignals(
-        signals, master_name, builder_name)
-                       if try_job_type == TryJobType.COMPILE else None)
-    suspected_revisions = _GetSuspectsFromHeuristicResult(heuristic_result)
-
-    pipeline = (
-        swarming_tasks_to_try_job_pipeline.SwarmingTasksToTryJobPipeline(
-            master_name, builder_name, build_number,
-            builds[str(last_pass)]['chromium_revision'],
-            builds[str(build_number)]['chromium_revision'],
-            builds[str(build_number)]['blame_list'],
-            try_job_type, compile_targets, targeted_tests, suspected_revisions,
-            force_try_job))
-
-    pipeline.target = appengine_util.GetTargetNameForModule(
-        constants.WATERFALL_BACKEND)
-    pipeline.start(queue_name=constants.WATERFALL_TRY_JOB_QUEUE)
-
-    if try_job_type == TryJobType.TEST:  # pragma: no cover
-      logging_str = (
-          'Trying to schedule swarming task(s) for build %s, %s, %s: %s'
-          ' because of %s failure. A try job may be triggered if some reliable'
-          ' failure is detected in task(s).') % (
-              master_name, builder_name, build_number,
-              pipeline.pipeline_status_path, try_job_type)
-    else:  # pragma: no cover
-      logging_str = (
-          'Try job was scheduled for build %s, %s, %s: %s because of %s '
-          'failure.') % (
-              master_name, builder_name, build_number,
-              pipeline.pipeline_status_path, try_job_type)
-    logging.info(logging_str)
-
-  return failure_result_map