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
-# ``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 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.
+# Some notes about the classes below, for people who are not familiar
+# with AppEngine pipelines:
+#
+# The pipeline library is designed in a strange way which requires that

[stgao, 2016/11/03 04:48:10]

It might be just me, but "strange" seems a little strong in wording.

IMHO, the design of the appengine pipeline looks good. It might just not expect
users to create a constructor in subclasses like what we do here.
The only reason we need to add the constructor here is because we want to update
an NDB entity in the ``finalized`` function to deal with errors.

[wrengr, 2016/11/03 17:40:34]

I've done a lot of software engineering on a lot of different projects in a lot
of different programming languages, and this set of constraints is extremely
peculiar. I can think of no other projects that have similar design
restrictions. Given that this design is both counterintuitive and entirely
unique to someone with over a decade of engineering experience, I think it's
fair to remark on it being "strange".
(a) Not allowing/expecting subclasses to define constructors is assuredly
"strange"; indeed, it deserves far stronger words than that. (b) If they don't
want us doing that, then maybe we shouldn't be doing it.

[stgao, 2016/11/03 18:48:30]

I would say "special" instead. But it's fine as is. :)
Maybe we should look into alternative solution and avoid adding the constructor.

[wrengr, 2016/11/03 20:04:21]

filed http://crbug.com/662145

+# all the ``__init__`` and ``run`` methods in this file take the exact
+# same arguments. This arises from the interaction between a few
+# different constraints. First, for any given pipeline, its ``__init__``
+# and ``run`` must take the same arguments. Second, all the objects that
+# ``CrashWrapperPipeline.run`` yields must take the same arguments as that
+# ``run`` method itself. For more about all this, see:

[stgao, 2016/11/03 04:48:10]

I don't understand why the yielded objects must take the same arguments as the
run itself.

We have examples that is against this statement.
https://chromium.googlesource.com/infra/infra/+/master/appengine/findit/water...

The official documentation also has similar example
https://github.com/GoogleCloudPlatform/appengine-pipelines/wiki/Python#defini...

[wrengr, 2016/11/03 17:40:34]

I have no idea. This is what katesonia has said repeatedly, and she's showed me
crash logs for the version where ``run`` accepts no arguments, so I believe her.

[stgao, 2016/11/03 18:17:37]

Sharu, would you mind clarifying here? I'm confused.

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

The comment "``CrashWrapperPipeline.run`` yields must take the same arguments as
that ``run`` method itself is not correct and not what I meant, it should be
updated to yield CrashWrapperPipeline(*args, **kwargs), *args and **kwargs
should be the same as CrashWrapperPipeline.run", will update this in following
which mocks CrashWrapperPipeline.

+# https://github.com/GoogleCloudPlatform/appengine-pipelines/wiki/Python
+# For our use case, we pass all the important data to ``__init__``,
+# since we need it to be available in ``finalized``; thus we ignore the
+# arguments passed to ``run`` (which will be the same values that were
+# passed to ``__init__``).
+#
+# In addition, whatever objects ``CrashWrapperPipeline.run`` yields must
+# be JSON-serializable. The pipeline class handles most of the details,

[stgao, 2016/11/03 04:48:10]

For the yielded objects by ``CrashWrapperPipeline.run``, they just must be an
instance of Pipeline; for the __return__ objects by the function, they must be
JSON-serializable.

[wrengr, 2016/11/03 17:40:34]

The bug I encountered previously about JSON-serializability was because of
passing a Findit object to the CrashAnalysisPipeline and/or
CrashPublishPipeline. Replacing that object with a string fixed the problem;
what any of the methods return was unaltered. This also agrees with what you say
in the start of the next comment...

[stgao, 2016/11/03 18:48:30]

This is true.

But my comment above is for "In addition, whatever objects
``CrashWrapperPipeline.run`` yields must be JSON-serializable." which is
misleading to me.

[wrengr, 2016/11/03 20:04:21]

The issue I encountered was that ``yield SomePipeline(stuff)`` will eventually
lead to a crash if ``stuff`` is not JSON-serializable. Perhaps I've misdiagnosed
why exactly that crashes things. Maybe the problem isn't passing ``stuff`` to
``SomePipeline.__init__`` but rather the fact that ``stuff`` will also be passed
to ``SomePipeline.run`` because of what the coroutine does with the yielded
SomePipeline; I dunno.

If you know of a document that explains exactly what the constraints are and why
that example crashes, I'll update the discussion here to point to them.

[stgao, 2016/11/03 21:07:07]

The problem most likely is due to passing ``stuff`` to
``SomePipeline.__init__``, and it could be verified by checking whether the
crash occurs when ``SomePipelineInstance.start`` is called. Without a successful
``start`` (parameters are saved to NDB and a taskqueue task is queued), ``run``
won't be executed.

However, it has nothing to do with __yield__.

Both of the code below should fail:

  def some_caller_func_in_app_layer(...):
     instance = SomePipeline(not-JSON-serializable-parameter)
     instance.start()

and

  class RootPipeline(pipeline.Pipeline):
     def run(...):
        yield SomePipeline(not-JSON-serializable-parameter)
I don't know the exact doc on the constraints, but
https://github.com/GoogleCloudPlatform/appengine-pipelines/wiki/Python is the
best doc to check.

+# so the force of this constraint is that whatever arguments we pass

[stgao, 2016/11/03 04:48:10]

It is true that parameters passed over to the constructor of Pipeline must be
JSON-serializable, but I don't see a cause-result relation between this and the
constraint of returned values by Pipeline.run though.

[wrengr, 2016/11/03 17:40:34]

Which agrees with what I wrote...
I don't say anything about the return values. (Because CrashAnalysisPipeline.run
and CrashPublishPipeline don't actually return anything, so it doesn't matter.)

[stgao, 2016/11/03 18:48:30]

Sorry for the typo again. I meant to say __yielded__ objects.

In my understanding, the constraint for the parameters to the constructor and
the constraints for the __yielded__ objects or __returned__ values by ``run``
are independent from each other, while the constraints are the same for
parameters and __returned__ values -- must be JSON-serializable. (__yielded__
objects are different, they must be instances of Pipeline.)

But from reading the words here, it's more like a cause-result relation.

+# to their ``__init__`` must themselves be JSON-serializable. Alas,
+# in Python, JSON-serializability isn't a property of classes themselves,
+# but rather a property of the JSON-encoder object used to do the
+# serialization. Thus, we cannot pass a ``Findit`` object directly to
+# any of the methods here, but rather must instead pass the ``client_id``
+# (or whatever JSON dict), and then reconstruct the ``Findit`` object
+# from that data.
+#
+# Moreover, the ``run`` and ``finalized`` methods are executed in separate

[stgao, 2016/11/03 04:48:10]

Actually, this is one of the goodies from appengine pipeline. A pipeline is
executed in a few phases so that we could do different things in different
phases: in ``run``, we could just deal with the main logic; in ``finalized``, we
could deal with the abortion due to too many retries or other errors.

Of course, it doesn't introduce the problem as described here.

[wrengr, 2016/11/03 17:40:34]

Does it not? How do changes from inside ``run`` get propagated to the
newly-constructed object on which ``finalized`` is called? If it does actually
work I'll remove the comment.

[stgao, 2016/11/03 18:17:37]

Sorry for the typo. I meant to say it __does__ introduce the problem here.

[wrengr, 2016/11/03 18:20:20]

Cool, I understood it right :)

+# processes, so we'll actually end up reconstructing the ``Findit`` object
+# twice. This also means ``run`` can't store anything in the pipeline
+# object and expect it to still be available in the ``finalized`` method.
 
 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):
+    """Call predator to do the analysis of the given crash.
+
+    N.B., due to the structure of AppEngine pipelines, this method must
+    accept the same arguments as are passed to ``__init__``; however,
+    because they were already passed to ``__init__`` there's no use in
+    recieving them here. Thus, we discard all the arguments to this method
+    (except for ``self``, naturally).
+    """
     # 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
+    """Publish the results of our analysis back into the ndb.Model.
+
+    N.B., due to the structure of AppEngine pipelines, this method must
+    accept the same arguments as are passed to ``__init__``; however,
+    because they were already passed to ``__init__`` there's no use in
+    recieving them here. Thus, we discard all the arguments to this method
+    (except for ``self``, naturally).
+    """
     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):
+    """Fire off pipelines to run the analysis and publish its results.
+
+    N.B., due to the structure of AppEngine pipelines, this method must
+    accept the same arguments as are passed to ``__init__``; however,
+    because they were already passed to ``__init__`` there's no use in
+    recieving them here. Thus, we discard all the arguments to this method
+    (except for ``self``, naturally).
+    """
     run_analysis = yield CrashAnalysisPipeline(
         self._client_id, self._crash_identifiers)
     with pipeline.After(run_analysis):
       yield PublishResultPipeline(self._client_id, self._crash_identifiers)