[Notes] [Git][BuildStream/buildstream][chandan/sourcetransform] Allow so

Chandan Singh pushed to branch chandan/sourcetransform at BuildStream / buildstream

Commits:

f5a2267e

by Chandan Singh at 2018-07-30T16:17:08Z

Allow source plugins to access previous sources

Source plugin implementations can now specify that they need access to
previously staged sources by specifying
`requires_previous_sources_track` and/or
`requires_previous_sources_fetch`, corresponding to access at `track`
and `fetch` times respectively.

Fixes #381.

Replaces !505. For relevant discussion, see this discussion:
https://gitlab.com/BuildStream/buildstream/merge_requests/505#note_83780747

9 changed files:

buildstream/_loader/loader.py
buildstream/_scheduler/queues/fetchqueue.py
buildstream/element.py
buildstream/source.py
+ tests/sources/previous_source_access.py
+ tests/sources/previous_source_access/elements/target.bst
+ tests/sources/previous_source_access/files/file
+ tests/sources/previous_source_access/plugins/sources/foo_transform.py
+ tests/sources/previous_source_access/project.conf

Changes:

buildstream/_loader/loader.py

@@ -506,14 +506,15 @@ class Loader():
          element = Element._new_from_meta(meta_element, platform.artifactcache)
          element._preflight()
 -        for source in element.sources():
 +        sources = list(element.sources())
 +        for idx, source in enumerate(sources):
              # Handle the case where a subproject needs to be fetched
+             #
              if source.get_consistency() == Consistency.RESOLVED:
                  if self._fetch_subprojects:
                      if ticker:
                          ticker(filename, 'Fetching subproject from {} source'.format(source.get_kind()))
 -                    source._fetch()
 +                    source._fetch(sources[0:idx])
                  else:
                      detail = "Try fetching the project with `bst fetch {}`".format(filename)
                      raise LoadError(LoadErrorReason.SUBPROJECT_FETCH_NEEDED,

buildstream/_scheduler/queues/fetchqueue.py

@@ -40,8 +40,10 @@ class FetchQueue(Queue):
          self._skip_cached = skip_cached
      def process(self, element):
 +        previous_sources = []
          for source in element.sources():
 -            source._fetch()
 +            source._fetch(previous_sources)
 +            previous_sources.append(source)
      def status(self, element):
          # state of dependencies may have changed, recalculate element state

buildstream/element.py

@@ -1195,6 +1195,11 @@ class Element(Plugin):
              # Prepend provenance to the error
              raise ElementError("{}: {}".format(self, e), reason=e.reason) from e
 +        # Ensure that the first source does not need access to previous soruces
 +        if self.__sources and self.__sources[0]._is_transform():
 +            raise ElementError("{}: A Source Transform plugin cannot be the first source"
 +                               .format(self))
++
          # Preflight the sources
          for source in self.sources():
              source._preflight()
@@ -1238,9 +1243,9 @@ class Element(Plugin):
+     #
      def _track(self):
          refs = []
 -        for source in self.__sources:
 +        for index, source in enumerate(self.__sources):
              old_ref = source.get_ref()
 -            new_ref = source._track()
 +            new_ref = source._track(self.__sources[0:index])
              refs.append((source._get_unique_id(), new_ref))
              # Complimentary warning that the new ref will be unused.

buildstream/source.py

@@ -156,7 +156,7 @@ class SourceFetcher():
      #############################################################
      #                      Abstract Methods                     #
      #############################################################
 -    def fetch(self, alias_override=None):
 +    def fetch(self, alias_override=None, **kwargs):
          """Fetch remote sources and mirror them locally, ensuring at least
          that the specific reference is cached locally.
@@ -209,6 +209,28 @@ class Source(Plugin):
      __defaults = {}          # The defaults from the project
      __defaults_set = False   # Flag, in case there are not defaults at all
 +    requires_previous_sources_track = False
 +    """Whether access to previous sources is required during track
++
 +    When set to True:
 +      * all sources listed before this source in the given element will be
 +        fetched before this source is tracked
 +      * Source.track() will be called with an additional keywork argument
 +        `previous_sources_dir` that will contain a list of sources
 +      * this source can not be the first source for an element
 +    """
++
 +    requires_previous_sources_fetch = False
 +    """Whether access to previous sources is required during fetch
++
 +    When set to True:
 +      * all sources listed before this source in the given element will be
 +        fetched before this source is fetched
 +      * Source.fetch() will be called with an additional keyword argument
 +        `previous_sources_dir` that will contain a list of sources
 +      * this source can not be the first source for an element
 +    """
++
      def __init__(self, context, project, meta, *, alias_override=None):
          provenance = _yaml.node_get_provenance(meta.config)
          super().__init__("{}-{}".format(meta.element_name, meta.element_index),
@@ -303,7 +325,7 @@ class Source(Plugin):
          """
          raise ImplError("Source plugin '{}' does not implement set_ref()".format(self.get_kind()))
 -    def track(self):
 +    def track(self, **kwargs):
          """Resolve a new ref from the plugin's track option
          Returns:
@@ -324,7 +346,7 @@ class Source(Plugin):
          # Allow a non implementation
          return None
 -    def fetch(self):
 +    def fetch(self, **kwargs):
          """Fetch remote sources and mirror them locally, ensuring at least
          that the specific reference is cached locally.
@@ -517,46 +539,59 @@ class Source(Plugin):
      # Wrapper function around plugin provided fetch method
+     #
 -    def _fetch(self):
 -        project = self._get_project()
 -        source_fetchers = self.get_source_fetchers()
 -        if source_fetchers:
 -            for fetcher in source_fetchers:
 -                alias = fetcher._get_alias()
 -                success = False
 +    # Args:
 +    #   previous_sources (list): List of Sources listed prior to this source
 +    #
 +    def _fetch(self, previous_sources):
 +        def __fetch(**kwargs):
 +            project = self._get_project()
 +            source_fetchers = self.get_source_fetchers()
 +            if source_fetchers:
 +                for fetcher in source_fetchers:
 +                    alias = fetcher._get_alias()
 +                    success = False
 +                    for uri in project.get_alias_uris(alias):
 +                        try:
 +                            fetcher.fetch(uri)
 +                        # FIXME: Need to consider temporary vs. permanent failures,
 +                        #        and how this works with retries.
 +                        except BstError as e:
 +                            last_error = e
 +                            continue
 +                        success = True
 +                        break
 +                    if not success:
 +                        raise last_error
 +            else:
 +                alias = self._get_alias()
 +                if not project.mirrors or not alias:
 +                    self.fetch(**kwargs)
 +                    return
++
 +                context = self._get_context()
 +                source_kind = type(self)
                  for uri in project.get_alias_uris(alias):
 +                    new_source = source_kind(context, project, self.__meta,
 +                                             alias_override=(alias, uri))
 +                    new_source._preflight()
                      try:
 -                        fetcher.fetch(uri)
 +                        new_source.fetch(**kwargs)
                      # FIXME: Need to consider temporary vs. permanent failures,
                      #        and how this works with retries.
                      except BstError as e:
                          last_error = e
                          continue
 -                    success = True
 -                    break
 -                if not success:
 -                    raise last_error
 +                    return
 +                raise last_error
++
 +        if self.requires_previous_sources_fetch:
 +            self.__ensure_previous_sources(previous_sources)
 +            with self.tempdir() as staging_directory:
 +                for src in previous_sources:
 +                    src._stage(staging_directory)
 +                __fetch(previous_sources_dir=staging_directory)
          else:
 -            alias = self._get_alias()
 -            if not project.mirrors or not alias:
 -                self.fetch()
 -                return
+-
 -            context = self._get_context()
 -            source_kind = type(self)
 -            for uri in project.get_alias_uris(alias):
 -                new_source = source_kind(context, project, self.__meta,
 -                                         alias_override=(alias, uri))
 -                new_source._preflight()
 -                try:
 -                    new_source.fetch()
 -                # FIXME: Need to consider temporary vs. permanent failures,
 -                #        and how this works with retries.
 -                except BstError as e:
 -                    last_error = e
 -                    continue
 -                return
 -            raise last_error
 +            __fetch()
      # Wrapper for stage() api which gives the source
      # plugin a fully constructed path considering the
@@ -762,8 +797,19 @@ class Source(Plugin):
      # Wrapper for track()
+     #
 -    def _track(self):
 -        new_ref = self.__do_track()
 +    # Args:
 +    #   previous_sources (list): List of Sources listed prior to this source
 +    #
 +    def _track(self, previous_sources):
 +        if self.requires_previous_sources_track:
 +            self.__ensure_previous_sources(previous_sources)
 +            with self.tempdir() as staging_directory:
 +                for src in previous_sources:
 +                    src._stage(staging_directory)
 +                new_ref = self.__do_track(previous_sources_dir=staging_directory)
 +        else:
 +            new_ref = self.__do_track()
++
          current_ref = self.get_ref()
          if new_ref is None:
@@ -775,6 +821,18 @@ class Source(Plugin):
          return new_ref
 +    # _is_transform()
 +    #
 +    # A plugin is considered a source transform plugin if it requires access to
 +    # previous sources for its tracking or fetching. Such sources cannot be the
 +    # first source of any element.
 +    #
 +    # Returns:
 +    #   (bool): Whetner this is a source transform plugin.
 +    #
 +    def _is_transform(self):
 +        return self.requires_previous_sources_track or self.requires_previous_sources_fetch
++
      # Returns the alias if it's defined in the project
      def _get_alias(self):
          alias = self.__expected_alias
@@ -791,12 +849,12 @@ class Source(Plugin):
      #############################################################
      # Tries to call track for every mirror, stopping once it succeeds
 -    def __do_track(self):
 +    def __do_track(self, **kwargs):
          project = self._get_project()
          # If there are no mirrors, or no aliases to replace, there's nothing to do here.
          alias = self._get_alias()
          if not project.mirrors or not alias:
 -            return self.track()
 +            return self.track(**kwargs)
          context = self._get_context()
          source_kind = type(self)
@@ -808,7 +866,7 @@ class Source(Plugin):
                                       alias_override=(alias, uri))
              new_source._preflight()
              try:
 -                ref = new_source.track()
 +                ref = new_source.track(**kwargs)
              # FIXME: Need to consider temporary vs. permanent failures,
              #        and how this works with retries.
              except BstError as e:
@@ -849,3 +907,14 @@ class Source(Plugin):
          _yaml.node_final_assertions(config)
          return config
++
 +    # Ensures that previous sources have been tracked and fetched.
 +    #
 +    def __ensure_previous_sources(self, previous_sources):
 +        for index, src in enumerate(previous_sources):
 +            if src.get_consistency() == Consistency.RESOLVED:
 +                src._fetch(previous_sources[0:index])
 +            elif src.get_consistency() == Consistency.INCONSISTENT:
 +                new_ref = src._track(previous_sources[0:index])
 +                src._save_ref(new_ref)
 +                src._fetch(previous_sources[0:index])

tests/sources/previous_source_access.py

 +import os
 +import pytest
++
 +from tests.testutils import cli
++
 +DATA_DIR = os.path.join(
 +    os.path.dirname(os.path.realpath(__file__)),
 +    'previous_source_access'
 +)
++
++
 +##################################################################
 +#                              Tests                             #
 +##################################################################
 +# Test that plugins can access data from previous sources
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_custom_transform_source(cli, tmpdir, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
++
 +    # Ensure we can track
 +    result = cli.run(project=project, args=[
 +        'track', 'target.bst'
 +    ])
 +    result.assert_success()
++
 +    # Ensure we can fetch
 +    result = cli.run(project=project, args=[
 +        'fetch', 'target.bst'
 +    ])
 +    result.assert_success()
++
 +    # Ensure we get correct output from foo_transform
 +    result = cli.run(project=project, args=[
 +        'build', 'target.bst'
 +    ])
 +    destpath = os.path.join(cli.directory, 'checkout')
 +    result = cli.run(project=project, args=[
 +        'checkout', 'target.bst', destpath
 +    ])
 +    result.assert_success()
 +    # Assert that files from both sources exist, and that they have
 +    # the same content
 +    assert os.path.exists(os.path.join(destpath, 'file'))
 +    assert os.path.exists(os.path.join(destpath, 'filetransform'))
 +    with open(os.path.join(destpath, 'file')) as file1:
 +        with open(os.path.join(destpath, 'filetransform')) as file2:
 +            assert file1.read() == file2.read()

tests/sources/previous_source_access/elements/target.bst

 +kind: import
++
 +sources:
 +- kind: local
 +  path: files/file
 +- kind: foo_transform

tests/sources/previous_source_access/files/file

+Hello World!

tests/sources/previous_source_access/plugins/sources/foo_transform.py

 +"""
 +foo_transform - transform "file" from previous sources into "filetransform"
 +===========================================================================
++
 +This is a test source plugin that looks for a file named "file" staged by
 +previous sources, and copies its contents to a file called "filetransform".
++
 +"""
++
 +import os
 +import hashlib
++
 +from buildstream import Consistency, Source, SourceError, utils
++
++
 +class FooTransformSource(Source):
++
 +    # We need access to previous both at track time and fetch time
 +    requires_previous_sources_track = True
 +    requires_previous_sources_fetch = True
++
 +    @property
 +    def mirror(self):
 +        """Directory where this source should stage its files
++
 +        """
 +        path = os.path.join(self.get_mirror_directory(), self.name,
 +                            self.ref.strip())
 +        os.makedirs(path, exist_ok=True)
 +        return path
++
 +    def configure(self, node):
 +        self.node_validate(node, ['ref'] + Source.COMMON_CONFIG_KEYS)
 +        self.ref = self.node_get_member(node, str, 'ref', None)
++
 +    def preflight(self):
 +        pass
++
 +    def get_unique_key(self):
 +        return (self.ref,)
++
 +    def get_consistency(self):
 +        if self.ref is None:
 +            return Consistency.INCONSISTENT
 +        # If we have a file called "filetransform", verify that its checksum
 +        # matches our ref. Otherwise, it resolved but not cached.
 +        fpath = os.path.join(self.mirror, 'filetransform')
 +        try:
 +            with open(fpath, 'rb') as f:
 +                if hashlib.sha256(f.read()).hexdigest() == self.ref.strip():
 +                    return Consistency.CACHED
 +        except Exception:
 +            pass
 +        return Consistency.RESOLVED
++
 +    def get_ref(self):
 +        return self.ref
++
 +    def set_ref(self, ref, node):
 +        self.ref = node['ref'] = ref
++
 +    def track(self, previous_sources_dir):
 +        # Store the checksum of the file from previous source as our ref
 +        fpath = os.path.join(previous_sources_dir, 'file')
 +        with open(fpath, 'rb') as f:
 +            return hashlib.sha256(f.read()).hexdigest()
++
 +    def fetch(self, previous_sources_dir):
 +        fpath = os.path.join(previous_sources_dir, 'file')
 +        # Verify that the checksum of the file from previous source matches
 +        # our ref
 +        with open(fpath, 'rb') as f:
 +            if hashlib.sha256(f.read()).hexdigest() != self.ref.strip():
 +                raise SourceError("Element references do not match")
++
 +        # Copy "file" as "filetransform"
 +        newfpath = os.path.join(self.mirror, 'filetransform')
 +        utils.safe_copy(fpath, newfpath)
++
 +    def stage(self, directory):
 +        # Simply stage the "filetransform" file
 +        utils.safe_copy(os.path.join(self.mirror, 'filetransform'),
 +                        os.path.join(directory, 'filetransform'))
++
++
 +def setup():
 +    return FooTransformSource

tests/sources/previous_source_access/project.conf

 +# Project with local source plugins
 +name: foo
++
 +element-path: elements
++
 +plugins:
 +- origin: local
 +  path: plugins/sources
 +  sources:
 +    foo_transform: 0

[Notes] [Git][BuildStream/buildstream][chandan/sourcetransform] Allow source plugins to access previous sources

Chandan Singh pushed to branch chandan/sourcetransform at BuildStream / buildstream

Commits:

9 changed files:

Changes: