[Findit] Adding spike detection for try job requests to data query script

appengine/findit/util_scripts/remote_queries/try_job_data_metrics.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.
 
 """Pulls historical try job metadata from Findit and prints a report."""
 
 import argparse
 from collections import defaultdict
 import datetime
 import json
 import numpy
 import os
 import sys
 
+try:
+  from matplotlib import pyplot
+except ImportError:
+  pyplot = None
+
 _REMOTE_API_DIR = os.path.join(os.path.dirname(__file__), os.path.pardir)
 sys.path.insert(1, _REMOTE_API_DIR)
 
 import remote_api
 
 from model.wf_try_job_data import WfTryJobData
 
 
 NOT_AVAILABLE = 'N/A'
 
@@ skipping 63 matching lines @@
 def _FormatSecondsAsHMS(seconds):
   """Formats the number of seconds into hours, minutes, seconds."""
   if seconds == NOT_AVAILABLE:
     return NOT_AVAILABLE
 
   minutes, seconds = divmod(seconds, 60)
   hours, minutes = divmod(minutes, 60)
   return '%d:%02d:%02d' % (hours, minutes, seconds)
 
 
+def _GetRequestSpikes(request_times, time_window_seconds=30*60,
+                      minimum_spike_size=3, show_plot=False):
+  """Calculates and plots try jobs by request time.
+
+  Args:
+    request_time: List of datetime objects representing try job request times.
+    time_window_seconds: Maximum number of seconds between requests to count
+      as a spike.
+    minimum_spike_size: Minimum number of requests within the specified time
+      window needed to count as a spike.
+    show_plot: Boolean whether to display visual graphs of the request times.
+
+  Returns:
+    spike_count: The number of spikes found.
+    average_spike_size: The average number of requests in each spike.
+    maximum_spike_size: The number of requests in the biggest spike.
+  """
+  request_times = sorted(request_times)
+
+  if show_plot:
+    if pyplot:
+      pyplot.plot(request_times, [i for i in range(len(request_times))], 'x')
+      pyplot.show()
+    else:
+      print ('In order to show plots, matplotlib needs to be installed. To '
+             'install, please run \'sudo pip install matplotlib\'')
+
+  candidate_spike_start = request_times[0]
+  points_in_spike = 1
+  spike_count = 0
+  spike_sizes = []
+
+  for point_being_examined in request_times[1:]:

[Sharu Jiang, 2016/07/18 22:15:31]

I have one concern, in worst case, this approach would only detect half of the
spike load, it may be a problem.

[lijeffrey, 2016/07/19 00:33:11]

Yes, with this vanilla approach it is possible some spikes are improperly
interpreted. For example in a 30 minute window at minute 0 we have a request,
then at 28 and 29 we have 2 more requests. This is a spike of size 3. But at 30,
31, 32, 33, there are 4 more requests. This algorithm will detect 2 spikes, one
of size 3 (0, 28, 29), and one of size 4 (30, 31, 32, 33), when the real spike
is actually 28-33 of size 6. But this is expected to be a corner case. Since
this code is to estimate how many bots are needed, we are only considering
average spike size over a large period (3 months) of many spikes to minimize
compile try jobs being queued up under theoretical expected load. 

+    if ((point_being_examined - candidate_spike_start).total_seconds() <
+        time_window_seconds):
+      points_in_spike += 1
+    else:
+      # The time window has passed. Need a new starting point.
+      if points_in_spike >= minimum_spike_size:
+        spike_count += 1
+        spike_sizes.append(points_in_spike)
+
+      candidate_spike_start = point_being_examined
+      points_in_spike = 1  # Start over.
+
+  return (spike_count, _GetAverageOfNumbersInList(spike_sizes),
+          max(spike_sizes) if spike_sizes else 0)
+
+
 def _GetReportInformation(try_job_data_list, start_date, end_date):
   """Computes and returns try job metadata.
 
   Args:
     try_job_data_list: A list of WfTryJobData entities.
     start_date: The earliest request date to compute data.
     end_date: The latest request date to compute data.
 
   Returns:
     A dict in the following format:
@@ skipping 47 matching lines @@
   under_five_minutes_rate = NOT_AVAILABLE
   under_fifteen_minutes_rate = NOT_AVAILABLE
   under_thirty_minutes_rate = NOT_AVAILABLE
   over_thirty_minutes_rate = NOT_AVAILABLE
 
   if try_job_data_list:
     try_jobs_per_day = (
         len(try_job_data_list) / float((end_date - start_date).days))
     regression_range_sizes = []
     execution_times_seconds = []
+    request_times = []
     in_queue_times = []
     end_to_end_times = []
     commits_analyzed = []
     culprits_detected = 0
     errors_detected = 0
     number_under_five_minutes = 0
     number_under_fifteen_minutes = 0
     number_under_thirty_minutes = 0
     number_over_thirty_minutes = 0
     total_number_of_try_jobs = len(try_job_data_list)
@@ skipping 11 matching lines @@
         execution_times_seconds.append(execution_time)
 
       # In-queue time.
       if try_job_data.start_time and try_job_data.request_time:
         in_queue_time_delta = (
             try_job_data.start_time - try_job_data.request_time)
         in_queue_time = in_queue_time_delta.total_seconds()
         in_queue_times.append(in_queue_time)
 
       # Total time end-to-end.
-      if try_job_data.request_time and try_job_data.end_time:
-        total_time_delta = try_job_data.end_time - try_job_data.start_time
-        total_time_seconds = total_time_delta.total_seconds()
-        end_to_end_times.append(total_time_seconds)
+      if try_job_data.request_time:
+        request_times.append(try_job_data.request_time)
 
-        if total_time_seconds < 300:  # Under 5 minutes.
-          number_under_five_minutes += 1
-        elif total_time_seconds < 900:  # Under 15 minutes.
-          number_under_fifteen_minutes += 1
-        elif total_time_seconds < 1800:  # Under 30 minutes.
-          number_under_thirty_minutes += 1
-        else:  # Over 30 minutes.
-          number_over_thirty_minutes += 1
+        if try_job_data.end_time:
+          total_time_delta = try_job_data.end_time - try_job_data.start_time
+          total_time_seconds = total_time_delta.total_seconds()
+          end_to_end_times.append(total_time_seconds)
+
+          if total_time_seconds < 300:  # Under 5 minutes.
+            number_under_five_minutes += 1
+          elif total_time_seconds < 900:  # Under 15 minutes.
+            number_under_fifteen_minutes += 1
+          elif total_time_seconds < 1800:  # Under 30 minutes.
+            number_under_thirty_minutes += 1
+          else:  # Over 30 minutes.
+            number_over_thirty_minutes += 1
 
       # Number of commits analyzed.
       if try_job_data.number_of_commits_analyzed:
         commits_analyzed.append(try_job_data.number_of_commits_analyzed)
 
       # Culprit detection rate.
       if try_job_data.culprits:
         culprits_detected += 1
 
       if try_job_data.error:
@@ skipping 39 matching lines @@
 
     under_five_minutes_rate = (
         float(number_under_five_minutes) / total_number_of_try_jobs)
     under_fifteen_minutes_rate = (
         float(number_under_fifteen_minutes) / total_number_of_try_jobs)
     under_thirty_minutes_rate = (
         float(number_under_thirty_minutes) / total_number_of_try_jobs)
     over_thirty_minutes_rate = (
         float(number_over_thirty_minutes) / total_number_of_try_jobs)
 
+  # Calculate try job spikes.
+  spike_count, average_spike_size, maximum_spike_size = _GetRequestSpikes(
+      request_times, time_window_seconds=30*60, minimum_spike_size=3,
+      show_plot=False)
+
   return {
       'try_jobs_per_day': _FormatDigits(try_jobs_per_day),
       'average_regression_range_size': _FormatDigits(
           average_regression_range_size),
       'median_regression_range_size': median_regression_range_size,
       'average_execution_time': _FormatSecondsAsHMS(_FormatDigits(
           average_execution_time)),
       'median_execution_time': _FormatSecondsAsHMS(_FormatDigits(
           median_execution_time)),
       'average_end_to_end_time': _FormatSecondsAsHMS(_FormatDigits(
           average_end_to_end_time)),
       'median_end_to_end_time': _FormatSecondsAsHMS(_FormatDigits(
           median_end_to_end_time)),
       'average_time_in_queue': _FormatSecondsAsHMS(
           _FormatDigits(average_time_in_queue)),
       'median_time_in_queue': _FormatSecondsAsHMS(_FormatDigits(
           median_time_in_queue)),
       'average_commits_analyzed': _FormatDigits(average_commits_analyzed),
       'median_commits_analyzed': median_commits_analyzed,
       'longest_execution_time': longest_execution_time,
       'shortest_execution_time': shortest_execution_time,
       'number_of_try_jobs': number_of_try_jobs,
       'detection_rate': _FormatDigits(detection_rate),
       'error_rate': _FormatDigits(error_rate),
       'time_per_revision': _FormatSecondsAsHMS(
           _FormatDigits(time_per_revision)),
       'under_five_minutes_rate': _FormatDigits(under_five_minutes_rate),
       'under_fifteen_minutes_rate': _FormatDigits(under_fifteen_minutes_rate),
       'under_thirty_minutes_rate': _FormatDigits(under_thirty_minutes_rate),
-      'over_thirty_minutes_rate': _FormatDigits(over_thirty_minutes_rate)
+      'over_thirty_minutes_rate': _FormatDigits(over_thirty_minutes_rate),
+      'request_spike_count': spike_count,
+      'request_spike_average_size': average_spike_size,
+      'request_spike_maximum_size': maximum_spike_size,
   }
 
 
 def PrintCommonStats(try_job_data_list, start_date, end_date, indent):
   """Takes a list of WfTryJobData entities and prints their stats."""
   spaces = ''
   for _ in range(indent):
     spaces += ' '
 
   report_info = _GetReportInformation(try_job_data_list, start_date, end_date)
@@ skipping 208 matching lines @@
     if args_dict[parsed_arg]:
       ordered_args.append(parsed_arg)
 
   return ordered_args
 
 
 if __name__ == '__main__':
   # Set up the Remote API to use services on the live App Engine.
   remote_api.EnableRemoteApi(app_id='findit-for-me')
 
-  START_DATE = datetime.datetime(2016, 5, 1)
-  END_DATE = datetime.datetime(2016, 6, 23)
+  START_DATE = datetime.datetime(2016, 4, 17)
+  END_DATE = datetime.datetime(2016, 7, 15)
 
   try_job_data_query = WfTryJobData.query(
       WfTryJobData.request_time >= START_DATE,
       WfTryJobData.request_time < END_DATE)
   categorized_data = try_job_data_query.fetch()
 
   args = GetArgsInOrder()
   for arg in args:
     categorized_data = SplitStructByOption(categorized_data, arg)
 
   # TODO(lijeffrey): Display data in an html page instead of printing.
   PrettyPrint(categorized_data, START_DATE, END_DATE)