[Findit] Modify Findit API to return more information to Sheriff-O-Matic.

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 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
 from common.waterfall import failure_type
+from model import analysis_approach_type
 from model.flake.flake_analysis_request import FlakeAnalysisRequest
+from model.suspected_cl_confidence import SuspectedCLConfidence
 from model.wf_analysis import WfAnalysis
+from model.wf_suspected_cl import WfSuspectedCL
 from model.wf_swarming_task import WfSwarmingTask
 from model.wf_try_job import WfTryJob
 from waterfall import buildbot
+from waterfall import suspected_cl_util
 from waterfall import waterfall_config
 from waterfall.flake import flake_analysis_service
 from waterfall.flake import triggering_sources
 
 
 # This is used by the underlying ProtoRpc when creating names for the ProtoRPC
 # messages below. This package name will show up as a prefix to the message
 # class names in the discovery doc and client libraries.
 package = 'FindIt'
 
 
 # These subclasses of Message are basically definitions of Protocol RPC
 # messages. https://cloud.google.com/appengine/docs/python/tools/protorpc/
 class _BuildFailure(messages.Message):
   master_url = messages.StringField(1, required=True)
   builder_name = messages.StringField(2, required=True)
   build_number = messages.IntegerField(3, variant=messages.Variant.INT32,
                                        required=True)
   # All failed steps of the build reported by the client.
   failed_steps = messages.StringField(4, repeated=True, required=False)
 
 
 class _BuildFailureCollection(messages.Message):
   """Represents a request from a client, eg. builder_alerts."""
   builds = messages.MessageField(_BuildFailure, 1, repeated=True)
 
 
+class _AnalysisApproach(messages.Enum):
+  HEURISTIC = analysis_approach_type.HEURISTIC
+  TRY_JOB = analysis_approach_type.TRY_JOB
+
+
 class _SuspectedCL(messages.Message):
   repo_name = messages.StringField(1, required=True)
   revision = messages.StringField(2, required=True)
   commit_position = messages.IntegerField(3, variant=messages.Variant.INT32)
+  confidence = messages.IntegerField(4, variant=messages.Variant.INT32)
+  analysis_approach = messages.EnumField(_AnalysisApproach, 5)
 
 
-class _AnalysisApproach(messages.Enum):
-  HEURISTIC = 1
-  TRY_JOB = 2
+class _TryJobStatus(messages.Enum):

[stgao, 2016/10/19 00:43:08]

More status here:
1. not supported
2. pending
3. error when finished
4. no culprit found when finished

[chanli, 2016/10/19 03:00:59]

I was thinking the try job status is to let sheriffs know should they wait for
try job results or not. That's why I only have 2 status. If the message showed
on SoM are the same for the status you mentioned, maybe we don't need to add the
status you mentioned.

[stgao, 2016/10/21 01:51:22]

OK, I agree 2-4 could be merged into RUNNING or FINISHED.
But for case 1, how do we expect sheriff to know it is not supported, and don't
wait?

[stgao, 2016/10/22 00:24:14]

Per discussion offline, we will go with the current approach, and make needed
change later.

+  # Try job is pending or running. Can expect result from try job.
+  RUNNING = 1
+  # There is no try job, try job completed or try job finished with error.
+  # Result from try job is ready or no need to continue waiting for it.
+  FINISHED = 2
 
 
 class _BuildFailureAnalysisResult(messages.Message):
   master_url = messages.StringField(1, required=True)
   builder_name = messages.StringField(2, required=True)
   build_number = messages.IntegerField(3, variant=messages.Variant.INT32,
                                        required=True)
   step_name = messages.StringField(4, required=True)
   is_sub_test = messages.BooleanField(5, variant=messages.Variant.BOOL,
                                       required=True)
   test_name = messages.StringField(6)
   first_known_failed_build_number = messages.IntegerField(
       7, variant=messages.Variant.INT32)
   suspected_cls = messages.MessageField(_SuspectedCL, 8, repeated=True)
   analysis_approach = messages.EnumField(_AnalysisApproach, 9)
+  try_job_status = messages.EnumField(_TryJobStatus, 10)
+  is_flaky_test = messages.BooleanField(11, variant=messages.Variant.BOOL)
 
 
 class _BuildFailureAnalysisResultCollection(messages.Message):
   """Represents a response to the client, eg. builder_alerts."""
   results = messages.MessageField(_BuildFailureAnalysisResult, 1, repeated=True)
 
 
 class _BuildStep(messages.Message):
   master_name = messages.StringField(1, required=True)
   builder_name = messages.StringField(2, required=True)
@@ skipping 30 matching lines @@
       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 _GetConfidenceAndApproachForCL(
+      self, repo_name, revision, confidences, build, first_failure):
+    cl = WfSuspectedCL.Get(repo_name, revision)
+    if not cl:
+      return None, None
+
+    master_name = buildbot.GetMasterNameFromUrl(build.master_url)
+    builder_name = build.builder_name
+    current_build = build.build_number
+
+    # If the CL is found by a try job, only the first failure will be recorded.
+    # So we might need to go to the first failure to get CL information.
+    build_info = (cl.GetBuildInfo(master_name, builder_name, current_build) or
+                  cl.GetBuildInfo(master_name, builder_name, first_failure))
+
+    confidence = suspected_cl_util.GetSuspectedCLConfidenceScore(
+        confidences, build_info)
+
+    cl_approach = (
+      _AnalysisApproach.TRY_JOB if analysis_approach_type.TRY_JOB in
+      build_info['approaches'] else _AnalysisApproach.HEURISTIC)
+
+    return confidence, cl_approach
+
   def _GenerateBuildFailureAnalysisResult(
-      self, build, suspected_cls_in_result, step_name,
-      first_failure, test_name=None,
-      analysis_approach=_AnalysisApproach.HEURISTIC):
+      self, build, suspected_cls_in_result, step_name, first_failure, test_name,
+      analysis_approach, confidences, try_job_status, is_flaky_test):
+
     suspected_cls = []
     for suspected_cl in suspected_cls_in_result:
+      repo_name = suspected_cl['repo_name']
+      revision = suspected_cl['revision']
+      commit_position = suspected_cl['commit_position']
+      confidence, cl_approach = self._GetConfidenceAndApproachForCL(
+          repo_name, revision, confidences, build, first_failure)
+      cl_approach = cl_approach or analysis_approach
+
       suspected_cls.append(_SuspectedCL(
-          repo_name=suspected_cl['repo_name'],
-          revision=suspected_cl['revision'],
-          commit_position=suspected_cl['commit_position']))
+          repo_name=repo_name, revision=revision,
+          commit_position=commit_position, confidence=confidence,
+          analysis_approach=cl_approach))
 
     return _BuildFailureAnalysisResult(
         master_url=build.master_url,
         builder_name=build.builder_name,
         build_number=build.build_number,
         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)
+        analysis_approach=analysis_approach,
+        try_job_status=try_job_status,
+        is_flaky_test=is_flaky_test)
 
-  def _GetCulpritFromTryJob(
+  def _GetStatusAndCulpritFromTryJob(
       self, try_job_map, 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 None
+      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 None
+      return _TryJobStatus.FINISHED, None
 
     try_job = WfTryJob.Get(*try_job_key.split('/'))
-    if not try_job or not try_job.completed or try_job.failed:
-      return None
+    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 None
-      return try_job.compile_results[-1].get('culprit', {}).get(step_name)
+        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 None
+      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 None
-      return step_info
+        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)
-    return try_job.test_results[-1].get('culprit', {}).get(
-        ref_name or step_name, {}).get('tests', {}).get(test_name)
+    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):
+    """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('/'),

[stgao, 2016/10/22 00:24:14]

Do we have function for the encode and decode of key like this one?

[chanli, 2016/10/24 18:30:44]

Done.

+                                       step_name=step_name)
+    if not swarming_task or not swarming_task.classified_tests:
+      return False
+
+    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, test_name=None):
+      heuristic_result, step_name, confidences, 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 analysis result from try-job.
-    culprit = self._GetCulpritFromTryJob(
-        try_job_map, build_failure_type, step_name, test_name=test_name)
-    if culprit:
-      suspected_cls = [culprit]
-      analysis_approach = _AnalysisApproach.TRY_JOB
+    # Check if the test is flaky.
+    is_flaky_test = self._CheckIsFlaky(try_job_map, step_name, test_name)
 
-    if not suspected_cls:
+    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)
+      if culprit:
+        suspected_cls = [culprit]
+        analysis_approach = _AnalysisApproach.TRY_JOB
+
+    if not is_flaky_test and not suspected_cls:
       return
 
     results.append(self._GenerateBuildFailureAnalysisResult(

[stgao, 2016/10/19 00:43:08]

Do we have an estimation of how many ndb read in order to generate one result?

[chanli, 2016/10/19 03:00:59]

For cases with the most reads:
1. 1 ndb read for confidence which can be shared by all results
2. 1 ndb read for WfAnalysis which will be shared by all results for this
analysis
For each step/test
3. 2 ndb reads for WfSwarmingTasks - first check if flaky, second check for
ref_name (can be improved)
3. 1 ndb read for WfTryJob
4. x ndb reads for WfSuspectedCL (based on how many suspected cls, normally
should be 1)

So  >5 reads.

There are some obvious improvements:
1. reduce the number of WfSwarmingTask reads (not always 1, based on the first
failures) by sharing all tasks among results of the same analysis
2. reduce the number of WfTryJob reads - same approach as WfSwarmingTask
3. Reduce the number of WfSuspectedCL reads -maybe need to handle try-job
results for non-first-time-failure differently.

Since there are several places need to be improved, I'll implement them in a
separate CL.

[stgao, 2016/10/21 01:51:22]

Acknowledged.

         build, suspected_cls, step_name, heuristic_result['first_failure'],
-        test_name=test_name, analysis_approach=analysis_approach))
+        test_name, analysis_approach, confidences, try_job_status,
+        is_flaky_test))
 
-  def _GenerateResultsForBuild(self, build, heuristic_analysis, results):
+  def _GenerateResultsForBuild(
+      self, build, heuristic_analysis, results, confidences):
     for failure in heuristic_analysis.result['failures']:
       if failure.get('tests'):  # Test-level analysis.
         for test in failure['tests']:
           self._PopulateResult(
               results, build, heuristic_analysis.failure_result_map,
               heuristic_analysis.failure_type, test,
-              failure['step_name'], test_name=test['test_name'])
+              failure['step_name'], confidences, test_name=test['test_name'])
       else:
         self._PopulateResult(
             results, build, heuristic_analysis.failure_result_map,
-            heuristic_analysis.failure_type, failure, failure['step_name'])
+            heuristic_analysis.failure_type, failure, failure['step_name'],
+            confidences)
 
   @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:
       request (_BuildFailureCollection): A list of build failures.
 
     Returns:
       _BuildFailureAnalysisResultCollection
       A list of analysis results for the given build failures.
     """
     results = []
     supported_builds = []
+    confidences = SuspectedCLConfidence.Get()
 
     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
       # scheduled to analyze new failed steps, the returned WfAnalysis will
       # still have the result from last completed analysis.
       # If there is no analysis yet, no result is returned.
       heuristic_analysis = WfAnalysis.Get(
           master_name, build.builder_name, build.build_number)
       if not heuristic_analysis:
         continue
 
       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)
+      self._GenerateResultsForBuild(
+          build, heuristic_analysis, results, confidences)
 
     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 add analysis request to task queue: %s',
                         repr(supported_builds))
@@ skipping 18 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)