[Findit] Implement retry within swarming_util.py when making server calls

appengine/findit/waterfall/swarming_util.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.
 
 import base64
 from collections import defaultdict
 import json
 import logging
+import time
 import urllib
 import zlib
 
 from google.appengine.api.urlfetch_errors import DeadlineExceededError
 from google.appengine.api.urlfetch_errors import DownloadError
 from google.appengine.api.urlfetch_errors import ConnectionClosedError
 from google.appengine.ext import ndb
 
 from common import auth_util
 from model.wf_step import WfStep
@@ skipping 39 matching lines @@
 TASK_FAILED = 2
 
 # Swarming task exit code descriptions.
 EXIT_CODE_DESCRIPTIONS = {
     ALL_TESTS_PASSED: 'All tests passed',
     SOME_TESTS_FAILED: 'Some tests failed',
     TASK_FAILED: 'Swarming task failed',
 }
 
 
+def _GetBackoffSeconds(retry_backoff, tries, maximum_retry_interval):
+  """Returns how many seconds to wait before next retry.
+
+  Params:
+    retry_backoff (int): The base backoff in seconds.
+    tries (int): Indicates how many tries have been done.
+    maximum_retry_interval (int): The upper limit in seconds of how long to wait
+      between retries.
+  """
+  return min(retry_backoff * (2 ** (tries - 1)), maximum_retry_interval)
+
+
 def _SendRequestToServer(url, http_client, post_data=None):
-  """Sends GET/POST request to arbitrary url and returns response content."""
+  """Sends GET/POST request to arbitrary url and returns response content.
+
+  Because the Swarming and Isolated servers that _SendRequestToServer tries to
+  contact are prone to outages, exceptions trying to reach them may occur thus
+  this method should retry. We want to monitor and document these occurrences
+  even if the request eventually succeeds after retrying, with the last error
+  encountered being the one that is reported.
+
+  Args:
+    url (str): The url to send the request to.
+    http_client (HttpClient): The httpclient object with which to make the
+      server calls.
+    post_data (dict): Data/params to send with the request, if any.
+    swarming_task (WfSwarmingTask, FlakeSwarmingTask): An optional swarming
+      task with which to capture errors.
+
+  Returns:
+    content (dict), error (dict): The content from the server and the last error
+    encountered trying to retrieve it.
+  """
   headers = {'Authorization': 'Bearer ' + auth_util.GetAuthToken()}
+  swarming_settings = waterfall_config.GetSwarmingSettings()
+  should_retry = swarming_settings.get('should_retry_server')

[chanli, 2016/11/23 23:45:32]

I have a question: How long is the server down typically? How do we decide the
time we wait and retry for a task?

[lijeffrey, 2016/11/28 23:55:24]

It's difficult to say just from the logs, which using data from October 2016
doesn't seem to be available anymore :(. Originally when I was seeing these
errors we only get 1 at a time, since there is no retry logic that 1 flake
analysis that happened to have been requested at that 1 time would see it, then
it would be a period of several hours until the next request came in which may
or may not end in the same error.

I plan to hook up an upcoming CL to use ts_mon for each retry to log the errors
encountered, so it should be possible to detect a continuous outage and we can
use that to adjust our timeouts accordingly.

[chanli, 2016/11/29 18:49:38]

Acknowledged.

+  timeout_seconds = swarming_settings.get('server_retry_timeout') * 60 * 60
+  maximum_retry_interval = swarming_settings.get(
+      'maximum_server_contact_retry_interval_seconds')
+  deadline = time.time() + timeout_seconds
+  retry_backoff = 60
+  tries = 1
   error = None
 
-  try:
-    if post_data:
-      post_data = json.dumps(post_data, sort_keys=True, separators=(',', ':'))
-      headers['Content-Type'] = 'application/json; charset=UTF-8'
-      headers['Content-Length'] = len(post_data)
-      status_code, content = http_client.Post(url, post_data, headers=headers)
+  while True:
+    try:
+      if post_data:
+        post_data = json.dumps(post_data, sort_keys=True, separators=(',', ':'))
+        headers['Content-Type'] = 'application/json; charset=UTF-8'
+        headers['Content-Length'] = len(post_data)

[chanli, 2016/11/30 18:53:29]

Nit: This part can be done only once outside of the loop.

[lijeffrey, 2016/11/30 20:12:11]

Done.

+        status_code, content = http_client.Post(url, post_data, headers=headers)
+      else:
+        status_code, content = http_client.Get(url, headers=headers)
+    except ConnectionClosedError as e:
+      error = {
+          'code': URLFETCH_CONNECTION_CLOSED_ERROR,
+          'message': e.message
+      }
+    except DeadlineExceededError as e:
+      error = {
+          'code': URLFETCH_DEADLINE_EXCEEDED_ERROR,
+          'message': e.message
+      }
+    except DownloadError as e:
+      error = {
+          'code': URLFETCH_DOWNLOAD_ERROR,
+          'message': e.message
+      }
+    except Exception as e:  # pragma: no cover
+      error = {
+          'code': UNKNOWN,
+          'message': e.message
+      }
+
+    if error or status_code != 200:
+      # The retry upon 50x (501 excluded) is automatically handled in the
+      # underlying http_client.
+      # By default, it retries 5 times with exponential backoff.
+      error = error or {
+          'code': EXCEEDED_MAX_RETRIES_ERROR,
+          'message': 'Max retries exceeded trying to reach %s' % url
+      }
+      logging.error(error['message'])
     else:
-      status_code, content = http_client.Get(url, headers=headers)
-  except ConnectionClosedError as e:
-    error = {
-        'code': URLFETCH_CONNECTION_CLOSED_ERROR,
-        'message': e.message
-    }
-  except DeadlineExceededError as e:
-    error = {
-        'code': URLFETCH_DEADLINE_EXCEEDED_ERROR,
-        'message': e.message
-    }
-  except DownloadError as e:
-    error = {
-        'code': URLFETCH_DOWNLOAD_ERROR,
-        'message': e.message
-    }
-  except Exception as e:  # pragma: no cover
-    error = {
-        'code': UNKNOWN,
-        'message': e.message
-    }
-    # Still raise an exception here for cases not encountered before in order
-    # to see what went wrong in the logs.
-    raise e
+      # Even if the call is successful, still return the last error encountered.
+      return content, error
 
-  if error or status_code != 200:
-    # The retry upon 50x (501 excluded) is automatically handled in the
-    # underlying http_client.
-    # By default, it retries 5 times with exponential backoff.
-    error = error or {
-        'code': EXCEEDED_MAX_RETRIES_ERROR,
-        'message': 'Max retries exceeded trying to reach %s' % url
-    }
-    logging.error(error['message'])
-    return None, error
+    if should_retry and time.time() < deadline:  # pragma: no cover
+      # Wait, then retry if applicable.
+      wait_time = _GetBackoffSeconds(
+          retry_backoff, tries, maximum_retry_interval)
+      logging.info('Retrying connection to %s in %d seconds', url, wait_time)
+      time.sleep(wait_time)
+      tries += 1
+    else:
+      if should_retry:

[chanli, 2016/11/23 23:45:32]

Shouldn't you also handle case when should_retry == False and should break the
loop?

[lijeffrey, 2016/11/28 23:55:24]

If I'm understanding your comment correctly, the should_retry == False too will
break the loop. We only want to include more information in the error dict that
Findit eventually gave up on trying after retrying for a period of time.

+        # Indicate in the error that the retry timeout was reached.
+        error['retry_timeout'] = True
+      break
 
-  return content, None
+  logging.error('Failed to get an adequate response from %s. No data could be '
+                'retrieved', url)
+  return None, error
 
 
 def GetSwarmingTaskRequest(task_id, http_client):
   """Returns an instance of SwarmingTaskRequest representing the given task."""
   swarming_server_host = waterfall_config.GetSwarmingSettings().get(
       'server_host')
   url = ('https://%s/_ah/api/swarming/v1/task/%s/request') % (
       swarming_server_host, task_id)
   content, error = _SendRequestToServer(url, http_client)
 
@@ skipping 20 matching lines @@
   request.priority = max(100, swarming_settings.get('default_request_priority'))
   request.expiration_secs = request_expiration_hours * 60 * 60
 
   request.tags.extend(['findit:1', 'project:Chromium', 'purpose:post-commit'])
 
   url = 'https://%s/_ah/api/swarming/v1/tasks/new' % swarming_settings.get(
       'server_host')
   response_data, error = _SendRequestToServer(
       url, http_client, request.Serialize())
 
-  # TODO(lijeffrey): Handle error in calling functions.
   if not error:
     return json.loads(response_data)['task_id'], None
 
   return None, error
 
 
 def ListSwarmingTasksDataByTags(
     master_name, builder_name, build_number, http_client,
     additional_tag_filters=None):
   """Downloads tasks data from swarming server.
@@ skipping 303 matching lines @@
   for isolated_data in list_isolated_data:
     output_json, _ = _DownloadTestResults(isolated_data, http_client)
     if not output_json:
       # TODO(lijeffrey): Report/handle error returned from _DownloadTestResults.
       return None
     shard_results.append(output_json)
 
   if len(list_isolated_data) == 1:
     return shard_results[0]
   return _MergeSwarmingTestShards(shard_results)