This is an automated email from the ASF dual-hosted git repository. not-in-ldap pushed a commit to branch tpollard/artifacttemp in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 2aaa91776f22816c2c2c1404adbf054822863a07 Author: Raoul Hidalgo Charman <[email protected]> AuthorDate: Thu Mar 28 12:31:50 2019 +0000 _artifact.py: Rework to use artifact proto This will replace the previous use of a directory structure. Quite a lot is changed here, predominantly _artifact and _artifactcache modules. Part of #974 --- buildstream/_artifact.py | 304 ++++++++++++++++------------------- buildstream/_artifactcache.py | 301 ++++++++++++++++++++++++++-------- buildstream/_basecache.py | 31 ++++ buildstream/_cas/cascache.py | 246 +++++++++++++--------------- buildstream/_context.py | 2 +- buildstream/_sourcecache.py | 41 ++++- buildstream/_stream.py | 4 +- buildstream/element.py | 10 +- buildstream/testing/runcli.py | 38 +++-- tests/artifactcache/junctions.py | 2 + tests/artifactcache/pull.py | 3 +- tests/artifactcache/push.py | 17 +- tests/frontend/artifact.py | 10 +- tests/frontend/pull.py | 46 ++++-- tests/frontend/remote-caches.py | 5 +- tests/integration/artifact.py | 31 ++-- tests/integration/pullbuildtrees.py | 15 +- tests/integration/shellbuildtrees.py | 3 + tests/sourcecache/fetch.py | 1 + tests/testutils/artifactshare.py | 44 ++++- 20 files changed, 714 insertions(+), 440 deletions(-) diff --git a/buildstream/_artifact.py b/buildstream/_artifact.py index 6cf51ee..ed81a4e 100644 --- a/buildstream/_artifact.py +++ b/buildstream/_artifact.py @@ -29,11 +29,11 @@ artifact composite interaction away from Element class """ import os -import shutil +import tempfile +from ._protos.buildstream.v2.artifact_pb2 import Artifact as ArtifactProto from . import _yaml from . import utils -from ._exceptions import ArtifactError from .types import Scope from .storage._casbaseddirectory import CasBasedDirectory @@ -49,12 +49,17 @@ from .storage._casbaseddirectory import CasBasedDirectory # class Artifact(): + version = 0 + def __init__(self, element, context, *, strong_key=None, weak_key=None): self._element = element self._context = context self._artifacts = context.artifactcache self._cache_key = strong_key self._weak_cache_key = weak_key + self._artifactdir = context.artifactdir + self._cas = context.get_cascache() + self._tmpdir = context.tmpdir self._metadata_keys = None # Strong and weak key tuple extracted from the artifact self._metadata_dependencies = None # Dictionary of dependency strong keys from the artifact @@ -69,7 +74,9 @@ class Artifact(): # (Directory): The virtual directory object # def get_files(self): - return self._get_subdirectory("files") + files_digest = self._get_artifact_field("files") + + return CasBasedDirectory(self._cas, digest=files_digest) # get_buildtree(): # @@ -79,7 +86,9 @@ class Artifact(): # (Directory): The virtual directory object # def get_buildtree(self): - return self._get_subdirectory("buildtree") + buildtree_digest = self._get_artifact_field("buildtree") + + return CasBasedDirectory(self._cas, digest=buildtree_digest) # get_extract_key(): # @@ -100,7 +109,6 @@ class Artifact(): # sandbox_build_dir (Directory): Virtual Directory object for the sandbox build-root # collectvdir (Directory): Virtual Directoy object from within the sandbox for collection # buildresult (tuple): bool, short desc and detailed desc of result - # keys (list): list of keys for the artifact commit metadata # publicdata (dict): dict of public data to commit to artifact metadata # # Returns: @@ -110,80 +118,78 @@ class Artifact(): context = self._context element = self._element + size = 0 - assemblevdir = CasBasedDirectory(cas_cache=self._artifacts.cas) - logsvdir = assemblevdir.descend("logs", create=True) - metavdir = assemblevdir.descend("meta", create=True) + filesvdir = None + buildtreevdir = None - # Create artifact directory structure - assembledir = os.path.join(rootdir, 'artifact') - logsdir = os.path.join(assembledir, 'logs') - metadir = os.path.join(assembledir, 'meta') - os.mkdir(assembledir) - os.mkdir(logsdir) - os.mkdir(metadir) + artifact = ArtifactProto() - if collectvdir is not None: - filesvdir = assemblevdir.descend("files", create=True) - filesvdir.import_files(collectvdir) + artifact.version = self.version - if sandbox_build_dir: - buildtreevdir = assemblevdir.descend("buildtree", create=True) - buildtreevdir.import_files(sandbox_build_dir) + # Store result + artifact.build_success = buildresult[0] + artifact.build_error = buildresult[1] + artifact.build_error_details = "" if not buildresult[2] else buildresult[2] - # Write some logs out to normal directories: logsdir and metadir - # Copy build log - log_filename = context.get_log_filename() - element._build_log_path = os.path.join(logsdir, 'build.log') - if log_filename: - shutil.copyfile(log_filename, element._build_log_path) + # Store keys + artifact.strong_key = self._cache_key + artifact.weak_key = self._weak_cache_key + + artifact.was_workspaced = bool(element._get_workspace()) + + # Store files + if collectvdir: + filesvdir = CasBasedDirectory(cas_cache=self._cas) + filesvdir.import_files(collectvdir) + artifact.files.CopyFrom(filesvdir._get_digest()) + size += filesvdir.get_size() # Store public data - _yaml.dump(_yaml.node_sanitize(publicdata), os.path.join(metadir, 'public.yaml')) + with tempfile.NamedTemporaryFile(dir=self._tmpdir) as tmp: + _yaml.dump(_yaml.node_sanitize(publicdata), tmp.name) + public_data_digest = self._cas.add_object(path=tmp.name, link_directly=True) + artifact.public_data.CopyFrom(public_data_digest) + size += public_data_digest.size_bytes + + # store build dependencies + for e in element.dependencies(Scope.BUILD): + new_build = artifact.build_deps.add() + new_build.element_name = e.name + new_build.cache_key = e._get_cache_key() + new_build.was_workspaced = bool(e._get_workspace()) + + # Store log file + log_filename = context.get_log_filename() + if log_filename: + digest = self._cas.add_object(path=log_filename) + element._build_log_path = self._cas.objpath(digest) + log = artifact.logs.add() + log.name = os.path.basename(log_filename) + log.digest.CopyFrom(digest) + size += log.digest.size_bytes + + # Store build tree + if sandbox_build_dir: + buildtreevdir = CasBasedDirectory(cas_cache=self._cas) + buildtreevdir.import_files(sandbox_build_dir) + artifact.buildtree.CopyFrom(buildtreevdir._get_digest()) + size += buildtreevdir.get_size() - # Store result - build_result_dict = {"success": buildresult[0], "description": buildresult[1]} - if buildresult[2] is not None: - build_result_dict["detail"] = buildresult[2] - _yaml.dump(build_result_dict, os.path.join(metadir, 'build-result.yaml')) - - # Store keys.yaml - _yaml.dump(_yaml.node_sanitize({ - 'strong': self._cache_key, - 'weak': self._weak_cache_key, - }), os.path.join(metadir, 'keys.yaml')) - - # Store dependencies.yaml - _yaml.dump(_yaml.node_sanitize({ - e.name: e._get_cache_key() for e in element.dependencies(Scope.BUILD) - }), os.path.join(metadir, 'dependencies.yaml')) - - # Store workspaced.yaml - _yaml.dump(_yaml.node_sanitize({ - 'workspaced': bool(element._get_workspace()) - }), os.path.join(metadir, 'workspaced.yaml')) - - # Store workspaced-dependencies.yaml - _yaml.dump(_yaml.node_sanitize({ - 'workspaced-dependencies': [ - e.name for e in element.dependencies(Scope.BUILD) - if e._get_workspace() - ] - }), os.path.join(metadir, 'workspaced-dependencies.yaml')) - - metavdir.import_files(metadir) - logsvdir.import_files(logsdir) - - artifact_size = assemblevdir.get_size() + os.makedirs(os.path.dirname(os.path.join( + self._artifactdir, element.get_artifact_name())), exist_ok=True) keys = utils._deduplicate([self._cache_key, self._weak_cache_key]) - self._artifacts.commit(element, assemblevdir, keys) + for key in keys: + path = os.path.join(self._artifactdir, element.get_artifact_name(key=key)) + with open(path, mode='w+b') as f: + f.write(artifact.SerializeToString()) - return artifact_size + return size # cached_buildtree() # # Check if artifact is cached with expected buildtree. A - # buildtree will not be present if the res tof the partial artifact + # buildtree will not be present if the rest of the partial artifact # is not cached. # # Returns: @@ -193,14 +199,12 @@ class Artifact(): # def cached_buildtree(self): - element = self._element - - key = self.get_extract_key() - if not self._artifacts.contains_subdir_artifact(element, key, 'buildtree'): + buildtree_digest = self._get_artifact_field("buildtree") + if buildtree_digest: + return self._cas.contains_directory(buildtree_digest, with_files=True) + else: return False - return True - # buildtree_exists() # # Check if artifact was created with a buildtree. This does not check @@ -211,8 +215,8 @@ class Artifact(): # def buildtree_exists(self): - artifact_vdir = self._get_directory() - return artifact_vdir._exists('buildtree') + artifact = self._get_proto() + return bool(str(artifact.buildtree)) # load_public_data(): # @@ -224,8 +228,8 @@ class Artifact(): def load_public_data(self): # Load the public data from the artifact - meta_vdir = self._get_subdirectory('meta') - meta_file = meta_vdir._objpath('public.yaml') + artifact = self._get_proto() + meta_file = self._cas.objpath(artifact.public_data) data = _yaml.load(meta_file, shortname='public.yaml') return data @@ -241,20 +245,10 @@ class Artifact(): # def load_build_result(self): - meta_vdir = self._get_subdirectory('meta') - - meta_file = meta_vdir._objpath('build-result.yaml') - if not os.path.exists(meta_file): - build_result = (True, "succeeded", None) - return build_result - - data = _yaml.load(meta_file, shortname='build-result.yaml') - - success = _yaml.node_get(data, bool, 'success') - description = _yaml.node_get(data, str, 'description', default_value=None) - detail = _yaml.node_get(data, str, 'detail', default_value=None) - - build_result = (success, description, detail) + artifact = self._get_proto() + build_result = (artifact.build_success, + artifact.build_error, + artifact.build_error_details) return build_result @@ -271,14 +265,11 @@ class Artifact(): if self._metadata_keys is not None: return self._metadata_keys - # Extract the metadata dir - meta_vdir = self._get_subdirectory('meta') + # Extract proto + artifact = self._get_proto() - # Parse the expensive yaml now and cache the result - meta_file = meta_vdir._objpath('keys.yaml') - meta = _yaml.load(meta_file, shortname='keys.yaml') - strong_key = _yaml.node_get(meta, str, 'strong') - weak_key = _yaml.node_get(meta, str, 'weak') + strong_key = artifact.strong_key + weak_key = artifact.weak_key self._metadata_keys = (strong_key, weak_key) @@ -296,14 +287,10 @@ class Artifact(): if self._metadata_dependencies is not None: return self._metadata_dependencies - # Extract the metadata dir - meta_vdir = self._get_subdirectory('meta') - - # Parse the expensive yaml now and cache the result - meta_file = meta_vdir._objpath('dependencies.yaml') - meta = _yaml.load(meta_file, shortname='dependencies.yaml') + # Extract proto + artifact = self._get_proto() - self._metadata_dependencies = meta + self._metadata_dependencies = {dep.element_name: dep.cache_key for dep in artifact.build_deps} return self._metadata_dependencies @@ -319,14 +306,10 @@ class Artifact(): if self._metadata_workspaced is not None: return self._metadata_workspaced - # Extract the metadata dir - meta_vdir = self._get_subdirectory('meta') + # Extract proto + artifact = self._get_proto() - # Parse the expensive yaml now and cache the result - meta_file = meta_vdir._objpath('workspaced.yaml') - meta = _yaml.load(meta_file, shortname='workspaced.yaml') - - self._metadata_workspaced = _yaml.node_get(meta, bool, 'workspaced') + self._metadata_workspaced = artifact.was_workspaced return self._metadata_workspaced @@ -342,15 +325,11 @@ class Artifact(): if self._metadata_workspaced_dependencies is not None: return self._metadata_workspaced_dependencies - # Extract the metadata dir - meta_vdir = self._get_subdirectory('meta') - - # Parse the expensive yaml now and cache the result - meta_file = meta_vdir._objpath('workspaced-dependencies.yaml') - meta = _yaml.load(meta_file, shortname='workspaced-dependencies.yaml') + # Extract proto + artifact = self._get_proto() - self._metadata_workspaced_dependencies = _yaml.node_sanitize(_yaml.node_get(meta, list, - 'workspaced-dependencies')) + self._metadata_workspaced_dependencies = [dep.element_name for dep in artifact.build_deps + if dep.was_workspaced] return self._metadata_workspaced_dependencies @@ -369,30 +348,21 @@ class Artifact(): def cached(self): context = self._context - try: - vdir = self._get_directory() - except ArtifactError: - # Either ref or top-level artifact directory missing - return False + artifact = self._get_proto() - # Check whether all metadata is available - metadigest = vdir._get_child_digest('meta') - if not self._artifacts.cas.contains_directory(metadigest, with_files=True): + if not artifact: return False - # Additional checks only relevant if artifact was created with 'files' subdirectory - if vdir._exists('files'): - # Determine whether directories are required - require_directories = context.require_artifact_directories - # Determine whether file contents are required as well - require_files = context.require_artifact_files or self._element._artifact_files_required() - - filesdigest = vdir._get_child_digest('files') + # Determine whether directories are required + require_directories = context.require_artifact_directories + # Determine whether file contents are required as well + require_files = (context.require_artifact_files or + self._element._artifact_files_required()) - # Check whether 'files' subdirectory is available, with or without file contents - if (require_directories and - not self._artifacts.cas.contains_directory(filesdigest, with_files=require_files)): - return False + # Check whether 'files' subdirectory is available, with or without file contents + if (require_directories and str(artifact.files) and + not self._cas.contains_directory(artifact.files, with_files=require_files)): + return False return True @@ -408,46 +378,46 @@ class Artifact(): if not self._element._cached(): return False - log_vdir = self._get_subdirectory('logs') + artifact = self._get_proto() - logsdigest = log_vdir._get_digest() - return self._artifacts.cas.contains_directory(logsdigest, with_files=True) + for logfile in artifact.logs: + if not self._cas.contains(logfile.digest.hash): + return False - # _get_directory(): - # - # Get a virtual directory for the artifact contents + return True + + # _get_proto() # # Args: - # key (str): The key for the artifact to extract, - # or None for the default key + # key (str): Key to use, or None for the default key # # Returns: - # (Directory): The virtual directory object + # (Artifact): Artifact proto # - def _get_directory(self, key=None): - - element = self._element + def _get_proto(self): + key = self.get_extract_key() - if key is None: - key = self.get_extract_key() + proto_path = os.path.join(self._artifactdir, + self._element.get_artifact_name(key=key)) + artifact = ArtifactProto() + try: + with open(proto_path, mode='r+b') as f: + artifact.ParseFromString(f.read()) + except FileNotFoundError: + return None - return self._artifacts.get_artifact_directory(element, key) + os.utime(proto_path) + return artifact - # _get_subdirectory(): - # - # Get a virtual directory for the artifact subdir contents - # - # Args: - # subdir (str): The specific artifact subdir - # key (str): The key for the artifact to extract, - # or None for the default key + # _get_artifact_field() # # Returns: - # (Directory): The virtual subdirectory object + # (Digest): Digest of field specified # - def _get_subdirectory(self, subdir, key=None): - - artifact_vdir = self._get_directory(key) - sub_vdir = artifact_vdir.descend(subdir) + def _get_artifact_field(self, field): + artifact_proto = self._get_proto() + digest = getattr(artifact_proto, field) + if not str(digest): + return None - return sub_vdir + return digest diff --git a/buildstream/_artifactcache.py b/buildstream/_artifactcache.py index eae1a93..e54449d 100644 --- a/buildstream/_artifactcache.py +++ b/buildstream/_artifactcache.py @@ -18,14 +18,16 @@ # Tristan Maat <[email protected]> import os +import grpc from ._basecache import BaseCache from .types import _KeyStrength -from ._exceptions import ArtifactError, CASCacheError, CASError +from ._exceptions import ArtifactError, CASError, CASCacheError +from ._protos.buildstream.v2 import artifact_pb2, artifact_pb2_grpc from ._cas import CASRemoteSpec from .storage._casbaseddirectory import CasBasedDirectory -from .storage.directory import VirtualDirectoryError +from ._artifact import Artifact # An ArtifactCacheSpec holds the user configuration for a single remote @@ -61,8 +63,11 @@ class ArtifactCache(BaseCache): self.artifactdir = context.artifactdir os.makedirs(self.artifactdir, exist_ok=True) - self.casquota.add_ref_callbacks(self.required_artifacts) - self.casquota.add_remove_callbacks((lambda x: not x.startswith('@'), self.remove)) + self.casquota.add_remove_callbacks(self.unrequired_artifacts, self.remove) + self.casquota.add_list_refs_callback(self.list_artifacts) + + self.cas.add_reachable_directories_callback(self._reachable_directories) + self.cas.add_reachable_digests_callback(self._reachable_digests) # mark_required_elements(): # @@ -98,13 +103,34 @@ class ArtifactCache(BaseCache): weak_key = element._get_cache_key(strength=_KeyStrength.WEAK) for key in (strong_key, weak_key): if key: - try: - ref = element.get_artifact_name(key) + ref = element.get_artifact_name(key) - self.cas.update_mtime(ref) - except CASError: + try: + self.update_mtime(ref) + except ArtifactError: pass + def update_mtime(self, ref): + try: + os.utime(os.path.join(self.artifactdir, ref)) + except FileNotFoundError as e: + raise ArtifactError("Couldn't find artifact: {}".format(ref)) from e + + # unrequired_artifacts() + # + # Returns iterator over artifacts that are not required in the build plan + # + # Returns: + # (iter): Iterator over tuples of (float, str) where float is the time + # and str is the artifact ref + # + def unrequired_artifacts(self): + required_artifacts = set(map(lambda x: x.get_artifact_name(), + self._required_elements)) + for (mtime, artifact) in self._list_refs_mtimes(self.artifactdir): + if artifact not in required_artifacts: + yield (mtime, artifact) + def required_artifacts(self): # Build a set of the cache keys which are required # based on the required elements at cleanup time @@ -153,7 +179,7 @@ class ArtifactCache(BaseCache): def contains(self, element, key): ref = element.get_artifact_name(key) - return self.cas.contains(ref) + return os.path.exists(os.path.join(self.artifactdir, ref)) # contains_subdir_artifact(): # @@ -183,9 +209,7 @@ class ArtifactCache(BaseCache): # ([str]) - A list of artifact names as generated in LRU order # def list_artifacts(self, *, glob=None): - return list(filter( - lambda x: not x.startswith('@'), - self.cas.list_refs(glob=glob))) + return [ref for _, ref in sorted(list(self._list_refs_mtimes(self.artifactdir, glob_expr=glob)))] # remove(): # @@ -202,7 +226,10 @@ class ArtifactCache(BaseCache): # (int): The amount of space recovered in the cache, in bytes # def remove(self, ref, *, defer_prune=False): - return self.cas.remove(ref, defer_prune=defer_prune) + try: + return self.cas.remove(ref, basedir=self.artifactdir, defer_prune=defer_prune) + except CASCacheError as e: + raise ArtifactError("{}".format(e)) from e # prune(): # @@ -211,47 +238,6 @@ class ArtifactCache(BaseCache): def prune(self): return self.cas.prune() - # get_artifact_directory(): - # - # Get virtual directory for cached artifact of the specified Element. - # - # Assumes artifact has previously been fetched or committed. - # - # Args: - # element (Element): The Element to extract - # key (str): The cache key to use - # - # Raises: - # ArtifactError: In cases there was an OSError, or if the artifact - # did not exist. - # - # Returns: virtual directory object - # - def get_artifact_directory(self, element, key): - ref = element.get_artifact_name(key) - try: - digest = self.cas.resolve_ref(ref, update_mtime=True) - return CasBasedDirectory(self.cas, digest=digest) - except (CASCacheError, VirtualDirectoryError) as e: - raise ArtifactError('Directory not in local cache: {}'.format(e)) from e - - # commit(): - # - # Commit built artifact to cache. - # - # Args: - # element (Element): The Element commit an artifact for - # content (Directory): The element's content directory - # keys (list): The cache keys to use - # - def commit(self, element, content, keys): - refs = [element.get_artifact_name(key) for key in keys] - - tree = content._get_digest() - - for ref in refs: - self.cas.set_ref(ref, tree) - # diff(): # # Return a list of files that have been added or modified between @@ -264,10 +250,19 @@ class ArtifactCache(BaseCache): # subdir (str): A subdirectory to limit the comparison to # def diff(self, element, key_a, key_b, *, subdir=None): - ref_a = element.get_artifact_name(key_a) - ref_b = element.get_artifact_name(key_b) + context = self.context + artifact_a = Artifact(element, context, strong_key=key_a) + artifact_b = Artifact(element, context, strong_key=key_b) + digest_a = artifact_a._get_proto().files + digest_b = artifact_b._get_proto().files + + added = [] + removed = [] + modified = [] - return self.cas.diff(ref_a, ref_b, subdir=subdir) + self.cas.diff_trees(digest_a, digest_b, added=added, removed=removed, modified=modified) + + return modified, removed, added # push(): # @@ -283,21 +278,25 @@ class ArtifactCache(BaseCache): # Raises: # (ArtifactError): if there was an error # - def push(self, element, keys): - refs = [element.get_artifact_name(key) for key in list(keys)] - + def push(self, element, artifact=None): project = element._get_project() push_remotes = [r for r in self._remotes[project] if r.spec.push] pushed = False + # An Artifact instance will have been passed if called via Element, + # only needs to be instantiated if invoked externally. + if artifact is None: + context = self.context + artifact = Artifact(element, context, strong_key=element._get_cache_key()) + for remote in push_remotes: remote.init() display_key = element._get_brief_display_key() element.status("Pushing artifact {} -> {}".format(display_key, remote.spec.url)) - if self.cas.push(refs, remote): + if self._push_artifact(element, artifact, remote): element.info("Pushed artifact {} -> {}".format(display_key, remote.spec.url)) pushed = True else: @@ -322,16 +321,17 @@ class ArtifactCache(BaseCache): # (bool): True if pull was successful, False if artifact was not available # def pull(self, element, key, *, progress=None, subdir=None, excluded_subdirs=None): - ref = element.get_artifact_name(key) display_key = key[:self.context.log_key_length] project = element._get_project() + pull_buildtrees = "buildtree" not in excluded_subdirs if excluded_subdirs else True for remote in self._remotes[project]: + remote.init() try: element.status("Pulling artifact {} <- {}".format(display_key, remote.spec.url)) - if self.cas.pull(ref, remote, progress=progress, subdir=subdir, excluded_subdirs=excluded_subdirs): + if self._pull_artifact(element, key, remote, pull_buildtrees=pull_buildtrees): element.info("Pulled artifact {} <- {}".format(display_key, remote.spec.url)) # no need to pull from additional remotes return True @@ -405,7 +405,9 @@ class ArtifactCache(BaseCache): oldref = element.get_artifact_name(oldkey) newref = element.get_artifact_name(newkey) - self.cas.link_ref(oldref, newref) + if not os.path.exists(os.path.join(self.artifactdir, newref)): + os.link(os.path.join(self.artifactdir, oldref), + os.path.join(self.artifactdir, newref)) # get_artifact_logs(): # @@ -469,3 +471,174 @@ class ArtifactCache(BaseCache): remote_missing_blobs_set.update(remote_missing_blobs) return list(remote_missing_blobs_set) + + ################################################ + # Local Private Methods # + ################################################ + + # _reachable_directories() + # + # Returns: + # (iter): Iterator over directories digests available from artifacts. + # + def _reachable_directories(self): + for root, _, files in os.walk(self.artifactdir): + for artifact_file in files: + artifact = artifact_pb2.Artifact() + with open(os.path.join(root, artifact_file), 'r+b') as f: + artifact.ParseFromString(f.read()) + + if str(artifact.files): + yield artifact.files + + if str(artifact.buildtree): + yield artifact.buildtree + + # _reachable_digests() + # + # Returns: + # (iter): Iterator over single file digests in artifacts + # + def _reachable_digests(self): + for root, _, files in os.walk(self.artifactdir): + for artifact_file in files: + artifact = artifact_pb2.Artifact() + with open(os.path.join(root, artifact_file), 'r+b') as f: + artifact.ParseFromString(f.read()) + + if str(artifact.public_data): + yield artifact.public_data + + for log_file in artifact.logs: + yield log_file.digest + + # _push_artifact() + # + # Pushes relevant directories and then artifact proto to remote. + # + # Args: + # element (Element): element + # keys ([str]): keys to push + # artifact (Artifact): the artifact being pushed + # remote (CASRemote): remote to push to + # + # Returns: + # (bool): whether the push was successful + # + def _push_artifact(self, element, artifact, remote): + + artifact = artifact._get_proto() + + keys = [artifact.strong_key, artifact.weak_key] + + # Check whether the artifact is on the server + present = False + for key in keys: + get_artifact = artifact_pb2.GetArtifactRequest() + get_artifact.cache_key = element.get_artifact_name(key) + try: + artifact_service = artifact_pb2_grpc.ArtifactServiceStub(remote.channel) + artifact_service.GetArtifact(get_artifact) + except grpc.RpcError as e: + if e.code() != grpc.StatusCode.NOT_FOUND: + raise ArtifactError("Error checking artifact cache: {}" + .format(e.details())) + else: + present = True + if present: + return False + + try: + self.cas._send_directory(remote, artifact.files) + + if str(artifact.buildtree): + try: + self.cas._send_directory(remote, artifact.buildtree) + except FileNotFoundError: + pass + + digests = [] + if str(artifact.public_data): + digests.append(artifact.public_data) + + for log_file in artifact.logs: + digests.append(log_file.digest) + + self.cas.send_blobs(remote, digests) + + except grpc.RpcError as e: + if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED: + raise ArtifactError("Failed to push artifact blobs: {}".format(e.details())) + return False + + # finally need to send the artifact proto + for key in keys: + update_artifact = artifact_pb2.UpdateArtifactRequest() + update_artifact.cache_key = element.get_artifact_name(key) + update_artifact.artifact.CopyFrom(artifact) + + try: + artifact_service = artifact_pb2_grpc.ArtifactServiceStub(remote.channel) + artifact_service.UpdateArtifact(update_artifact) + except grpc.RpcError as e: + raise ArtifactError("Failed to push artifact: {}".format(e.details())) + + return True + + # _pull_artifact() + # + # Args: + # element (Element): element to pull + # key (str): specific key of element to pull + # remote (CASRemote): remote to pull from + # pull_buildtree (bool): whether to pull buildtrees or not + # + # Returns: + # (bool): whether the pull was successful + # + def _pull_artifact(self, element, key, remote, pull_buildtrees=False): + + def __pull_digest(digest): + self.cas._fetch_directory(remote, digest) + required_blobs = self.cas.required_blobs_for_directory(digest) + missing_blobs = self.cas.local_missing_blobs(required_blobs) + if missing_blobs: + self.cas.fetch_blobs(remote, missing_blobs) + + request = artifact_pb2.GetArtifactRequest() + request.cache_key = element.get_artifact_name(key=key) + try: + artifact_service = artifact_pb2_grpc.ArtifactServiceStub(remote.channel) + artifact = artifact_service.GetArtifact(request) + except grpc.RpcError as e: + if e.code() != grpc.StatusCode.NOT_FOUND: + raise ArtifactError("Failed to pull artifact: {}".format(e.details())) + return False + + try: + if str(artifact.files): + __pull_digest(artifact.files) + + if pull_buildtrees and str(artifact.buildtree): + __pull_digest(artifact.buildtree) + + digests = [] + if str(artifact.public_data): + digests.append(artifact.public_data) + + for log_digest in artifact.logs: + digests.append(log_digest.digest) + + self.cas.fetch_blobs(remote, digests) + except grpc.RpcError as e: + if e.code() != grpc.StatusCode.NOT_FOUND: + raise ArtifactError("Failed to pull artifact: {}".format(e.details())) + return False + + # Write the artifact proto to cache + artifact_path = os.path.join(self.artifactdir, request.cache_key) + os.makedirs(os.path.dirname(artifact_path), exist_ok=True) + with open(artifact_path, 'w+b') as f: + f.write(artifact.SerializeToString()) + + return True diff --git a/buildstream/_basecache.py b/buildstream/_basecache.py index af3fe9b..68654b2 100644 --- a/buildstream/_basecache.py +++ b/buildstream/_basecache.py @@ -17,6 +17,8 @@ # Raoul Hidalgo Charman <[email protected]> # import multiprocessing +import os +from fnmatch import fnmatch from . import utils from . import _yaml @@ -274,3 +276,32 @@ class BaseCache(): with self.context.timed_activity("Initializing remote caches", silent_nested=True): self.initialize_remotes(on_failure=remote_failed) + + # _list_refs_mtimes() + # + # List refs in a directory, given a base path. Also returns the + # associated mtimes + # + # Args: + # base_path (str): Base path to traverse over + # glob_expr (str|None): Optional glob expression to match against files + # + # Returns: + # (iter (mtime, filename)]): iterator of tuples of mtime and refs + # + def _list_refs_mtimes(self, base_path, *, glob_expr=None): + path = base_path + if glob_expr is not None: + globdir = os.path.dirname(glob_expr) + if not any(c in "*?[" for c in globdir): + # path prefix contains no globbing characters so + # append the glob to optimise the os.walk() + path = os.path.join(base_path, globdir) + + for root, _, files in os.walk(path): + for filename in files: + ref_path = os.path.join(root, filename) + relative_path = os.path.relpath(ref_path, base_path) # Relative to refs head + if not glob_expr or fnmatch(relative_path, glob_expr): + # Obtain the mtime (the time a file was last modified) + yield (os.path.getmtime(ref_path), relative_path) diff --git a/buildstream/_cas/cascache.py b/buildstream/_cas/cascache.py index 1a99c2f..c45f56f 100644 --- a/buildstream/_cas/cascache.py +++ b/buildstream/_cas/cascache.py @@ -24,7 +24,6 @@ import stat import errno import uuid import contextlib -from fnmatch import fnmatch import grpc @@ -89,6 +88,9 @@ class CASCache(): os.makedirs(os.path.join(self.casdir, 'objects'), exist_ok=True) os.makedirs(self.tmpdir, exist_ok=True) + self.__reachable_directory_callbacks = [] + self.__reachable_digest_callbacks = [] + # preflight(): # # Preflight check. @@ -502,44 +504,6 @@ class CASCache(): except FileNotFoundError as e: raise CASCacheError("Attempt to access unavailable ref: {}".format(e)) from e - # list_refs(): - # - # List refs in Least Recently Modified (LRM) order. - # - # Args: - # glob (str) - An optional glob expression to be used to list refs satisfying the glob - # - # Returns: - # (list) - A list of refs in LRM order - # - def list_refs(self, *, glob=None): - # string of: /path/to/repo/refs/heads - ref_heads = os.path.join(self.casdir, 'refs', 'heads') - path = ref_heads - - if glob is not None: - globdir = os.path.dirname(glob) - if not any(c in "*?[" for c in globdir): - # path prefix contains no globbing characters so - # append the glob to optimise the os.walk() - path = os.path.join(ref_heads, globdir) - - refs = [] - mtimes = [] - - for root, _, files in os.walk(path): - for filename in files: - ref_path = os.path.join(root, filename) - relative_path = os.path.relpath(ref_path, ref_heads) # Relative to refs head - if not glob or fnmatch(relative_path, glob): - refs.append(relative_path) - # Obtain the mtime (the time a file was last modified) - mtimes.append(os.path.getmtime(ref_path)) - - # NOTE: Sorted will sort from earliest to latest, thus the - # first ref of this list will be the file modified earliest. - return [ref for _, ref in sorted(zip(mtimes, refs))] - # list_objects(): # # List cached objects in Least Recently Modified (LRM) order. @@ -581,6 +545,8 @@ class CASCache(): # # Args: # ref (str): A symbolic ref + # basedir (str): Path of base directory the ref is in, defaults to + # CAS refs heads # defer_prune (bool): Whether to defer pruning to the caller. NOTE: # The space won't be freed until you manually # call prune. @@ -589,10 +555,12 @@ class CASCache(): # (int|None) The amount of space pruned from the repository in # Bytes, or None if defer_prune is True # - def remove(self, ref, *, defer_prune=False): + def remove(self, ref, *, basedir=None, defer_prune=False): + if basedir is None: + basedir = os.path.join(self.casdir, 'refs', 'heads') # Remove cache ref - self._remove_ref(ref) + self._remove_ref(ref, basedir) if not defer_prune: pruned = self.prune() @@ -600,6 +568,14 @@ class CASCache(): return None + # adds callback of iterator over reachable directory digests + def add_reachable_directories_callback(self, callback): + self.__reachable_directory_callbacks.append(callback) + + # adds callbacks of iterator over reachable file digests + def add_reachable_digests_callback(self, callback): + self.__reachable_digest_callbacks.append(callback) + # prune(): # # Prune unreachable objects from the repo. @@ -619,6 +595,16 @@ class CASCache(): tree = self.resolve_ref(ref) self._reachable_refs_dir(reachable, tree) + # check callback directory digests that are reachable + for digest_callback in self.__reachable_directory_callbacks: + for digest in digest_callback(): + self._reachable_refs_dir(reachable, digest) + + # check callback file digests that are reachable + for digest_callback in self.__reachable_digest_callbacks: + for digest in digest_callback(): + reachable.add(digest.hash) + # Prune unreachable objects for root, _, files in os.walk(os.path.join(self.casdir, 'objects')): for filename in files: @@ -786,22 +772,24 @@ class CASCache(): # # Args: # ref (str): The ref to remove + # basedir (str): Path of base directory the ref is in # # Raises: # (CASCacheError): If the ref didnt exist, or a system error # occurred while removing it # - def _remove_ref(self, ref): + def _remove_ref(self, ref, basedir): # Remove the ref itself - refpath = self._refpath(ref) + refpath = os.path.join(basedir, ref) + try: os.unlink(refpath) except FileNotFoundError as e: raise CASCacheError("Could not find ref '{}'".format(ref)) from e # Now remove any leading directories - basedir = os.path.join(self.casdir, 'refs', 'heads') + components = list(os.path.split(ref)) while components: components.pop() @@ -1148,8 +1136,8 @@ class CASQuota: self._message = context.message - self._ref_callbacks = [] # Call backs to get required refs - self._remove_callbacks = [] # Call backs to remove refs + self._remove_callbacks = [] # Callbacks to remove unrequired refs and their remove method + self._list_refs_callbacks = [] # Callbacks to all refs self._calculate_cache_quota() @@ -1227,6 +1215,21 @@ class CASQuota: return False + # add_remove_callbacks() + # + # This adds tuples of iterators over unrequired objects (currently + # artifacts and source refs), and a callback to remove them. + # + # Args: + # callback (iter(unrequired), remove): tuple of iterator and remove + # method associated. + # + def add_remove_callbacks(self, list_unrequired, remove_method): + self._remove_callbacks.append((list_unrequired, remove_method)) + + def add_list_refs_callback(self, list_callback): + self._list_refs_callbacks.append(list_callback) + ################################################ # Local Private Methods # ################################################ @@ -1383,28 +1386,25 @@ class CASQuota: removed_ref_count = 0 space_saved = 0 - # get required refs - refs = self.cas.list_refs() - required_refs = set( - required - for callback in self._ref_callbacks - for required in callback() - ) + total_refs = 0 + for refs in self._list_refs_callbacks: + total_refs += len(list(refs())) # Start off with an announcement with as much info as possible volume_size, volume_avail = self._get_cache_volume_size() self._message(Message( None, MessageType.STATUS, "Starting cache cleanup", - detail=("Elements required by the current build plan: {}\n" + + detail=("Elements required by the current build plan:\n" + "{}\n" + "User specified quota: {} ({})\n" + "Cache usage: {}\n" + "Cache volume: {} total, {} available") - .format(len(required_refs), - context.config_cache_quota, - utils._pretty_size(self._cache_quota, dec_places=2), - utils._pretty_size(self.get_cache_size(), dec_places=2), - utils._pretty_size(volume_size, dec_places=2), - utils._pretty_size(volume_avail, dec_places=2)))) + .format( + total_refs, + context.config_cache_quota, + utils._pretty_size(self._cache_quota, dec_places=2), + utils._pretty_size(self.get_cache_size(), dec_places=2), + utils._pretty_size(volume_size, dec_places=2), + utils._pretty_size(volume_avail, dec_places=2)))) # Do a real computation of the cache size once, just in case self.compute_cache_size() @@ -1412,67 +1412,63 @@ class CASQuota: self._message(Message(None, MessageType.STATUS, "Cache usage recomputed: {}".format(usage))) - while self.get_cache_size() >= self._cache_lower_threshold: - try: - to_remove = refs.pop(0) - except IndexError: - # If too many artifacts are required, and we therefore - # can't remove them, we have to abort the build. - # - # FIXME: Asking the user what to do may be neater - # - default_conf = os.path.join(os.environ['XDG_CONFIG_HOME'], - 'buildstream.conf') - detail = ("Aborted after removing {} refs and saving {} disk space.\n" - "The remaining {} in the cache is required by the {} references in your build plan\n\n" - "There is not enough space to complete the build.\n" - "Please increase the cache-quota in {} and/or make more disk space." - .format(removed_ref_count, - utils._pretty_size(space_saved, dec_places=2), - utils._pretty_size(self.get_cache_size(), dec_places=2), - len(required_refs), - (context.config_origin or default_conf))) - - if self.full(): - raise CASCacheError("Cache too full. Aborting.", - detail=detail, - reason="cache-too-full") - else: - break - - key = to_remove.rpartition('/')[2] - if key not in required_refs: + # Collect digests and their remove method + all_unrequired_refs = [] + for (unrequired_refs, remove) in self._remove_callbacks: + for (mtime, ref) in unrequired_refs(): + all_unrequired_refs.append((mtime, ref, remove)) - # Remove the actual artifact, if it's not required. - size = 0 - removed_ref = False - for (pred, remove) in self._remove_callbacks: - if pred(to_remove): - size = remove(to_remove) - removed_ref = True - break + # Pair refs and their remove method sorted in time order + all_unrequired_refs = [(ref, remove) for (_, ref, remove) in sorted(all_unrequired_refs)] - if not removed_ref: - continue + # Go through unrequired refs and remove them, oldest first + made_space = False + for (ref, remove) in all_unrequired_refs: + size = remove(ref) + removed_ref_count += 1 + space_saved += size - removed_ref_count += 1 - space_saved += size - - self._message(Message( - None, MessageType.STATUS, - "Freed {: <7} {}".format( - utils._pretty_size(size, dec_places=2), - to_remove))) - - self.set_cache_size(self._cache_size - size) + self._message(Message( + None, MessageType.STATUS, + "Freed {: <7} {}".format( + utils._pretty_size(size, dec_places=2), + ref))) + + self.set_cache_size(self._cache_size - size) + + # User callback + # + # Currently this process is fairly slow, but we should + # think about throttling this progress() callback if this + # becomes too intense. + if progress: + progress() + + if self.get_cache_size() < self._cache_lower_threshold: + made_space = True + break - # User callback - # - # Currently this process is fairly slow, but we should - # think about throttling this progress() callback if this - # becomes too intense. - if progress: - progress() + if not made_space and self.full(): + # If too many artifacts are required, and we therefore + # can't remove them, we have to abort the build. + # + # FIXME: Asking the user what to do may be neater + # + default_conf = os.path.join(os.environ['XDG_CONFIG_HOME'], + 'buildstream.conf') + detail = ("Aborted after removing {} refs and saving {} disk space.\n" + "The remaining {} in the cache is required by the {} references in your build plan\n\n" + "There is not enough space to complete the build.\n" + "Please increase the cache-quota in {} and/or make more disk space." + .format(removed_ref_count, + utils._pretty_size(space_saved, dec_places=2), + utils._pretty_size(self.get_cache_size(), dec_places=2), + total_refs, + (context.config_origin or default_conf))) + + raise CASCacheError("Cache too full. Aborting.", + detail=detail, + reason="cache-too-full") # Informational message about the side effects of the cleanup self._message(Message( @@ -1485,22 +1481,6 @@ class CASQuota: return self.get_cache_size() - # add_ref_callbacks() - # - # Args: - # callback (Iterator): function that gives list of required refs - def add_ref_callbacks(self, callback): - self._ref_callbacks.append(callback) - - # add_remove_callbacks() - # - # Args: - # callback (predicate, callback): The predicate says whether this is the - # correct type to remove given a ref and the callback does actual - # removing. - def add_remove_callbacks(self, callback): - self._remove_callbacks.append(callback) - def _grouper(iterable, n): while True: diff --git a/buildstream/_context.py b/buildstream/_context.py index 93e3f62..151ea63 100644 --- a/buildstream/_context.py +++ b/buildstream/_context.py @@ -233,7 +233,7 @@ class Context(): self.tmpdir = os.path.join(self.cachedir, 'tmp') self.casdir = os.path.join(self.cachedir, 'cas') self.builddir = os.path.join(self.cachedir, 'build') - self.artifactdir = os.path.join(self.cachedir, 'artifacts') + self.artifactdir = os.path.join(self.cachedir, 'artifacts', 'refs') # Move old artifact cas to cas if it exists and create symlink old_casdir = os.path.join(self.cachedir, 'artifacts', 'cas') diff --git a/buildstream/_sourcecache.py b/buildstream/_sourcecache.py index 36f75d0..ecd0234 100644 --- a/buildstream/_sourcecache.py +++ b/buildstream/_sourcecache.py @@ -17,6 +17,8 @@ # Authors: # Raoul Hidalgo Charman <[email protected]> # +import os + from ._cas import CASRemoteSpec from .storage._casbaseddirectory import CasBasedDirectory from ._basecache import BaseCache @@ -53,8 +55,8 @@ class SourceCache(BaseCache): self._required_sources = set() - self.casquota.add_ref_callbacks(self.required_sources) - self.casquota.add_remove_callbacks((lambda x: x.startswith('@sources/'), self.cas.remove)) + self.casquota.add_remove_callbacks(self.unrequired_sources, self.cas.remove) + self.casquota.add_list_refs_callback(self.list_sources) # mark_required_sources() # @@ -81,14 +83,43 @@ class SourceCache(BaseCache): # required_sources() # - # Yields the keys of all sources marked as required + # Yields the keys of all sources marked as required by the current build + # plan # # Returns: - # iterable (str): iterable over the source keys + # iterable (str): iterable over the required source refs # def required_sources(self): for source in self._required_sources: - yield source._key + yield source._get_source_name() + + # unrequired_sources() + # + # Yields the refs of all sources not required by the current build plan + # + # Returns: + # iter (str): iterable over unrequired source keys + # + def unrequired_sources(self): + required_source_names = set(map( + lambda x: x._get_source_name(), self._required_sources)) + for (mtime, source) in self._list_refs_mtimes( + os.path.join(self.cas.casdir, 'refs', 'heads'), + glob_expr="@sources/*"): + if source not in required_source_names: + yield (mtime, source) + + # list_sources() + # + # Get list of all sources in the `cas/refs/heads/@sources/` folder + # + # Returns: + # ([str]): iterable over all source refs + # + def list_sources(self): + return [ref for _, ref in self._list_refs_mtimes( + os.path.join(self.cas.casdir, 'refs', 'heads'), + glob_expr="@sources/*")] # contains() # diff --git a/buildstream/_stream.py b/buildstream/_stream.py index d4f26e4..2343c55 100644 --- a/buildstream/_stream.py +++ b/buildstream/_stream.py @@ -32,7 +32,7 @@ from contextlib import contextmanager, suppress from fnmatch import fnmatch from ._artifactelement import verify_artifact_ref -from ._exceptions import StreamError, ImplError, BstError, ArtifactElementError, CASCacheError +from ._exceptions import StreamError, ImplError, BstError, ArtifactElementError, ArtifactError from ._message import Message, MessageType from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, \ SourcePushQueue, BuildQueue, PullQueue, ArtifactPushQueue @@ -587,7 +587,7 @@ class Stream(): for ref in remove_refs: try: self._artifacts.remove(ref, defer_prune=True) - except CASCacheError as e: + except ArtifactError as e: self._message(MessageType.WARN, str(e)) continue diff --git a/buildstream/element.py b/buildstream/element.py index cc31d3f..b465693 100644 --- a/buildstream/element.py +++ b/buildstream/element.py @@ -768,12 +768,12 @@ class Element(Plugin): if dep.name in old_dep_keys: key_new = dep._get_cache_key() - key_old = _yaml.node_get(old_dep_keys, str, dep.name) + key_old = old_dep_keys[dep.name] # We only need to worry about modified and added # files, since removed files will be picked up by # build systems anyway. - to_update, _, added = self.__artifacts.diff(dep, key_old, key_new, subdir='files') + to_update, _, added = self.__artifacts.diff(dep, key_old, key_new) workspace.add_running_files(dep.name, to_update + added) to_update.extend(workspace.running_files[dep.name]) @@ -1875,9 +1875,9 @@ class Element(Plugin): # in user context, as to complete a partial artifact subdir, _ = self.__pull_directories() - if self.__strong_cached and subdir: + if self.__strong_cached and subdir == 'buildtree': # 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): + if self.__artifact.cached_buildtree(): return False elif self.__strong_cached: return False @@ -1987,7 +1987,7 @@ class Element(Plugin): return False # Push all keys used for local commit - pushed = self.__artifacts.push(self, self.__get_cache_keys_for_commit()) + pushed = self.__artifacts.push(self, self.__artifact) if not pushed: return False diff --git a/buildstream/testing/runcli.py b/buildstream/testing/runcli.py index 934c312..f67d0a4 100644 --- a/buildstream/testing/runcli.py +++ b/buildstream/testing/runcli.py @@ -56,6 +56,7 @@ from buildstream._cas import CASCache # Special private exception accessor, for test case purposes from buildstream._exceptions import BstError, get_last_exception, get_last_task_error +from buildstream._protos.buildstream.v2 import artifact_pb2 # Wrapper for the click.testing result @@ -636,7 +637,8 @@ class TestArtifact(): # def remove_artifact_from_cache(self, cache_dir, element_name): - cache_dir = os.path.join(cache_dir, 'cas', 'refs', 'heads') + cache_dir = os.path.join(cache_dir, 'artifacts', 'refs') + normal_name = element_name.replace(os.sep, '-') cache_dir = os.path.splitext(os.path.join(cache_dir, 'test', normal_name))[0] shutil.rmtree(cache_dir) @@ -655,13 +657,13 @@ class TestArtifact(): # def is_cached(self, cache_dir, element, element_key): - cas = CASCache(str(cache_dir)) + # cas = CASCache(str(cache_dir)) artifact_ref = element.get_artifact_name(element_key) - return cas.contains(artifact_ref) + return os.path.exists(os.path.join(cache_dir, 'artifacts', 'refs', artifact_ref)) # get_digest(): # - # Get the digest for a given element's artifact + # Get the digest for a given element's artifact files # # Args: # cache_dir (str): Specific cache dir to check @@ -673,10 +675,12 @@ class TestArtifact(): # def get_digest(self, cache_dir, element, element_key): - cas = CASCache(str(cache_dir)) artifact_ref = element.get_artifact_name(element_key) - digest = cas.resolve_ref(artifact_ref) - return digest + artifact_dir = os.path.join(cache_dir, 'artifacts', 'refs') + artifact_proto = artifact_pb2.Artifact() + with open(os.path.join(artifact_dir, artifact_ref), 'rb') as f: + artifact_proto.ParseFromString(f.read()) + return artifact_proto.files # extract_buildtree(): # @@ -691,9 +695,19 @@ class TestArtifact(): # (str): path to extracted buildtree directory, does not guarantee # existence. @contextmanager - def extract_buildtree(self, tmpdir, digest): - with self._extract_subdirectory(tmpdir, digest, 'buildtree') as extract: - yield extract + def extract_buildtree(self, cache_dir, tmpdir, ref): + artifact = artifact_pb2.Artifact() + try: + with open(os.path.join(cache_dir, 'artifacts', 'refs', ref), 'rb') as f: + artifact.ParseFromString(f.read()) + except FileNotFoundError: + yield None + else: + if str(artifact.buildtree): + with self._extract_subdirectory(tmpdir, artifact.buildtree) as f: + yield f + else: + yield None # _extract_subdirectory(): # @@ -709,12 +723,12 @@ class TestArtifact(): # (str): path to extracted subdir directory, does not guarantee # existence. @contextmanager - def _extract_subdirectory(self, tmpdir, digest, subdir): + def _extract_subdirectory(self, tmpdir, digest): with tempfile.TemporaryDirectory() as extractdir: try: cas = CASCache(str(tmpdir)) cas.checkout(extractdir, digest) - yield os.path.join(extractdir, subdir) + yield extractdir except FileNotFoundError: yield None diff --git a/tests/artifactcache/junctions.py b/tests/artifactcache/junctions.py index 1eb67b6..94aaccd 100644 --- a/tests/artifactcache/junctions.py +++ b/tests/artifactcache/junctions.py @@ -70,6 +70,8 @@ def test_push_pull(cli, tmpdir, datafiles): # cas = os.path.join(cli.directory, 'cas') shutil.rmtree(cas) + artifact_dir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifact_dir) # Assert that nothing is cached locally anymore state = cli.get_element_state(project, 'target.bst') diff --git a/tests/artifactcache/pull.py b/tests/artifactcache/pull.py index 40fed76..c29afd5 100644 --- a/tests/artifactcache/pull.py +++ b/tests/artifactcache/pull.py @@ -259,7 +259,8 @@ def test_pull_tree(cli, tmpdir, datafiles): utils._kill_process_tree(process.pid) raise - assert directory_hash and directory_size + # Directory size now zero with AaaP and stack element commit #1cbc5e63dc + assert directory_hash and not directory_size directory_digest = remote_execution_pb2.Digest(hash=directory_hash, size_bytes=directory_size) diff --git a/tests/artifactcache/push.py b/tests/artifactcache/push.py index 4c1d2cd..53ce836 100644 --- a/tests/artifactcache/push.py +++ b/tests/artifactcache/push.py @@ -4,7 +4,7 @@ import signal import pytest -from buildstream import _yaml, _signals, utils +from buildstream import _yaml, _signals, utils, Scope from buildstream._context import Context from buildstream._project import Project from buildstream._protos.build.bazel.remote.execution.v2 import remote_execution_pb2 @@ -85,7 +85,7 @@ def test_push(cli, tmpdir, datafiles): # See https://github.com/grpc/grpc/blob/master/doc/fork_support.md for details process = multiprocessing.Process(target=_queue_wrapper, args=(_test_push, queue, user_config_file, project_dir, - 'target.bst', element_key)) + 'target.bst')) try: # Keep SIGINT blocked in the child process @@ -102,7 +102,7 @@ def test_push(cli, tmpdir, datafiles): assert share.has_artifact('test', 'target.bst', element_key) -def _test_push(user_config_file, project_dir, element_name, element_key, queue): +def _test_push(user_config_file, project_dir, element_name, queue): # Fake minimal context context = Context() context.load(config=user_config_file) @@ -118,13 +118,22 @@ def _test_push(user_config_file, project_dir, element_name, element_key, queue): # Load the target element element = project.load_elements([element_name])[0] + # Ensure the element's artifact memeber is initialised + # This is duplicated from Pipeline.resolve_elements() + # as this test does not use the cli frontend. + for e in element.dependencies(Scope.ALL): + # Preflight + e._preflight() + # Determine initial element state. + e._update_state() + # Manually setup the CAS remotes artifactcache.setup_remotes(use_config=True) artifactcache.initialize_remotes() if artifactcache.has_push_remotes(plugin=element): # Push the element's artifact - if not artifactcache.push(element, [element_key]): + if not artifactcache.push(element): queue.put("Push operation failed") else: queue.put(None) diff --git a/tests/frontend/artifact.py b/tests/frontend/artifact.py index 10cb4f5..b28c516 100644 --- a/tests/frontend/artifact.py +++ b/tests/frontend/artifact.py @@ -92,7 +92,7 @@ def test_artifact_delete_artifact(cli, tmpdir, datafiles): element = 'target.bst' # Configure a local cache - local_cache = os.path.join(str(tmpdir), 'artifacts') + local_cache = os.path.join(str(tmpdir), 'cache') cli.configure({'cachedir': local_cache}) # First build an element so that we can find its artifact @@ -104,7 +104,7 @@ def test_artifact_delete_artifact(cli, tmpdir, datafiles): artifact = os.path.join('test', os.path.splitext(element)[0], cache_key) # Explicitly check that the ARTIFACT exists in the cache - assert os.path.exists(os.path.join(local_cache, 'cas', 'refs', 'heads', artifact)) + assert os.path.exists(os.path.join(local_cache, 'artifacts', 'refs', artifact)) # Delete the artifact result = cli.run(project=project, args=['artifact', 'delete', artifact]) @@ -122,7 +122,7 @@ def test_artifact_delete_element_and_artifact(cli, tmpdir, datafiles): dep = 'compose-all.bst' # Configure a local cache - local_cache = os.path.join(str(tmpdir), 'artifacts') + local_cache = os.path.join(str(tmpdir), 'cache') cli.configure({'cachedir': local_cache}) # First build an element so that we can find its artifact @@ -138,14 +138,14 @@ def test_artifact_delete_element_and_artifact(cli, tmpdir, datafiles): artifact = os.path.join('test', os.path.splitext(element)[0], cache_key) # Explicitly check that the ARTIFACT exists in the cache - assert os.path.exists(os.path.join(local_cache, 'cas', 'refs', 'heads', artifact)) + assert os.path.exists(os.path.join(local_cache, 'artifacts', 'refs', artifact)) # Delete the artifact result = cli.run(project=project, args=['artifact', 'delete', artifact, dep]) result.assert_success() # Check that the ARTIFACT is no longer in the cache - assert not os.path.exists(os.path.join(local_cache, 'cas', 'refs', 'heads', artifact)) + assert not os.path.exists(os.path.join(local_cache, 'artifacts', artifact)) # Check that the dependency ELEMENT is no longer cached assert cli.get_element_state(project, dep) != 'cached' diff --git a/tests/frontend/pull.py b/tests/frontend/pull.py index cc62afe..4034de8 100644 --- a/tests/frontend/pull.py +++ b/tests/frontend/pull.py @@ -68,8 +68,10 @@ def test_push_pull_all(cli, tmpdir, datafiles): # Now we've pushed, delete the user's local artifact cache # directory and try to redownload it from the share # - cas = os.path.join(cli.directory, 'cas') - shutil.rmtree(cas) + casdir = os.path.join(cli.directory, 'cas') + shutil.rmtree(casdir) + artifactdir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifactdir) # Assert that nothing is cached locally anymore states = cli.get_element_states(project, all_elements) @@ -118,8 +120,10 @@ def test_push_pull_default_targets(cli, tmpdir, datafiles): # Now we've pushed, delete the user's local artifact cache # directory and try to redownload it from the share # - artifacts = os.path.join(cli.directory, 'cas') - shutil.rmtree(artifacts) + casdir = os.path.join(cli.directory, 'cas') + shutil.rmtree(casdir) + artifactdir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifactdir) # Assert that nothing is cached locally anymore states = cli.get_element_states(project, all_elements) @@ -160,8 +164,10 @@ def test_pull_secondary_cache(cli, tmpdir, datafiles): assert_shared(cli, share2, project, 'target.bst') # Delete the user's local artifact cache. - cas = os.path.join(cli.directory, 'cas') - shutil.rmtree(cas) + casdir = os.path.join(cli.directory, 'cas') + shutil.rmtree(casdir) + artifactdir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifactdir) # Assert that the element is not cached anymore. assert cli.get_element_state(project, 'target.bst') != 'cached' @@ -214,8 +220,10 @@ def test_push_pull_specific_remote(cli, tmpdir, datafiles): # Now we've pushed, delete the user's local artifact cache # directory and try to redownload it from the good_share. # - cas = os.path.join(cli.directory, 'cas') - shutil.rmtree(cas) + casdir = os.path.join(cli.directory, 'cas') + shutil.rmtree(casdir) + artifactdir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifactdir) result = cli.run(project=project, args=['artifact', 'pull', 'target.bst', '--remote', good_share.repo]) @@ -245,7 +253,7 @@ def test_push_pull_non_strict(cli, tmpdir, datafiles): result.assert_success() assert cli.get_element_state(project, 'target.bst') == 'cached' - # Assert that everything is now cached in the remote. + # Assert that everything is now cached in the reote. all_elements = ['target.bst', 'import-bin.bst', 'import-dev.bst', 'compose-all.bst'] for element_name in all_elements: assert_shared(cli, share, project, element_name) @@ -253,8 +261,10 @@ def test_push_pull_non_strict(cli, tmpdir, datafiles): # Now we've pushed, delete the user's local artifact cache # directory and try to redownload it from the share # - cas = os.path.join(cli.directory, 'cas') - shutil.rmtree(cas) + casdir = os.path.join(cli.directory, 'cas') + shutil.rmtree(casdir) + artifactdir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifactdir) # Assert that nothing is cached locally anymore for element_name in all_elements: @@ -303,8 +313,10 @@ def test_push_pull_track_non_strict(cli, tmpdir, datafiles): # Now we've pushed, delete the user's local artifact cache # directory and try to redownload it from the share # - cas = os.path.join(cli.directory, 'cas') - shutil.rmtree(cas) + casdir = os.path.join(cli.directory, 'cas') + shutil.rmtree(casdir) + artifactdir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifactdir) # Assert that nothing is cached locally anymore for element_name in all_elements: @@ -341,6 +353,8 @@ def test_push_pull_cross_junction(cli, tmpdir, datafiles): cache_dir = os.path.join(project, 'cache', 'cas') shutil.rmtree(cache_dir) + artifact_dir = os.path.join(project, 'cache', 'artifacts') + shutil.rmtree(artifact_dir) assert cli.get_element_state(project, 'junction.bst:import-etc.bst') == 'buildable' @@ -374,8 +388,10 @@ def test_pull_missing_blob(cli, tmpdir, datafiles): # Now we've pushed, delete the user's local artifact cache # directory and try to redownload it from the share # - cas = os.path.join(cli.directory, 'cas') - shutil.rmtree(cas) + casdir = os.path.join(cli.directory, 'cas') + shutil.rmtree(casdir) + artifactdir = os.path.join(cli.directory, 'artifacts') + shutil.rmtree(artifactdir) # Assert that nothing is cached locally anymore for element_name in all_elements: diff --git a/tests/frontend/remote-caches.py b/tests/frontend/remote-caches.py index e3f10e6..089cf96 100644 --- a/tests/frontend/remote-caches.py +++ b/tests/frontend/remote-caches.py @@ -80,8 +80,9 @@ def test_source_artifact_caches(cli, tmpdir, datafiles): # remove the artifact from the repo and check it pulls sources, builds # and then pushes the artifacts shutil.rmtree(os.path.join(cachedir, 'cas')) - print(os.listdir(os.path.join(share.repodir, 'cas', 'refs', 'heads'))) - shutil.rmtree(os.path.join(share.repodir, 'cas', 'refs', 'heads', 'test')) + shutil.rmtree(os.path.join(cachedir, 'artifacts')) + print(os.listdir(os.path.join(share.repodir, 'artifacts', 'refs'))) + shutil.rmtree(os.path.join(share.repodir, 'artifacts', 'refs', 'test')) res = cli.run(project=project_dir, args=['build', 'repo.bst']) res.assert_success() diff --git a/tests/integration/artifact.py b/tests/integration/artifact.py index a5e1f4d..d36785a 100644 --- a/tests/integration/artifact.py +++ b/tests/integration/artifact.py @@ -27,6 +27,7 @@ import tempfile from buildstream import utils from buildstream.testing import cli_integration as cli from tests.testutils import create_artifact_share +from tests.testutils.element_name import element_ref_name from tests.testutils.site import HAVE_SANDBOX from buildstream._cas import CASCache @@ -48,6 +49,7 @@ DATA_DIR = os.path.join( def test_cache_buildtrees(cli, tmpdir, datafiles): project = str(datafiles) element_name = 'autotools/amhello.bst' + cwd = str(tmpdir) # Create artifact shares for pull & push testing with create_artifact_share(os.path.join(str(tmpdir), 'share1')) as share1,\ @@ -69,19 +71,23 @@ def test_cache_buildtrees(cli, tmpdir, datafiles): # The buildtree dir should not exist, as we set the config to not cache buildtrees. cache_key = cli.get_element_key(project, element_name) - elementdigest = share1.has_artifact('test', element_name, cache_key) - with cli.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: - assert not os.path.isdir(buildtreedir) + artifact_ref = "{}/{}/{}".format('test', element_ref_name(element_name), cache_key) + + assert share1.has_artifact('test', element_name, cache_key) + with cli.artifact.extract_buildtree(cwd, cwd, artifact_ref) as buildtreedir: + assert not buildtreedir # Delete the local cached artifacts, and assert the when pulled with --pull-buildtrees # that is was cached in share1 as expected without a buildtree dir shutil.rmtree(os.path.join(str(tmpdir), 'cas')) + shutil.rmtree(os.path.join(str(tmpdir), 'artifacts')) assert cli.get_element_state(project, element_name) != 'cached' result = cli.run(project=project, args=['--pull-buildtrees', 'artifact', 'pull', element_name]) assert element_name in result.get_pulled_elements() - with cli.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: - assert not os.path.isdir(buildtreedir) + with cli.artifact.extract_buildtree(cwd, cwd, artifact_ref) as buildtreedir: + assert not buildtreedir shutil.rmtree(os.path.join(str(tmpdir), 'cas')) + shutil.rmtree(os.path.join(str(tmpdir), 'artifacts')) # Assert that the default behaviour of pull to not include buildtrees on the artifact # in share1 which was purposely cached with an empty one behaves as expected. As such the @@ -89,9 +95,10 @@ def test_cache_buildtrees(cli, tmpdir, datafiles): # leading to no buildtreedir being extracted result = cli.run(project=project, args=['artifact', 'pull', element_name]) assert element_name in result.get_pulled_elements() - with cli.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: - assert not os.path.isdir(buildtreedir) + with cli.artifact.extract_buildtree(cwd, cwd, artifact_ref) as buildtreedir: + assert not buildtreedir shutil.rmtree(os.path.join(str(tmpdir), 'cas')) + shutil.rmtree(os.path.join(str(tmpdir), 'artifacts')) # Repeat building the artifacts, this time with cache-buildtrees set to # 'always' via the cli, as such the buildtree dir should not be empty @@ -105,21 +112,22 @@ def test_cache_buildtrees(cli, tmpdir, datafiles): assert share2.has_artifact('test', element_name, cli.get_element_key(project, element_name)) # Cache key will be the same however the digest hash will have changed as expected, so reconstruct paths - elementdigest = share2.has_artifact('test', element_name, cache_key) - with cli.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: + with cli.artifact.extract_buildtree(cwd, cwd, artifact_ref) as buildtreedir: assert os.path.isdir(buildtreedir) assert os.listdir(buildtreedir) # Delete the local cached artifacts, and assert that when pulled with --pull-buildtrees # that it was cached in share2 as expected with a populated buildtree dir shutil.rmtree(os.path.join(str(tmpdir), 'cas')) + shutil.rmtree(os.path.join(str(tmpdir), 'artifacts')) assert cli.get_element_state(project, element_name) != 'cached' result = cli.run(project=project, args=['--pull-buildtrees', 'artifact', 'pull', element_name]) assert element_name in result.get_pulled_elements() - with cli.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: + with cli.artifact.extract_buildtree(cwd, cwd, artifact_ref) as buildtreedir: assert os.path.isdir(buildtreedir) assert os.listdir(buildtreedir) shutil.rmtree(os.path.join(str(tmpdir), 'cas')) + shutil.rmtree(os.path.join(str(tmpdir), 'artifacts')) # Clarify that the user config option for cache-buildtrees works as the cli # main option does. Point to share3 which does not have the artifacts cached to force @@ -133,7 +141,6 @@ def test_cache_buildtrees(cli, tmpdir, datafiles): assert result.exit_code == 0 assert cli.get_element_state(project, element_name) == 'cached' cache_key = cli.get_element_key(project, element_name) - elementdigest = share3.has_artifact('test', element_name, cache_key) - with cli.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: + with cli.artifact.extract_buildtree(cwd, cwd, artifact_ref) as buildtreedir: assert os.path.isdir(buildtreedir) assert os.listdir(buildtreedir) diff --git a/tests/integration/pullbuildtrees.py b/tests/integration/pullbuildtrees.py index 91acff4..8c34c5a 100644 --- a/tests/integration/pullbuildtrees.py +++ b/tests/integration/pullbuildtrees.py @@ -6,6 +6,7 @@ import tempfile from tests.testutils import create_artifact_share from tests.testutils.site import HAVE_SANDBOX +from tests.testutils.element_name import element_ref_name from buildstream import utils from buildstream.testing import cli, cli_integration as cli2 @@ -40,6 +41,7 @@ def default_state(cli, tmpdir, share): def test_pullbuildtrees(cli2, tmpdir, datafiles): project = str(datafiles) element_name = 'autotools/amhello.bst' + cwd = str(tmpdir) # Create artifact shares for pull & push testing with create_artifact_share(os.path.join(str(tmpdir), 'share1')) as share1,\ @@ -58,6 +60,10 @@ def test_pullbuildtrees(cli2, tmpdir, datafiles): assert share1.has_artifact('test', element_name, cli2.get_element_key(project, element_name)) default_state(cli2, tmpdir, share1) + element_ref = "test/{}/{}".format( + element_ref_name(element_name), + cli2.get_element_key(project, element_name)) + # 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. @@ -75,12 +81,11 @@ def test_pullbuildtrees(cli2, tmpdir, datafiles): # Also assert that the buildtree is added to the local CAS. result = cli2.run(project=project, args=['artifact', 'pull', element_name]) assert element_name in result.get_pulled_elements() - elementdigest = share1.has_artifact('test', element_name, cli2.get_element_key(project, element_name)) - with cli2.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: + with cli2.artifact.extract_buildtree(cwd, cwd, element_ref) as buildtreedir: assert not buildtreedir result = cli2.run(project=project, args=['--pull-buildtrees', 'artifact', 'pull', element_name]) assert element_name in result.get_pulled_elements() - with cli2.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: + with cli2.artifact.extract_buildtree(cwd, cwd, element_ref) as buildtreedir: assert os.path.isdir(buildtreedir) default_state(cli2, tmpdir, share1) @@ -139,7 +144,7 @@ def test_pullbuildtrees(cli2, tmpdir, datafiles): result = cli2.run(project=project, args=['--pull-buildtrees', 'artifact', 'push', element_name]) assert "Attempting to fetch missing artifact buildtrees" in result.stderr assert element_name not in result.get_pulled_elements() - with cli2.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: + with cli2.artifact.extract_buildtree(cwd, cwd, element_ref) as buildtreedir: assert not buildtreedir assert element_name not in result.get_pushed_elements() assert not share3.has_artifact('test', element_name, cli2.get_element_key(project, element_name)) @@ -152,7 +157,7 @@ def test_pullbuildtrees(cli2, tmpdir, datafiles): result = cli2.run(project=project, args=['--pull-buildtrees', 'artifact', 'push', element_name]) assert "Attempting to fetch missing artifact buildtrees" in result.stderr assert element_name in result.get_pulled_elements() - with cli2.artifact.extract_buildtree(tmpdir, elementdigest) as buildtreedir: + with cli2.artifact.extract_buildtree(cwd, cwd, element_ref) as buildtreedir: assert os.path.isdir(buildtreedir) assert element_name in result.get_pushed_elements() assert share3.has_artifact('test', element_name, cli2.get_element_key(project, element_name)) diff --git a/tests/integration/shellbuildtrees.py b/tests/integration/shellbuildtrees.py index 3d59c78..2bdb644 100644 --- a/tests/integration/shellbuildtrees.py +++ b/tests/integration/shellbuildtrees.py @@ -196,6 +196,7 @@ def test_buildtree_pulled(cli, tmpdir, datafiles): # Discard the cache shutil.rmtree(str(os.path.join(str(tmpdir), 'cache', 'cas'))) + shutil.rmtree(str(os.path.join(str(tmpdir), 'cache', 'artifacts'))) assert cli.get_element_state(project, element_name) != 'cached' # Pull from cache, ensuring cli options is set to pull the buildtree @@ -229,6 +230,7 @@ def test_buildtree_options(cli, tmpdir, datafiles): # Discard the cache shutil.rmtree(str(os.path.join(str(tmpdir), 'cache', 'cas'))) + shutil.rmtree(str(os.path.join(str(tmpdir), 'cache', 'artifacts'))) assert cli.get_element_state(project, element_name) != 'cached' # Pull from cache, but do not include buildtrees. @@ -269,6 +271,7 @@ def test_buildtree_options(cli, tmpdir, datafiles): assert 'Attempting to fetch missing artifact buildtree' in res.stderr assert 'Hi' in res.output shutil.rmtree(os.path.join(os.path.join(str(tmpdir), 'cache', 'cas'))) + shutil.rmtree(os.path.join(os.path.join(str(tmpdir), 'cache', 'artifacts'))) assert cli.get_element_state(project, element_name) != 'cached' # Check it's not loading the shell at all with always set for the buildtree, when the diff --git a/tests/sourcecache/fetch.py b/tests/sourcecache/fetch.py index bc025cb..899e162 100644 --- a/tests/sourcecache/fetch.py +++ b/tests/sourcecache/fetch.py @@ -95,6 +95,7 @@ def test_source_fetch(cli, tmpdir, datafiles): os.path.join(str(tmpdir), 'cache', 'cas'), os.path.join(str(tmpdir), 'sourceshare', 'repo')) shutil.rmtree(os.path.join(str(tmpdir), 'cache', 'sources')) + shutil.rmtree(os.path.join(str(tmpdir), 'cache', 'artifacts')) digest = share.cas.resolve_ref(source._get_source_name()) assert share.has_object(digest) diff --git a/tests/testutils/artifactshare.py b/tests/testutils/artifactshare.py index 80959e9..d4025d2 100644 --- a/tests/testutils/artifactshare.py +++ b/tests/testutils/artifactshare.py @@ -10,6 +10,7 @@ from buildstream._cas import CASCache from buildstream._cas.casserver import create_server from buildstream._exceptions import CASError from buildstream._protos.build.bazel.remote.execution.v2 import remote_execution_pb2 +from buildstream._protos.buildstream.v2 import artifact_pb2 from tests.testutils.element_name import element_ref_name @@ -44,6 +45,8 @@ class ArtifactShare(): # self.repodir = os.path.join(self.directory, 'repo') os.makedirs(self.repodir) + self.artifactdir = os.path.join(self.repodir, 'artifacts', 'refs') + os.makedirs(self.artifactdir) self.cas = CASCache(self.repodir) @@ -140,17 +143,44 @@ class ArtifactShare(): element_name = element_ref_name(element_name) artifact_key = '{0}/{1}/{2}'.format(project_name, element_name, cache_key) + artifact_proto = artifact_pb2.Artifact() + artifact_path = os.path.join(self.artifactdir, artifact_key) + try: - tree = self.cas.resolve_ref(artifact_key) - reachable = set() - try: - self.cas._reachable_refs_dir(reachable, tree, update_mtime=False, check_exists=True) - except FileNotFoundError: - return None - return tree + with open(artifact_path, 'rb') as f: + artifact_proto.ParseFromString(f.read()) + except FileNotFoundError: + return None + + reachable = set() + + def reachable_dir(digest): + self.cas._reachable_refs_dir( + reachable, digest, update_mtime=False, check_exists=True) + + try: + if str(artifact_proto.files): + reachable_dir(artifact_proto.files) + + if str(artifact_proto.buildtree): + reachable_dir(artifact_proto.buildtree) + + if str(artifact_proto.public_data): + if not os.path.exists(self.cas.objpath(artifact_proto.public_data)): + return None + + for log_file in artifact_proto.logs: + if not os.path.exists(self.cas.objpath(log_file.digest)): + return None + + return artifact_proto.files + except CASError: return None + except FileNotFoundError: + return None + # close(): # # Remove the artifact share.
