[Findit] Hook up analysis for CQ flakes.

appengine/findit/waterfall/flake/flake_analysis_service.py

Index: appengine/findit/waterfall/flake/flake_analysis_service.py
diff --git a/appengine/findit/waterfall/flake/flake_analysis_service.py b/appengine/findit/waterfall/flake/flake_analysis_service.py
index ea36ab9f1533c0fd160327bd6f422fad786b8442..d3bbfa8cf245ca3d2e808704c0d94879053ed0ee 100644
--- a/appengine/findit/waterfall/flake/flake_analysis_service.py
+++ b/appengine/findit/waterfall/flake/flake_analysis_service.py
@@ -2,9 +2,148 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+import logging
 
-def ScheduleAnalysisForFlake(
-    _request, _user_email, _is_admin):  # pragma: no cover.
+from common import constants
+from model.flake.flake_analysis_request import FlakeAnalysisRequest
+from waterfall.flake import initialize_flake_pipeline
+from waterfall.flake import step_mapper
+
+
+def _CheckFlakeSwarmedAndSupported(request):
+  """Checks if the flake is Swarmed and supported in any build step.
+
+  Args:
+    request (FlakeAnalysisRequest): The request to analyze a flake.
+
+  Returns:
+    (swarmed, supported, build_step)
+    swarmed(bool): True if any step is Swarmed.
+    supported(bool): True if any step is supported (Swarmed Gtest).
+    build_step(BuildStep): The representative step that is Swarmed Gtest.
+  """
+  build_step = None
+  swarmed = False
+  supported = False
+  for step in request.build_steps:
+    swarmed = swarmed or step.swarmed
+    supported = supported or step.supported
+    if step.supported:
+      build_step = step
+      break
+  return swarmed, supported, build_step
+
+
+def _NeedNewAnalysis(request):

[chanli, 2016/10/07 21:40:37]

Nit: This kind of name indicates a True/False return value, could you change the
function name to be more clear?

[stgao, 2016/10/07 23:07:06]

Renamed to CheckForNewAnalyais. Hope it is better, although still not ideal.

+  """Checks if a new analysis is needed for the requested flake.
+
+  Args:
+    request (FlakeAnalysisRequest): The request to analyze a flake.
+
+  Returns:
+    (version_number, build_step)
+    version_number(int): The version of the FlakeAnalysisRequest if a new
+        analysis is needed; otherwise 0.

[lijeffrey, 2016/10/07 20:45:07]

nit: space before (. Same with build_step

[stgao, 2016/10/07 23:07:05]

Done.

+    build_step(BuildStep): a BuildStep instance if a new analysis is needed;
+        otherwise None.
+  """
+  previous_request = FlakeAnalysisRequest.GetVersion(key=request.name)
+  if not previous_request or (previous_request.bug_id and request.bug_id and
+                              previous_request.bug_id != request.bug_id):
+    # If no existing analysis or last analysis was for a different bug, randomly
+    # pick one configuration for a new analysis.
+    if previous_request:
+      # Make a copy to preserve the version number of previous analysis and
+      # prevent concurrent analyses of the same flake.
+      previous_request.CopyFrom(request)
+      request = previous_request
+
+    swarmed, supported, supported_build_step = _CheckFlakeSwarmedAndSupported(
+        request)
+    request.swarmed = swarmed
+    request.supported = supported
+
+    if supported_build_step and not request.is_step:
+      supported_build_step.analyzed = True  # This step will be analyzed.

[chanli, 2016/10/07 21:40:37]

Based on flake_analysis_request.py, analyzed 'Indicate whether the flake on this
configuration is analyzed.'? I thought it means this step has been analyzed
rather than will be analyzed?

[stgao, 2016/10/07 23:07:06]

Good point.
Renamed to "scheduled".

+    # For unsupported or step-level flakes, still save them for monitoring.

[lijeffrey, 2016/10/07 20:45:07]

nit: add 1 empty line before the comment

[stgao, 2016/10/07 23:07:06]

Done.

+    _, saved = request.Save(retry_on_conflict=False)  # Create a new version.
+
+    if not saved or not supported_build_step or request.is_step:
+      # No new analysis if:
+      # 1. Another analysis was just triggered.
+      # 2. No representative step is Swarmed Gtest.
+      # 3. The flake is a step-level one.
+      return 0, None
+
+    return request.version_number, supported_build_step
+  else:
+    # If no bug is attached to the existing analysis or the new request, or both
+    # are attached to the same bug, start a new analysis with a different
+    # configuration. For a configuration that was analyzed 7 days ago, reset it
+    # to use the new reported step of the same configuration.
+    # TODO: move this setting to config.
+    seconds_n_days = 7 * 24 * 60 * 60  # 7 days.
+    candidate_supported_steps = []
+    need_updating = False
+    for step in request.build_steps:

[lijeffrey, 2016/10/07 20:45:07]

is it possible to move this for loop to a separate function that returns
existing_step? In general it looks like this else branch is a bit long

[stgao, 2016/10/07 23:07:06]

I just moved the else branch to another function.

+      existing_step = None
+      for s in previous_request.build_steps:
+        if (step.master_name == s.master_name and
+            step.builder_name == s.builder_name):
+          existing_step = s
+          break
+
+      if existing_step:
+        # If last reported flake at the existing step was too long ago, drop it
+        # so that the new one is recorded.
+        time_diff = step.reported_time - existing_step.reported_time
+        if time_diff.total_seconds() > seconds_n_days:
+          previous_request.build_steps.remove(existing_step)
+          existing_step = None
+
+      if not existing_step:
+        need_updating = True
+        previous_request.build_steps.append(step)
+        if step.supported:
+          candidate_supported_steps.append(step)
+
+    if not candidate_supported_steps:
+      # Find some existing configuration that is not analyzed yet.
+      for s in previous_request.build_steps:
+        if not s.analyzed and s.supported:
+          candidate_supported_steps.append(s)
+
+    supported_build_step = None
+    if candidate_supported_steps:
+      supported_build_step = candidate_supported_steps[0]
+      previous_request.swarmed = (previous_request.swarmed or
+                                  supported_build_step.swarmed)
+      previous_request.supported = True
+      need_updating = True
+
+    if supported_build_step and not previous_request.is_step:
+      supported_build_step.analyzed = True  # This will be analyzed.
+
+    if not previous_request.bug_id:  # No bug was attached before.
+      previous_request.bug_id = request.bug_id
+      need_updating = True
+
+    if need_updating:
+      # TODO: update in a transaction.
+      previous_request.put()
+
+    if not supported_build_step or previous_request.is_step:
+      # No new analysis if:
+      # 1. All analyzed steps are fresh enough and cover all the steps in the
+      #    request.
+      # 2. No representative step is Swarmed Gtest.
+      # 3. The flake is a step-level one.
+      return 0, None
+
+    return previous_request.version_number, supported_build_step

[chanli, 2016/10/07 21:40:37]

Just to be clear:
If there is not previous analysis, or previous analysis has a different bug id,
save a new request; If there is a previous analysis, and it has the same bug id,
or only one of the request has bug_id, think it as the same request and use
another configuration (or use the same if previous is too old) and use the
previous request?

What if later a request comes in with the same bug_id as the previous request?

[stgao, 2016/10/07 23:07:06]

Assume only one bug is files for the same flake. And if a new bug is filed, that
means the old bug was fixed.

+
+
+def ScheduleAnalysisForFlake(request, user_email, is_admin):
   """Schedules an analysis on the flake in the given request if needed.
 
   Args:
@@ -16,5 +155,47 @@ def ScheduleAnalysisForFlake(
     An instance of MasterFlakeAnalysis if an analysis was scheduled; otherwise
     None if no analysis was scheduled before and the user has no permission to.
   """
-  # TODO (stgao): hook up with analysis.
+  assert len(request.build_steps) > 0, 'At least 1 build step is needed!'
+
+  # TODO (stgao): maybe move this to config.
+  whitelisted_accounts = [

[chanli, 2016/10/07 21:40:36]

Move this to constant for now?

[stgao, 2016/10/07 23:07:06]

Done.

+      'chromium-try-flakes@appspot.gserviceaccount.com',

[lijeffrey, 2016/10/07 21:22:47]

does findit-for-me@appspot.gserviceaccount.com count as admin too?

[stgao, 2016/10/07 23:07:06]

No. Admin means admin of the project.

+  ]
+
+  # Check whether the requester is authorized for access.
+  if not user_email or not (

[lijeffrey, 2016/10/07 20:45:07]

nit: I think this should be a separate function _IsUserAuthorized(user_email) or
similar

[stgao, 2016/10/07 23:07:06]

Done.

+      user_email.endswith('@google.com') or
+      user_email in whitelisted_accounts or is_admin):
+    return None
+  request.user_emails = [user_email]
+
+  manually_triggered = user_email.endswith('@google.com')
+
+  for build_step in request.build_steps:
+    step_mapper.FindMatchingWaterfallStep(build_step)
+
+  version_number, build_step = _NeedNewAnalysis(request)
+  if version_number and build_step:
+    # A new analysis is needed.
+    logging.info('A new analysis is needed for: %s', build_step)
+    analysis = initialize_flake_pipeline.ScheduleAnalysisIfNeeded(
+        build_step.wf_master_name, build_step.wf_builder_name,
+        build_step.wf_build_number, build_step.wf_step_name,
+        request.name, allow_new_analysis=True,
+        manually_triggered=manually_triggered,
+        queue_name=constants.WATERFALL_ANALYSIS_QUEUE)
+    if analysis:
+      # TODO: put this in a transaction.
+      request = FlakeAnalysisRequest.GetVersion(
+          key=request.name, version=version_number)
+      request.analyses.append(analysis.key)
+      request.put()
+      logging.info('A new analysis was triggered successfully: %s',
+                   analysis.key)
+      return True

[chanli, 2016/10/07 21:40:37]

Shouldn't return an instance of MasterFlakeAnalysis? According to docstring...

[stgao, 2016/10/07 23:07:06]

Updated docstring.

+    else:
+      logging.info('But no new analysis was not triggered!')
+  else:
+    logging.info('No new analysis is needed: %s', request)
+
   return False

[chanli, 2016/10/07 21:40:37]

Return None?

[stgao, 2016/10/07 23:07:06]

Acknowledged.