Commit 82d7ae42 authored by Tom Pollard's avatar Tom Pollard
Browse files

WIP: Don't pull artifact buildtrees by default 

- Set default pull to not include buildtree artifact dir
- PullQueue configurable buildtree attribute
- Add 'pullbuildtrees' option to existing artifact artifactcache()
  config
- pullbuildtrees set on a per server basis from user conf
- Add --pull-buildtrees flag to bst build cli
- Add --pull-buildtrees flag to bst pull cli
- Add helper function _fetch_subdir to cascache, to fetch buildtree
  or any other subdir digest
- Make element._pull_pending not assume no need to process pull if
  artifact is cached if buildtrees are set to be pulled
- Ensure cascache.py doesn't try to checkout/extract a dangling ref

ToDo:
- Tests
parent 232662f1

buildstream/_artifactcache/artifactcache.py

+24 −5
Original line number Diff line number Diff line
@@ -38,8 +38,9 @@ CACHE_SIZE_FILE = "cache_size" 
#     url (str): Location of the remote artifact cache

#     push (bool): Whether we should attempt to push artifacts to this cache,

#                  in addition to pulling from it.

#     buildtrees (bool): Whether the default action of pull should include the artifact buildtree

#

class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push server_cert client_key client_cert')):

class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push buildtrees server_cert client_key client_cert')):



    # _new_from_config_node

    #
@@ -47,9 +48,10 @@ class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push server_cert cl 
    #

    @staticmethod

    def _new_from_config_node(spec_node, basedir=None):

        _yaml.node_validate(spec_node, ['url', 'push', 'server-cert', 'client-key', 'client-cert'])

        _yaml.node_validate(spec_node, ['url', 'push', 'pullbuildtrees', 'server-cert', 'client-key', 'client-cert'])

        url = _yaml.node_get(spec_node, str, 'url')

        push = _yaml.node_get(spec_node, bool, 'push', default_value=False)

        buildtrees = _yaml.node_get(spec_node, bool, 'pullbuildtrees', default_value=False)

        if not url:

            provenance = _yaml.node_get_provenance(spec_node, 'url')

            raise LoadError(LoadErrorReason.INVALID_DATA,
@@ -77,10 +79,10 @@ class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push server_cert cl 
            raise LoadError(LoadErrorReason.INVALID_DATA,

                            "{}: 'client-cert' was specified without 'client-key'".format(provenance))



        return ArtifactCacheSpec(url, push, server_cert, client_key, client_cert)

        return ArtifactCacheSpec(url, push, buildtrees, server_cert, client_key, client_cert)





ArtifactCacheSpec.__new__.__defaults__ = (None, None, None)

ArtifactCacheSpec.__new__.__defaults__ = (None, None, None, None)





# An ArtifactCache manages artifacts.
@@ -426,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.
@@ -551,11 +569,12 @@ 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

    #

    # 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

+32 −8
Original line number Diff line number Diff line
@@ -92,6 +92,16 @@ class CASCache(ArtifactCache): 
        # This assumes that the repository doesn't have any dangling pointers

        return os.path.exists(refpath)



    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):

        ref = self.get_artifact_fullname(element, key)
@@ -228,7 +238,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 +257,14 @@ 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 subdir

                if subdir and os.path.exists(self.objpath(tree)):

                    self._fetch_subdir(remote, tree, 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))
@@ -649,7 +665,6 @@ class CASCache(ArtifactCache): 
    ################################################

    #             Local Private Methods            #

    ################################################



    def _checkout(self, dest, tree):

        os.makedirs(dest, exist_ok=True)
@@ -668,6 +683,8 @@ class CASCache(ArtifactCache): 
                         stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH)



        for dirnode in directory.directories:

            # 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)
@@ -948,10 +965,12 @@ class CASCache(ArtifactCache): 
    #     remote (Remote): The remote to use.

    #     dir_digest (Digest): Digest object for the directory to 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:
@@ -966,6 +985,7 @@ class CASCache(ArtifactCache): 
                directory.ParseFromString(f.read())



            for dirnode in directory.directories:

                if dirnode.name not in excluded_subdirs:

                    batch = self._fetch_directory_node(remote, dirnode.digest, batch,

                                                       fetch_queue, fetch_next_queue, recursive=True)
@@ -976,6 +996,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/_frontend/cli.py

+9 −4
Original line number Diff line number Diff line
@@ -305,10 +305,12 @@ def init(app, project_name, format_version, element_path, force): 
              help="Allow tracking to cross junction boundaries")

@click.option('--track-save', default=False, is_flag=True,

              help="Deprecated: This is ignored")

@click.option('--pull-buildtrees', default=False, is_flag=True,

              help="Pull buildtrees from a remote cache server")

@click.argument('elements', nargs=-1,

                type=click.Path(readable=False))

@click.pass_obj

def build(app, elements, all_, track_, track_save, track_all, track_except, track_cross_junctions):

def build(app, elements, all_, track_, track_save, track_all, track_except, track_cross_junctions, pull_buildtrees):

    """Build elements in a pipeline"""



    if (track_except or track_cross_junctions) and not (track_ or track_all):
@@ -327,7 +329,8 @@ def build(app, elements, all_, track_, track_save, track_all, track_except, trac 
                         track_targets=track_,

                         track_except=track_except,

                         track_cross_junctions=track_cross_junctions,

                         build_all=all_)

                         build_all=all_,

                         pull_buildtrees=pull_buildtrees)





##################################################################
@@ -429,10 +432,12 @@ def track(app, elements, deps, except_, cross_junctions): 
              help='The dependency artifacts to pull (default: none)')

@click.option('--remote', '-r',

              help="The URL of the remote cache (defaults to the first configured cache)")

@click.option('--pull-buildtrees', default=False, is_flag=True,

              help="Pull buildtrees from a remote cache server")

@click.argument('elements', nargs=-1,

                type=click.Path(readable=False))

@click.pass_obj

def pull(app, elements, deps, remote):

def pull(app, elements, deps, remote, pull_buildtrees):

    """Pull a built artifact from the configured remote artifact cache.



    By default the artifact will be pulled one of the configured caches
@@ -446,7 +451,7 @@ def pull(app, elements, deps, remote): 
        all:   All dependencies

    """

    with app.initialized(session_name="Pull"):

        app.stream.pull(elements, selection=deps, remote=remote)

        app.stream.pull(elements, selection=deps, remote=remote, pull_buildtrees=pull_buildtrees)





##################################################################

buildstream/_scheduler/queues/pullqueue.py

+13 −2
Original line number Diff line number Diff line
@@ -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

+16 −4
Original line number Diff line number Diff line
@@ -160,12 +160,14 @@ 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,

              track_except=None,

              track_cross_junctions=False,

              build_all=False):

              build_all=False,

              pull_buildtrees=False):



        if build_all:

            selection = PipelineSelection.ALL
@@ -195,7 +197,11 @@ class Stream(): 
            self._add_queue(track_queue, track=True)



        if self._artifacts.has_fetch_remotes():

            self._add_queue(PullQueue(self._scheduler))

            # Query if any of the user defined artifact servers have buildtrees set

            for cache in self._context.artifact_cache_specs:

                if cache.buildtrees:

                    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))
@@ -295,7 +301,8 @@ class Stream(): 
    #

    def pull(self, targets, *,

             selection=PipelineSelection.NONE,

             remote=None):

             remote=None,

             pull_buildtrees=False):



        use_config = True

        if remote:
@@ -310,8 +317,13 @@ class Stream(): 
        if not self._artifacts.has_fetch_remotes():

            raise StreamError("No artifact caches available for pulling artifacts")



        # Query if any of the user defined artifact servers have buildtrees set

        for cache in self._context.artifact_cache_specs:

            if cache.buildtrees:

                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()