[Findit] Process analysis requests of Waterfall failures concurrently.

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.
@@ skipping 99 matching lines @@
 
 class _FlakeAnalysis(messages.Message):
   analysis_triggered = messages.BooleanField(1, required=True)
 
 
 def _TriggerNewAnalysesOnDemand(builds):
   """Pushes a task to run on the backend to trigger new analyses on demand."""
   target = appengine_util.GetTargetNameForModule(constants.WATERFALL_BACKEND)
   payload = json.dumps({'builds': builds})
   taskqueue.add(
-      url=constants.WATERFALL_TRIGGER_ANALYSIS_URL, payload=payload,
-      target=target, queue_name=constants.WATERFALL_SERIAL_QUEUE)
+      url=constants.WATERFALL_TRIGGER_ANALYSIS_URL,
+      payload=payload, target=target,
+      queue_name=constants.WATERFALL_FAILURE_ANALYSIS_REQUEST_QUEUE)
 
 
 # Create a Cloud Endpoints API.
 # https://cloud.google.com/appengine/docs/python/endpoints/create_api
 @endpoints.api(name='findit', version='v1', description='FindIt API')
 class FindItApi(remote.Service):
   """FindIt API v1."""
 
   def _GenerateBuildFailureAnalysisResult(
       self, build, suspected_cls_in_result, step_name,
@@ skipping 109 matching lines @@
     Returns:
       _BuildFailureAnalysisResultCollection
       A list of analysis results for the given build failures.
     """
     results = []
     supported_builds = []
 
     for build in request.builds:
       master_name = buildbot.GetMasterNameFromUrl(build.master_url)
       if not (master_name and waterfall_config.MasterIsSupported(master_name)):
+        logging.info('%s/%s/%s is not supported',
+                     build.master_url, build.builder_name, build.build_number)
         continue
 
       supported_builds.append({
           'master_name': master_name,
           'builder_name': build.builder_name,
           'build_number': build.build_number,
           'failed_steps': build.failed_steps,
       })
 
       # If the build failure was already analyzed and a new analysis is
@@ skipping 11 matching lines @@
 
       self._GenerateResultsForBuild(build, heuristic_analysis, results)
 
     logging.info('%d build failure(s), while %d are supported',
                  len(request.builds), len(supported_builds))
     try:
       _TriggerNewAnalysesOnDemand(supported_builds)
     except Exception:  # pragma: no cover.
       # If we fail to post a task to the task queue, we ignore and wait for next
       # request.
-      logging.exception('Failed to trigger new analyses on demand.')
+      logging.exception('Failed to add analysis request to task queue: %s',
+                        repr(supported_builds))
 
     return _BuildFailureAnalysisResultCollection(results=results)
 
   @endpoints.method(_Flake, _FlakeAnalysis, path='flake', name='flake')
   def AnalyzeFlake(self, request):
     """Analyze a flake on Commit Queue. Currently only supports flaky tests."""
     user_email = auth_util.GetUserEmail()
     is_admin = auth_util.IsCurrentUserAdmin()
 
     def CreateFlakeAnalysisRequest(flake):
       analysis_request = FlakeAnalysisRequest.Create(
           flake.name, flake.is_step, flake.bug_id)
       for step in flake.build_steps:
         analysis_request.AddBuildStep(step.master_name, step.builder_name,
                                       step.build_number, step.step_name,
                                       time_util.GetUTCNow())
       return analysis_request
 
     logging.info('Flake: %s', CreateFlakeAnalysisRequest(request))
     analysis_triggered = flake_analysis_service.ScheduleAnalysisForFlake(
         CreateFlakeAnalysisRequest(request), user_email, is_admin)
 
     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)