[Findit] Asynchronously process flake reports from chromium-try-flakes.

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.
 """
 
 import json
 import logging
+import pickle
 
 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
 from common import constants
 from common import time_util
@@ skipping 77 matching lines @@
 
 
 class _Build(messages.Message):
   master_name = messages.StringField(1, required=True)
   builder_name = messages.StringField(2, required=True)
   build_number = messages.IntegerField(
       3, variant=messages.Variant.INT32, required=True)
 
 
 class _FlakeAnalysis(messages.Message):
-  analysis_triggered = messages.BooleanField(1, required=True)
+  queued = messages.BooleanField(1, required=True)
 
 
-def _TriggerNewAnalysesOnDemand(builds):
-  """Pushes a task to run on the backend to trigger new analyses on demand."""
+def _AsyncProcessFailureAnalysisRequests(builds):
+  """Pushes a task on the backend to process requests of failure analysis."""
   target = appengine_util.GetTargetNameForModule(constants.WATERFALL_BACKEND)
   payload = json.dumps({'builds': builds})
   taskqueue.add(
-      url=constants.WATERFALL_TRIGGER_ANALYSIS_URL,
+      url=constants.WATERFALL_PROCESS_FAILURE_ANALYSIS_REQUESTS_URL,
       payload=payload, target=target,
       queue_name=constants.WATERFALL_FAILURE_ANALYSIS_REQUEST_QUEUE)
 
 
+def _AsyncProcessFlakeReport(flake_analysis_request, user_email, is_admin):
+  """Pushes a task on the backend to process the flake report."""
+  target = appengine_util.GetTargetNameForModule(constants.WATERFALL_BACKEND)
+  payload = pickle.dumps((flake_analysis_request, user_email, is_admin))
+  taskqueue.add(
+      url=constants.WATERFALL_PROCESS_FLAKE_ANALYSIS_REQUEST_URL,
+      payload=payload, target=target,
+      queue_name=constants.WATERFALL_FLAKE_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,
       first_failure, test_name=None,
       analysis_approach=_AnalysisApproach.HEURISTIC):
@@ skipping 136 matching lines @@
 
       if heuristic_analysis.failed or not heuristic_analysis.result:
         # Bail out if the analysis failed or there is no result yet.
         continue
 
       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)
+      _AsyncProcessFailureAnalysisRequests(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 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()
 
+    if not flake_analysis_service.IsAuthorizedUser(user_email, is_admin):
+      raise endpoints.UnauthorizedException(
+          'No permission to run a new analysis! User is %s' % user_email)
+
     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)
+    flake_analysis_request = CreateFlakeAnalysisRequest(request)
+    logging.info('Flake: %s', flake_analysis_request)
 
-    if analysis_triggered is None:
-      raise endpoints.UnauthorizedException(
-          'No permission for a new analysis! User is %s' % user_email)
+    try:
+      _AsyncProcessFlakeReport(flake_analysis_request, user_email, is_admin)
+      queued = True
+    except Exception:
+      # Ignore the report when fail to queue it for async processing.
+      queued = False
+      logging.exception('Failed to queue flake report for async processing')
 
-    return _FlakeAnalysis(analysis_triggered=analysis_triggered)
+    return _FlakeAnalysis(queued=queued)