Initial implementation of _casbaseddirectory.py

1 files changed, 563 insertions, 0 deletions

diff --git a/buildstream/storage/_casbaseddirectory.py b/buildstream/storage/_casbaseddirectory.py
new file mode 100644
index 000000000..5ca100793
--- /dev/null
+++ b/buildstream/storage/_casbaseddirectory.py
@@ -0,0 +1,563 @@
+#
+#  Copyright (C) 2018 Bloomberg LP
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Authors:
+#        Jim MacArthur <jim.macarthur@codethink.co.uk>
+
+"""
+CasBasedDirectory
+=========
+
+Implementation of the Directory class which backs onto a Merkle-tree based content
+addressable storage system.
+
+See also: :ref:`sandboxing`.
+"""
+
+from collections import OrderedDict
+
+import os
+import tempfile
+import stat
+
+from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
+from .._exceptions import BstError
+from .directory import Directory, VirtualDirectoryError
+from ._filebaseddirectory import FileBasedDirectory
+from ..utils import FileListResult, safe_copy, list_relative_paths
+from .._artifactcache.cascache import CASCache
+
+
+class IndexEntry():
+    """ Used in our index of names to objects to store the 'modified' flag
+    for directory entries. Because we need both the remote_execution_pb2 object
+    and our own Directory object for directory entries, we store both. For files
+    and symlinks, only pb_object is used. """
+    def __init__(self, pb_object, buildstream_object=None, modified=False):
+        self.pb_object = pb_object  # Short for 'protocol buffer object')
+        self.buildstream_object = buildstream_object
+        self.modified = modified
+
+
+# CasBasedDirectory intentionally doesn't call its superclass constuctor,
+# which is meant to be unimplemented.
+# pylint: disable=super-init-not-called
+
+class CasBasedDirectory(Directory):
+    """
+    CAS-based directories can have two names; one is a 'common name' which has no effect
+    on functionality, and the 'filename'. If a CasBasedDirectory has a parent, then 'filename'
+    must be the name of an entry in the parent directory's index which points to this object.
+    This is used to inform a parent directory that it must update the given hash for this
+    object when this object changes.
+
+    Typically a top-level CasBasedDirectory will have a common_name and no filename, and
+    subdirectories wil have a filename and no common_name. common_name can used to identify
+    CasBasedDirectory objects in a log file, since they have no unique position in a file
+    system.
+    """
+
+    # Two constants which define the separators used by the remote execution API.
+    _pb2_path_sep = "/"
+    _pb2_absolute_path_prefix = "/"
+
+    def __init__(self, context, ref=None, parent=None, common_name="untitled", filename=None):
+        self.context = context
+        self.cas_directory = os.path.join(context.artifactdir, 'cas')
+        self.filename = filename
+        self.common_name = common_name
+        self.pb2_directory = remote_execution_pb2.Directory()
+        self.cas_cache = CASCache(context)
+        if ref:
+            with open(self.cas_cache.objpath(ref), 'rb') as f:
+                self.pb2_directory.ParseFromString(f.read())
+
+        self.ref = ref
+        self.index = OrderedDict()
+        self.parent = parent
+        self._directory_read = False
+        self._populate_index()
+
+    def _populate_index(self):
+        if self._directory_read:
+            return
+        for entry in self.pb2_directory.directories:
+            buildStreamDirectory = CasBasedDirectory(self.context, ref=entry.digest,
+                                                     parent=self, filename=entry.name)
+            self.index[entry.name] = IndexEntry(entry, buildstream_object=buildStreamDirectory)
+        for entry in self.pb2_directory.files:
+            self.index[entry.name] = IndexEntry(entry)
+        for entry in self.pb2_directory.symlinks:
+            self.index[entry.name] = IndexEntry(entry)
+        self._directory_read = True
+
+    def _recalculate_recursing_up(self, caller=None):
+        """Recalcuate the hash for this directory and store the results in
+        the cache.  If this directory has a parent, tell it to
+        recalculate (since changing this directory changes an entry in
+        the parent).
+
+        """
+        self.ref = self.cas_cache.add_object(buffer=self.pb2_directory.SerializeToString())
+        if caller:
+            old_dir = self._find_pb2_entry(caller.filename)
+            self.cas_cache.add_object(digest=old_dir.digest, buffer=caller.pb2_directory.SerializeToString())
+        if self.parent:
+            self.parent._recalculate_recursing_up(self)
+
+    def _recalculate_recursing_down(self, parent=None):
+        """Recalcuate the hash for this directory and any
+        subdirectories. Hashes for subdirectories should be calculated
+        and stored after a significant operation (e.g. an
+        import_files() call) but not after adding each file, as that
+        is extremely wasteful.
+
+        """
+        for entry in self.pb2_directory.directories:
+            self.index[entry.name].buildstream_object._recalculate_recursing_down(entry)
+
+        if parent:
+            self.ref = self.cas_cache.add_object(digest=parent.digest, buffer=self.pb2_directory.SerializeToString())
+        else:
+            self.ref = self.cas_cache.add_object(buffer=self.pb2_directory.SerializeToString())
+        # We don't need to do anything more than that; files were already added ealier, and symlinks are
+        # part of the directory structure.
+
+    def _find_pb2_entry(self, name):
+        if name in self.index:
+            return self.index[name].pb_object
+        return None
+
+    def _find_self_in_parent(self):
+        assert self.parent is not None
+        parent = self.parent
+        for (k, v) in parent.index.items():
+            if v.buildstream_object == self:
+                return k
+        return None
+
+    def _add_directory(self, name):
+        if name in self.index:
+            newdir = self.index[name].buildstream_object
+            if not isinstance(newdir, CasBasedDirectory):
+                # TODO: This may not be an actual error; it may actually overwrite it
+                raise VirtualDirectoryError("New directory {} in {} would overwrite existing non-directory of type {}"
+                                            .format(name, str(self), type(newdir)))
+            dirnode = self._find_pb2_entry(name)
+        else:
+            newdir = CasBasedDirectory(self.context, parent=self, filename=name)
+            dirnode = self.pb2_directory.directories.add()
+
+        dirnode.name = name
+
+        # Calculate the hash for an empty directory
+        new_directory = remote_execution_pb2.Directory()
+        self.cas_cache.add_object(digest=dirnode.digest, buffer=new_directory.SerializeToString())
+        self.index[name] = IndexEntry(dirnode, buildstream_object=newdir)
+        return newdir
+
+    def _add_new_file(self, basename, filename):
+        filenode = self.pb2_directory.files.add()
+        filenode.name = filename
+        self.cas_cache.add_object(digest=filenode.digest, path=os.path.join(basename, filename))
+        is_executable = os.access(os.path.join(basename, filename), os.X_OK)
+        filenode.is_executable = is_executable
+        self.index[filename] = IndexEntry(filenode, modified=(filename in self.index))
+
+    def _add_new_link(self, basename, filename):
+        existing_link = self._find_pb2_entry(filename)
+        if existing_link:
+            symlinknode = existing_link
+        else:
+            symlinknode = self.pb2_directory.symlinks.add()
+        symlinknode.name = filename
+        # A symlink node has no digest.
+        symlinknode.target = os.readlink(os.path.join(basename, filename))
+        self.index[filename] = IndexEntry(symlinknode, modified=(existing_link is not None))
+
+    def delete_entry(self, name):
+        for collection in [self.pb2_directory.files, self.pb2_directory.symlinks, self.pb2_directory.directories]:
+            if name in collection:
+                collection.remove(name)
+        if name in self.index:
+            del self.index[name]
+
+    def descend(self, subdirectory_spec, create=False):
+        """Descend one or more levels of directory hierarchy and return a new
+        Directory object for that directory.
+
+        Arguments:
+        * subdirectory_spec (list of strings): A list of strings which are all directory
+          names.
+        * create (boolean): If this is true, the directories will be created if
+          they don't already exist.
+
+        Note: At the moment, creating a directory by descending does
+        not update this object in the CAS cache. However, performing
+        an import_files() into a subdirectory of any depth obtained by
+        descending from this object *will* cause this directory to be
+        updated and stored.
+
+        """
+
+        # It's very common to send a directory name instead of a list and this causes
+        # bizarre errors, so check for it here
+        if not isinstance(subdirectory_spec, list):
+            subdirectory_spec = [subdirectory_spec]
+
+        # Because of the way split works, it's common to get a list which begins with
+        # an empty string. Detect these and remove them.
+        while subdirectory_spec and subdirectory_spec[0] == "":
+            subdirectory_spec.pop(0)
+
+        # Descending into [] returns the same directory.
+        if not subdirectory_spec:
+            return self
+
+        if subdirectory_spec[0] in self.index:
+            entry = self.index[subdirectory_spec[0]].buildstream_object
+            if isinstance(entry, CasBasedDirectory):
+                return entry.descend(subdirectory_spec[1:], create)
+            else:
+                error = "Cannot descend into {}, which is a '{}' in the directory {}"
+                raise VirtualDirectoryError(error.format(subdirectory_spec[0],
+                                                         type(entry).__name__,
+                                                         self))
+        else:
+            if create:
+                newdir = self._add_directory(subdirectory_spec[0])
+                return newdir.descend(subdirectory_spec[1:], create)
+            else:
+                error = "No entry called '{}' found in {}. There are directories called {}."
+                directory_list = ",".join([entry.name for entry in self.pb2_directory.directories])
+                raise VirtualDirectoryError(error.format(subdirectory_spec[0], str(self),
+                                                         directory_list))
+        return None
+
+    def find_root(self):
+        """ Finds the root of this directory tree by following 'parent' until there is
+        no parent. """
+        if self.parent:
+            return self.parent.find_root()
+        else:
+            return self
+
+    def _resolve_symlink_or_directory(self, name):
+        """Used only by _import_files_from_directory. Tries to resolve a
+        directory name or symlink name. 'name' must be an entry in this
+        directory. It must be a single symlink or directory name, not a path
+        separated by path separators. If it's an existing directory name, it
+        just returns the Directory object for that. If it's a symlink, it will
+        attempt to find the target of the symlink and return that as a
+        Directory object.
+
+        If a symlink target doesn't exist, it will attempt to create it
+        as a directory as long as it's within this directory tree.
+        """
+
+        if isinstance(self.index[name].buildstream_object, Directory):
+            return self.index[name].buildstream_object
+        # OK then, it's a symlink
+        symlink = self._find_pb2_entry(name)
+        absolute = symlink.target.startswith(CasBasedDirectory._pb2_absolute_path_prefix)
+        if absolute:
+            root = self.find_root()
+        else:
+            root = self
+        directory = root
+        components = symlink.target.split(CasBasedDirectory._pb2_path_sep)
+        for c in components:
+            if c == "..":
+                directory = directory.parent
+            else:
+                directory = directory.descend(c, create=True)
+        return directory
+
+    def _check_replacement(self, name, path_prefix, fileListResult):
+        """ Checks whether 'name' exists, and if so, whether we can overwrite it.
+        If we can, add the name to 'overwritten_files' and delete the existing entry.
+        Returns 'True' if the import should go ahead.
+        fileListResult.overwritten and fileListResult.ignore are updated depending
+        on the result. """
+        existing_entry = self._find_pb2_entry(name)
+        relative_pathname = os.path.join(path_prefix, name)
+        if existing_entry is None:
+            return True
+        if (isinstance(existing_entry,
+                       (remote_execution_pb2.FileNode, remote_execution_pb2.SymlinkNode))):
+            fileListResult.overwritten.append(relative_pathname)
+            return True
+        elif isinstance(existing_entry, remote_execution_pb2.DirectoryNode):
+            # If 'name' maps to a DirectoryNode, then there must be an entry in index
+            # pointing to another Directory.
+            if self.index[name].buildstream_object.is_empty():
+                self.delete_entry(name)
+                fileListResult.overwritten.append(relative_pathname)
+                return True
+            else:
+                # We can't overwrite a non-empty directory, so we just ignore it.
+                fileListResult.ignored.append(relative_pathname)
+                return False
+        assert False, ("Entry '{}' is not a recognised file/link/directory and not None; it is {}"
+                       .format(name, type(existing_entry)))
+        return False  # In case asserts are disabled
+
+    def _import_directory_recursively(self, directory_name, source_directory, remaining_path, path_prefix):
+        """ _import_directory_recursively and _import_files_from_directory will be called alternately
+        as a directory tree is descended. """
+        if directory_name in self.index:
+            subdir = self._resolve_symlink_or_directory(directory_name)
+        else:
+            subdir = self._add_directory(directory_name)
+        new_path_prefix = os.path.join(path_prefix, directory_name)
+        subdir_result = subdir._import_files_from_directory(os.path.join(source_directory, directory_name),
+                                                            [os.path.sep.join(remaining_path)],
+                                                            path_prefix=new_path_prefix)
+        return subdir_result
+
+    def _import_files_from_directory(self, source_directory, files, path_prefix=""):
+        """ Imports files from a traditional directory """
+        result = FileListResult()
+        for entry in sorted(files):
+            split_path = entry.split(os.path.sep)
+            # The actual file on the FS we're importing
+            import_file = os.path.join(source_directory, entry)
+            # The destination filename, relative to the root where the import started
+            relative_pathname = os.path.join(path_prefix, entry)
+            if len(split_path) > 1:
+                directory_name = split_path[0]
+                # Hand this off to the importer for that subdir. This will only do one file -
+                # a better way would be to hand off all the files in this subdir at once.
+                subdir_result = self._import_directory_recursively(directory_name, source_directory,
+                                                                   split_path[1:], path_prefix)
+                result.combine(subdir_result)
+            elif os.path.islink(import_file):
+                if self._check_replacement(entry, path_prefix, result):
+                    self._add_new_link(source_directory, entry)
+                    result.files_written.append(relative_pathname)
+            elif os.path.isdir(import_file):
+                # A plain directory which already exists isn't a problem; just ignore it.
+                if entry not in self.index:
+                    self._add_directory(entry)
+            elif os.path.isfile(import_file):
+                if self._check_replacement(entry, path_prefix, result):
+                    self._add_new_file(source_directory, entry)
+                    result.files_written.append(relative_pathname)
+        return result
+
+    def import_files(self, external_pathspec, *, files=None,
+                     report_written=True, update_utimes=False,
+                     can_link=False):
+        """Imports some or all files from external_path into this directory.
+
+        Keyword arguments: external_pathspec: Either a string
+        containing a pathname, or a Directory object, to use as the
+        source.
+
+        files (list of strings): A list of all the files relative to
+        the external_pathspec to copy. If 'None' is supplied, all
+        files are copied.
+
+        report_written (bool): Return the full list of files
+        written. Defaults to true. If false, only a list of
+        overwritten files is returned.
+
+        update_utimes (bool): Currently ignored, since CAS does not store utimes.
+
+        can_link (bool): Ignored, since hard links do not have any meaning within CAS.
+        """
+        if isinstance(external_pathspec, FileBasedDirectory):
+            source_directory = external_pathspec._get_underlying_directory()
+        elif isinstance(external_pathspec, CasBasedDirectory):
+            # TODO: This transfers from one CAS to another via the
+            # filesystem, which is very inefficient. Alter this so it
+            # transfers refs across directly.
+            with tempfile.TemporaryDirectory(prefix="roundtrip") as tmpdir:
+                external_pathspec.export_files(tmpdir)
+                if files is None:
+                    files = list_relative_paths(tmpdir)
+                result = self._import_files_from_directory(tmpdir, files=files)
+            return result
+        else:
+            source_directory = external_pathspec
+
+        if files is None:
+            files = list_relative_paths(source_directory)
+
+        # TODO: No notice is taken of report_written, update_utimes or can_link.
+        # Current behaviour is to fully populate the report, which is inefficient,
+        # but still correct.
+        result = self._import_files_from_directory(source_directory, files=files)
+
+        # We need to recalculate and store the hashes of all directories both
+        # up and down the tree; we have changed our directory by importing files
+        # which changes our hash and all our parents' hashes of us. The trees
+        # lower down need to be stored in the CAS as they are not automatically
+        # added during construction.
+        self._recalculate_recursing_down()
+        if self.parent:
+            self.parent._recalculate_recursing_up(self)
+        return result
+
+    def set_deterministic_mtime(self):
+        """ Sets a static modification time for all regular files in this directory.
+        Since we don't store any modification time, we don't need to do anything.
+        """
+        pass
+
+    def set_deterministic_user(self):
+        """ Sets all files in this directory to the current user's euid/egid.
+        We also don't store user data, so this can be ignored.
+        """
+        pass
+
+    def export_files(self, to_directory, *, can_link=False, can_destroy=False):
+        """Copies everything from this into to_directory, which must be the name
+        of a traditional filesystem directory.
+
+        Arguments:
+
+        to_directory (string): a path outside this directory object
+        where the contents will be copied to.
+
+        can_link (bool): Whether we can create hard links in to_directory
+        instead of copying.
+
+        can_destroy (bool): Whether we can destroy elements in this
+        directory to export them (e.g. by renaming them as the
+        target).
+
+        """
+
+        if not os.path.exists(to_directory):
+            os.mkdir(to_directory)
+
+        for entry in self.pb2_directory.directories:
+            if entry.name not in self.index:
+                raise VirtualDirectoryError("CasDir {} contained {} in directories but not in the index"
+                                            .format(str(self), entry.name))
+            if not self._directory_read:
+                raise VirtualDirectoryError("CasDir {} has not been indexed yet".format(str(self)))
+            dest_dir = os.path.join(to_directory, entry.name)
+            if not os.path.exists(dest_dir):
+                os.mkdir(dest_dir)
+            target = self.descend([entry.name])
+            target.export_files(dest_dir)
+        for entry in self.pb2_directory.files:
+            # Extract the entry to a single file
+            dest_name = os.path.join(to_directory, entry.name)
+            src_name = self.cas_cache.objpath(entry.digest)
+            safe_copy(src_name, dest_name)
+            if entry.is_executable:
+                os.chmod(dest_name, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR |
+                         stat.S_IRGRP | stat.S_IXGRP |
+                         stat.S_IROTH | stat.S_IXOTH)
+        for entry in self.pb2_directory.symlinks:
+            src_name = os.path.join(to_directory, entry.name)
+            target_name = entry.target
+            try:
+                os.symlink(target_name, src_name)
+            except FileExistsError as e:
+                raise BstError(("Cannot create a symlink named {} pointing to {}." +
+                                " The original error was: {}").
+                               format(src_name, entry.target, e))
+
+    def export_to_tar(self, tarfile, destination_dir, mtime=0):
+        raise NotImplementedError()
+
+    def mark_changed(self):
+        """ It should not be possible to externally modify a CAS-based
+        directory at the moment."""
+        raise NotImplementedError()
+
+    def is_empty(self):
+        """ Return true if this directory has no files, subdirectories or links in it.
+        """
+        return len(self.index) == 0
+
+    def _mark_directory_unmodified(self):
+        # Marks all entries in this directory and all child directories as unmodified.
+        for i in self.index.values():
+            i.modified = False
+            if isinstance(i.buildstream_object, CasBasedDirectory):
+                i.buildstream_object._mark_directory_unmodified()
+
+    def _mark_entry_unmodified(self, name):
+        # Marks an entry as unmodified. If the entry is a directory, it will
+        # recursively mark all its tree as unmodified.
+        self.index[name].modified = False
+        if self.index[name].buildstream_object:
+            self.index[name].buildstream_object._mark_directory_unmodified()
+
+    def mark_unmodified(self):
+        """ Marks all files in this directory (recursively) as unmodified.
+        If we have a parent, we mark our own entry as unmodified in that parent's
+        index.
+        """
+        if self.parent:
+            self.parent._mark_entry_unmodified(self._find_self_in_parent())
+        else:
+            self._mark_directory_unmodified()
+
+    def list_modified_paths(self):
+        """Provide a list of relative paths which have been modified since the
+        last call to mark_unmodified.
+
+        Return value: List(str) - list of modified paths
+        """
+
+        filelist = []
+        for (k, v) in self.index.items():
+            if isinstance(v.buildstream_object, CasBasedDirectory):
+                filelist.extend([k + os.path.sep + x for x in v.buildstream_object.list_modified_paths()])
+            elif isinstance(v.pb_object, remote_execution_pb2.FileNode) and v.modified:
+                filelist.append(k)
+        return filelist
+
+    def list_relative_paths(self):
+        """Provide a list of all relative paths.
+
+        NOTE: This list is not in the same order as utils.list_relative_paths.
+
+        Return value: List(str) - list of all paths
+        """
+
+        filelist = []
+        for (k, v) in self.index.items():
+            if isinstance(v.buildstream_object, CasBasedDirectory):
+                filelist.extend([k + os.path.sep + x for x in v.buildstream_object.list_relative_paths()])
+            elif isinstance(v.pb_object, remote_execution_pb2.FileNode):
+                filelist.append(k)
+        return filelist
+
+    def _get_identifier(self):
+        path = ""
+        if self.parent:
+            path = self.parent._get_identifier()
+        if self.filename:
+            path += "/" + self.filename
+        else:
+            path += "/" + self.common_name
+        return path
+
+    def __str__(self):
+        return "[CAS:{}]".format(self._get_identifier())
+
+    def _get_underlying_directory(self):
+        """ There is no underlying directory for a CAS-backed directory, so
+        throw an exception. """
+        raise VirtualDirectoryError("_get_underlying_directory was called on a CAS-backed directory," +
+                                    " which has no underlying directory.")