[Findit] Reduce redundant ndb reads by querying necessary entities ahead of time and share them amo…

appengine/findit/findit_api.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.
 
 """This module is to provide Findit service APIs through Cloud Endpoints:
 
 Current APIs include:
 1. Analysis of compile/test failures in Chromium waterfalls.
    Analyzes failures and detects suspected CLs.
 2. Analysis of flakes on Commit Queue.
 """
 
+from collections import defaultdict
 import json
 import logging
 
 import endpoints
 from google.appengine.api import taskqueue
 from protorpc import messages
 from protorpc import remote
 
 from common import appengine_util
 from common import auth_util
@@ skipping 171 matching lines @@
         step_name=step_name,
         is_sub_test=test_name is not None,
         test_name=test_name,
         first_known_failed_build_number=first_failure,
         suspected_cls=suspected_cls,
         analysis_approach=analysis_approach,
         try_job_status=try_job_status,
         is_flaky_test=is_flaky_test)
 
   def _GetStatusAndCulpritFromTryJob(
-      self, try_job_map, build_failure_type, step_name, test_name=None):
+      self, try_job, swarming_task, build_failure_type, step_name,
+      test_name=None):
     """Returns the culprit found by try-job for the given step or test."""
-    if not try_job_map:
-      return _TryJobStatus.FINISHED, None
 
-    if test_name is None:
-      try_job_key = try_job_map.get(step_name)
-    else:
-      try_job_key = try_job_map.get(step_name, {}).get(test_name)
-
-    if not try_job_key:
-      return _TryJobStatus.FINISHED, None
-
-    try_job = WfTryJob.Get(*try_job_key.split('/'))
     if not try_job or try_job.failed:
       return _TryJobStatus.FINISHED, None
 
     if not try_job.completed:
       return _TryJobStatus.RUNNING, None
 
     if build_failure_type == failure_type.COMPILE:
       if not try_job.compile_results:  # pragma: no cover.
         return _TryJobStatus.FINISHED, None
       return (
           _TryJobStatus.FINISHED,
           try_job.compile_results[-1].get('culprit', {}).get(step_name))
 
     if not try_job.test_results:  # pragma: no cover.
       return _TryJobStatus.FINISHED, None
 
     if test_name is None:
       step_info = try_job.test_results[-1].get('culprit', {}).get(step_name)
       if not step_info or step_info.get('tests'):  # pragma: no cover.
         # TODO(chanli): For some steps like checkperms/sizes/etc, the culprit
         # finding try-job might have test-level results.
         return _TryJobStatus.FINISHED, None
       return _TryJobStatus.FINISHED, step_info
 
-    task = WfSwarmingTask.Get(*try_job_key.split('/'), step_name=step_name)
-    ref_name = (task.parameters.get('ref_name') if task and task.parameters
-                else None)
+    ref_name = (swarming_task.parameters.get('ref_name') if swarming_task and
+                swarming_task.parameters else None)
     return (
         _TryJobStatus.FINISHED, try_job.test_results[-1].get('culprit', {}).get(
             ref_name or step_name, {}).get('tests', {}).get(test_name))
 
-  def _CheckIsFlaky(self, try_job_map, step_name, test_name):
+  def _CheckIsFlaky(self, swarming_task, test_name):
     """Checks if the test is flaky."""
-    if not try_job_map or not test_name:
-      return False
-
-    try_job_key = try_job_map.get(step_name, {}).get(test_name)
-    if not try_job_key:
-      return False
-
-    swarming_task = WfSwarmingTask.Get(*try_job_key.split('/'),
-                                       step_name=step_name)
     if not swarming_task or not swarming_task.classified_tests:
       return False

[stgao, 2016/10/22 00:29:03]

So for tests, there are three status:
1. Reliable
2. Flaky
3. Unclassified

It seems Reliable + Unclassified are treated as Reliable here.

[chanli, 2016/10/24 18:42:34]

We don't have swarming tasks and try jobs for unclassified tests, so we cannot
say they are flaky; and if we have heuristic results for them, we have to treat
them as reliable tests and display what we found on SoM. What makes it better is
that we now have confidence score, so it should provide some level of guidance
to sheriffs about how trustworthy the result is.

[stgao, 2016/10/25 17:24:04]

Sorry for the confusion. My question here is: what "False" mean here? If it
means reliable, why Unclassified ones are treated as reliable even we don't know
for sure?
As it is determined to be flaky by Swarm rerun, what's the consideration for a
flaky test to be treated as reliable one here?

[chanli, 2016/10/26 03:37:00]

Basically true means Findit thinks the failure is flaky, and false means
everything else.

And for unclassified failures, the results are not necessarily to be wrong, so
we should show the result with confidence score

 
     return test_name in swarming_task.classified_tests.get('flaky_tests', [])
 
   def _PopulateResult(
-      self, results, build, try_job_map, build_failure_type,
-      heuristic_result, step_name, confidences, test_name=None):
+      self, results, build, build_failure_type,heuristic_result, step_name,
+      confidences, swarming_task, try_job, test_name=None):
     """Appends an analysis result for the given step or test.
 
     Try-job results are always given priority over heuristic results.
     """
     # Default to heuristic analysis.
     suspected_cls = heuristic_result['suspected_cls']
     analysis_approach = _AnalysisApproach.HEURISTIC
 
     # Check if the test is flaky.
-    is_flaky_test = self._CheckIsFlaky(try_job_map, step_name, test_name)
+    is_flaky_test = self._CheckIsFlaky(swarming_task, test_name)
 
     if is_flaky_test:
       suspected_cls = []
       try_job_status = _TryJobStatus.FINISHED  # There will be no try job.
     else:
       # Check analysis result from try-job.
       try_job_status, culprit = self._GetStatusAndCulpritFromTryJob(
-          try_job_map, build_failure_type, step_name, test_name=test_name)
+          try_job, swarming_task, build_failure_type, step_name,
+          test_name=test_name)
       if culprit:
         suspected_cls = [culprit]
         analysis_approach = _AnalysisApproach.TRY_JOB
 
     if not is_flaky_test and not suspected_cls:
       return
 
     results.append(self._GenerateBuildFailureAnalysisResult(
         build, suspected_cls, step_name, heuristic_result['first_failure'],
         test_name, analysis_approach, confidences, try_job_status,
         is_flaky_test))
 
+  def _GetAllSwarmingTasks(self, failure_result_map):
+    """Returns all swarming tasks related to one build.
+
+    Returns:
+      A dict of swarming tasks like below:
+      {
+          'step1': {
+              'm/b/1': WfSwarmingTask(
+                  key=Key('WfBuild', 'm/b/1', 'WfSwarmingTask', 'step1'),...)
+          },
+          ...
+      }
+    """
+    if not failure_result_map:
+      return {}
+
+    swarming_tasks = defaultdict(dict)
+    for step_name, step_map in failure_result_map.iteritems():
+      if isinstance(step_map, basestring):
+        swarming_tasks[step_name][step_map] = (
+            WfSwarmingTask.Get(*step_map.split('/'), step_name=step_name))

[stgao, 2016/10/22 00:29:03]

again: encode & decode should have a function.

[chanli, 2016/10/24 18:42:34]

Done.

+      else:

[lijeffrey, 2016/10/19 20:35:25]

nit: add a comment here for the type of structure of step_map

[chanli, 2016/10/24 18:42:34]

Added Args in docstring.

+        for task_key in step_map.values():
+          if not swarming_tasks[step_name].get(task_key):
+            swarming_tasks[step_name][task_key] = (
+              WfSwarmingTask.Get(*task_key.split('/'), step_name=step_name))
+
+    return swarming_tasks
+
+  def _GetAllTryJobs(self, failure_result_map):
+    """Returns all try jobs related to one build.
+
+    Returns:
+      A dict of try jobs like below:
+      {
+          'm/b/1': WfTryJob(
+              key=Key('WfBuild', 'm/b/1'),...)
+          ...
+      }
+    """
+    if not failure_result_map:
+      return {}
+
+    try_jobs = {}
+    for step_map in failure_result_map.values():
+      if isinstance(step_map, basestring):
+        try_jobs[step_map] = WfTryJob.Get(*step_map.split('/'))
+      else:

[lijeffrey, 2016/10/19 20:35:24]

nit: same here

[chanli, 2016/10/24 18:42:34]

Done.

+        for task_key in step_map.values():
+          if not try_jobs.get(task_key):
+            try_jobs[task_key] = WfTryJob.Get(*task_key.split('/'))
+
+    return try_jobs
+
+  def _GetSwarmingTaskAndTryJobForFailure(
+      self, step_name, test_name, failure_result_map, swarming_tasks, try_jobs):

[lijeffrey, 2016/10/19 20:35:25]

nit: please add a docstring for this function

[chanli, 2016/10/24 18:42:34]

Done.

+    if not failure_result_map:
+      return None, None
+
+    if test_name:
+      try_job_key = failure_result_map.get(step_name, {}).get(test_name)
+    else:
+      try_job_key = failure_result_map.get(step_name)
+
+    # Gets the swarming task for the test.
+    swarming_task = swarming_tasks.get(step_name, {}).get(try_job_key)
+    # Get the try job for the step/ test.

[lijeffrey, 2016/10/19 20:35:24]

nit: remove extra space before 'test' and add 1 empty line before this comment

[chanli, 2016/10/24 18:42:34]

Done.

+    try_job = try_jobs.get(try_job_key)
+
+    return swarming_task, try_job
+
   def _GenerateResultsForBuild(
       self, build, heuristic_analysis, results, confidences):
+
+    swarming_tasks = self._GetAllSwarmingTasks(
+        heuristic_analysis.failure_result_map)
+    try_jobs = self._GetAllTryJobs(heuristic_analysis.failure_result_map)
+
     for failure in heuristic_analysis.result['failures']:
+      step_name = failure['step_name']
       if failure.get('tests'):  # Test-level analysis.
         for test in failure['tests']:
+          test_name = test['test_name']
+          swarming_task, try_job = self._GetSwarmingTaskAndTryJobForFailure(
+              step_name, test_name, heuristic_analysis.failure_result_map,
+              swarming_tasks, try_jobs)
+
           self._PopulateResult(
-              results, build, heuristic_analysis.failure_result_map,
-              heuristic_analysis.failure_type, test,
-              failure['step_name'], confidences, test_name=test['test_name'])
+              results, build, heuristic_analysis.failure_type, test,
+              step_name, confidences, swarming_task, try_job,
+              test_name=test_name)
       else:
+        swarming_task, try_job = self._GetSwarmingTaskAndTryJobForFailure(
+            step_name, None, heuristic_analysis.failure_result_map,
+            swarming_tasks, try_jobs)
         self._PopulateResult(
-            results, build, heuristic_analysis.failure_result_map,
-            heuristic_analysis.failure_type, failure, failure['step_name'],
-            confidences)
+            results, build, heuristic_analysis.failure_type, failure,
+            step_name, confidences, swarming_task, try_job)
 
   @endpoints.method(
       _BuildFailureCollection, _BuildFailureAnalysisResultCollection,
       path='buildfailure', name='buildfailure')
   def AnalyzeBuildFailures(self, request):
     """Returns analysis results for the given build failures in the request.
 
     Analysis of build failures will be triggered automatically on demand.
 
     Args:
@@ skipping 66 matching lines @@
 
     logging.info('Flake: %s', CreateFlakeAnalysisRequest(request))
     analysis_triggered = flake_analysis_service.ScheduleAnalysisForFlake(
         CreateFlakeAnalysisRequest(request), user_email, is_admin,
         triggering_sources.FINDIT_API)
 
     if analysis_triggered is None:
       raise endpoints.UnauthorizedException(
           'No permission for a new analysis! User is %s' % user_email)
 
-    return _FlakeAnalysis(analysis_triggered=analysis_triggered)
+    return _FlakeAnalysis(analysis_triggered=analysis_triggered)