[Notes] [Git][BuildStream/buildstream][tpollard/494] Don't pull artifact

Tom Pollard pushed to branch tpollard/494 at BuildStream / buildstream

Commits:

b690d1f9

by Tom Pollard at 2018-10-23T17:09:26Z

Don't pull artifact buildtrees by default.

The addition of cached buildtrees being included in element
artifacts has led to mostly redundant download overheads when
pulling from a remote artifact server. As such the default behaviour
of pull shouldn't fetch the buildtree object if available.

_stream.py: propagate if pullbuildtrees has been set via cli or
user config to the PullQueue.

pullqueue.py: add extensible attributes to determine required/
excluded directories for element pull jobs.

element.py: extend relevant pull logic for specified subdir
consideration and ensure push logic does not lead to partial
artifact pushes.

_artifactcache/: artifactcache.py & cascache.py inclusion of
helper functions for subdir artifact checking & fetching, fetch
logic extended to only pull required artifact directories.
extract & checkout updated to handle full/partial operation.

tests/: addition of integration test pullbuildtrees.py,
testutils/artifactshare.py has_artifact changed to return
artifact digest on true condition.

8 changed files:

NEWS
buildstream/_artifactcache/artifactcache.py
buildstream/_artifactcache/cascache.py
buildstream/_scheduler/queues/pullqueue.py
buildstream/_stream.py
buildstream/element.py
+ tests/integration/pullbuildtrees.py
tests/testutils/artifactshare.py

Changes:

NEWS

@@ -31,6 +31,17 @@ buildstream 1.3.1
      new the `conf-root` variable to make the process easier. And there has been
      a bug fix to workspaces so they can be build in workspaces too.
 +  o Due to the element `buildtree` being cached in the respective artifact their
 +    size in some cases has significantly increased. In *most* cases the buildtree
 +    is not utilised when building targets, as such by default bst 'pull' & 'build'
 +    will not fetch buildtrees from remotes. This behaviour can be overriden with
 +    the cli option '--pull-buildtrees', or the user configuration option
 +    'pullbuildtrees = True'. The override will also add the buildtree to already
 +    cached artifacts. When attempting to populate an artifactcache server with
 +    cached artifacts, only 'complete' elements can be pushed. If the element is
 +    expected to have a populated buildtree then it must be cached before pushing.
++
++
  =================
  buildstream 1.1.5
  =================

buildstream/_artifactcache/artifactcache.py

@@ -428,6 +428,22 @@ class ArtifactCache():
          raise ImplError("Cache '{kind}' does not implement contains()"
                          .format(kind=type(self).__name__))
 +    # contains_subdir_artifact():
 +    #
 +    # Check whether an artifact element contains a digest for a subdir
 +    # which is populated in the cache, i.e non dangling.
 +    #
 +    # Args:
 +    #     element (Element): The Element to check
 +    #     key (str): The cache key to use
 +    #     subdir (str): The subdir to check
 +    #
 +    # Returns: True if the subdir exists & is populated in the cache, False otherwise
 +    #
 +    def contains_subdir_artifact(self, element, key, subdir):
 +        raise ImplError("Cache '{kind}' does not implement contains_subdir_artifact()"
 +                        .format(kind=type(self).__name__))
++
      # list_artifacts():
+     #
      # List artifacts in this cache in LRU order.
@@ -464,6 +480,7 @@ class ArtifactCache():
      # Args:
      #     element (Element): The Element to extract
      #     key (str): The cache key to use
 +    #     subdir (str): Optional specific dir to extract
+     #
      # Raises:
      #     ArtifactError: In cases there was an OSError, or if the artifact
@@ -471,7 +488,7 @@ class ArtifactCache():
+     #
      # Returns: path to extracted artifact
+     #
 -    def extract(self, element, key):
 +    def extract(self, element, key, subdir=None):
          raise ImplError("Cache '{kind}' does not implement extract()"
                          .format(kind=type(self).__name__))
@@ -553,11 +570,13 @@ class ArtifactCache():
      #     element (Element): The Element whose artifact is to be fetched
      #     key (str): The cache key to use
      #     progress (callable): The progress callback, if any
 +    #     subdir (str): The optional specific subdir to pull
 +    #     excluded_subdirs (list): The optional list of subdirs to not pull
+     #
      # Returns:
      #   (bool): True if pull was successful, False if artifact was not available
+     #
 -    def pull(self, element, key, *, progress=None):
 +    def pull(self, element, key, *, progress=None, subdir=None, excluded_subdirs=None):
          raise ImplError("Cache '{kind}' does not implement pull()"
                          .format(kind=type(self).__name__))

buildstream/_artifactcache/cascache.py

@@ -92,16 +92,37 @@ class CASCache(ArtifactCache):
          # This assumes that the repository doesn't have any dangling pointers
          return os.path.exists(refpath)
 -    def extract(self, element, key):
 +    def contains_subdir_artifact(self, element, key, subdir):
 +        tree = self.resolve_ref(self.get_artifact_fullname(element, key))
++
 +        # This assumes that the subdir digest is present in the element tree
 +        subdirdigest = self._get_subdir(tree, subdir)
 +        objpath = self.objpath(subdirdigest)
++
 +        # True if subdir content is cached or if empty as expected
 +        return os.path.exists(objpath)
++
 +    def extract(self, element, key, subdir=None):
          ref = self.get_artifact_fullname(element, key)
          tree = self.resolve_ref(ref, update_mtime=True)
          dest = os.path.join(self.extractdir, element._get_project().name,
                              element.normal_name, tree.hash)
++
++
 +        # If artifact is already extracted, check if the optional subdir
 +        # has also been extracted. If the artifact has not been extracted
 +        # assume that a full extraction would include the optional subdir
          if os.path.isdir(dest):
 -            # artifact has already been extracted
 -            return dest
 +            if subdir:
 +                if not os.path.isdir(os.path.join(dest, subdir)):
 +                    dest = os.path.join(dest, subdir)
 +                    tree = self._get_subdir(tree, subdir)
 +                else:
 +                    return dest
 +            else:
 +                return dest
          with tempfile.TemporaryDirectory(prefix='tmp', dir=self.extractdir) as tmpdir:
              checkoutdir = os.path.join(tmpdir, ref)
@@ -228,7 +249,7 @@ class CASCache(ArtifactCache):
              remotes_for_project = self._remotes[element._get_project()]
              return any(remote.spec.push for remote in remotes_for_project)
 -    def pull(self, element, key, *, progress=None):
 +    def pull(self, element, key, *, progress=None, subdir=None, excluded_subdirs=None):
          ref = self.get_artifact_fullname(element, key)
          project = element._get_project()
@@ -247,8 +268,18 @@ class CASCache(ArtifactCache):
                  tree.hash = response.digest.hash
                  tree.size_bytes = response.digest.size_bytes
 -                self._fetch_directory(remote, tree)
 +                # Check if the element artifact is present, if so just fetch the subdir
 +                # and extract into artifact extract dir if it already exists. If the
 +                # respective artifact extract dir does not exist a complete extraction
 +                # will complete
 +                if subdir and os.path.exists(self.objpath(tree)):
 +                    self._fetch_subdir(remote, tree, subdir)
 +                    self.extract(element, key, subdir)
 +                else:
 +                    # Fetch artifact, excluded_subdirs determined in pullqueue
 +                    self._fetch_directory(remote, tree, excluded_subdirs=excluded_subdirs)
 +                # tree is the remote value, so is the same without or without dangling ref locally
                  self.set_ref(ref, tree)
                  element.info("Pulled artifact {} <- {}".format(display_key, remote.spec.url))
@@ -672,8 +703,10 @@ class CASCache(ArtifactCache):
                           stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH)
          for dirnode in directory.directories:
 -            fullpath = os.path.join(dest, dirnode.name)
 -            self._checkout(fullpath, dirnode.digest)
 +            # Don't try to checkout a dangling ref
 +            if os.path.exists(self.objpath(dirnode.digest)):
 +                fullpath = os.path.join(dest, dirnode.name)
 +                self._checkout(fullpath, dirnode.digest)
          for symlinknode in directory.symlinks:
              # symlink
@@ -951,11 +984,14 @@ class CASCache(ArtifactCache):
      # Args:
      #     remote (Remote): The remote to use.
      #     dir_digest (Digest): Digest object for the directory to fetch.
 +    #     excluded_subdirs (list): The optional list of subdirs to not fetch
+     #
 -    def _fetch_directory(self, remote, dir_digest):
 +    def _fetch_directory(self, remote, dir_digest, *, excluded_subdirs=None):
          fetch_queue = [dir_digest]
          fetch_next_queue = []
          batch = _CASBatchRead(remote)
 +        if not excluded_subdirs:
 +            excluded_subdirs = []
          while len(fetch_queue) + len(fetch_next_queue) > 0:
              if len(fetch_queue) == 0:
@@ -970,8 +1006,9 @@ class CASCache(ArtifactCache):
                  directory.ParseFromString(f.read())
              for dirnode in directory.directories:
 -                batch = self._fetch_directory_node(remote, dirnode.digest, batch,
 -                                                   fetch_queue, fetch_next_queue, recursive=True)
 +                if dirnode.name not in excluded_subdirs:
 +                    batch = self._fetch_directory_node(remote, dirnode.digest, batch,
 +                                                       fetch_queue, fetch_next_queue, recursive=True)
              for filenode in directory.files:
                  batch = self._fetch_directory_node(remote, filenode.digest, batch,
@@ -980,6 +1017,10 @@ class CASCache(ArtifactCache):
          # Fetch final batch
          self._fetch_directory_batch(remote, batch, fetch_queue, fetch_next_queue)
 +    def _fetch_subdir(self, remote, tree, subdir):
 +        subdirdigest = self._get_subdir(tree, subdir)
 +        self._fetch_directory(remote, subdirdigest)
++
      def _fetch_tree(self, remote, digest):
          # download but do not store the Tree object
          with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:

buildstream/_scheduler/queues/pullqueue.py

@@ -32,9 +32,20 @@ class PullQueue(Queue):
      complete_name = "Pulled"
      resources = [ResourceType.DOWNLOAD, ResourceType.CACHE]
 +    def __init__(self, scheduler, buildtrees=False):
 +        super().__init__(scheduler)
++
 +        # Current default exclusions on pull
 +        self._excluded_subdirs = ["buildtree"]
 +        self._subdir = None
 +        # If buildtrees are to be pulled, remove the value from exclusion list
 +        if buildtrees:
 +            self._subdir = "buildtree"
 +            self._excluded_subdirs.remove(self._subdir)
++
      def process(self, element):
          # returns whether an artifact was downloaded or not
 -        if not element._pull():
 +        if not element._pull(subdir=self._subdir, excluded_subdirs=self._excluded_subdirs):
              raise SkipJob(self.action_name)
      def status(self, element):
@@ -49,7 +60,7 @@ class PullQueue(Queue):
          if not element._can_query_cache():
              return QueueStatus.WAIT
 -        if element._pull_pending():
 +        if element._pull_pending(subdir=self._subdir):
              return QueueStatus.READY
          else:
              return QueueStatus.SKIP

buildstream/_stream.py

@@ -160,6 +160,7 @@ class Stream():
      #    track_cross_junctions (bool): Whether tracking should cross junction boundaries
      #    build_all (bool): Whether to build all elements, or only those
      #                      which are required to build the target.
 +    #    pull_buildtrees (bool): Whether to pull buildtrees from a remote cache server
+     #
      def build(self, targets, *,
                track_targets=None,
@@ -196,7 +197,10 @@ class Stream():
              self._add_queue(track_queue, track=True)
          if self._artifacts.has_fetch_remotes():
 -            self._add_queue(PullQueue(self._scheduler))
 +            # Query if pullbuildtrees has been set globally in user config
 +            if self._context.pullbuildtrees:
 +                pull_buildtrees = True
 +            self._add_queue(PullQueue(self._scheduler, buildtrees=pull_buildtrees))
          self._add_queue(FetchQueue(self._scheduler, skip_cached=True))
          self._add_queue(BuildQueue(self._scheduler))
@@ -312,8 +316,12 @@ class Stream():
          if not self._artifacts.has_fetch_remotes():
              raise StreamError("No artifact caches available for pulling artifacts")
 +        # Query if pullbuildtrees has been set globally in user config
 +        if self._context.pullbuildtrees:
 +            pull_buildtrees = True
++
          self._pipeline.assert_consistent(elements)
 -        self._add_queue(PullQueue(self._scheduler))
 +        self._add_queue(PullQueue(self._scheduler, buildtrees=pull_buildtrees))
          self._enqueue_plan(elements)
          self._run()

buildstream/element.py

@@ -1693,18 +1693,26 @@ class Element(Plugin):
      # _pull_pending()
+     #
 -    # Check whether the artifact will be pulled.
 +    # Check whether the artifact will be pulled. If the pull operation is to
 +    # include a specific subdir of the element artifact (from cli or user conf)
 +    # then the local cache is queried for the subdirs existence.
 +    #
 +    # Args:
 +    #    subdir (str): Whether the pull has been invoked with a specific subdir set
+     #
      # Returns:
      #   (bool): Whether a pull operation is pending
+     #
 -    def _pull_pending(self):
 +    def _pull_pending(self, subdir=None):
          if self._get_workspace():
              # Workspace builds are never pushed to artifact servers
              return False
 -        if self.__strong_cached:
 -            # Artifact already in local cache
 +        if self.__strong_cached and subdir:
 +            # If we've specified a subdir, check if the subdir is cached locally
 +            if self.__artifacts.contains_subdir_artifact(self, self.__strict_cache_key, subdir):
 +                return False
 +        elif self.__strong_cached:
              return False
          # Pull is pending if artifact remote server available
@@ -1726,11 +1734,10 @@ class Element(Plugin):
          self._update_state()
 -    def _pull_strong(self, *, progress=None):
 +    def _pull_strong(self, *, progress=None, subdir=None, excluded_subdirs=None):
          weak_key = self._get_cache_key(strength=_KeyStrength.WEAK)
+-
          key = self.__strict_cache_key
 -        if not self.__artifacts.pull(self, key, progress=progress):
 +        if not self.__artifacts.pull(self, key, progress=progress, subdir=subdir, excluded_subdirs=excluded_subdirs):
              return False
          # update weak ref by pointing it to this newly fetched artifact
@@ -1738,10 +1745,10 @@ class Element(Plugin):
          return True
 -    def _pull_weak(self, *, progress=None):
 +    def _pull_weak(self, *, progress=None, subdir=None, excluded_subdirs=None):
          weak_key = self._get_cache_key(strength=_KeyStrength.WEAK)
+-
 -        if not self.__artifacts.pull(self, weak_key, progress=progress):
 +        if not self.__artifacts.pull(self, weak_key, progress=progress, subdir=subdir,
 +                                     excluded_subdirs=excluded_subdirs):
              return False
          # extract strong cache key from this newly fetched artifact
@@ -1759,17 +1766,17 @@ class Element(Plugin):
+     #
      # Returns: True if the artifact has been downloaded, False otherwise
+     #
 -    def _pull(self):
 +    def _pull(self, subdir=None, excluded_subdirs=None):
          context = self._get_context()
          def progress(percent, message):
              self.status(message)
          # Attempt to pull artifact without knowing whether it's available
 -        pulled = self._pull_strong(progress=progress)
 +        pulled = self._pull_strong(progress=progress, subdir=subdir, excluded_subdirs=excluded_subdirs)
          if not pulled and not self._cached() and not context.get_strict():
 -            pulled = self._pull_weak(progress=progress)
 +            pulled = self._pull_weak(progress=progress, subdir=subdir, excluded_subdirs=excluded_subdirs)
          if not pulled:
              return False
@@ -1792,10 +1799,21 @@ class Element(Plugin):
          if not self._cached():
              return True
 -        # Do not push tained artifact
 +        # Do not push tainted artifact
          if self.__get_tainted():
              return True
 +        # Do not push elements that have a dangling buildtree artifact unless element type is
 +        # expected to have an empty buildtree directory
 +        if not self.__artifacts.contains_subdir_artifact(self, self.__strict_cache_key, 'buildtree'):
 +            return True
++
 +        # strict_cache_key can't be relied on to be available when running in non strict mode
 +        context = self._get_context()
 +        if not context.get_strict():
 +            if not self.__artifacts.contains_subdir_artifact(self, self.__weak_cache_key, 'buildtree'):
 +                return True
++
          return False
      # _push():

tests/integration/pullbuildtrees.py

 +import os
 +import shutil
 +import pytest
++
 +from tests.testutils import cli_integration as cli, create_artifact_share
 +from tests.testutils.integration import assert_contains
++
++
 +DATA_DIR = os.path.join(
 +    os.path.dirname(os.path.realpath(__file__)),
 +    "project"
 +)
++
++
 +# Remove artifact cache & set cli.config value of pullbuildtrees
 +# to false, which is the default user context. The cache has to be
 +# cleared as just forcefully removing the refpath leaves dangling objects.
 +def default_state(cli, tmpdir, share):
 +    shutil.rmtree(os.path.join(str(tmpdir), 'artifacts'))
 +    cli.configure({
 +        'pullbuildtrees': False,
 +        'artifacts': {'url': share.repo, 'push': False},
 +        'artifactdir': os.path.join(str(tmpdir), 'artifacts')
 +    })
++
++
 +# A test to capture the integration of the pullbuildtrees
 +# behaviour, which by default is to not include the buildtree
 +# directory of an element.
 +@pytest.mark.integration
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_pullbuildtrees(cli, tmpdir, datafiles, integration_cache):
++
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
 +    element_name = 'autotools/amhello.bst'
++
 +    # Create artifact shares for pull & push testing
 +    with create_artifact_share(os.path.join(str(tmpdir), 'share1')) as share1,\
 +        create_artifact_share(os.path.join(str(tmpdir), 'share2')) as share2:
 +        cli.configure({
 +            'artifacts': {'url': share1.repo, 'push': True},
 +            'artifactdir': os.path.join(str(tmpdir), 'artifacts')
 +        })
++
 +        # Build autotools element, checked pushed, delete local
 +        result = cli.run(project=project, args=['build', element_name])
 +        assert result.exit_code == 0
 +        assert cli.get_element_state(project, element_name) == 'cached'
 +        assert share1.has_artifact('test', element_name, cli.get_element_key(project, element_name))
 +        default_state(cli, tmpdir, share1)
++
 +        # Pull artifact with default config, assert that pulling again
 +        # doesn't create a pull job, then assert with buildtrees user
 +        # config set creates a pull job.
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name not in result.get_pulled_elements()
 +        cli.configure({'pullbuildtrees': True})
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        default_state(cli, tmpdir, share1)
++
 +        # Pull artifact with default config, then assert that pulling
 +        # with buildtrees cli flag set creates a pull job.
 +        # Also assert that the buildtree is added to the artifact's
 +        # extract dir
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        elementdigest = share1.has_artifact('test', element_name, cli.get_element_key(project, element_name))
 +        buildtreedir = os.path.join(str(tmpdir), 'artifacts', 'extract', 'test', 'autotools-amhello',
 +                                    elementdigest.hash, 'buildtree')
 +        assert not os.path.isdir(buildtreedir)
 +        result = cli.run(project=project, args=['pull', '--pull-buildtrees', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        assert os.path.isdir(buildtreedir)
 +        default_state(cli, tmpdir, share1)
++
 +        # Pull artifact with pullbuildtrees set in user config, then assert
 +        # that pulling with the same user config doesn't creates a pull job,
 +        # or when buildtrees cli flag is set.
 +        cli.configure({'pullbuildtrees': True})
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name not in result.get_pulled_elements()
 +        result = cli.run(project=project, args=['pull', '--pull-buildtrees', element_name])
 +        assert element_name not in result.get_pulled_elements()
 +        default_state(cli, tmpdir, share1)
++
 +        # Pull artifact with default config and buildtrees cli flag set, then assert
 +        # that pulling with pullbuildtrees set in user config doesn't create a pull
 +        # job.
 +        result = cli.run(project=project, args=['pull', '--pull-buildtrees', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        cli.configure({'pullbuildtrees': True})
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name not in result.get_pulled_elements()
 +        default_state(cli, tmpdir, share1)
++
 +        # Assert that a partial build element (not containing a populated buildtree dir)
 +        # can't be pushed to an artifact share, then assert that a complete build element
 +        # can be. This will attempt a partial pull from share1 and then a partial push
 +        # to share2
 +        result = cli.run(project=project, args=['pull', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        cli.configure({'artifacts': {'url': share2.repo, 'push': True}})
 +        result = cli.run(project=project, args=['push', element_name])
 +        assert element_name not in result.get_pushed_elements()
 +        assert not share2.has_artifact('test', element_name, cli.get_element_key(project, element_name))
++
 +        # Assert that after pulling the missing buildtree the element artifact can be
 +        # successfully pushed to the remote. This will attempt to pull the buildtree
 +        # from share1 and then a 'complete' push to share2
 +        cli.configure({'artifacts': {'url': share1.repo, 'push': False}})
 +        result = cli.run(project=project, args=['pull', '--pull-buildtrees', element_name])
 +        assert element_name in result.get_pulled_elements()
 +        cli.configure({'artifacts': {'url': share2.repo, 'push': True}})
 +        result = cli.run(project=project, args=['push', element_name])
 +        assert element_name in result.get_pushed_elements()
 +        assert share2.has_artifact('test', element_name, cli.get_element_key(project, element_name))
 +        default_state(cli, tmpdir, share1)

tests/testutils/artifactshare.py

@@ -114,7 +114,7 @@ class ArtifactShare():
      #    cache_key (str): The cache key
+     #
      # Returns:
 -    #    (bool): True if the artifact exists in the share, otherwise false.
 +    #    (str): artifact digest if the artifact exists in the share, otherwise false.
      def has_artifact(self, project_name, element_name, cache_key):
          # NOTE: This should be kept in line with our
@@ -134,7 +134,7 @@ class ArtifactShare():
          try:
              tree = self.cas.resolve_ref(artifact_key)
 -            return True
 +            return tree
          except ArtifactError:
              return False

[Notes] [Git][BuildStream/buildstream][tpollard/494] Don't pull artifact buildtrees by default.

Tom Pollard pushed to branch tpollard/494 at BuildStream / buildstream

Commits:

8 changed files:

Changes: