[Findit] For automatic analyses of flaky tests, run the Swarming tasks off PST peak hours.

appengine/findit/waterfall/flake/recursive_flake_pipeline.py

 # Copyright 2016 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.
 
 import logging
+from datetime import timedelta
+import random
 
 from common import appengine_util
 from common import constants
 from common import time_util
 from common.pipeline_wrapper import BasePipeline
 
 from model import analysis_status
 from model.flake.flake_swarming_task import FlakeSwarmingTask
 from model.flake.master_flake_analysis import MasterFlakeAnalysis
 from waterfall import waterfall_config
 from waterfall.process_flake_swarming_task_result_pipeline import (
     ProcessFlakeSwarmingTaskResultPipeline)
 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()
 
 
+def _GetETAToStartAnalysis(manually_triggered):
+  """Returns an ETA as of a UTC datetime.datetime to start the analysis.
+
+  If not urgent, Swarming tasks should be run off PST peak hours from 11am to
+  6pm on workdays.
+
+  Args:
+    manually_triggered (bool): True if the analysis is from manual request, like
+        by a Chromium sheriff.
+
+  Returns:
+    The ETA as of a UTC datetime.datetime to start the analysis.
+  """
+  if manually_triggered:
+    # If the analysis is manually triggered, run it right away.
+    return time_util.GetUTCNow()
+
+  now_at_mtv = time_util.GetDatetimeInTimezone(
+      'US/Pacific', time_util.GetUTCNowWithTimezone())
+  if now_at_mtv.weekday() >= 5:  # PST Saturday or Sunday.
+    return time_util.GetUTCNow()
+
+  if now_at_mtv.hour < 11 or now_at_mtv.hour >= 18:  # Before 11am or after 6pm.
+    return time_util.GetUTCNow()
+
+  # Set ETA time to 6pm, and also with a random latency within 30 minutes to
+  # avoid sudden burst traffic to Swarming.
+  diff = timedelta(hours=18 - now_at_mtv.hour,
+                   minutes=-now_at_mtv.minute,
+                   seconds=-now_at_mtv.second + random.randint(0, 30 * 60),
+                   microseconds=-now_at_mtv.microsecond)
+  eta = now_at_mtv + diff
+
+  # Convert back to UTC.
+  return time_util.GetDatetimeInTimezone('UTC', eta)
+
+
 class RecursiveFlakePipeline(BasePipeline):
 
+  def __init__(self, *args, **kwargs):
+    super(RecursiveFlakePipeline, self).__init__(*args, **kwargs)
+    self.manually_triggered = kwargs.get('manually_triggered', False)
+
+  def StartOffPSTPeakHours(self, *args, **kwargs):
+    """Starts the pipeline off PST peak hours if not triggered manually."""
+    kwargs['eta'] = _GetETAToStartAnalysis(self.manually_triggered)
+    self.start(*args, **kwargs)
+
   # Arguments number differs from overridden method - pylint: disable=W0221
   def run(self, master_name, builder_name, run_build_number, step_name,
           test_name, version_number, master_build_number,
-          flakiness_algorithm_results_dict, queue_name=constants.DEFAULT_QUEUE):
+          flakiness_algorithm_results_dict, manually_triggered=False):
     """Pipeline to determine the regression range of a flaky test.
 
     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.
       version_number (int): The version to save analysis results and data to.
       master_build_number (int): The build number of the Master_Flake_analysis.
       flakiness_algorithm_results_dict (dict): A dictionary used by
         NextBuildNumberPipeline
-      queue_name (str): Which queue to run on.
+      manually_triggered (bool): True if the analysis is from manual request,
+          like by a Chromium sheriff.
 
     Returns:
       A dict of lists for reliable/flaky tests.
     """
     flake_analysis = MasterFlakeAnalysis.GetVersion(
         master_name, builder_name, master_build_number, step_name, test_name,
         version=version_number)
     logging.info(
         'Running RecursiveFlakePipeline on MasterFlakeAnalysis %s/%s/%s/%s/%s',
         master_name, builder_name, master_build_number, step_name, test_name)
     logging.info(
         'MasterFlakeAnalysis %s version %s', flake_analysis, version_number)
 
     if flake_analysis.status != analysis_status.RUNNING:  # pragma: no branch
       flake_analysis.status = analysis_status.RUNNING
       flake_analysis.put()
 
     # Call trigger pipeline (flake style).
     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, version_number)
     yield NextBuildNumberPipeline(
         master_name, builder_name, master_build_number, run_build_number,
-        step_name, test_name, version_number, test_result_future, queue_name,
-        flakiness_algorithm_results_dict)
+        step_name, test_name, version_number, test_result_future,
+        flakiness_algorithm_results_dict, manually_triggered=manually_triggered)
 
 
 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_flake_analysis.data_points[-1].pass_rate
   cur_run = min([d.build_number for d in master_flake_analysis.data_points])
   flake_settings = waterfall_config.GetCheckFlakeSettings()
   lower_flake_threshold = flake_settings.get('lower_flake_threshold')
@@ skipping 82 matching lines @@
   return (flakiness_algorithm_results_dict['lower_boundary'] +
           flakiness_algorithm_results_dict['sequential_run_index'])
 
 
 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,
           run_build_number, step_name, test_name, version_number,
-          test_result_future, queue_name, flakiness_algorithm_results_dict):
+          test_result_future, flakiness_algorithm_results_dict,
+          manually_triggered=False):
 
     # Get MasterFlakeAnalysis success list corresponding to parameters.
     master_flake_analysis = MasterFlakeAnalysis.GetVersion(
         master_name, builder_name, master_build_number, step_name, test_name,
         version=version_number)
     # 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:
@@ skipping 18 matching lines @@
 
     if next_run < flakiness_algorithm_results_dict['last_build_number']:
       next_run = 0
     elif next_run >= master_build_number:
       next_run = 0
 
     if next_run:
       pipeline_job = RecursiveFlakePipeline(
           master_name, builder_name, next_run, step_name, test_name,
           version_number, master_build_number,
-          flakiness_algorithm_results_dict=flakiness_algorithm_results_dict)
+          flakiness_algorithm_results_dict=flakiness_algorithm_results_dict,
+          manually_triggered=manually_triggered)
       # pylint: disable=W0201
       pipeline_job.target = appengine_util.GetTargetNameForModule(
           constants.WATERFALL_BACKEND)
-      pipeline_job.start(queue_name=queue_name)
+      pipeline_job.StartOffPSTPeakHours(
+          queue_name=self.queue_name or constants.DEFAULT_QUEUE)
     else:
       _UpdateAnalysisStatusUponCompletion(
           master_flake_analysis, analysis_status.COMPLETED, None)