GNOME.org

[Notes] [Git][BuildStream/buildstream][jmac/remote_execution_rebase] 24 commits: Add warning to git track if track and ref are not present

From: Jim MacArthur <gitlab mg gitlab com>
To: buildstream-notifications-list gnome org
Subject: [Notes] [Git][BuildStream/buildstream][jmac/remote_execution_rebase] 24 commits: Add warning to git track if track and ref are not present
Date: Tue, 14 Aug 2018 16:53:58 +0000

Title: GitLab

Jim MacArthur pushed to branch jmac/remote_execution_rebase at BuildStream / buildstream

Commits:

42aa3999

by William Salmon at 2018-08-07T13:41:02Z

Add warning to git track if track and ref are not present

This is to address https://gitlab.com/BuildStream/buildstream/issues/471 that
documented unhelpfull behavour when tracking git sources.

2ceb5dec

by Will Salmon at 2018-08-07T14:46:36Z

Merge branch 'willsalmon/trackWarning' into 'master'

Add warning to git track if track and ref are not present

Closes #471

See merge request BuildStream/buildstream!580

eee4b674

by Jürg Billeter at 2018-08-07T15:36:35Z

Revert "Restrict version of pylint"

This reverts commit 4f168b9b6a02216e2fae24d758ae6b778e545869.

The latest version of pytest_pylint works fine with pylint 2, which
means there is no longer a reason to restrict the pylint version.

pylint 2 is required for Python 3.7.

ea27e389

by Jürg Billeter at 2018-08-07T15:36:35Z

_artifactcache/cascache.py: Fix for PEP 479 / Python 3.7

Do not rely on `StopIteration` bubbling up.

https://www.python.org/dev/peps/pep-0479/

fa5a59f0

by Jürg Billeter at 2018-08-07T16:48:21Z

Merge branch 'juerg/python3.7' into 'master'

Python 3.7 support

See merge request BuildStream/buildstream!615

ed653fbc

by Chandan Singh at 2018-08-08T00:06:28Z

.gitlab-ci.yml: Remove unused script "install.sh"

65f382f1

by Chandan Singh at 2018-08-08T10:52:32Z

Merge branch 'chandan/remove-dead-code' into 'master'

.gitlab-ci.yml: Remove unused script "install.sh"

See merge request BuildStream/buildstream!617

7de520f8

by Jim MacArthur at 2018-08-08T12:52:26Z

Convert uses of external_directory to get_underlying_directory()

421c6969

by Jim MacArthur at 2018-08-08T12:52:26Z

_filebaseddirectory/directory.py: Move VirtualDirectoryError to Directory

0aba926c

by Jim MacArthur at 2018-08-08T12:52:26Z

Move KeyStrength enum out to element_enums.py

a626f146

by Jim MacArthur at 2018-08-08T12:52:26Z

Initial implementation of _casbaseddirectory.py

b4351094
by Jim MacArthur at 2018-08-08T12:52:26Z
```
Directory API: add mark_changed
```

f0bce7da

by Jim MacArthur at 2018-08-08T12:52:26Z

Sandboxes: Call _vdir.mark_changed() at the end of run()

45e96e5a

by Jim MacArthur at 2018-08-08T12:52:26Z

filebaseddirectory: Update index after descend

0054727e

by Jim MacArthur at 2018-08-08T12:52:26Z

Sandbox: Return a CasBasedDirectory when an environment variable is set

3fcdcf3a
by Jim MacArthur at 2018-08-08T12:52:26Z
```
Add basic storage test 'storage-test.py'
```

197f081d

by Jim MacArthur at 2018-08-09T15:29:01Z

_casbaseddirectory: Alter _save to save the serialised digest object, not the text digest

612a33a2

by Jim MacArthur at 2018-08-09T15:29:12Z

cascache: Modifications necessary for remote execution.

fe9b657a

by Jim MacArthur at 2018-08-14T16:29:31Z

cascache.py: Update to new protocol buffers spec

47b81e9c

by Jim MacArthur at 2018-08-14T16:29:31Z

sandbox.py: Allow setting the virtual directory (for use after remote execution has finished)

08565f1e
by Jim MacArthur at 2018-08-14T16:29:54Z
```
_sandboxremote.py: New file.
```

403cd1a8

by Jim MacArthur at 2018-08-14T16:29:54Z

element.py: Hack to get the updated virtual directory after remote execution updates it

5d7751dd
by Jim MacArthur at 2018-08-14T16:29:54Z
```
sandbox/__init__.py: Add SandboxRemote
```

cc6319b2

by Jim MacArthur at 2018-08-14T16:29:54Z

Make SandboxRemote the default for linux targets

25 changed files:

.gitlab-ci.yml
buildstream/__init__.py
buildstream/_artifactcache/artifactcache.py
buildstream/_artifactcache/cascache.py
buildstream/_pipeline.py
buildstream/_platform/linux.py
buildstream/buildelement.py
buildstream/element.py
+ buildstream/element_enums.py
buildstream/plugins/sources/git.py
buildstream/sandbox/__init__.py
buildstream/sandbox/_mount.py
buildstream/sandbox/_sandboxbwrap.py
buildstream/sandbox/_sandboxchroot.py
+ buildstream/sandbox/_sandboxremote.py
buildstream/sandbox/sandbox.py
+ buildstream/storage/_casbaseddirectory.py
buildstream/storage/_filebaseddirectory.py
buildstream/storage/directory.py
setup.py
+ tests/sandboxes/storage-test/original/bin/bash
+ tests/sandboxes/storage-test/original/bin/hello
+ tests/sandboxes/storage-test/overlay/bin/bash
+ tests/sandboxes/storage-tests.py
tests/sources/git.py

Changes:

.gitlab-ci.yml

@@ -26,15 +26,6 @@ source_dist:
    - tar -ztf dist/*
    - tarball=$(cd dist && echo $(ls *))
 -  # Create an installer script
 -  - |
 -    cat > dist/install.sh << EOF
 -    #!/bin/sh
 -    tar -zxf ${tarball}
 -    cd ${tarball%.tar.gz}
 -    pip3 install --no-index .
 -    EOF
+-
    # unpack tarball as `dist/buildstream` directory
    - |
      cat > dist/unpack.sh << EOF
@@ -44,7 +35,6 @@ source_dist:
      EOF
    # Make our helpers executable
 -  - chmod +x dist/install.sh
    - chmod +x dist/unpack.sh
    artifacts:
      paths:

buildstream/__init__.py

@@ -30,6 +30,7 @@ if "_BST_COMPLETION" not in os.environ:
      from .sandbox import Sandbox, SandboxFlags
      from .plugin import Plugin
      from .source import Source, SourceError, Consistency, SourceFetcher
 -    from .element import Element, ElementError, Scope
 +    from .element import Element, ElementError
 +    from .element_enums import Scope
      from .buildelement import BuildElement
      from .scriptelement import ScriptElement

buildstream/_artifactcache/artifactcache.py

@@ -21,7 +21,7 @@ import os
  import string
  from collections import Mapping, namedtuple
 -from ..element import _KeyStrength
 +from ..element_enums import _KeyStrength
  from .._exceptions import ArtifactError, ImplError, LoadError, LoadErrorReason
  from .._message import Message, MessageType
  from .. import utils

buildstream/_artifactcache/cascache.py

@@ -213,6 +213,29 @@ class CASCache(ArtifactCache):
              remotes_for_project = self._remotes[element._get_project()]
              return any(remote.spec.push for remote in remotes_for_project)
 +    def pull_key(self, key, size_bytes, project):
 +        """ Pull a single key rather than an artifact.
 +        Does not update local refs. """
++
 +        for remote in self._remotes[project]:
 +            try:
 +                remote.init()
++
 +                tree = remote_execution_pb2.Digest()
 +                tree.hash = key
 +                tree.size_bytes = size_bytes
++
 +                self._fetch_directory(remote, tree)
++
 +                # no need to pull from additional remotes
 +                return True
++
 +            except grpc.RpcError as e:
 +                if e.code() != grpc.StatusCode.NOT_FOUND:
 +                    raise
++
 +        return False
++
      def pull(self, element, key, *, progress=None):
          ref = self.get_artifact_fullname(element, key)
@@ -254,10 +277,93 @@ class CASCache(ArtifactCache):
          self.set_ref(newref, tree)
 +    def _push_refs_to_remote(self, refs, remote, may_have_dependencies):
 +        skipped_remote = True
++
 +        try:
 +            for ref in refs:
 +                tree = self.resolve_ref(ref)
++
 +                # Check whether ref is already on the server in which case
 +                # there is no need to push the artifact
 +                try:
 +                    request = buildstream_pb2.GetReferenceRequest()
 +                    request.key = ref
 +                    response = remote.ref_storage.GetReference(request)
++
 +                    if response.digest.hash == tree.hash and response.digest.size_bytes == tree.size_bytes:
 +                        # ref is already on the server with the same tree
 +                        continue
++
 +                except grpc.RpcError as e:
 +                    if e.code() != grpc.StatusCode.NOT_FOUND:
 +                        # Intentionally re-raise RpcError for outer except block.
 +                        raise
++
 +                missing_blobs = {}
 +                required_blobs = self._required_blobs(tree, may_have_dependencies)
++
 +                # Limit size of FindMissingBlobs request
 +                for required_blobs_group in _grouper(required_blobs, 512):
 +                    request = remote_execution_pb2.FindMissingBlobsRequest()
++
 +                    for required_digest in required_blobs_group:
 +                        d = request.blob_digests.add()
 +                        d.hash = required_digest.hash
 +                        d.size_bytes = required_digest.size_bytes
++
 +                    response = remote.cas.FindMissingBlobs(request)
 +                    for digest in response.missing_blob_digests:
 +                        d = remote_execution_pb2.Digest()
 +                        d.hash = digest.hash
 +                        d.size_bytes = digest.size_bytes
 +                        missing_blobs[d.hash] = d
++
 +                # Upload any blobs missing on the server
 +                skipped_remote = False
 +                for digest in missing_blobs.values():
 +                    def request_stream():
 +                        resource_name = os.path.join(digest.hash, str(digest.size_bytes))
 +                        with open(self.objpath(digest), 'rb') as f:
 +                            assert os.fstat(f.fileno()).st_size == digest.size_bytes
 +                            offset = 0
 +                            finished = False
 +                            remaining = digest.size_bytes
 +                            while not finished:
 +                                chunk_size = min(remaining, 64 * 1024)
 +                                remaining -= chunk_size
++
 +                                request = bytestream_pb2.WriteRequest()
 +                                request.write_offset = offset
 +                                # max. 64 kB chunks
 +                                request.data = f.read(chunk_size)
 +                                request.resource_name = resource_name
 +                                request.finish_write = remaining <= 0
 +                                yield request
 +                                offset += chunk_size
 +                                finished = request.finish_write
 +                    response = remote.bytestream.Write(request_stream())
++
 +                request = buildstream_pb2.UpdateReferenceRequest()
 +                request.keys.append(ref)
 +                request.digest.hash = tree.hash
 +                request.digest.size_bytes = tree.size_bytes
 +                remote.ref_storage.UpdateReference(request)
++
 +        except grpc.RpcError as e:
 +            if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED:
 +                raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
++
 +        return not skipped_remote
++
      def push(self, element, keys):
 +        keys = list(keys)
          refs = [self.get_artifact_fullname(element, key) for key in keys]
          project = element._get_project()
 +        return self.push_refs(refs, project, element=element)
++
 +    def push_refs(self, refs, project, may_have_dependencies=True, element=None):
          push_remotes = [r for r in self._remotes[project] if r.spec.push]
@@ -265,94 +371,52 @@ class CASCache(ArtifactCache):
          for remote in push_remotes:
              remote.init()
 -            skipped_remote = True
 -            element.info("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
+-
 -            try:
 -                for ref in refs:
 -                    tree = self.resolve_ref(ref)
+-
 -                    # Check whether ref is already on the server in which case
 -                    # there is no need to push the artifact
 -                    try:
 -                        request = buildstream_pb2.GetReferenceRequest()
 -                        request.key = ref
 -                        response = remote.ref_storage.GetReference(request)
+-
 -                        if response.digest.hash == tree.hash and response.digest.size_bytes == tree.size_bytes:
 -                            # ref is already on the server with the same tree
 -                            continue
+-
 -                    except grpc.RpcError as e:
 -                        if e.code() != grpc.StatusCode.NOT_FOUND:
 -                            # Intentionally re-raise RpcError for outer except block.
 -                            raise
+-
 -                    missing_blobs = {}
 -                    required_blobs = self._required_blobs(tree)
+-
 -                    # Limit size of FindMissingBlobs request
 -                    for required_blobs_group in _grouper(required_blobs, 512):
 -                        request = remote_execution_pb2.FindMissingBlobsRequest()
+-
 -                        for required_digest in required_blobs_group:
 -                            d = request.blob_digests.add()
 -                            d.hash = required_digest.hash
 -                            d.size_bytes = required_digest.size_bytes
+-
 -                        response = remote.cas.FindMissingBlobs(request)
 -                        for digest in response.missing_blob_digests:
 -                            d = remote_execution_pb2.Digest()
 -                            d.hash = digest.hash
 -                            d.size_bytes = digest.size_bytes
 -                            missing_blobs[d.hash] = d
+-
 -                    # Upload any blobs missing on the server
 -                    skipped_remote = False
 -                    for digest in missing_blobs.values():
 -                        def request_stream():
 -                            resource_name = os.path.join(digest.hash, str(digest.size_bytes))
 -                            with open(self.objpath(digest), 'rb') as f:
 -                                assert os.fstat(f.fileno()).st_size == digest.size_bytes
 -                                offset = 0
 -                                finished = False
 -                                remaining = digest.size_bytes
 -                                while not finished:
 -                                    chunk_size = min(remaining, 64 * 1024)
 -                                    remaining -= chunk_size
+-
 -                                    request = bytestream_pb2.WriteRequest()
 -                                    request.write_offset = offset
 -                                    # max. 64 kB chunks
 -                                    request.data = f.read(chunk_size)
 -                                    request.resource_name = resource_name
 -                                    request.finish_write = remaining <= 0
 -                                    yield request
 -                                    offset += chunk_size
 -                                    finished = request.finish_write
 -                        response = remote.bytestream.Write(request_stream())
+-
 -                    request = buildstream_pb2.UpdateReferenceRequest()
 -                    request.keys.append(ref)
 -                    request.digest.hash = tree.hash
 -                    request.digest.size_bytes = tree.size_bytes
 -                    remote.ref_storage.UpdateReference(request)
+-
 -                    pushed = True
+-
 -            except grpc.RpcError as e:
 -                if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED:
 -                    raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
+-
 -            if skipped_remote:
 +            if self._push_refs_to_remote(refs, remote, may_have_dependencies):
 +                pushed = True
 +            elif element:
                  self.context.message(Message(
                      None,
                      MessageType.SKIPPED,
                      "Remote ({}) already has {} cached".format(
                          remote.spec.url, element._get_brief_display_key())
                  ))
++
          return pushed
 +    def verify_key_pushed(self, key, project):
 +        ref = key
 +        push_remotes = [r for r in self._remotes[project] if r.spec.push]
++
 +        pushed = False
++
 +        for remote in push_remotes:
 +            remote.init()
++
 +            if self._verify_ref_on_remote(ref, remote):
 +                pushed = True
++
 +        return pushed
++
 +    def _verify_ref_on_remote(self, ref, remote):
 +        tree = self.resolve_ref(ref)
++
 +        # Check whether ref is already on the server in which case
 +        # there is no need to push the artifact
 +        try:
 +            request = buildstream_pb2.GetReferenceRequest()
 +            request.key = ref
 +            response = remote.ref_storage.GetReference(request)
++
 +            if response.digest.hash == tree.hash and response.digest.size_bytes == tree.size_bytes:
 +                # ref is already on the server with the same tree
 +                return True
++
 +        except grpc.RpcError as e:
 +            if e.code() != grpc.StatusCode.NOT_FOUND:
 +                raise
++
 +        return False
++
      ################################################
      #                API Private Methods           #
      ################################################
@@ -726,26 +790,27 @@ class CASCache(ArtifactCache):
+             #
              q.put(str(e))
 -    def _required_blobs(self, tree):
 +    def _required_blobs(self, tree, may_have_dependencies=True):
          # parse directory, and recursively add blobs
          d = remote_execution_pb2.Digest()
          d.hash = tree.hash
          d.size_bytes = tree.size_bytes
          yield d
 -        directory = remote_execution_pb2.Directory()
 +        if may_have_dependencies:
 +            directory = remote_execution_pb2.Directory()
 -        with open(self.objpath(tree), 'rb') as f:
 -            directory.ParseFromString(f.read())
 +            with open(self.objpath(tree), 'rb') as f:
 +                directory.ParseFromString(f.read())
 -        for filenode in directory.files:
 -            d = remote_execution_pb2.Digest()
 -            d.hash = filenode.digest.hash
 -            d.size_bytes = filenode.digest.size_bytes
 -            yield d
 +            for filenode in directory.files:
 +                d = remote_execution_pb2.Digest()
 +                d.hash = filenode.digest.hash
 +                d.size_bytes = filenode.digest.size_bytes
 +                yield d
 -        for dirnode in directory.directories:
 -            yield from self._required_blobs(dirnode.digest)
 +            for dirnode in directory.directories:
 +                yield from self._required_blobs(dirnode.digest)
      def _fetch_blob(self, remote, digest, out):
          resource_name = os.path.join(digest.hash, str(digest.size_bytes))
@@ -846,6 +911,9 @@ class _CASRemote():
  def _grouper(iterable, n):
 -    # pylint: disable=stop-iteration-return
      while True:
 -        yield itertools.chain([next(iterable)], itertools.islice(iterable, n - 1))
 +        try:
 +            current = next(iterable)
 +        except StopIteration:
 +            return
 +        yield itertools.chain([current], itertools.islice(iterable, n - 1))

buildstream/_pipeline.py

@@ -358,10 +358,24 @@ class Pipeline():
                      inconsistent.append(element)
          if inconsistent:
 -            detail = "Exact versions are missing for the following elements\n" + \
 -                     "Try tracking these elements first with `bst track`\n\n"
 +            detail = "Exact versions are missing for the following elements:\n\n"
++
 +            missingTrack = 0
              for element in inconsistent:
 -                detail += "  " + element._get_full_name() + "\n"
 +                detail += "  " + element._get_full_name()
 +                for source in element.sources():
 +                    if not source._get_consistency() and not source.get_ref():
 +                        if hasattr(source, 'tracking') and source.tracking is None:
 +                            detail += ": Source {} is missing ref and track. ".format(source._get_full_name()) + \
 +                                      "Please specify a ref or branch/tag to track."
 +                            missingTrack = 1
++
 +                detail += "\n"
++
 +            if missingTrack:
 +                detail += "\nThen track these elements with `bst track`\n"
 +            else:
 +                detail += "\nTry tracking these elements first with `bst track`\n"
              raise PipelineError("Inconsistent pipeline", detail=detail, reason="inconsistent-pipeline")
      #############################################################

buildstream/_platform/linux.py

@@ -24,6 +24,7 @@ from .. import utils
  from .._artifactcache.cascache import CASCache
  from .._message import Message, MessageType
  from ..sandbox import SandboxBwrap
 +from ..sandbox import SandboxRemote
  from . import Platform
@@ -46,7 +47,7 @@ class Linux(Platform):
          # Inform the bubblewrap sandbox as to whether it can use user namespaces or not
          kwargs['user_ns_available'] = self._user_ns_available
          kwargs['die_with_parent_available'] = self._die_with_parent_available
 -        return SandboxBwrap(*args, **kwargs)
 +        return SandboxRemote(*args, **kwargs)
      ################################################
      #              Private Methods                 #

buildstream/buildelement.py

@@ -155,6 +155,9 @@ class BuildElement(Element):
              command_dir = build_root
          sandbox.set_work_directory(command_dir)
 +        # Tell sandbox which directory is preserved in the finished artifact
 +        sandbox.set_output_directory(install_root)
++
          # Setup environment
          sandbox.set_environment(self.get_environment())

buildstream/element.py

@@ -78,7 +78,6 @@ import stat
  import copy
  from collections import Mapping, OrderedDict
  from contextlib import contextmanager
 -from enum import Enum
  import tempfile
  import shutil
@@ -97,41 +96,9 @@ from ._platform import Platform
  from .sandbox._config import SandboxConfig
  from .storage.directory import Directory
 -from .storage._filebaseddirectory import FileBasedDirectory, VirtualDirectoryError
+-
+-
 -# _KeyStrength():
 -#
 -# Strength of cache key
 -#
 -class _KeyStrength(Enum):
+-
 -    # Includes strong cache keys of all build dependencies and their
 -    # runtime dependencies.
 -    STRONG = 1
+-
 -    # Includes names of direct build dependencies but does not include
 -    # cache keys of dependencies.
 -    WEAK = 2
+-
+-
 -class Scope(Enum):
 -    """Types of scope for a given element"""
+-
 -    ALL = 1
 -    """All elements which the given element depends on, following
 -    all elements required for building. Including the element itself.
 -    """
+-
 -    BUILD = 2
 -    """All elements required for building the element, including their
 -    respective run dependencies. Not including the given element itself.
 -    """
+-
 -    RUN = 3
 -    """All elements required for running the element. Including the element
 -    itself.
 -    """
 +from .storage._filebaseddirectory import FileBasedDirectory
 +from .storage.directory import VirtualDirectoryError
 +from .element_enums import _KeyStrength, Scope
  class ElementError(BstError):
@@ -1579,6 +1546,8 @@ class Element(Plugin):
                  finally:
                      if collect is not None:
                          try:
 +                            # Sandbox will probably have replaced its virtual directory, so get it again
 +                            sandbox_vroot = sandbox.get_virtual_directory()
                              collectvdir = sandbox_vroot.descend(collect.lstrip(os.sep).split(os.sep))
                          except VirtualDirectoryError:
                              # No collect directory existed

buildstream/element_enums.py

 +#
 +#  Copyright (C) 2018 Bloomberg LLC
 +#
 +#  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:
 +#        Tristan Van Berkom <tristan vanberkom codethink co uk>
 +#        Jim MacArthur <jim macarthur codethink co uk>
++
 +"""
 +Element - Globally visible enumerations
 +=======================================
++
 +"""
++
 +from enum import Enum
++
++
 +# KeyStrength():
 +#
 +# Strength of cache key
 +#
 +class _KeyStrength(Enum):
++
 +    # Includes strong cache keys of all build dependencies and their
 +    # runtime dependencies.
 +    STRONG = 1
++
 +    # Includes names of direct build dependencies but does not include
 +    # cache keys of dependencies.
 +    WEAK = 2
++
++
 +class Scope(Enum):
 +    """Types of scope for a given element"""
++
 +    ALL = 1
 +    """All elements which the given element depends on, following
 +    all elements required for building. Including the element itself.
 +    """
++
 +    BUILD = 2
 +    """All elements required for building the element, including their
 +    respective run dependencies. Not including the given element itself.
 +    """
++
 +    RUN = 3
 +    """All elements required for running the element. Including the element
 +    itself.
 +    """

buildstream/plugins/sources/git.py

@@ -363,6 +363,12 @@ class GitSource(Source):
          # If self.tracking is not specified it's not an error, just silently return
          if not self.tracking:
 +            # Is there a better way to check if a ref is given.
 +            if self.mirror.ref is None:
 +                detail = 'Without a tracking branch ref can not be updated. Please ' + \
 +                         'provide a ref or a track.'
 +                raise SourceError("{}: No track or ref".format(self),
 +                                  detail=detail, reason="track-attempt-no-track")
              return None
          with self.timed_activity("Tracking {} from {}"

buildstream/sandbox/__init__.py

@@ -20,3 +20,4 @@
  from .sandbox import Sandbox, SandboxFlags
  from ._sandboxchroot import SandboxChroot
  from ._sandboxbwrap import SandboxBwrap
 +from ._sandboxremote import SandboxRemote

buildstream/sandbox/_mount.py

@@ -32,8 +32,10 @@ from .._fuse import SafeHardlinks
  class Mount():
      def __init__(self, sandbox, mount_point, safe_hardlinks):
          scratch_directory = sandbox._get_scratch_directory()
 -        # Getting external_directory here is acceptable as we're part of the sandbox code.
 -        root_directory = sandbox.get_virtual_directory().external_directory
 +        # Getting _get_underlying_directory() here is acceptable as
 +        # we're part of the sandbox code. This will fail if our
 +        # directory is CAS-based.
 +        root_directory = sandbox.get_virtual_directory()._get_underlying_directory()
          self.mount_point = mount_point
          self.safe_hardlinks = safe_hardlinks

buildstream/sandbox/_sandboxbwrap.py

@@ -58,7 +58,7 @@ class SandboxBwrap(Sandbox):
          stdout, stderr = self._get_output()
          # Allowable access to underlying storage as we're part of the sandbox
 -        root_directory = self.get_virtual_directory().external_directory
 +        root_directory = self.get_virtual_directory()._get_underlying_directory()
          # Fallback to the sandbox default settings for
          # the cwd and env.
@@ -243,6 +243,7 @@ class SandboxBwrap(Sandbox):
                          # a bug, bwrap mounted a tempfs here and when it exits, that better be empty.
                          pass
 +        self._vdir.mark_changed()
          return exit_code
      def run_bwrap(self, argv, stdin, stdout, stderr, env, interactive):

buildstream/sandbox/_sandboxchroot.py

@@ -106,6 +106,7 @@ class SandboxChroot(Sandbox):
              status = self.chroot(rootfs, command, stdin, stdout,
                                   stderr, cwd, env, flags)
 +        self._vdir.mark_changed()
          return status
      # chroot()

buildstream/sandbox/_sandboxremote.py

 +#!/usr/bin/env python3
 +#
 +#  Copyright (C) 2018 Codethink Limited
 +#
 +#  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>
++
 +import os
 +import time
++
 +import grpc
++
 +from . import Sandbox
 +from ..storage._filebaseddirectory import FileBasedDirectory
 +from ..storage._casbaseddirectory import CasBasedDirectory
 +from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
 +from .._protos.google.longrunning import operations_pb2, operations_pb2_grpc
++
 +from .._artifactcache.cascache import CASCache
++
++
 +class SandboxError(Exception):
 +    pass
++
++
 +# SandboxRemote()
 +#
 +# This isn't really a sandbox, it's a stub which sends all the source
 +# to a remote server and retrieves the results from it.
 +#
 +class SandboxRemote(Sandbox):
++
 +    def __init__(self, *args, **kwargs):
 +        super().__init__(*args, **kwargs)
 +        self.user_ns_available = kwargs['user_ns_available']
 +        self.die_with_parent_available = kwargs['die_with_parent_available']
 +        self.cascache = None
++
 +    def _get_cascache(self):
 +        if self.cascache is None:
 +            self.cascache = CASCache(self._get_context())
 +            self.cascache.setup_remotes(use_config=True)
 +        return self.cascache
++
 +    def __run_remote_command(self, cascache, command, input_root_digest, environment):
++
 +        environment_variables = [ remote_execution_pb2.Command.
 +                                  EnvironmentVariable(name=k, value=v)
 +                                  for (k,v) in environment.items() ]
++
 +        # Create and send the Command object.
 +        remote_command = remote_execution_pb2.Command(arguments=command, environment_variables=environment_variables,
 +                                                      output_files=[],
 +                                                      output_directories=[self._output_directory],
 +                                                      platform=None)
 +        command_digest = cascache.add_object(buffer=remote_command.SerializeToString())
 +        command_ref = 'worker-command/{}'.format(command_digest.hash)
 +        cascache.set_ref(command_ref, command_digest)
++
 +        command_push_successful = cascache.push_refs([command_ref], self._get_project(), may_have_dependencies=False)
 +        if not command_push_successful and not cascache.verify_key_pushed(command_ref, self._get_project()):
 +            # Command push failed
 +            return None
++
 +        # Create and send the action.
++
 +        action = remote_execution_pb2.Action(command_digest=command_digest,
 +                                             input_root_digest=input_root_digest,
 +                                             timeout=None,
 +                                             do_not_cache=True)
++
 +        action_digest = cascache.add_object(buffer=action.SerializeToString())
 +        action_ref = 'worker-action/{}'.format(command_digest.hash)
 +        cascache.set_ref(action_ref, action_digest)
 +        action_push_successful = cascache.push_refs([action_ref], self._get_project(), may_have_dependencies=False)
++
 +        if not action_push_successful and not cascache.verify_key_pushed(action_ref, self._get_project()):
 +            # Action push failed
 +            return None
++
 +        # Next, try to create a communication channel to the BuildGrid server.
 +        port = 50051
 +        channel = grpc.insecure_channel('dekatron.office.codethink.co.uk:{}'.format(port))
 +        stub = remote_execution_pb2_grpc.ExecutionStub(channel)
++
 +        request = remote_execution_pb2.ExecuteRequest(instance_name='default',
 +                                                      action_digest=action_digest,
 +                                                      skip_cache_lookup=True)
++
 +        operation_iterator = stub.Execute(request)
++
 +        for operation in operation_iterator:
 +            if operation.done:
 +                break
 +            # TODO: Do we need a sleep here?
 +        return operation
++
 +    def process_job_output(self, output_directories, output_files):
 +        # output_directories is an array of OutputDirectory objects.
 +        # output_files is an array of OutputFile objects.
 +        #
 +        # We only specify one output_directory, so it's an error
 +        # for there to be any output files or more than one directory at the moment.
++
 +        if output_files:
 +            raise SandboxError("Output files were returned when we didn't request any.")
 +        elif len(output_directories) > 1:
 +            error_text = "More than one output directory was returned from the build server: {}"
 +            raise SandboxError(error_text.format(output_directories))
 +        elif len(output_directories) < 1:
 +            error_text = "No output directory was returned from the build server."
 +            raise SandboxError(error_test)
++
 +        digest = output_directories[0].tree_digest
 +        if digest is None or digest.hash is None or digest.hash == "":
 +            raise SandboxError("Output directory structure had no digest attached.")
++
 +        # Now do a pull to ensure we have the necessary parts.
 +        cascache = self._get_cascache()
 +        cascache.pull_key(digest.hash, digest.size_bytes, self._get_project())
 +        path_components = os.path.split(self._output_directory)
++
 +        # Now what we have is a digest for the output. Once we return, the calling process will
 +        # attempt to descend into our directory and find that directory, so we need to overwrite
 +        # that.
++
 +        if not path_components:
 +            # The artifact wants the whole directory; we could just return the returned hash in its
 +            # place, but we don't have a means to do that yet.
 +            raise SandboxError("Unimplemented: Output directory is empty or equal to the sandbox root.")
++
 +        # At the moment, we will get the whole directory back in the first directory argument and we need
 +        # to replace the sandbox's virtual directory with that. Creating a new virtual directory object
 +        # from another hash will be interesting, though...
++
 +        new_dir = CasBasedDirectory(self._get_context(), ref=digest)
 +        self.set_virtual_directory(new_dir)
++
 +    def run(self, command, flags, *, cwd=None, env=None):
 +        # Upload sources
 +        upload_vdir = self.get_virtual_directory()
++
 +        if isinstance(upload_vdir, FileBasedDirectory):
 +            # Make a new temporary directory to put source in
 +            upload_vdir = CasBasedDirectory(self._get_context(), ref=None)
 +            upload_vdir.import_files(self.get_virtual_directory()._get_underlying_directory())
++
++
 +        # Now, push that key (without necessarily needing a ref) to the remote.
 +        cascache = self._get_cascache()
++
 +        ref = 'worker-source/{}'.format(upload_vdir.ref.hash)
 +        upload_vdir._save(ref)
 +        source_push_successful = cascache.push_refs([ref], self._get_project())
++
 +        # Set up environment and PWD
 +        if env is None:
 +            env = self._get_environment()
 +        if 'PWD' not in env:
 +            env['PWD'] = self._get_work_directory()
++
 +        # We want command args as a list of strings
 +        if isinstance(command, str):
 +            command = [command]
++
 +        # Now transmit the command to execute
 +        if source_push_successful or cascache.verify_key_pushed(ref, self._get_project()):
 +            response = self.__run_remote_command(cascache, command, upload_vdir.ref, env)
++
 +            if response is None or response.HasField("error"):
 +                # Build failed, so return a failure code
 +                return 1
 +            else:
++
 +                # At the moment, response can either be an
 +                # ExecutionResponse containing an ActionResult, or an
 +                # ActionResult directly.
 +                executeResponse = remote_execution_pb2.ExecuteResponse()
 +                if response.response.Is(executeResponse.DESCRIPTOR):
 +                    # Unpack ExecuteResponse and set response to its response
 +                    response.response.Unpack(executeResponse)
 +                    response = executeResponse
++
 +                actionResult = remote_execution_pb2.ActionResult()
 +                if response.response.Is(actionResult.DESCRIPTOR):
 +                    response.response.Unpack(actionResult)
 +                    self.process_job_output(actionResult.output_directories, actionResult.output_files)
 +                else:
 +                    raise SandboxError("Received unknown message from server (expected ExecutionResponse).")
 +        else:
 +            raise SandboxError("Failed to verify that source has been pushed to the remote artifact cache.")
 +        return 0

buildstream/sandbox/sandbox.py

@@ -31,6 +31,7 @@ See also: :ref:`sandboxing`.
  import os
  from .._exceptions import ImplError, BstError
  from ..storage._filebaseddirectory import FileBasedDirectory
 +from ..storage._casbaseddirectory import CasBasedDirectory
  class SandboxFlags():
@@ -98,13 +99,16 @@ class Sandbox():
          self.__stdout = kwargs['stdout']
          self.__stderr = kwargs['stderr']
 -        # Setup the directories. Root should be available to subclasses, hence
 -        # being single-underscore. The others are private to this class.
 +        # Setup the directories. Root and output_directory should be
 +        # available to subclasses, hence being single-underscore. The
 +        # others are private to this class.
          self._root = os.path.join(directory, 'root')
 +        self._output_directory = None
          self.__directory = directory
          self.__scratch = os.path.join(self.__directory, 'scratch')
          for directory_ in [self._root, self.__scratch]:
              os.makedirs(directory_, exist_ok=True)
 +        self._vdir = None
      def get_directory(self):
          """Fetches the sandbox root directory
@@ -132,15 +136,39 @@ class Sandbox():
          Returns:
             (str): The sandbox root directory
 +        """
 +        if not self._vdir:
 +            # BST_CAS_DIRECTORIES is a deliberately hidden environment variable which
 +            # can be used to switch on CAS-based directories for testing.
 +            if 'BST_CAS_DIRECTORIES' in os.environ:
 +                self._vdir = CasBasedDirectory(self.__context, ref=None)
 +            else:
 +                self._vdir = FileBasedDirectory(self._root)
 +        return self._vdir
++
 +    def set_virtual_directory(self, vdir):
 +        """ Sets virtual directory. Useful after remote execution
 +        has rewritten the working directory. """
 +        self._vdir = vdir
++
 +    def get_virtual_toplevel_directory(self):
 +        """Fetches the sandbox's toplevel directory
++
 +        The toplevel directory contains 'root', 'scratch' and later
 +        'artifact' where output is copied to.
++
 +        Returns:
 +           (str): The sandbox toplevel directory
++
          """
          # For now, just create a new Directory every time we're asked
 -        return FileBasedDirectory(self._root)
 +        return FileBasedDirectory(self.__directory)
      def set_environment(self, environment):
          """Sets the environment variables for the sandbox
          Args:
 -           directory (dict): The environment variables to use in the sandbox
 +           environment (dict): The environment variables to use in the sandbox
          """
          self.__env = environment
@@ -152,6 +180,15 @@ class Sandbox():
          """
          self.__cwd = directory
 +    def set_output_directory(self, directory):
 +        """Sets the output directory - the directory which is preserved
 +        as an artifact after assembly.
++
 +        Args:
 +           directory (str): An absolute path within the sandbox
 +        """
 +        self._output_directory = directory
++
      def mark_directory(self, directory, *, artifact=False):
          """Marks a sandbox directory and ensures it will exist

buildstream/storage/_casbaseddirectory.py

 +#
 +#  Copyright (C) 2018 Bloomberg LLC
 +#
 +#  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 _save(self, name):
 +        """ Saves this directory into the content cache as a named ref. This function is not
 +        currently in use, but may be useful later. """
 +        self._recalculate_recursing_up()
 +        self._recalculate_recursing_down()
 +        (rel_refpath, refname) = os.path.split(name)
 +        refdir = os.path.join(self.cas_directory, 'refs', 'heads', rel_refpath)
 +        refname = os.path.join(refdir, refname)
++
 +        if not os.path.exists(refdir):
 +            os.makedirs(refdir)
 +        with open(refname, "wb") as f:
 +            f.write(self.ref.SerializeToString())
++
 +    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.")

buildstream/storage/_filebaseddirectory.py

@@ -29,25 +29,12 @@ See also: :ref:`sandboxing`.
  import os
  import time
 -from .._exceptions import BstError, ErrorDomain
 -from .directory import Directory
 +from .directory import Directory, VirtualDirectoryError
  from ..utils import link_files, copy_files, list_relative_paths, _get_link_mtime, _magic_timestamp
  from ..utils import _set_deterministic_user, _set_deterministic_mtime
+-
 -class VirtualDirectoryError(BstError):
 -    """Raised by Directory functions when system calls fail.
 -    This will be handled internally by the BuildStream core,
 -    if you need to handle this error, then it should be reraised,
 -    or either of the :class:`.ElementError` or :class:`.SourceError`
 -    exceptions should be raised from this error.
 -    """
 -    def __init__(self, message, reason=None):
 -        super().__init__(message, domain=ErrorDomain.VIRTUAL_FS, reason=reason)
+-
+-
  # FileBasedDirectory intentionally doesn't call its superclass constuctor,
 -# which is mean to be unimplemented.
 +# which is meant to be unimplemented.
  # pylint: disable=super-init-not-called
@@ -108,7 +95,8 @@ class FileBasedDirectory(Directory):
              if create:
                  new_path = os.path.join(self.external_directory, subdirectory_spec[0])
                  os.makedirs(new_path, exist_ok=True)
 -                return FileBasedDirectory(new_path).descend(subdirectory_spec[1:], create)
 +                self.index[subdirectory_spec[0]] = FileBasedDirectory(new_path).descend(subdirectory_spec[1:], create)
 +                return self.index[subdirectory_spec[0]]
              else:
                  error = "No entry called '{}' found in the directory rooted at {}"
                  raise VirtualDirectoryError(error.format(subdirectory_spec[0], self.external_directory))
@@ -134,8 +122,12 @@ class FileBasedDirectory(Directory):
              for f in import_result.files_written:
                  os.utime(os.path.join(self.external_directory, f), times=(cur_time, cur_time))
 +        self.mark_changed()
          return import_result
 +    def mark_changed(self):
 +        self._directory_read = False
++
      def set_deterministic_mtime(self):
          _set_deterministic_mtime(self.external_directory)
@@ -214,3 +206,8 @@ class FileBasedDirectory(Directory):
          # which exposes the sandbox directory; we will have to assume for the time being
          # that people will not abuse __str__.
          return self.external_directory
++
 +    def _get_underlying_directory(self) -> str:
 +        """ Returns the underlying (real) file system directory this
 +        object refers to. """
 +        return self.external_directory

buildstream/storage/directory.py

@@ -31,6 +31,19 @@ See also: :ref:`sandboxing`.
  """
 +from .._exceptions import BstError, ErrorDomain
++
++
 +class VirtualDirectoryError(BstError):
 +    """Raised by Directory functions when system calls fail.
 +    This will be handled internally by the BuildStream core,
 +    if you need to handle this error, then it should be reraised,
 +    or either of the :class:`.ElementError` or :class:`.SourceError`
 +    exceptions should be raised from this error.
 +    """
 +    def __init__(self, message, reason=None):
 +        super().__init__(message, domain=ErrorDomain.VIRTUAL_FS, reason=reason)
++
  class Directory():
      def __init__(self, external_directory=None):
@@ -111,6 +124,14 @@ class Directory():
          """
          raise NotImplementedError()
 +    def mark_changed(self):
 +        """ Mark this directory as having been changed outside this API. This
 +        normally can only happen by calling the Sandbox's `run`
 +        method.
++
 +        """
 +        raise NotImplementedError()
++
      # Convenience functions
      def is_empty(self):
          """ Return true if this directory has no files, subdirectories or links in it.

@@ -272,6 +272,5 @@ setup(name='BuildStream',
                       'pytest-cov >= 2.5.0',
                       # Provide option to run tests in parallel, less reliable
                       'pytest-xdist',
 -                     'pytest >= 3.1.0',
 -                     'pylint >= 1.8 , < 2'],
 +                     'pytest >= 3.1.0'],
        zip_safe=False)

tests/sandboxes/storage-test/original/bin/bash

1

+This is the original /bin/bash.

tests/sandboxes/storage-test/original/bin/hello

1

+This is the original /bin/hello.

tests/sandboxes/storage-test/overlay/bin/bash

	1	+This is the replacement /bin/bash.
		\ No newline at end of file

tests/sandboxes/storage-tests.py

 +import os
 +import pytest
++
 +from buildstream._exceptions import ErrorDomain
++
 +from buildstream._context import Context
 +from buildstream.storage._casbaseddirectory import CasBasedDirectory
 +from buildstream.storage._filebaseddirectory import FileBasedDirectory
++
 +DATA_DIR = os.path.join(
 +    os.path.dirname(os.path.realpath(__file__)),
 +    "storage-test"
 +)
++
++
 +def setup_backend(backend_class, tmpdir):
 +    if backend_class == FileBasedDirectory:
 +        return backend_class(os.path.join(tmpdir, "vdir"))
 +    else:
 +        context = Context()
 +        context.artifactdir = os.path.join(tmpdir, "cas")
 +        return backend_class(context)
++
++
 +@pytest.mark.parametrize("backend", [
 +    FileBasedDirectory, CasBasedDirectory])
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_import(tmpdir, datafiles, backend):
 +    original = os.path.join(str(datafiles), "original")
++
 +    c = setup_backend(backend, str(tmpdir))
++
 +    c.import_files(original)
++
 +    assert("bin/bash" in c.list_relative_paths())
 +    assert("bin/hello" in c.list_relative_paths())
++
++
 +@pytest.mark.parametrize("backend", [
 +    FileBasedDirectory, CasBasedDirectory])
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_modified_file_list(tmpdir, datafiles, backend):
 +    original = os.path.join(str(datafiles), "original")
 +    overlay = os.path.join(str(datafiles), "overlay")
++
 +    c = setup_backend(backend, str(tmpdir))
++
 +    c.import_files(original)
++
 +    c.mark_unmodified()
++
 +    c.import_files(overlay)
++
 +    print("List of all paths in imported results: {}".format(c.list_relative_paths()))
 +    assert("bin/bash" in c.list_relative_paths())
 +    assert("bin/bash" in c.list_modified_paths())
 +    assert("bin/hello" not in c.list_modified_paths())

tests/sources/git.py

@@ -359,3 +359,45 @@ def test_submodule_track_ignore_inconsistent(cli, tmpdir, datafiles):
      # Assert that we are just fine without it, and emit a warning to the user.
      assert "Ignoring inconsistent submodule" in result.stderr
++
++
 +@pytest.mark.skipif(HAVE_GIT is False, reason="git is not available")
 +@pytest.mark.datafiles(os.path.join(DATA_DIR, 'template'))
 +def test_submodule_track_no_ref_or_track(cli, tmpdir, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
++
 +    # Create the repo from 'repofiles' subdir
 +    repo = create_repo('git', str(tmpdir))
 +    ref = repo.create(os.path.join(project, 'repofiles'))
++
 +    # Write out our test target
 +    gitsource = repo.source_config(ref=None)
 +    gitsource.pop('track')
 +    element = {
 +        'kind': 'import',
 +        'sources': [
 +            gitsource
 +        ]
 +    }
++
 +    _yaml.dump(element, os.path.join(project, 'target.bst'))
++
 +    # Track will encounter an inconsistent submodule without any ref
 +    result = cli.run(project=project, args=['track', 'target.bst'])
 +    result.assert_main_error(ErrorDomain.STREAM, None)
 +    result.assert_task_error(ErrorDomain.SOURCE, 'track-attempt-no-track')
++
 +    # Assert that we are just fine without it, and emit a warning to the user.
 +    assert "FAILURE git source at" in result.stderr
 +    assert "Without a tracking branch ref can not be updated. Please " + \
 +        "provide a ref or a track." in result.stderr
++
 +    # Track will encounter an inconsistent submodule without any ref
 +    result = cli.run(project=project, args=['build', 'target.bst'])
 +    result.assert_main_error(ErrorDomain.PIPELINE, 'inconsistent-pipeline')
 +    result.assert_task_error(None, None)
++
 +    # Assert that we are just fine without it, and emit a warning to the user.
 +    assert "Exact versions are missing for the following elements" in result.stderr
 +    assert "is missing ref and track." in result.stderr
 +    assert "Then track these elements with `bst track`" in result.stderr

[Date Prev][Date Next] [Thread Prev][Thread Next] [Thread Index] [Date Index] [Author Index]