[Findit] Implementing swarming task error detection

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 urllib
 import zlib
 
+from google.appengine.api.urlfetch_errors import (
+    DeadlineExceededError, DownloadError, ConnectionClosedError)

[chanli, 2016/11/11 00:05:13]

Nit: according to
https://engdoc.corp.google.com/eng/doc/devguide/py/styleguide.shtml?cl=head#s...

Imports should be on separate lines.

[lijeffrey, 2016/11/11 20:55:41]

Done.

 from google.appengine.ext import ndb
 
 from common import auth_util
 from model.wf_step import WfStep
 from waterfall import waterfall_config
 from waterfall.swarming_task_request import SwarmingTaskRequest
 
 
 STATES_RUNNING = ('RUNNING', 'PENDING')
 STATE_COMPLETED = 'COMPLETED'
 STATES_NOT_RUNNING = (
     'EXPIRED', 'TIMED_OUT', 'BOT_DIED', 'CANCELED', 'COMPLETED')
 
+# General error codes.
+UNKNOWN = 0

[chanli, 2016/11/11 00:05:13]

Based on process_base_swarming_task_result_pipeline.py, this value is being used
as exit_code as well?

[lijeffrey, 2016/11/11 20:55:41]

Yes that probably is confusing so i made UNKNOWN its own thing separate from
exit codes

+

[chanli, 2016/11/11 00:05:13]

In my opinion, since they are all error codes so the 2 empty lines could be
removed?

[lijeffrey, 2016/11/11 20:55:41]

Done.

+
+# Urlfetch error codes.
+URLFETCH_DOWNLOAD_ERROR = 100
+URLFETCH_DEADLINE_EXCEEDED_ERROR = 110
+URLFETCH_CONNECTION_CLOSED_ERROR = 120
+EXCEEDED_MAX_RETRIES_ERROR = 210

[chanli, 2016/11/11 00:05:13]

Maybe we can use enum for error codes?

[lijeffrey, 2016/11/11 20:55:41]

I don't think there's a python enum?

+
+
+# Swarming task exit codes.

[chanli, 2016/11/11 00:05:13]

Should we use enum here?

[chanli, 2016/11/11 00:05:13]

I was confused then I realize you mixed swarming states and exit codes and then
used them as error codes. 

Actually TIMED_OUT, EXPIRED, BOT_DIED and CANCELED don't belong to here. 

And in the rest of exit_codes, only TASK_FAILED should be reported as error.

My suggestion is adding new error codes for the failing states and TASK_FAILED.

[lijeffrey, 2016/11/11 20:55:41]

Moved to a separate dict

 ALL_TESTS_PASSED = 0
 SOME_TESTS_FAILED = 1
 TASK_FAILED = 2
+TIMED_OUT = 3
+EXPIRED = 4
+BOT_DIED = 5
+CANCELED = 6
 
+
+# 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'
+    UNKNOWN: 'Unknown',
+    ALL_TESTS_PASSED: 'All tests passed',
+    SOME_TESTS_FAILED: 'Some tests failed',
+    TASK_FAILED: 'Swarming task failed',
+    TIMED_OUT: 'Swarming task timed out',
+    EXPIRED: 'Swarming task expired',
+    BOT_DIED: 'Swarming task bot died',
+    CANCELED: 'Swarming task was canceled',
 }
 
 
 def _SendRequestToServer(url, http_client, post_data=None):
   """Sends GET/POST request to arbitrary url and returns response content."""
   headers = {'Authorization': 'Bearer ' + auth_util.GetAuthToken()}
-  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)
-  else:
-    status_code, content = http_client.Get(url, headers=headers)
+  error = None
 
-  if status_code != 200:
+  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)
+    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
+
+  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.
-    return None
-  return content
+    error = error or {
+        'code': EXCEEDED_MAX_RETRIES_ERROR,
+        'message': 'Max retries exceeded trying to reach %s' % url
+    }
+    logging.error(error['message'])
+    return None, error
+
+  return content, None
 
 
 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)
-  json_data = json.loads(_SendRequestToServer(url, http_client))
-  return SwarmingTaskRequest.Deserialize(json_data)
+  content, error = _SendRequestToServer(url, http_client)
+
+  # TODO(lijeffrey): Handle/report error in calling functions.
+  if not error:
+    json_data = json.loads(content)
+    return SwarmingTaskRequest.Deserialize(json_data)
+  return None
 
 
 def TriggerSwarmingTask(request, http_client):
   """Triggers a new Swarming task for the given request.
 
   The Swarming task priority will be overwritten, and extra tags might be added.
   Args:
     request (SwarmingTaskRequest): A Swarming task request.
     http_client (RetryHttpClient): An http client with automatic retry.
   """
   # Use a priority much lower than CQ for now (CQ's priority is 30).
   # Later we might use a higher priority -- a lower value here.
   # Note: the smaller value, the higher priority.
   swarming_settings = waterfall_config.GetSwarmingSettings()
   request_expiration_hours = swarming_settings.get('request_expiration_hours')
   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 = _SendRequestToServer(url, http_client, request.Serialize())
-  return json.loads(response_data)['task_id']
+  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.
 
   Args:
     master_name(str): Value of the master tag.
     builder_name(str): Value of the buildername tag.
@@ skipping 12 matching lines @@
     base_url += '&tags=%s' % urllib.quote('%s:%s' % (tag_name, tag_value))
 
   items = []
   cursor = None
 
   while True:
     if not cursor:
       url = base_url
     else:
       url = base_url + '&cursor=%s' % urllib.quote(cursor)
-    new_data = _SendRequestToServer(url, http_client)
+    new_data, _ = _SendRequestToServer(url, http_client)
+
+    # TODO(lijeffrey): handle error in calling functions.
     if not new_data:
       break
 
     new_data_json = json.loads(new_data)
     if new_data_json.get('items'):
       items.extend(new_data_json['items'])
 
     if new_data_json.get('cursor'):
       cursor = new_data_json['cursor']
     else:
       break
 
   return items
 
 
 def _GenerateIsolatedData(outputs_ref):
   if not outputs_ref:
     return {}
   return {
       'digest': outputs_ref['isolated'],
       'namespace': outputs_ref['namespace'],
       'isolatedserver': outputs_ref['isolatedserver']
   }
 
 
 def GetSwarmingTaskResultById(task_id, http_client):
   """Gets swarming result, checks state and returns outputs ref if needed."""
   base_url = ('https://%s/_ah/api/swarming/v1/task/%s/result') % (
       waterfall_config.GetSwarmingSettings().get('server_host'), task_id)
-  data = _SendRequestToServer(base_url, http_client)
-  json_data = json.loads(data)
+  json_data = {}
 
-  return json_data
+  data, error = _SendRequestToServer(base_url, http_client)
+
+  if not error:
+    json_data = json.loads(data)
+
+  return json_data, error
 
 
 def GetSwarmingTaskFailureLog(outputs_ref, http_client):
   """Downloads failure log from isolated server."""
   isolated_data = _GenerateIsolatedData(outputs_ref)
   return _DownloadTestResults(isolated_data, http_client)
 
 
 def GetTagValue(tags, tag_name):
   """Returns the content for a specific tag."""
@@ skipping 74 matching lines @@
     return None
 
   post_data = {
       'digest': isolated_data['digest'],
       'namespace': {
           'namespace': isolated_data['namespace']
       }
   }
   url = '%s/_ah/api/isolateservice/v1/retrieve' % (
       isolated_data['isolatedserver'])
-  content = _SendRequestToServer(url, http_client, post_data)
-  return content
+
+  return _SendRequestToServer(url, http_client, post_data)
 
 
 def _GetOutputJsonHash(content):
   """Gets hash for output.json.
 
   Parses response content of the request using hash for .isolated file and
   returns the hash for output.json file.
 
   Args:
     content (string): Content returned by the POST request to isolated server
         for hash to output.json.
   """
   content_json = json.loads(content)
   content_string = zlib.decompress(base64.b64decode(content_json['content']))
   json_result = json.loads(content_string)
   return json_result.get('files', {}).get('output.json', {}).get('h')
 
 
 def _RetrieveOutputJsonFile(output_json_content, http_client):
   """Downloads output.json file from isolated server or process it directly.
 
   If there is a url provided, send get request to that url to download log;
   else the log would be in content so use it directly.
   """
   json_content = json.loads(output_json_content)
   output_json_url = json_content.get('url')
   if output_json_url:
-    get_content = _SendRequestToServer(output_json_url, http_client)
+    get_content, _ = _SendRequestToServer(output_json_url, http_client)
+    # TODO(lijeffrey): handle error in calling function.
   elif json_content.get('content'):
     get_content = base64.b64decode(json_content['content'])
   else:  # pragma: no cover
     get_content = None  # Just for precausion.
   try:
     return json.loads(zlib.decompress(get_content)) if get_content else None
   except ValueError:  # pragma: no cover
     logging.info(
         'swarming result is invalid: %s' % zlib.decompress(get_content))
     return None
 
+
 def _DownloadTestResults(isolated_data, http_client):
   """Downloads the output.json file and returns the json object."""
   # First POST request to get hash for the output.json file.
-  content = _FetchOutputJsonInfoFromIsolatedServer(
+  content, error = _FetchOutputJsonInfoFromIsolatedServer(
       isolated_data, http_client)
-  if not content:
-    return None
+  if error:
+    return None, error
+
   output_json_hash = _GetOutputJsonHash(content)
   if not output_json_hash:
-    return None
+    return None, None
 
   # Second POST request to get the redirect url for the output.json file.
   data_for_output_json = {
       'digest': output_json_hash,
       'namespace': isolated_data['namespace'],
       'isolatedserver': isolated_data['isolatedserver']
   }
-  output_json_content = _FetchOutputJsonInfoFromIsolatedServer(
+
+  output_json_content, error = _FetchOutputJsonInfoFromIsolatedServer(
       data_for_output_json, http_client)
-  if not output_json_content:
-    return None
+  if error:
+    return None, error
 
   # GET Request to get output.json file.
-  return _RetrieveOutputJsonFile(
-      output_json_content, http_client)
+  return _RetrieveOutputJsonFile(output_json_content, http_client), None
 
 
 def _MergeListsOfDicts(merged, shard):
   output = []
   for i in xrange(max(len(merged), len(shard))):
     merged_dict = merged[i] if i < len(merged) else {}
     shard_dict = shard[i] if i < len(shard) else {}
     output_dict = merged_dict.copy()
     output_dict.update(shard_dict)
     output.append(output_dict)
@@ skipping 42 matching lines @@
         shard_result.get('per_iteration_data', []))
   merged_results['all_tests'] = sorted(merged_results['all_tests'])
   return merged_results
 
 
 def RetrieveShardedTestResultsFromIsolatedServer(
     list_isolated_data, http_client):
   """Gets test results from isolated server and merge the results."""
   shard_results = []
   for isolated_data in list_isolated_data:
-    output_json = _DownloadTestResults(isolated_data, http_client)
+    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)