[Findit] Added algorithm to analysis

appengine/findit/waterfall/flake/recursive_flake_pipeline.py

Index: appengine/findit/waterfall/flake/recursive_flake_pipeline.py
diff --git a/appengine/findit/waterfall/flake/recursive_flake_pipeline.py b/appengine/findit/waterfall/flake/recursive_flake_pipeline.py
index 332a54897f63632062efe28164a47b8612d01d08..3e6ce000aab4e473a8a613a840f0450d0d85a180 100644
--- a/appengine/findit/waterfall/flake/recursive_flake_pipeline.py
+++ b/appengine/findit/waterfall/flake/recursive_flake_pipeline.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.
 
+import copy
 from datetime import datetime
 
 from common import appengine_util
@@ -10,47 +11,99 @@ from common.pipeline_wrapper import BasePipeline
 
 from model import analysis_status
 from model.flake.master_flake_analysis import MasterFlakeAnalysis
+from model.flake.flake_swarming_task import FlakeSwarmingTask
 from waterfall.trigger_flake_swarming_task_pipeline import (
     TriggerFlakeSwarmingTaskPipeline)
 from waterfall.process_flake_swarming_task_result_pipeline import (
     ProcessFlakeSwarmingTaskResultPipeline)
 
+# TODO(lijeffrey): Move to config.
+LOWER_FLAKE_THRESHOLD = .02
+UPPER_FLAKE_THRESHOLD = .98
+MAX_FLAKE_IN_A_ROW = 4
+MAX_STABLE_IN_A_ROW = 4
+
 
 class RecursiveFlakePipeline(BasePipeline):
 
   # Arguments number differs from overridden method - pylint: disable=W0221
   def run(self, master_name, builder_name, run_build_number, step_name,
-          test_name, master_build_number, queue_name=constants.DEFAULT_QUEUE):
+          test_name, master_build_number, algo_dict,
+          queue_name=constants.DEFAULT_QUEUE):
+    """
+    Args:
+      master_name (str): The master name.
+      builder_name (str): The builder name.
+      run_build_number (int): The build number of the current swarming rerun.
+      step_name (str): The step name.
+      test_name (str): The test name.
+      master_build_number (int): The build number of the Master_Flake_analysis.
+      algo_dict (dict): A dictionary used by NextBuildNumberPipeline
+      queue_name (str): Which queue to run on.
+    Returns:
+      A dict of lists for reliable/flaky tests.
+    """
+
     # Call trigger pipeline (flake style).
-    task_id = yield TriggerFlakeSwarmingTaskPipeline(master_name, builder_name,
-                          run_build_number, step_name, [test_name])
+    task_id = yield TriggerFlakeSwarmingTaskPipeline(
+        master_name, builder_name, run_build_number, step_name, [test_name])
     # Pass the trigger pipeline into a process pipeline.
     test_result_future = yield ProcessFlakeSwarmingTaskResultPipeline(
         master_name, builder_name, run_build_number,
         step_name, task_id, master_build_number, test_name)
     yield NextBuildNumberPipeline(
-        master_name, builder_name, master_build_number,
-        step_name, test_name, test_result_future, queue_name)
+        master_name, builder_name, master_build_number, run_build_number,
+        step_name, test_name, test_result_future, queue_name, algo_dict)
+
 
 class NextBuildNumberPipeline(BasePipeline):
+
   # Arguments number differs from overridden method - pylint: disable=W0221
   # Unused argument - pylint: disable=W0613
-  def run(self, master_name, builder_name, master_build_number, step_name,
-          test_name, test_result_future, queue_name):
+  def run(self, master_name, builder_name, master_build_number,
+          run_build_number, step_name, test_name, test_result_future,
+          queue_name, algo_dict):
+
     # Get MasterFlakeAnalysis success list corresponding to parameters.
     master = MasterFlakeAnalysis.Get(master_name, builder_name,
                                      master_build_number, step_name, test_name)
+
+    # Don't call another pipeline if we fail.
+    flake_swarming_task = FlakeSwarmingTask.Get(

[lijeffrey, 2016/08/15 04:14:07]

nit: it looks like this can be moved to before getting master =
MasterFlakeAnalysis since master isn't needed till later

[caiw, 2016/08/17 08:36:18]

Done.

+        master_name, builder_name, run_build_number, step_name, test_name)
+    if flake_swarming_task.status == analysis_status.ERROR:
+      return
+
     # Figure out what build_number we should call, if any
     # This is a placeholder for testing:
-    next_run = False
-    if len(master.build_numbers) < 10:
-      # TODO(caiw): Develop algorithm to optimize this.
-      next_run = min(master.build_numbers) - 10
+
+    # Get the last result.
+    last_result = master.success_rates[-1]

[lijeffrey, 2016/08/15 04:14:08]

I would move this stuff into a new function that returns next_run and possibly
new_algo_dict too to make it more testable and this run method less long

[caiw, 2016/08/17 08:36:18]

Done.

+    if (last_result < LOWER_FLAKE_THRESHOLD or
+        last_result > UPPER_FLAKE_THRESHOLD):
+      algo_dict['stable_in_a_row'] += 1

[lijeffrey, 2016/08/15 04:14:07]

just an idea but it looks like algo_dict can be implemented as a class with a
ToDict() method that translates it. This is in case you have multiple algorithms
that may produce different results or capture different data we can compare
which is more effective in the future.

Just a thought though, no code change needed for now.

[caiw, 2016/08/17 08:36:18]

Acknowledged.

+      if algo_dict['stable_in_a_row'] > MAX_STABLE_IN_A_ROW:
+        algo_dict['stabled_out'] = True
+      algo_dict['flakes_in_a_row'] = 0
+      step_size = algo_dict['stable_in_a_row'] + 1
+    else:
+      algo_dict['flakes_in_a_row'] += 1
+      if algo_dict['flakes_in_a_row'] > MAX_FLAKE_IN_A_ROW:
+        algo_dict['flaked_out'] = True
+      algo_dict['stable_in_a_row'] = 0
+      step_size = algo_dict['flakes_in_a_row'] + 1
+    next_run = min(master.build_numbers) - step_size

[chanli, 2016/08/15 23:01:05]

Just a thought: Say after 10 flakes in a row, we start to get stables in a row.
that means there are 10 builds between last flake and first stable, right? Do we
need to run swarming runs for this 10 builds to further narrow down the actual
build where the culprit's at?

[caiw, 2016/08/17 08:36:18]

Yes Shuotao and I talked about this today.  I will add it.

+
+    if (next_run < algo_dict['last_build_number'] or
+        (algo_dict['stabled_out'] and algo_dict['flaked_out'])):
+      next_run = 0
+
     if next_run:
+      new_algo_dict = copy.deepcopy(algo_dict)
       pipeline_job = RecursiveFlakePipeline(
           master_name, builder_name, next_run, step_name, test_name,
-          master_build_number)
-      #pylint: disable=W0201
+          master_build_number, algo_dict=new_algo_dict)
+      # pylint: disable=W0201
       pipeline_job.target = appengine_util.GetTargetNameForModule(
           constants.WATERFALL_BACKEND)
       pipeline_job.start(queue_name=queue_name)