[Findit] Capture versionized metadata for master_flake_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 4b1e0efa45d8f7963ca97d3ab60f5f20b76d4129..4264fd7159ce7d53b2cbd60ffb97bba3460153b2 100644
--- a/appengine/findit/waterfall/flake/recursive_flake_pipeline.py
+++ b/appengine/findit/waterfall/flake/recursive_flake_pipeline.py
@@ -4,8 +4,10 @@
 
 from common import appengine_util
 from common import constants
+from common import time_util
 from common.pipeline_wrapper import BasePipeline
 
+

[chanli, 2016/09/27 22:15:24]

Romove this line?

[lijeffrey, 2016/09/28 03:12:31]

Done.

 from model import analysis_status
 from model.flake.flake_swarming_task import FlakeSwarmingTask
 from model.flake.master_flake_analysis import MasterFlakeAnalysis
@@ -16,6 +18,16 @@ from waterfall.trigger_flake_swarming_task_pipeline import (
     TriggerFlakeSwarmingTaskPipeline)
 
 
+def _UpdateAnalysisStatusUponCompletion(master_flake_analysis, status, error):
+  master_flake_analysis.completed_time = time_util.GetUTCNow()
+  master_flake_analysis.status = status
+
+  if error:
+    master_flake_analysis.error = error
+
+  master_flake_analysis.put()
+
+
 class RecursiveFlakePipeline(BasePipeline):
 
   # Arguments number differs from overridden method - pylint: disable=W0221
@@ -38,11 +50,12 @@ class RecursiveFlakePipeline(BasePipeline):
     Returns:
       A dict of lists for reliable/flaky tests.
     """
-    master = MasterFlakeAnalysis.Get(master_name, builder_name,
-                                     master_build_number, step_name, test_name)
-    if master.status != analysis_status.RUNNING:  # pragma: no branch
-      master.status = analysis_status.RUNNING
-      master.put()
+    analysis = MasterFlakeAnalysis.GetVersion(

[chanli, 2016/09/27 22:15:24]

Nit: maybe a longer name so we don't confuse it with WfAnalysis?

[lijeffrey, 2016/09/28 03:12:31]

Unfortunately it seems the #pragma: no branch doesn't work if this is on 2 lines
so the longest we can make it is flake_analysis

+        master_name, builder_name, master_build_number, step_name, test_name)
+
+    if analysis.status != analysis_status.RUNNING:  # pragma: no branch
+      analysis.status = analysis_status.RUNNING
+      analysis.put()
 
     # Call trigger pipeline (flake style).
     task_id = yield TriggerFlakeSwarmingTaskPipeline(
@@ -57,32 +70,32 @@ class RecursiveFlakePipeline(BasePipeline):
         flakiness_algorithm_results_dict)
 
 
-def get_next_run(master, flakiness_algorithm_results_dict):
+def get_next_run(master_flake_analysis, flakiness_algorithm_results_dict):
   # A description of this algorithm can be found at:
   # https://docs.google.com/document/d/1wPYFZ5OT998Yn7O8wGDOhgfcQ98mknoX13AesJaS6ig/edit
   # Get the last result.
-  last_result = master.success_rates[-1]
-  cur_run = min(master.build_numbers)
+  last_result = master_flake_analysis.pass_rates[-1]
+  cur_run = min(master_flake_analysis.build_numbers)
   flake_settings = waterfall_config.GetCheckFlakeSettings()
   lower_flake_threshold = flake_settings.get('lower_flake_threshold')
   upper_flake_threshold = flake_settings.get('upper_flake_threshold')
   max_stable_in_a_row = flake_settings.get('max_stable_in_a_row')
   max_flake_in_a_row = flake_settings.get('max_flake_in_a_row')
 
-  if last_result < 0: # Test doesn't exist in the current build number.
+  if last_result < 0:  # Test doesn't exist in the current build number.
     flakiness_algorithm_results_dict['stable_in_a_row'] += 1
     flakiness_algorithm_results_dict['stabled_out'] = True
     flakiness_algorithm_results_dict['flaked_out'] = True
     flakiness_algorithm_results_dict['lower_boundary_result'] = 'STABLE'
 
-    lower_boundary = master.build_numbers[
+    lower_boundary = master_flake_analysis.build_numbers[
         -flakiness_algorithm_results_dict['stable_in_a_row']]
 
     flakiness_algorithm_results_dict['lower_boundary'] = lower_boundary
     flakiness_algorithm_results_dict['sequential_run_index'] += 1
     return lower_boundary + 1
   elif (last_result < lower_flake_threshold or
-      last_result > upper_flake_threshold):  # Stable result.
+        last_result > upper_flake_threshold):  # Stable result.
     flakiness_algorithm_results_dict['stable_in_a_row'] += 1
     if (flakiness_algorithm_results_dict['stable_in_a_row'] >
         max_stable_in_a_row):  # Identified a stable region.
@@ -127,8 +140,9 @@ def get_next_run(master, flakiness_algorithm_results_dict):
     return cur_run - step_size
 
 
-def sequential_next_run(master, flakiness_algorithm_results_dict):
-  last_result = master.success_rates[-1]
+def sequential_next_run(
+    master_flake_analysis, flakiness_algorithm_results_dict):
+  last_result = master_flake_analysis.pass_rates[-1]
   last_result_status = 'FLAKE'
   flake_settings = waterfall_config.GetCheckFlakeSettings()
   lower_flake_threshold = flake_settings.get('lower_flake_threshold')
@@ -140,10 +154,10 @@ def sequential_next_run(master, flakiness_algorithm_results_dict):
   if flakiness_algorithm_results_dict['sequential_run_index'] > 0:
     if (last_result_status !=
         flakiness_algorithm_results_dict['lower_boundary_result']):
-      master.suspected_flake_build_number = (
+      master_flake_analysis.suspected_flake_build_number = (
           flakiness_algorithm_results_dict['lower_boundary'] +
           flakiness_algorithm_results_dict['sequential_run_index'])
-      master.put()
+      master_flake_analysis.put()
       return 0
   flakiness_algorithm_results_dict['sequential_run_index'] += 1
   return (flakiness_algorithm_results_dict['lower_boundary'] +
@@ -159,23 +173,31 @@ class NextBuildNumberPipeline(BasePipeline):
           queue_name, flakiness_algorithm_results_dict):
 
     # Get MasterFlakeAnalysis success list corresponding to parameters.
-    master = MasterFlakeAnalysis.Get(master_name, builder_name,
-                                     master_build_number, step_name, test_name)
+    master_flake_analysis = MasterFlakeAnalysis.GetVersion(
+        master_name, builder_name, master_build_number, step_name, test_name)
     # Don't call another pipeline if we fail.
     flake_swarming_task = FlakeSwarmingTask.Get(
         master_name, builder_name, run_build_number, step_name, test_name)
 
     if flake_swarming_task.status == analysis_status.ERROR:
-      master.status = analysis_status.ERROR
-      master.put()
+      # TODO(lijeffrey): Implement more detailed error detection and reporting,
+      # such as timeouts, dead bots, etc.
+      error = {
+          'error': 'An error occurred',
+          'message': 'Unknown'

[stgao, 2016/09/28 00:03:24]

Can we say "Swarming task failed" for now?

[lijeffrey, 2016/09/28 03:12:31]

Done.

+      }
+      _UpdateAnalysisStatusUponCompletion(
+          master_flake_analysis, analysis_status.ERROR, error)
       return
 
     # Figure out what build_number we should call, if any
     if (flakiness_algorithm_results_dict['stabled_out'] and
         flakiness_algorithm_results_dict['flaked_out']):
-      next_run = sequential_next_run(master, flakiness_algorithm_results_dict)
+      next_run = sequential_next_run(
+          master_flake_analysis, flakiness_algorithm_results_dict)
     else:
-      next_run = get_next_run(master, flakiness_algorithm_results_dict)
+      next_run = get_next_run(
+          master_flake_analysis, flakiness_algorithm_results_dict)
 
     if next_run < flakiness_algorithm_results_dict['last_build_number']:
       next_run = 0
@@ -192,5 +214,5 @@ class NextBuildNumberPipeline(BasePipeline):
           constants.WATERFALL_BACKEND)
       pipeline_job.start(queue_name=queue_name)
     else:
-      master.status = analysis_status.COMPLETED
-      master.put()
+      _UpdateAnalysisStatusUponCompletion(
+          master_flake_analysis, analysis_status.COMPLETED, None)