[Findit] Refactoring monitor swarming task pipelines

appengine/findit/waterfall/process_base_swarming_task_result_pipeline.py

Index: appengine/findit/waterfall/process_base_swarming_task_result_pipeline.py
diff --git a/appengine/findit/waterfall/process_base_swarming_task_result_pipeline.py b/appengine/findit/waterfall/process_base_swarming_task_result_pipeline.py
index e493145670b4608e36b2180bb4624f6f9cf71dc5..34214f64d8a69537756ac1f57713afd6fd174908 100644
--- a/appengine/findit/waterfall/process_base_swarming_task_result_pipeline.py
+++ b/appengine/findit/waterfall/process_base_swarming_task_result_pipeline.py
@@ -28,19 +28,6 @@ class ProcessBaseSwarmingTaskResultPipeline(BasePipeline):
     raise NotImplementedError(
         '_CheckTestsRunStatuses should be implemented in the child class')
 
-  def _ConvertDateTime(self, time_string):
-    """Convert UTC time string to datetime.datetime."""
-    # Match the time conversion with swarming.py which elides the suffix
-    # when microseconds are 0.
-    if not time_string:
-      return None
-    for fmt in ('%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S'):
-      try:
-        return datetime.datetime.strptime(time_string, fmt)
-      except ValueError:
-        pass
-    raise ValueError('Failed to parse %s' % time_string)  # pragma: no cover
-
   def _GetSwarmingTask(self):
     # Get the appropriate kind of Swarming Task (Wf or Flake).
     # Should be overwritten by subclass.
@@ -53,87 +40,107 @@ class ProcessBaseSwarmingTaskResultPipeline(BasePipeline):
     raise NotImplementedError(
         '_GetArgs should be implemented in the child class')
 
-  # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self, master_name, builder_name, build_number,
-          step_name, task_id, *args):  # pragma: no cover.
-    """
-    Args:
-      master_name (str): The master name.
-      builder_name (str): The builder name.
-      build_number (str): The build number.
-      step_name (str): The failed test step name.
-      task_id (str): Id for the swarming task which is triggered by Findit.
+  def _ConvertDateTime(self, time_string):
+    """Convert UTC time string to datetime.datetime."""
+    if not time_string:
+      return None
+    for fmt in ('%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S'):
+      # When microseconds are 0, the '.123456' suffix is elided.
+      try:
+        return datetime.datetime.strptime(time_string, fmt)
+      except ValueError:
+        pass
+    raise ValueError('Failed to parse %s' % time_string)  # pragma: no cover
 
-    Returns:
-      A dict of lists for reliable/flaky tests.
-    """
-    call_args = self._GetArgs(master_name, builder_name, build_number,
-                              step_name, *args)
-    assert task_id
+  def _MonitorSwarmingTask(self, task_id, *call_args):
+    """Monitors the swarming task and waits for it to complete."""
     timeout_hours = waterfall_config.GetSwarmingSettings().get(
         'task_timeout_hours')
     deadline = time.time() + timeout_hours * 60 * 60
     server_query_interval_seconds = waterfall_config.GetSwarmingSettings().get(
         'server_query_interval_seconds')
+
     task_started = False
     task_completed = False
     tests_statuses = {}
     step_name_no_platform = None
+    task = self._GetSwarmingTask(*call_args)
 
     while not task_completed:
-      # Keeps monitoring the swarming task, waits for it to complete.
-      data = swarming_util.GetSwarmingTaskResultById(
-          task_id, self.HTTP_CLIENT)
+      data = swarming_util.GetSwarmingTaskResultById(task_id, self.HTTP_CLIENT)
       task_state = data['state']
-      exit_code = (data['exit_code'] if
+      exit_code = (data.get('exit_code') if
                    task_state == swarming_util.STATE_COMPLETED else None)
       step_name_no_platform = swarming_util.GetTagValue(

[chanli, 2016/11/09 00:55:24]

We can modify this to:
step_name_no_platform = step_name_no_platform or
swarming_util.GetTagValue(data.get('tags', {}), 'ref_name') 

to reduce function calls

[lijeffrey, 2016/11/09 01:03:21]

Done.

           data.get('tags', {}), 'ref_name')
+
       if task_state not in swarming_util.STATES_RUNNING:
         task_completed = True
-        task = self._GetSwarmingTask(*call_args)
+
         if (task_state == swarming_util.STATE_COMPLETED and
             int(exit_code) != swarming_util.TASK_FAILED):
           outputs_ref = data.get('outputs_ref')
           output_json = swarming_util.GetSwarmingTaskFailureLog(
               outputs_ref, self.HTTP_CLIENT)
-          tests_statuses = self._CheckTestsRunStatuses(
-              output_json, *call_args)
+          tests_statuses = self._CheckTestsRunStatuses(output_json, *call_args)
           task.status = analysis_status.COMPLETED
           task.tests_statuses = tests_statuses
         else:
           task.status = analysis_status.ERROR
           logging_str = 'Swarming task stopped with status: %s' % task_state
-          if exit_code:
+          if exit_code:  # pragma: no cover
             logging_str += ' and exit_code: %s - %s' % (
                 exit_code, swarming_util.EXIT_CODE_DESCRIPTIONS[int(exit_code)])
           logging.error(logging_str)
 
-        priority_str = swarming_util.GetTagValue(
-            data.get('tags', {}), 'priority')
+        tags = data.get('tags', {})
+        priority_str = swarming_util.GetTagValue(tags, 'priority')
         if priority_str:
           task.parameters['priority'] = int(priority_str)
+
         task.put()
       else:  # pragma: no cover
         if task_state == 'RUNNING' and not task_started:
           # swarming task just starts, update status.
           task_started = True
-          task = self._GetSwarmingTask(*call_args)
           task.status = analysis_status.RUNNING
           task.put()
         time.sleep(server_query_interval_seconds)
+
+      # Timeout.
       if time.time() > deadline:
         # Updates status as ERROR.
-        task = self._GetSwarmingTask(*call_args)
         task.status = analysis_status.ERROR
         task.put()
         logging.error('Swarming task timed out after %d hours.' % timeout_hours)
         break  # Stops the loop and return.
-    # Update swarming task metadata.
-    task = self._GetSwarmingTask(*call_args)
+
+    # Update swarming task metadata timestamps.
     task.created_time = self._ConvertDateTime(data.get('created_ts'))
     task.started_time = self._ConvertDateTime(data.get('started_ts'))
     task.completed_time = self._ConvertDateTime(data.get('completed_ts'))
     task.put()
 
+    return step_name_no_platform
+
+  # Arguments number differs from overridden method - pylint: disable=W0221
+  def run(self, master_name, builder_name, build_number, step_name, task_id,
+          *args):
+    """Monitors a swarming task.
+
+    Args:
+      master_name (str): The master name.
+      builder_name (str): The builder name.
+      build_number (str): The build number.
+      step_name (str): The failed test step name.
+      task_id (str): The task id to query the swarming server on the progresss
+        of a swarming task.
+
+    Returns:
+      A dict of lists for reliable/flaky tests.
+    """
+    call_args = self._GetArgs(master_name, builder_name, build_number,
+                              step_name, *args)
+    step_name_no_platform = self._MonitorSwarmingTask(task_id, *call_args)
     return step_name, step_name_no_platform
+