[Findit] Capture and store latest buildbucket response with each query

appengine/findit/waterfall/monitor_try_job_pipeline.py

 # Copyright 2015 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.
 
 from datetime import datetime
 import json
 import time
 
 from google.appengine.ext import ndb
 
@@ skipping 127 matching lines @@
   error_dict, error_code = _GetError(
       buildbucket_response, buildbucket_error, timed_out)
 
   if error_dict:
     try_job_data.error = error_dict
     try_job_data.error_code = error_code
 
   try_job_data.put()
 
 
+def _DictsAreEqual(dict_1, dict_2, exclude_keys=None):
+  if dict_1 == dict_2:
+    return True
+
+  if dict_1 is None or dict_2 is None:
+    return False
+
+  if exclude_keys is None:
+    exclude_keys = []
+
+  for key, value in dict_1.iteritems():
+    if key not in exclude_keys and (key not in dict_2 or dict_2[key] != value):
+      return False
+
+  for key, value in dict_2.iteritems():
+    if key not in exclude_keys and (key not in dict_1 or dict_1[key] != value):
+      return False
+
+  return True
+
+
+def _UpdateLastBuildbucketResponse(try_job_data, build):
+  if not build or not build.response:
+    return
+
+  if not _DictsAreEqual(try_job_data.last_buildbucket_response,
+                        build.response, exclude_keys=['utcnow_ts']):
+    try_job_data.last_buildbucket_response = build.response
+    try_job_data.put()
+
+
 class MonitorTryJobPipeline(BasePipeline):
   """A pipeline for monitoring a try job and recording results when it's done.
 
   The result will be stored to compile_results or test_results according to
   which type of build failure we are running try job for.
   """
 
   UNKNOWN = 'UNKNOWN'
 
   @ndb.transactional
@@ skipping 41 matching lines @@
     allowed_response_error_times = max_error_times
 
     # TODO(chanli): Make sure total wait time equals to timeout_hours
     # regardless of retries.
     deadline = time.time() + timeout_hours * 60 * 60
     try_job_data = WfTryJobData.Get(try_job_id)
     already_set_started = False
     start_time = None
     while True:
       error, build = buildbucket_client.GetTryJobs([try_job_id])[0]
+
       if error:
         if allowed_response_error_times > 0:
           allowed_response_error_times -= 1
           pipeline_wait_seconds += default_pipeline_wait_seconds
         else:  # pragma: no cover
           # Buildbucket has responded error more than 5 times, retry pipeline.
           _UpdateTryJobMetadata(try_job_data, start_time, build, error, False)
           raise pipeline.Retry(
               'Error "%s" occurred. Reason: "%s"' % (error.message,
                                                      error.reason))
@@ skipping 27 matching lines @@
 
           already_set_started = True
 
       if time.time() > deadline:  # pragma: no cover
         _UpdateTryJobMetadata(try_job_data, start_time, build, error, True)
         # Explicitly abort the whole pipeline.
         raise pipeline.Abort(
             'Try job %s timed out after %d hours.' % (
                 try_job_id, timeout_hours))
 
+      # Ensure last_buildbucket_response is always the most recent
+      # whenever available during intermediate queries.
+      _UpdateLastBuildbucketResponse(try_job_data, build)
+
       time.sleep(pipeline_wait_seconds)  # pragma: no cover