Moving ScheduleNewAnalysis to break the cycle

appengine/findit/crash/crash_pipeline.py

 # Copyright 2016 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.
 
 import copy
 import json
 import logging
 
 from common import appengine_util
 from common import pubsub_util
 from common import time_util
 from common.http_client_appengine import HttpClientAppengine
 from common.pipeline_wrapper import BasePipeline
 from common.pipeline_wrapper import pipeline
 from crash import findit_for_chromecrash
 from crash import findit_for_clusterfuzz
 from crash.type_enums import CrashClient
 from lib.gitiles import gitiles_repository
 from model import analysis_status
 
 
-# TODO(http://crbug.com/659346): write complete coverage tests for this.
-def FinditForClientID(client_id): # pragma: no cover
+# TODO(http://crbug.com/659346): this needs complete coverage tests.
+def FinditForClientID(client_id):
   """Construct a Findit object from a client id string specifying the class.
 
   We cannot pass Findit objects to the various methods in
   ``crash.crash_pipeline``, because they are not JSON serializable. For
   now, we just serialize Findit objects as their ``client_id``, and then
   use this function to reconstruct them. Alas, this means we will lose
   various other information stored in the Findit object (i.e., stuff that
   comes from CrashConfig); which could lead to some hard-to-diagnose
   coherency bugs, since the new Findit object will be based on the
   CrashConfig at the time it's constructed, which may be different
   than the CrashConfig at the time the previous Findit object was
   constructed. In the future we should fix all this to serialize Findit
   objects in a more robust way.
   """
   assert isinstance(client_id, (str, unicode)), (
       'FinditForClientID: expected string or unicode, but got %s'
       % client_id.__class__.__name__)
   # TODO(wrengr): it'd be nice to replace this with a single lookup in
   # a dict; but that's buggy for some unknown reason.
   if client_id == CrashClient.FRACAS:
     cls = findit_for_chromecrash.FinditForFracas
-  elif client_id == CrashClient.CRACAS:
+  elif client_id == CrashClient.CRACAS: # pragma: no cover
     cls = findit_for_chromecrash.FinditForCracas
-  elif client_id == CrashClient.CLUSTERFUZZ:
+  elif client_id == CrashClient.CLUSTERFUZZ: # pragma: no cover
     cls = findit_for_clusterfuzz.FinditForClusterfuzz
-  else:
+  else: # pragma: no cover
     raise ValueError('FinditForClientID: '
         'unknown or unsupported client %s' % client_id)
 
-  return cls(
-      gitiles_repository.GitilesRepository(http_client=HttpClientAppengine()),
-      CrashWrapperPipeline)
+  return cls(gitiles_repository.GitilesRepository(
+      http_client=HttpClientAppengine()))
 
 
 # Some notes about the classes below, for people who are not
 # familiar with AppEngine. The thing that really kicks everything off
 # is ``CrashWrapperPipeline.run``. However, an important thing to bear in
 # mind is that whatever arguments are passed to that method will also
 # be passed to the ``run`` method on whatever objects it yields. Thus,
 # all the ``run`` methods across these different classes must have
 # the same type. In practice, we end up passing all the arguments to the
 # constructors, because we need to have the fields around for logging
 # (e.g., in ``finalized``); thus, there's nothing that needs to be passed
-# to ``run``. Another thing to bear in mind is that whatever objects
+# to ``run``. However, for some inscrutable reason, whatever arguments
+# are passed to the constructors will also be passed to ``run`` (it's
+# unclear what sort of terrible introspection AppEngine does in order to

[stgao, 2016/11/02 01:28:27]

It is not AppEngine, but the AppEngine Pipeline framework
https://github.com/GoogleCloudPlatform/appengine-pipelines/wiki/Python

That's by design in the AppEngine Pipeline. One reason I could think of is that
users of the framework don't have to create constructor unless other functions
like ``finalized`` also need to access those parameters.

[wrengr, 2016/11/02 23:52:26]

Acknowledged.

+# figure this out, but no matter); thus, even though the ``run`` methods
+# don't need them (and don't want them) they must still accept a bunch
+# of arguments. Another thing to bear in mind is that whatever objects
 # ``CrashWrapperPipeline.run`` yields must be JSON-serializable. The base
 # class handles most of that for us, so the force of this constraint is
 # that all the arguments to the constructors for those classes must be
-# JSON-serializable. Thus, we cannot actually pass a Findit object to
-# the constructor, but rather must pass only the ``client_id`` (or whatever
+# JSON-serializable. Thus, we cannot actually pass a Findit object to the
+# constructor, but rather must pass only the ``client_id`` (or whatever
 # JSON dict) and then reconstruct the Findit object from that. Moreover,
 # the ``run`` method and the ``finalized`` method will be run in different
 # processes, so we will actually end up reconstructing the Findit object
 # twice. Thus, we shouldn't store anything in the pipeline objects outside
 # of what their constructors store.
 
 class CrashBasePipeline(BasePipeline):
   def __init__(self, client_id, crash_identifiers):
     super(CrashBasePipeline, self).__init__(client_id, crash_identifiers)
     self._crash_identifiers = crash_identifiers
     self._findit = FinditForClientID(client_id)
 
   @property
   def client_id(self): # pragma: no cover
     return self._findit.client_id
 
   def run(self, *args, **kwargs):
     raise NotImplementedError()
 
 
 class CrashAnalysisPipeline(CrashBasePipeline):
-  def finalized(self): # pragma: no cover
-    if self.was_aborted:
+  def finalized(self):
+    if self.was_aborted: # pragma: no cover
       self._PutAbortedError()
 
   # N.B., this method must be factored out for unittest reasons; since
   # ``finalized`` takes no arguments (by AppEngine's spec) and
   # ``was_aborted`` can't be altered directly.
   def _PutAbortedError(self):
     """Update the ndb.Model to indicate that this pipeline was aborted."""
     logging.error('Aborted analysis for %s', repr(self._crash_identifiers))
     analysis = self._findit.GetAnalysis(self._crash_identifiers)
     analysis.status = analysis_status.ERROR
     analysis.put()
 
-  # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
-  # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self):
+  def run(self, *_args, **_kwargs):
     # TODO(wrengr): shouldn't this method somehow call _NeedsNewAnalysis
     # to guard against race conditions?
     analysis = self._findit.GetAnalysis(self._crash_identifiers)
 
     # Update the model's status to say we're in the process of doing analysis.
     analysis.pipeline_status_path = self.pipeline_status_path()
     analysis.status = analysis_status.RUNNING
     analysis.started_time = time_util.GetUTCNow()
     analysis.findit_version = appengine_util.GetCurrentVersion()
     analysis.put()
@@ skipping 10 matching lines @@
           'found_components': False,
           'has_regression_range': False,
           'solution': None,
       }
 
     # Update model's status to say we're done, and save the results.
     analysis.completed_time = time_util.GetUTCNow()
     analysis.result = result
     for tag_name, tag_value in tags.iteritems():
       # TODO(http://crbug.com/602702): make it possible to add arbitrary tags.
-      if hasattr(analysis, tag_name):
+      # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
+      if hasattr(analysis, tag_name): # pragma: no cover
         setattr(analysis, tag_name, tag_value)
     analysis.status = analysis_status.COMPLETED
     analysis.put()
 
 
 class PublishResultPipeline(CrashBasePipeline):
-  # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
   def finalized(self):
-    if self.was_aborted:  # pragma: no cover.
+    if self.was_aborted: # pragma: no cover.
       logging.error('Failed to publish %s analysis result for %s',
                     repr(self._crash_identifiers), self.client_id)
 
-  # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self):
+  # TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
+  def run(self, *_args, **_kwargs): # pragma: no cover

[Sharu Jiang, 2016/11/01 23:03:24]

This is a little hacky, we should at lease add doc saying we will throw out
parameters because of the yield statement. 

[stgao, 2016/11/02 01:28:27]

I don't understand this.
Sharu, do you mind elaborating a little bit? Which yield? Why the yield impacts
here?

[Sharu Jiang, 2016/11/02 17:01:35]

like the `yield` in #190 and #193
What `yield` does is to construct a pipeline and pass whatever parameters of the
construct function to run(*_args, **_kwargs). This way we can force the *_args,
**_kwargs to be the same as in construct function.

However, if we do:
pipe = PublishResultPipeline(client1, crash_identifiers1)
pipe.run(client2, crash_identifiers2, ...)

The pipe will run using (client1, crash_identifiers1) instead of using the
(client2, crash_identifiers2) passed into `run`. I think this may cause
confusion, so if we are to do something like #189-#193, we should at lease add
doc to notify users to use `yield` instead of creating a pipeline and call its
run separately.

[stgao, 2016/11/02 21:22:09]

Would you mind pointing me to the doc for the usage of calling ``run`` directly?
It's new to me and I'd like to learn more.

[wrengr, 2016/11/02 23:52:26]

I agree it's a terrible hack, but it's forced on us by the structure of
AppEngine pipelines. I'll add docstrings to reiterate the discussion from lines
57--81.

[Sharu Jiang, 2016/11/03 23:03:25]

Sorry for the late response, was busy working on my several pending cls, the doc
is in here:
https://sookocheff.com/post/appengine/pipelines/connecting-pipelines/

Saying: "The rule of thumb here is that anything you instantiate your pipeline
with (and subsequently pass to the run method) is accessible within your
pipeline. These are called immediate values and you can treat them as regular
Python values."

[Sharu Jiang, 2016/11/04 00:19:00]

Calling ``run`` directly is not a correct usage for a pipeline ( thus, we cannot
find an example in a doc), so that's why I was suggesting to add that note in
the doc to avoid that. However, if that's very obvious to all pipeline users,
then it's ok to not to bother adding that note in the doc :)

     analysis = self._findit.GetAnalysis(self._crash_identifiers)
     result = analysis.ToPublishableResult(self._crash_identifiers)
     messages_data = [json.dumps(result, sort_keys=True)]
 
     # TODO(http://crbug.com/659354): remove Findit's dependency on CrashConfig.
     client_config = self._findit.config
     # TODO(katesonia): Clean string uses in config.
     topic = client_config['analysis_result_pubsub_topic']
     pubsub_util.PublishMessagesToTopic(messages_data, topic)
     logging.info('Published %s analysis result for %s', self.client_id,
                  repr(self._crash_identifiers))
 
 
-class CrashWrapperPipeline(BasePipeline):
+# TODO(http://crbug.com/659346): we misplaced the coverage test; find it!
+class CrashWrapperPipeline(BasePipeline): # pragma: no cover
   """Fire off pipelines to (1) do the analysis and (2) publish results.
 
   The reason we have analysis and publishing as separate pipelines is
   because each of them can fail for independent reasons. E.g., if we
   successfully finish the analysis, but then the publishing fails due to
   network errors, we don't want to have to redo the analysis in order
   to redo the publishing. We could try to cache the fact that analysis
   succeeded in the pipeline object itself, but we'd have to be careful
   because the ``run`` and ``finalized`` methods are executed in different
   processes.
   """
   def __init__(self, client_id, crash_identifiers):
     super(CrashWrapperPipeline, self).__init__(client_id, crash_identifiers)
     self._crash_identifiers = crash_identifiers
     self._client_id = client_id
 
-  # TODO(http://crbug.com/659346): write coverage tests.
-  # Arguments number differs from overridden method - pylint: disable=W0221
-  def run(self): # pragma: no cover
+  def run(self, *_args, **_kwargs):
     run_analysis = yield CrashAnalysisPipeline(
         self._client_id, self._crash_identifiers)
     with pipeline.After(run_analysis):
       yield PublishResultPipeline(self._client_id, self._crash_identifiers)