[Notes] [Git][BuildStream/buildstream][jmac/remote_execution

Martin Blanchard pushed to branch jmac/remote_execution_client at BuildStream / buildstream

Commits:

680e4fe1

by knownexus at 2018-08-30T13:37:55Z

Replacing string 'bzr' with value from host tools

8cd719eb

by Phillip Smyth at 2018-08-30T15:47:06Z

Merge branch 'bzr_fix' into 'master'

Replacing string 'bzr' with value from host tools

See merge request BuildStream/buildstream!764

18d0bfb4

by Tom Pollard at 2018-08-30T19:14:33Z

buildstream/_project.py: Report if project.conf is missing name

Explicitly check that project.conf contains a name. This resolves
the issue of the provenance check from _yaml.py incorrectly reporting
the offending file as the default_config_node projectconfig.yaml
when no name key exists in the pre_config_node dict.

95121148

by Tiago Gomes at 2018-08-30T20:52:29Z

Merge branch 'tpollard/591' into 'master'

buildstream/_project.py: Report if project.conf is missing name

See merge request BuildStream/buildstream!680

775d3fca

by Tristan Van Berkom at 2018-08-31T06:05:32Z

_project.py: Adding FIXME comment to address #591 properly

0682aa1c

by Josh Smith at 2018-08-31T10:58:12Z

job.py: Modify retry messages to be FAIL

This adjusts the message handler for the child processes to no longer
override the message type.

This also removes the ability for unhandled non BstError exceptions to
retry.

68eff9ba

by Josh Smith at 2018-08-31T10:58:20Z

tests: Modify tests to ensure retry FAILURE

9b18e1bf

by Tristan Van Berkom at 2018-08-31T11:22:51Z

Merge branch 'Qinusty/retries-should-fail' into 'master'

Retries log as failures

See merge request BuildStream/buildstream!766

e2d8fedf

by Josh Smith at 2018-08-31T15:53:43Z

scheduler.py: Move cache_size logs into folder

This prevents the cache_size.pid.log files from cluttering the root
log directory.

6805a2ab

by Tristan Van Berkom at 2018-09-01T08:24:05Z

Merge branch 'Qinusty/cache-size-directory' into 'master'

Move cache_size.pid.log files into a subdirectory of logs

See merge request BuildStream/buildstream!769

f8b06acc

by Tristan Van Berkom at 2018-09-02T09:33:39Z

source.py: Document Source.get_source_fetchers() to return an iterable

Also highlight the fact that the plugin can rely on the fetcher's
fetch() method getting called before consuming the next item in the
list, which is the magick behavior that the git plugin relies on.

This is a part of #620

559b6fbc

by Tristan Van Berkom at 2018-09-02T09:34:10Z

source.py: Fixing docs link formatting to be consistent.

43ad22d7

by Tristan Van Berkom at 2018-09-02T09:34:10Z

source.py: Documenting that marking download URLs is mandatory

A download URL must be interpreted by the core at `Plugin.configure()`
time, even if only employed later on.

This is a part of #620

16462e9c

by Tristan Van Berkom at 2018-09-02T09:36:49Z

plugin.py: Added _configure() and _get_configuring() private APIs

Keeps track of whether the plugin is currently being configured.

Adjusted Element and Source classes to call _configure() in place
of calling configure() directly.

This is a part of #620

1f9c4147

by Tristan Van Berkom at 2018-09-02T09:37:12Z

source.py: Added `primary` argument to URL marking APIs

The Source must now mention whether the marked or translated
URL is "primary" or not. Even when a Source may have multiple
URLs, the auxilliary URLs are derived from the primary one, not
only is this true for git, but it is mandated by our tracking
API which assumes there is a primary URL.

This adjusts the `git` source and the test `fetch_source.py` source
to behave properly and advertize it's primary URL properly.

This is a part of #620

f9b2f1a9

by Tristan Van Berkom at 2018-09-02T09:37:21Z

source.py: Track marked URLs and assert they are marked during Plugin.configure()

This cannot test for unaliased URLs, as those can be discovered later
on outside of user provided element configuration; at least we
assert that if an alias was used, we have seen it at load time.

This will cause a BUG to occur for a plugin which marks an aliased
URL (or attempts to translate one) outside of `Plugin.configure()`,
if that URL was not previously seen.

This is a part of #620

88460cd2

by Tristan Van Berkom at 2018-09-03T06:23:38Z

Merge branch 'tristan/source-fetcher-changes' into 'master'

Source fetcher changes

See merge request BuildStream/buildstream!772

2339f0c4

by Tom Pollard at 2018-09-03T09:44:36Z

plugins/git.py: Warn if ref is not in given track

Add a helper function assert_ref_in_track to git.py GitMirror()
which is used when staging & initing the source workspace. It
checks the element's ref exists in the track (branch/tag) if it
has been specified, raising a warning if necessary. The warning makes
use of the warning token 'REF_NOT_IN_TRACK' from the configurable
CoreWarnings. If the element has been tracked with bst, it is assumed
that the value of ref exists in the track as it was generated from it
& as such is not asserted.

c96fec5d

by Tom Pollard at 2018-09-03T09:44:36Z

tests/sources/git.py: Add tests for REF_NOT_IN_TRACK

Add tests that cover assert_ref_in_track & the configurable
CoreWarnings REF_NOT_IN_TRACK warnings token.

6a0cdedf

by Tom Pollard at 2018-09-03T10:16:39Z

Merge branch 'tpollard/483' into 'master'

plugins/git.py: Warn if ref is not in given track

See merge request BuildStream/buildstream!564

9bd3a625

by Jim MacArthur at 2018-09-04T09:46:02Z

cascache.py: Modifications necessary for remote execution

Add helper method for object pushing and pulling: push_directory() and
pull_tree() and a verifier verify_key_pushed().

https://gitlab.com/BuildStream/buildstream/issues/454

7e1f9bb3

by Martin Blanchard at 2018-09-04T09:46:46Z

_casbaseddirectory.py: Add a method for hash recalculation

https://gitlab.com/BuildStream/buildstream/issues/454

8a033d85

by Jim MacArthur at 2018-09-04T09:46:53Z

sandbox.py: Allow setting the virtual directory

This is for use after remote execution has finished, since remote
execution produces a new output directory rather than modifying
the initial directory.

https://gitlab.com/BuildStream/buildstream/issues/454

32773bf7

by Jim MacArthur at 2018-09-04T09:47:09Z

Add "remote-execution" project configuration option.

This just adds one option, "remote-execution/url". Affects multiple files.

https://gitlab.com/BuildStream/buildstream/issues/454

e7b26e07

by Jim MacArthur at 2018-09-04T09:47:15Z

doc/source/format_project.rst: Document remote-execution option

https://gitlab.com/BuildStream/buildstream/issues/454

ee47f659

by Jim MacArthur at 2018-09-04T09:47:22Z

_sandboxremote.py: Implement the REAPI client

The remote execution client is implemented as a remote sandbox that
sends sources and build commands to a REAPI server and fetches results
once remotely executed. New file.

https://gitlab.com/BuildStream/buildstream/issues/454

28bd46d6

by Jim MacArthur at 2018-09-04T09:49:04Z

sandbox/__init__.py: Add SandboxRemote

https://gitlab.com/BuildStream/buildstream/issues/454

db1b357b

by Jim MacArthur at 2018-09-04T09:49:11Z

element.py: Get the updated virtual directory after running

Executing run() on a sandbox can now replace the virtual directory,
since remote execution returns a potentially different directory rather
than an update to the existing one. Call get_virtual_directory() again
after running to accout for this.

https://gitlab.com/BuildStream/buildstream/issues/454

2a96530e

by Jim MacArthur at 2018-09-04T09:49:17Z

autotools.py: Mark this as a BST_VIRTUAL_DIRECTORY plugin

https://gitlab.com/BuildStream/buildstream/issues/454

ac116aed

by Jim MacArthur at 2018-09-04T09:49:22Z

element.py: Switch to SandboxRemote if config option is set

https://gitlab.com/BuildStream/buildstream/issues/454

22 changed files:

buildstream/_artifactcache/cascache.py
buildstream/_project.py
buildstream/_scheduler/jobs/job.py
buildstream/_scheduler/scheduler.py
buildstream/buildelement.py
buildstream/data/projectconfig.yaml
buildstream/element.py
buildstream/plugin.py
buildstream/plugins/elements/autotools.py
buildstream/plugins/sources/git.py
buildstream/sandbox/__init__.py
+ buildstream/sandbox/_sandboxremote.py
buildstream/sandbox/sandbox.py
buildstream/source.py
buildstream/storage/_casbaseddirectory.py
doc/source/format_project.rst
tests/frontend/project/sources/fetch_source.py
tests/sources/deb.py
tests/sources/git.py
tests/sources/tar.py
tests/sources/zip.py
tests/testutils/repo/bzr.py

Changes:

buildstream/_artifactcache/cascache.py

@@ -19,6 +19,7 @@
  import hashlib
  import itertools
 +import io
  import multiprocessing
  import os
  import signal
@@ -66,6 +67,10 @@ class CASCache(ArtifactCache):
          self._calculate_cache_quota()
          self._enable_push = enable_push
 +        if self._enable_push:
 +            self._uuid = str(uuid.uuid4())
 +        else:
 +            self._uuid = None
          # Per-project list of _CASRemote instances.
          self._remotes = {}
@@ -76,6 +81,7 @@ class CASCache(ArtifactCache):
      ################################################
      #     Implementation of abstract methods       #
      ################################################
++
      def contains(self, element, key):
          refpath = self._refpath(self.get_artifact_fullname(element, key))
@@ -259,6 +265,25 @@ class CASCache(ArtifactCache):
          return False
 +    def pull_tree(self, project, digest):
 +        """ Pull a single Tree rather than an artifact.
 +        Does not update local refs. """
++
 +        for remote in self._remotes[project]:
 +            try:
 +                remote.init()
++
 +                digest = self._fetch_tree(remote, digest)
++
 +                # no need to pull from additional remotes
 +                return digest
++
 +            except grpc.RpcError as e:
 +                if e.code() != grpc.StatusCode.NOT_FOUND:
 +                    raise
++
 +        return None
++
      def link_key(self, element, oldkey, newkey):
          oldref = self.get_artifact_fullname(element, oldkey)
          newref = self.get_artifact_fullname(element, newkey)
@@ -267,106 +292,123 @@ class CASCache(ArtifactCache):
          self.set_ref(newref, tree)
 +    def _push_refs_to_remote(self, refs, remote):
 +        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
++
 +                self._send_directory(remote, tree)
++
 +                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):
 -        refs = [self.get_artifact_fullname(element, key) for key in keys]
          project = element._get_project()
 +        refs = [self.get_artifact_fullname(element, key) for key in list(keys)]
          push_remotes = [r for r in self._remotes[project] if r.spec.push]
          pushed = False
+-
          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():
 -                        uuid_ = uuid.uuid4()
 -                        resource_name = '/'.join(['uploads', str(uuid_), 'blobs',
 -                                                  digest.hash, str(digest.size_bytes)])
+-
 -                        def request_stream(resname):
 -                            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 = resname
 -                                    request.finish_write = remaining <= 0
 -                                    yield request
 -                                    offset += chunk_size
 -                                    finished = request.finish_write
 -                        response = remote.bytestream.Write(request_stream(resource_name))
+-
 -                    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 element:
 +                element.info("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
 +            if self._push_refs_to_remote(refs, remote):
 +                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 push_directory(self, project, directory):
++
 +        push_remotes = [r for r in self._remotes[project] if r.spec.push]
++
 +        if directory.ref is None:
 +            return None
++
 +        for remote in push_remotes:
 +            remote.init()
++
 +            self._send_directory(remote, directory.ref)
++
 +        return directory.ref
++
 +    def push_message(self, project, message):
++
 +        push_remotes = [r for r in self._remotes[project] if r.spec.push]
++
 +        message_buffer = message.SerializeToString()
 +        message_sha = hashlib.sha256(message_buffer)
 +        message_digest = remote_execution_pb2.Digest()
 +        message_digest.hash = message_sha.hexdigest()
 +        message_digest.size_bytes = len(message_buffer)
++
 +        for remote in push_remotes:
 +            remote.init()
++
 +            with io.BytesIO(message_buffer) as b:
 +                self._send_blob(remote, message_digest, b)
++
 +        return message_digest
++
 +    def _verify_digest_on_remote(self, remote, digest):
 +        # Check whether ref is already on the server in which case
 +        # there is no need to push the artifact
 +        request = remote_execution_pb2.FindMissingBlobsRequest()
 +        request.blob_digests.extend([digest])
++
 +        response = remote.cas.FindMissingBlobs(request)
 +        if digest in response.missing_blob_digests:
 +            return False
++
 +        return True
++
 +    def verify_key_pushed(self, project, digest):
++
 +        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_digest_on_remote(remote, digest):
 +                pushed = True
++
          return pushed
      ################################################
@@ -599,6 +641,7 @@ class CASCache(ArtifactCache):
      ################################################
      #             Local Private Methods            #
      ################################################
++
      def _checkout(self, dest, tree):
          os.makedirs(dest, exist_ok=True)
@@ -782,16 +825,16 @@ class CASCache(ArtifactCache):
          for dirnode in directory.directories:
              yield from self._required_blobs(dirnode.digest)
 -    def _fetch_blob(self, remote, digest, out):
 +    def _fetch_blob(self, remote, digest, stream):
          resource_name = '/'.join(['blobs', digest.hash, str(digest.size_bytes)])
          request = bytestream_pb2.ReadRequest()
          request.resource_name = resource_name
          request.read_offset = 0
          for response in remote.bytestream.Read(request):
 -            out.write(response.data)
 +            stream.write(response.data)
 +        stream.flush()
 -        out.flush()
 -        assert digest.size_bytes == os.fstat(out.fileno()).st_size
 +        assert digest.size_bytes == os.fstat(stream.fileno()).st_size
      def _fetch_directory(self, remote, tree):
          objpath = self.objpath(tree)
@@ -827,6 +870,92 @@ class CASCache(ArtifactCache):
              digest = self.add_object(path=out.name)
              assert digest.hash == tree.hash
 +    def _fetch_tree(self, remote, digest):
 +        # download but do not store the Tree object
 +        with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:
 +            self._fetch_blob(remote, digest, out)
++
 +            tree = remote_execution_pb2.Tree()
++
 +            with open(out.name, 'rb') as f:
 +                tree.ParseFromString(f.read())
++
 +            tree.children.extend([tree.root])
 +            for directory in tree.children:
 +                for filenode in directory.files:
 +                    fileobjpath = self.objpath(filenode.digest)
 +                    if os.path.exists(fileobjpath):
 +                        # already in local cache
 +                        continue
++
 +                    with tempfile.NamedTemporaryFile(dir=self.tmpdir) as f:
 +                        self._fetch_blob(remote, filenode.digest, f)
++
 +                        added_digest = self.add_object(path=f.name)
 +                        assert added_digest.hash == filenode.digest.hash
++
 +                # place directory blob only in final location when we've downloaded
 +                # all referenced blobs to avoid dangling references in the repository
 +                dirbuffer = directory.SerializeToString()
 +                dirdigest = self.add_object(buffer=dirbuffer)
 +                assert dirdigest.size_bytes == len(dirbuffer)
++
 +        return dirdigest
++
 +    def _send_blob(self, remote, digest, stream):
 +        resource_name = '/'.join(['uploads', self._uuid, 'blobs',
 +                              digest.hash, str(digest.size_bytes)])
++
 +        def request_stream(resname, instream):
 +            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 = instream.read(chunk_size)
 +                request.resource_name = resname
 +                request.finish_write = remaining <= 0
++
 +                yield request
++
 +                offset += chunk_size
 +                finished = request.finish_write
++
 +        response = remote.bytestream.Write(request_stream(resource_name, stream))
++
 +        assert response.committed_size == digest.size_bytes
++
 +    def _send_directory(self, remote, digest):
 +        required_blobs = self._required_blobs(digest)
++
 +        missing_blobs = dict()
 +        # 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
 +        for blob_digest in missing_blobs.values():
 +            with open(self.objpath(blob_digest), 'rb') as f:
 +                assert os.fstat(f.fileno()).st_size == blob_digest.size_bytes
 +                self._send_blob(remote, blob_digest, f)
++
  # Represents a single remote CAS cache.
+ #

buildstream/_project.py

@@ -129,6 +129,7 @@ class Project():
          self.artifact_cache_specs = None
          self._sandbox = None
 +        self._remote_execution = None
          self._splits = None
          self._context.add_project(self)
@@ -398,6 +399,17 @@ class Project():
                  "Project requested format version {}, but BuildStream {}.{} only supports up until format version {}"
                  .format(format_version, major, minor, BST_FORMAT_VERSION))
 +        # FIXME:
 +        #
 +        #   Performing this check manually in the absense
 +        #   of proper support from _yaml.node_get(), this should
 +        #   be removed in favor of a proper accessor function
 +        #   from the _yaml module when #591 is fixed.
 +        #
 +        if self._project_conf.get('name') is None:
 +            raise LoadError(LoadErrorReason.INVALID_DATA,
 +                            "{}: project.conf does not contain expected key '{}'".format(projectfile, 'name'))
++
          # The project name, element path and option declarations
          # are constant and cannot be overridden by option conditional statements
          self.name = _yaml.node_get(pre_config_node, str, 'name')
@@ -460,7 +472,7 @@ class Project():
              'aliases', 'name',
              'artifacts', 'options',
              'fail-on-overlap', 'shell', 'fatal-warnings',
 -            'ref-storage', 'sandbox', 'mirrors'
 +            'ref-storage', 'sandbox', 'mirrors', 'remote-execution'
          ])
+         #
@@ -478,6 +490,9 @@ class Project():
          # Load sandbox configuration
          self._sandbox = _yaml.node_get(config, Mapping, 'sandbox')
 +        # Load remote execution configuration
 +        self._remote_execution = _yaml.node_get(config, Mapping, 'remote-execution')
++
          # Load project split rules
          self._splits = _yaml.node_get(config, Mapping, 'split-rules')

buildstream/_scheduler/jobs/job.py

@@ -403,7 +403,7 @@ class Job():
                  if self._retry_flag and (self._tries <= self._max_retries):
                      self.message(MessageType.FAIL,
                                   "Try #{} failed, retrying".format(self._tries),
 -                                 elapsed=elapsed)
 +                                 elapsed=elapsed, logfile=filename)
                  else:
                      self.message(MessageType.FAIL, str(e),
                                   elapsed=elapsed, detail=e.detail,
@@ -430,7 +430,8 @@ class Job():
                  self.message(MessageType.BUG, self.action_name,
                               elapsed=elapsed, detail=detail,
                               logfile=filename)
 -                self._child_shutdown(RC_FAIL)
 +                # Unhandled exceptions should permenantly fail
 +                self._child_shutdown(RC_PERM_FAIL)
              else:
                  # No exception occurred in the action
@@ -509,11 +510,6 @@ class Job():
          message.action_name = self.action_name
          message.task_id = self._task_id
 -        if (message.message_type == MessageType.FAIL and
 -                self._tries <= self._max_retries and self._retry_flag):
 -            # Job will be retried, display failures as warnings in the frontend
 -            message.message_type = MessageType.WARN
+-
          # Send to frontend if appropriate
          if context.silent_messages() and (message.message_type not in unconditional_messages):
              return

buildstream/_scheduler/scheduler.py

@@ -329,7 +329,7 @@ class Scheduler():
          self.schedule_jobs([job])
      def _check_cache_size_real(self):
 -        job = CacheSizeJob(self, 'cache_size', 'cache_size',
 +        job = CacheSizeJob(self, 'cache_size', 'cache_size/cache_size',
                             resources=[ResourceType.CACHE,
                                        ResourceType.PROCESS],
                             exclusive_resources=[ResourceType.CACHE],

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/data/projectconfig.yaml

@@ -204,3 +204,6 @@ shell:
    # Command to run when `bst shell` does not provide a command
+   #
    command: [ 'sh', '-i' ]
++
 +remote-execution:
 +  url: ""
 \ No newline at end of file

buildstream/element.py

@@ -95,6 +95,7 @@ from . import _site
  from ._platform import Platform
  from .plugin import CoreWarnings
  from .sandbox._config import SandboxConfig
 +from .sandbox._sandboxremote import SandboxRemote
  from .storage.directory import Directory
  from .storage._filebaseddirectory import FileBasedDirectory
@@ -245,11 +246,14 @@ class Element(Plugin):
          # Collect the composited element configuration and
          # ask the element to configure itself.
          self.__config = self.__extract_config(meta)
 -        self.configure(self.__config)
 +        self._configure(self.__config)
          # Extract Sandbox config
          self.__sandbox_config = self.__extract_sandbox_config(meta)
 +        # Extract remote execution URL
 +        self.__remote_execution_url = self.__extract_remote_execution_config(meta)
++
      def __lt__(self, other):
          return self.name < other.name
@@ -1570,6 +1574,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
@@ -2146,7 +2152,32 @@ class Element(Plugin):
          project = self._get_project()
          platform = Platform.get_platform()
 -        if directory is not None and os.path.exists(directory):
 +        if self.__remote_execution_url and self.BST_VIRTUAL_DIRECTORY:
 +            if not self.__artifacts.has_push_remotes(element=self):
 +                # Give an early warning if remote execution will not work
 +                raise ElementError("Artifact {} is configured to use remote execution but has no push remotes. "
 +                                   .format(self.name) +
 +                                   "The remote artifact server(s) may not be correctly configured or contactable.")
++
 +            self.info("Using a remote sandbox for artifact {}".format(self.name))
++
 +            sandbox = SandboxRemote(context, project,
 +                                    directory,
 +                                    stdout=stdout,
 +                                    stderr=stderr,
 +                                    config=config,
 +                                    server_url=self.__remote_execution_url,
 +                                    allow_real_directory=False)
 +            yield sandbox
++
 +        elif directory is not None and os.path.exists(directory):
 +            if self.__remote_execution_url:
 +                self.warn("Artifact {} is configured to use remote execution but element plugin does not support it."
 +                          .format(self.name), detail="Element plugin '{kind}' does not support virtual directories."
 +                          .format(kind=self.get_kind()), warning_token="remote-failure")
++
 +                self.info("Falling back to local sandbox for artifact {}".format(self.name))
++
              sandbox = platform.create_sandbox(context, project,
                                                directory,
                                                stdout=stdout,
@@ -2318,6 +2349,20 @@ class Element(Plugin):
          return SandboxConfig(self.node_get_member(sandbox_config, int, 'build-uid'),
                               self.node_get_member(sandbox_config, int, 'build-gid'))
 +    # Remote execution configuration data (server URL), to be used by the remote sandbox.
 +    #
 +    def __extract_remote_execution_config(self, meta):
 +        if self.__is_junction:
 +            return None
 +        else:
 +            project = self._get_project()
 +            project.ensure_fully_loaded()
 +            if project._remote_execution:
 +                rexec_config = _yaml.node_chain_copy(project._remote_execution)
 +                return self.node_get_member(rexec_config, str, 'url')
 +            else:
 +                return None
++
      # This makes a special exception for the split rules, which
      # elements may extend but whos defaults are defined in the project.
+     #

buildstream/plugin.py

@@ -179,6 +179,7 @@ class Plugin():
          self.__provenance = provenance  # The Provenance information
          self.__type_tag = type_tag      # The type of plugin (element or source)
          self.__unique_id = _plugin_register(self)  # Unique ID
 +        self.__configuring = False      # Whether we are currently configuring
          # Infer the kind identifier
          modulename = type(self).__module__
@@ -682,7 +683,32 @@ class Plugin():
          else:
              yield log
 +    # _configure():
 +    #
 +    # Calls configure() for the plugin, this must be called by
 +    # the core instead of configure() directly, so that the
 +    # _get_configuring() state is up to date.
 +    #
 +    # Args:
 +    #    node (dict): The loaded configuration dictionary
 +    #
 +    def _configure(self, node):
 +        self.__configuring = True
 +        self.configure(node)
 +        self.__configuring = False
++
 +    # _get_configuring():
 +    #
 +    # Checks whether the plugin is in the middle of having
 +    # its Plugin.configure() method called
 +    #
 +    # Returns:
 +    #    (bool): Whether we are currently configuring
 +    def _get_configuring(self):
 +        return self.__configuring
++
      # _preflight():
 +    #
      # Calls preflight() for the plugin, and allows generic preflight
      # checks to be added
+     #
@@ -690,6 +716,7 @@ class Plugin():
      #    SourceError: If it's a Source implementation
      #    ElementError: If it's an Element implementation
      #    ProgramNotFoundError: If a required host tool is not found
 +    #
      def _preflight(self):
          self.preflight()

buildstream/plugins/elements/autotools.py

@@ -57,7 +57,7 @@ from buildstream import BuildElement
  # Element implementation for the 'autotools' kind.
  class AutotoolsElement(BuildElement):
 -    pass
 +    BST_VIRTUAL_DIRECTORY = True
  # Plugin entry point

buildstream/plugins/sources/git.py

@@ -74,6 +74,9 @@ This plugin provides the following configurable warnings:
  - 'git:inconsistent-submodule' - A submodule was found to be missing from the underlying git repository.
 +This plugin also utilises the following configurable core plugin warnings:
++
 +- 'ref-not-in-track' - The provided ref was not found in the provided track in the element's git repository.
  """
  import os
@@ -87,6 +90,7 @@ from configparser import RawConfigParser
  from buildstream import Source, SourceError, Consistency, SourceFetcher
  from buildstream import utils
 +from buildstream.plugin import CoreWarnings
  GIT_MODULES = '.gitmodules'
@@ -100,13 +104,14 @@ INCONSISTENT_SUBMODULE = "inconsistent-submodules"
+ #
  class GitMirror(SourceFetcher):
 -    def __init__(self, source, path, url, ref):
 +    def __init__(self, source, path, url, ref, *, primary=False):
          super().__init__()
          self.source = source
          self.path = path
          self.url = url
          self.ref = ref
 +        self.primary = primary
          self.mirror = os.path.join(source.get_mirror_directory(), utils.url_directory_name(url))
          self.mark_download_url(url)
@@ -124,7 +129,8 @@ class GitMirror(SourceFetcher):
              # system configured tmpdir is not on the same partition.
+             #
              with self.source.tempdir() as tmpdir:
 -                url = self.source.translate_url(self.url, alias_override=alias_override)
 +                url = self.source.translate_url(self.url, alias_override=alias_override,
 +                                                primary=self.primary)
                  self.source.call([self.source.host_git, 'clone', '--mirror', '-n', url, tmpdir],
                                   fail="Failed to clone git repository {}".format(url),
                                   fail_temporarily=True)
@@ -146,7 +152,9 @@ class GitMirror(SourceFetcher):
                                            .format(self.source, url, tmpdir, self.mirror, e)) from e
      def _fetch(self, alias_override=None):
 -        url = self.source.translate_url(self.url, alias_override=alias_override)
 +        url = self.source.translate_url(self.url,
 +                                        alias_override=alias_override,
 +                                        primary=self.primary)
          if alias_override:
              remote_name = utils.url_directory_name(alias_override)
@@ -199,7 +207,7 @@ class GitMirror(SourceFetcher):
              cwd=self.mirror)
          return output.rstrip('\n')
 -    def stage(self, directory):
 +    def stage(self, directory, track=None):
          fullpath = os.path.join(directory, self.path)
          # Using --shared here avoids copying the objects into the checkout, in any
@@ -213,10 +221,14 @@ class GitMirror(SourceFetcher):
                           fail="Failed to checkout git ref {}".format(self.ref),
                           cwd=fullpath)
 +        # Check that the user specified ref exists in the track if provided & not already tracked
 +        if track:
 +            self.assert_ref_in_track(fullpath, track)
++
          # Remove .git dir
          shutil.rmtree(os.path.join(fullpath, ".git"))
 -    def init_workspace(self, directory):
 +    def init_workspace(self, directory, track=None):
          fullpath = os.path.join(directory, self.path)
          url = self.source.translate_url(self.url)
@@ -232,6 +244,10 @@ class GitMirror(SourceFetcher):
                           fail="Failed to checkout git ref {}".format(self.ref),
                           cwd=fullpath)
 +        # Check that the user specified ref exists in the track if provided & not already tracked
 +        if track:
 +            self.assert_ref_in_track(fullpath, track)
++
      # List the submodules (path/url tuples) present at the given ref of this repo
      def submodule_list(self):
          modules = "{}:{}".format(self.ref, GIT_MODULES)
@@ -296,6 +312,28 @@ class GitMirror(SourceFetcher):
              return None
 +    # Assert that ref exists in track, if track has been specified.
 +    def assert_ref_in_track(self, fullpath, track):
 +        _, branch = self.source.check_output([self.source.host_git, 'branch', '--list', track,
 +                                              '--contains', self.ref],
 +                                             cwd=fullpath,)
 +        if branch:
 +            return True
 +        else:
 +            _, tag = self.source.check_output([self.source.host_git, 'tag', '--list', track,
 +                                               '--contains', self.ref],
 +                                              cwd=fullpath,)
 +            if tag:
 +                return True
++
 +        detail = "The ref provided for the element does not exist locally in the provided track branch / tag " + \
 +                 "'{}'.\nYou may wish to track the element to update the ref from '{}' ".format(track, track) + \
 +                 "with `bst track`,\nor examine the upstream at '{}' for the specific ref.".format(self.url)
++
 +        self.source.warn("{}: expected ref '{}' was not found in given track '{}' for staged repository: '{}'\n"
 +                         .format(self.source, self.ref, track, self.url),
 +                         detail=detail, warning_token=CoreWarnings.REF_NOT_IN_TRACK)
++
  class GitSource(Source):
      # pylint: disable=attribute-defined-outside-init
@@ -307,7 +345,7 @@ class GitSource(Source):
          self.node_validate(node, config_keys + Source.COMMON_CONFIG_KEYS)
          self.original_url = self.node_get_member(node, str, 'url')
 -        self.mirror = GitMirror(self, '', self.original_url, ref)
 +        self.mirror = GitMirror(self, '', self.original_url, ref, primary=True)
          self.tracking = self.node_get_member(node, str, 'track', None)
          # At this point we now know if the source has a ref and/or a track.
@@ -327,12 +365,18 @@ class GitSource(Source):
          for path, _ in self.node_items(modules):
              submodule = self.node_get_member(modules, Mapping, path)
              url = self.node_get_member(submodule, str, 'url', None)
++
 +            # Make sure to mark all URLs that are specified in the configuration
 +            if url:
 +                self.mark_download_url(url, primary=False)
++
              self.submodule_overrides[path] = url
              if 'checkout' in submodule:
                  checkout = self.node_get_member(submodule, bool, 'checkout')
                  self.submodule_checkout_overrides[path] = checkout
          self.mark_download_url(self.original_url)
 +        self.tracked = False
      def preflight(self):
          # Check if git is installed, get the binary at the same time
@@ -398,6 +442,8 @@ class GitSource(Source):
              # Update self.mirror.ref and node.ref from the self.tracking branch
              ret = self.mirror.latest_commit(self.tracking)
 +        # Set tracked attribute, parameter for if self.mirror.assert_ref_in_track is needed
 +        self.tracked = True
          return ret
      def init_workspace(self, directory):
@@ -405,7 +451,7 @@ class GitSource(Source):
          self.refresh_submodules()
          with self.timed_activity('Setting up workspace "{}"'.format(directory), silent_nested=True):
 -            self.mirror.init_workspace(directory)
 +            self.mirror.init_workspace(directory, track=(self.tracking if not self.tracked else None))
              for mirror in self.submodules:
                  mirror.init_workspace(directory)
@@ -421,7 +467,7 @@ class GitSource(Source):
          # Stage the main repo in the specified directory
+         #
          with self.timed_activity("Staging {}".format(self.mirror.url), silent_nested=True):
 -            self.mirror.stage(directory)
 +            self.mirror.stage(directory, track=(self.tracking if not self.tracked else None))
              for mirror in self.submodules:
                  if mirror.path in self.submodule_checkout_overrides:
                      checkout = self.submodule_checkout_overrides[mirror.path]

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/_sandboxremote.py

 +#!/usr/bin/env python3
 +#
 +#  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>
++
 +import os
 +import re
 +from urllib.parse import urlparse
++
 +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 .._artifactcache.cascache import CASCache
++
++
 +class SandboxError(Exception):
 +    pass
++
++
 +# SandboxRemote()
 +#
 +# This isn't really a sandbox, it's a stub which sends all the sources and build
 +# commands to a remote server and retrieves the results from it.
 +#
 +class SandboxRemote(Sandbox):
++
 +    def __init__(self, *args, **kwargs):
 +        super().__init__(*args, **kwargs)
 +        self.cascache = None
++
 +        url = urlparse(kwargs['server_url'])
 +        if not url.scheme or not url.hostname or not url.port:
 +            raise SandboxError("Configured remote URL '{}' does not match the expected layout. "
 +                               .format(self.server_url) +
 +                               "It should be of the form <protocol>://<domain name>:<port>.")
 +        elif url.scheme != 'http':
 +            raise SandboxError("Configured remote '{}' uses an unsupported protocol. "
 +                               "Only plain HTTP is currenlty supported (no HTTPS).")
++
 +        self.server_url = '{}:{}'.format(url.hostname, url.port)
++
 +    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, command, input_root_digest, working_directory, environment):
 +        # Sends an execution request to the remote execution server.
 +        #
 +        # This function blocks until it gets a response from the server.
 +        #
 +        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,
 +                                                      working_directory=working_directory,
 +                                                      environment_variables=environment_variables,
 +                                                      output_files=[],
 +                                                      output_directories=[self._output_directory],
 +                                                      platform=None)
++
 +        cascache = self._get_cascache()
 +        # Upload the Command message to the remote CAS server
 +        command_digest = cascache.push_message(self._get_project(), remote_command)
 +        if not command_digest or not cascache.verify_key_pushed(self._get_project(), command_digest):
 +            # 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)
++
 +        # Upload the Action message to the remote CAS server
 +        action_digest = cascache.push_message(self._get_project(), action)
 +        if not action_digest or not cascache.verify_key_pushed(self._get_project(), action_digest):
 +            # Action push failed
 +            return None
++
 +        # Next, try to create a communication channel to the BuildGrid server.
 +        channel = grpc.insecure_channel(self.server_url)
 +        stub = remote_execution_pb2_grpc.ExecutionStub(channel)
 +        request = remote_execution_pb2.ExecuteRequest(action_digest=action_digest,
 +                                                      skip_cache_lookup=True)
 +        try:
 +            operation_iterator = stub.Execute(request)
 +        except grpc.RpcError:
 +            return None
++
 +        operation = None
 +        with self._get_context().timed_activity("Waiting for the remote build to complete"):
 +            # It is advantageous to check operation_iterator.code() is grpc.StatusCode.OK here,
 +            # which will check the server is actually contactable. However, calling it when the
 +            # server is available seems to cause .code() to hang forever.
 +            for operation in operation_iterator:
 +                if operation.done:
 +                    break
++
 +        return operation
++
 +    def process_job_output(self, output_directories, output_files):
 +        # Reads the remote execution server response to an execution request.
 +        #
 +        # 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 not output_directories:
 +            error_text = "No output directory was returned from the build server."
 +            raise SandboxError(error_text)
 +        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))
++
 +        tree_digest = output_directories[0].tree_digest
 +        if tree_digest is None or not tree_digest.hash:
 +            raise SandboxError("Output directory structure had no digest attached.")
++
 +        cascache = self._get_cascache()
 +        # Now do a pull to ensure we have the necessary parts.
 +        dir_digest = cascache.pull_tree(self._get_project(), tree_digest)
 +        if dir_digest is None or not dir_digest.hash or not dir_digest.size_bytes:
 +            raise SandboxError("Output directory structure pulling from remote failed.")
++
 +        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=dir_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())
++
 +        upload_vdir.recalculate_hash()
++
 +        cascache = self._get_cascache()
 +        # Now, push that key (without necessarily needing a ref) to the remote.
 +        vdir_digest = cascache.push_directory(self._get_project(), upload_vdir)
 +        if not vdir_digest or not cascache.verify_key_pushed(self._get_project(), vdir_digest):
 +            raise SandboxError("Failed to verify that source has been pushed to the remote artifact cache.")
++
 +        # Set up environment and working directory
 +        if cwd is None:
 +            cwd = self._get_work_directory()
++
 +        if cwd is None:
 +            cwd = '/'
++
 +        if env is None:
 +            env = self._get_environment()
++
 +        # We want command args as a list of strings
 +        if isinstance(command, str):
 +            command = [command]
++
 +        # Now transmit the command to execute
 +        operation = self.run_remote_command(command, upload_vdir.ref, cwd, env)
++
 +        if operation is None:
 +            # Failure of remote execution, usually due to an error in BuildStream
 +            # NB This error could be raised in __run_remote_command
 +            raise SandboxError("No response returned from server")
++
 +        assert(not operation.HasField('error') and operation.HasField('response'))
++
 +        execution_response = remote_execution_pb2.ExecuteResponse()
 +        # The response is expected to be an ExecutionResponse message
 +        assert(operation.response.Is(execution_response.DESCRIPTOR))
++
 +        operation.response.Unpack(execution_response)
++
 +        if execution_response.status.code != 0:
 +            # A normal error during the build: the remote execution system
 +            # has worked correctly but the command failed.
 +            # execution_response.error also contains 'message' (str) and
 +            # 'details' (iterator of Any) which we ignore at the moment.
 +            return execution_response.status.code
++
 +        action_result = execution_response.result
++
 +        self.process_job_output(action_result.output_directories, action_result.output_files)
++
 +        return 0

buildstream/sandbox/sandbox.py

@@ -99,9 +99,11 @@ 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]:
@@ -144,11 +146,17 @@ class Sandbox():
                  self._vdir = FileBasedDirectory(self._root)
          return self._vdir
 +    def _set_virtual_directory(self, virtual_directory):
 +        """ Sets virtual directory. Useful after remote execution
 +        has rewritten the working directory.
 +        """
 +        self._vdir = virtual_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
@@ -160,6 +168,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/source.py

@@ -28,6 +28,18 @@ Abstract Methods
  For loading and configuration purposes, Sources must implement the
  :ref:`Plugin base class abstract methods <core_plugin_abstract_methods>`.
 +.. attention::
++
 +   In order to ensure that all configuration data is processed at
 +   load time, it is important that all URLs have been processed during
 +   :func:`Plugin.configure() <buildstream.plugin.Plugin.configure>`.
++
 +   Source implementations *must* either call
 +   :func:`Source.translate_url() <buildstream.source.Source.translate_url>` or
 +   :func:`Source.mark_download_url() <buildstream.source.Source.mark_download_url>`
 +   for every URL that has been specified in the configuration during
 +   :func:`Plugin.configure() <buildstream.plugin.Plugin.configure>`
++
  Sources expose the following abstract methods. Unless explicitly mentioned,
  these methods are mandatory to implement.
@@ -184,6 +196,13 @@ class SourceFetcher():
      fetching and substituting aliases.
      *Since: 1.2*
++
 +    .. attention::
++
 +       When implementing a SourceFetcher, remember to call
 +       :func:`Source.mark_download_url() <buildstream.source.Source.mark_download_url>`
 +       for every URL found in the configuration data at
 +       :func:`Plugin.configure() <buildstream.plugin.Plugin.configure>` time.
      """
      def __init__(self):
          self.__alias = None
@@ -206,7 +225,7 @@ class SourceFetcher():
          Implementors should raise :class:`.SourceError` if the there is some
          network error or if the source reference could not be matched.
          """
 -        raise ImplError("Source fetcher '{}' does not implement fetch()".format(type(self)))
 +        raise ImplError("SourceFetcher '{}' does not implement fetch()".format(type(self)))
      #############################################################
      #                       Public Methods                      #
@@ -277,8 +296,11 @@ class Source(Plugin):
          self.__element_kind = meta.element_kind         # The kind of the element owning this source
          self.__directory = meta.directory               # Staging relative directory
          self.__consistency = Consistency.INCONSISTENT   # Cached consistency state
++
 +        # The alias_override is only set on a re-instantiated Source
          self.__alias_override = alias_override          # Tuple of alias and its override to use instead
 -        self.__expected_alias = None                    # A hacky way to store the first alias used
 +        self.__expected_alias = None                    # The primary alias
 +        self.__marked_urls = set()                      # Set of marked download URLs
          # FIXME: Reconstruct a MetaSource from a Source instead of storing it.
          self.__meta = meta                              # MetaSource stored so we can copy this source later.
@@ -289,7 +311,7 @@ class Source(Plugin):
          self.__config = self.__extract_config(meta)
          self.__first_pass = meta.first_pass
 -        self.configure(self.__config)
 +        self._configure(self.__config)
      COMMON_CONFIG_KEYS = ['kind', 'directory']
      """Common source config keys
@@ -351,10 +373,10 @@ class Source(Plugin):
          Args:
             ref (simple object): The internal source reference to set, or ``None``
             node (dict): The same dictionary which was previously passed
 -                        to :func:`~buildstream.source.Source.configure`
 +                        to :func:`Plugin.configure() <buildstream.plugin.Plugin.configure>`
 -        See :func:`~buildstream.source.Source.get_ref` for a discussion on
 -        the *ref* parameter.
 +        See :func:`Source.get_ref() <buildstream.source.Source.get_ref>`
 +        for a discussion on the *ref* parameter.
          .. note::
@@ -384,8 +406,8 @@ class Source(Plugin):
          backend store allows one to query for a new ref from a symbolic
          tracking data without downloading then that is desirable.
 -        See :func:`~buildstream.source.Source.get_ref` for a discussion on
 -        the *ref* parameter.
 +        See :func:`Source.get_ref() <buildstream.source.Source.get_ref>`
 +        for a discussion on the *ref* parameter.
          """
          # Allow a non implementation
          return None
@@ -435,7 +457,7 @@ class Source(Plugin):
             :class:`.SourceError`
          Default implementation is to call
 -        :func:`~buildstream.source.Source.stage`.
 +        :func:`Source.stage() <buildstream.source.Source.stage>`.
          Implementors overriding this method should assume that *directory*
          already exists.
@@ -453,8 +475,15 @@ class Source(Plugin):
          is recommended.
          Returns:
 -           list: A list of SourceFetchers. If SourceFetchers are not supported,
 -                 this will be an empty list.
 +           iterable: The Source's SourceFetchers, if any.
++
 +        .. note::
++
 +           Implementors can implement this as a generator.
++
 +           The :func:`SourceFetcher.fetch() <buildstream.source.SourceFetcher.fetch>`
 +           method will be called on the returned fetchers one by one,
 +           before consuming the next fetcher in the list.
          *Since: 1.2*
          """
@@ -477,17 +506,27 @@ class Source(Plugin):
          os.makedirs(directory, exist_ok=True)
          return directory
 -    def translate_url(self, url, *, alias_override=None):
 +    def translate_url(self, url, *, alias_override=None, primary=True):
          """Translates the given url which may be specified with an alias
          into a fully qualified url.
          Args:
 -           url (str): A url, which may be using an alias
 +           url (str): A URL, which may be using an alias
             alias_override (str): Optionally, an URI to override the alias with. (*Since: 1.2*)
 +           primary (bool): Whether this is the primary URL for the source. (*Since: 1.2*)
          Returns:
 -           str: The fully qualified url, with aliases resolved
 +           str: The fully qualified URL, with aliases resolved
 +        .. note::
++
 +           This must be called for every URL in the configuration during
 +           :func:`Plugin.configure() <buildstream.plugin.Plugin.configure>` if
 +           :func:`Source.mark_download_url() <buildstream.source.Source.mark_download_url>`
 +           is not called.
          """
 +        # Ensure that the download URL is also marked
 +        self.mark_download_url(url, primary=primary)
++
          # Alias overriding can happen explicitly (by command-line) or
          # implicitly (the Source being constructed with an __alias_override).
          if alias_override or self.__alias_override:
@@ -506,25 +545,55 @@ class Source(Plugin):
                          url = override_url + url_body
              return url
          else:
 -            # Sneakily store the alias if it hasn't already been stored
 -            if not self.__expected_alias and url and utils._ALIAS_SEPARATOR in url:
 -                self.mark_download_url(url)
+-
              project = self._get_project()
              return project.translate_url(url, first_pass=self.__first_pass)
 -    def mark_download_url(self, url):
 +    def mark_download_url(self, url, *, primary=True):
          """Identifies the URL that this Source uses to download
 -        This must be called during :func:`~buildstream.plugin.Plugin.configure` if
 -        :func:`~buildstream.source.Source.translate_url` is not called.
+-
          Args:
 -           url (str): The url used to download
 +           url (str): The URL used to download
 +           primary (bool): Whether this is the primary URL for the source
++
 +        .. note::
++
 +           This must be called for every URL in the configuration during
 +           :func:`Plugin.configure() <buildstream.plugin.Plugin.configure>` if
 +           :func:`Source.translate_url() <buildstream.source.Source.translate_url>`
 +           is not called.
          *Since: 1.2*
          """
 -        self.__expected_alias = _extract_alias(url)
 +        # Only mark the Source level aliases on the main instance, not in
 +        # a reinstantiated instance in mirroring.
 +        if not self.__alias_override:
 +            if primary:
 +                expected_alias = _extract_alias(url)
++
 +                assert (self.__expected_alias is None or
 +                        self.__expected_alias == expected_alias), \
 +                    "Primary URL marked twice with different URLs"
++
 +                self.__expected_alias = expected_alias
++
 +        # Enforce proper behaviour of plugins by ensuring that all
 +        # aliased URLs have been marked at Plugin.configure() time.
 +        #
 +        if self._get_configuring():
 +            # Record marked urls while configuring
 +            #
 +            self.__marked_urls.add(url)
 +        else:
 +            # If an unknown aliased URL is seen after configuring,
 +            # this is an error.
 +            #
 +            # It is still possible that a URL that was not mentioned
 +            # in the element configuration can be marked, this is
 +            # the case for git submodules which might be automatically
 +            # discovered.
 +            #
 +            assert (url in self.__marked_urls or not _extract_alias(url)), \
 +                "URL was not seen at configure time: {}".format(url)
      def get_project_directory(self):
          """Fetch the project base directory

buildstream/storage/_casbaseddirectory.py

@@ -543,6 +543,15 @@ class CasBasedDirectory(Directory):
                  filelist.append(k)
          return filelist
 +    def recalculate_hash(self):
 +        """ Recalcuates 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). Hashes for subdirectories also get recalculated.
 +        """
 +        self._recalculate_recursing_up()
 +        self._recalculate_recursing_down()
++
      def _get_identifier(self):
          path = ""
          if self.parent:

doc/source/format_project.rst

@@ -204,6 +204,24 @@ with an artifact share.
  You can also specify a list of caches here; earlier entries in the list
  will have higher priority than later ones.
 +Remote execution
 +~~~~~~~~~~~~~~~~
 +BuildStream supports remote execution using the Google Remote Execution API
 +(REAPI). A description of how remote execution works is beyond the scope
 +of this document, but you can specify a remote server complying with the REAPI
 +using the `remote-execution` option:
++
 +.. code:: yaml
++
 +  remote-execution:
++
 +    # A url defining a remote execution server
 +    url: http://buildserver.example.com:50051
++
 +The url should contain a hostname and port separated by ':'. Only plain HTTP is
 +currently suported (no HTTPS).
++
 +The Remote Execution API can be found via https://github.com/bazelbuild/remote-apis.
  .. _project_essentials_mirrors:

tests/frontend/project/sources/fetch_source.py

@@ -15,14 +15,17 @@ from buildstream import Source, Consistency, SourceError, SourceFetcher
  class FetchFetcher(SourceFetcher):
 -    def __init__(self, source, url):
 +    def __init__(self, source, url, primary=False):
          super().__init__()
          self.source = source
          self.original_url = url
 +        self.primary = primary
          self.mark_download_url(url)
      def fetch(self, alias_override=None):
 -        url = self.source.translate_url(self.original_url, alias_override=alias_override)
 +        url = self.source.translate_url(self.original_url,
 +                                        alias_override=alias_override,
 +                                        primary=self.primary)
          with open(self.source.output_file, "a") as f:
              success = url in self.source.fetch_succeeds and self.source.fetch_succeeds[url]
              message = "Fetch {} {} from {}\n".format(self.original_url,
@@ -37,12 +40,21 @@ class FetchSource(Source):
      # Read config to know which URLs to fetch
      def configure(self, node):
          self.original_urls = self.node_get_member(node, list, 'urls')
 -        self.fetchers = [FetchFetcher(self, url) for url in self.original_urls]
          self.output_file = self.node_get_member(node, str, 'output-text')
          self.fetch_succeeds = {}
          if 'fetch-succeeds' in node:
              self.fetch_succeeds = {x[0]: x[1] for x in self.node_items(node['fetch-succeeds'])}
 +        # First URL is the primary one for this test
 +        #
 +        primary = True
 +        self.fetchers = []
 +        for url in self.original_urls:
 +            self.mark_download_url(url, primary=primary)
 +            fetcher = FetchFetcher(self, url, primary=primary)
 +            self.fetchers.append(fetcher)
 +            primary = False
++
      def get_source_fetchers(self):
          return self.fetchers

tests/sources/deb.py

@@ -56,7 +56,7 @@ def test_fetch_bad_url(cli, tmpdir, datafiles):
      result = cli.run(project=project, args=[
          'fetch', 'target.bst'
      ])
 -    assert "Try #" in result.stderr
 +    assert "FAILURE Try #" in result.stderr
      result.assert_main_error(ErrorDomain.STREAM, None)
      result.assert_task_error(ErrorDomain.SOURCE, None)

tests/sources/git.py

@@ -25,6 +25,7 @@ import pytest
  from buildstream._exceptions import ErrorDomain
  from buildstream import _yaml
 +from buildstream.plugin import CoreWarnings
  from tests.testutils import cli, create_repo
  from tests.testutils.site import HAVE_GIT
@@ -408,3 +409,70 @@ def test_submodule_track_no_ref_or_track(cli, tmpdir, datafiles):
      result = cli.run(project=project, args=['show', 'target.bst'])
      result.assert_main_error(ErrorDomain.SOURCE, "missing-track-and-ref")
      result.assert_task_error(None, None)
++
++
 +@pytest.mark.skipif(HAVE_GIT is False, reason="git is not available")
 +@pytest.mark.datafiles(os.path.join(DATA_DIR, 'template'))
 +def test_ref_not_in_track_warn(cli, tmpdir, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
++
 +    # Create the repo from 'repofiles', create a branch without latest commit
 +    repo = create_repo('git', str(tmpdir))
 +    ref = repo.create(os.path.join(project, 'repofiles'))
++
 +    gitsource = repo.source_config(ref=ref)
++
 +    # Overwrite the track value to the added branch
 +    gitsource['track'] = 'foo'
++
 +    # Write out our test target
 +    element = {
 +        'kind': 'import',
 +        'sources': [
 +            gitsource
 +        ]
 +    }
 +    _yaml.dump(element, os.path.join(project, 'target.bst'))
++
 +    # Assert the warning is raised as ref is not in branch foo.
 +    # Assert warning not error to the user, when not set as fatal.
 +    result = cli.run(project=project, args=['build', 'target.bst'])
 +    assert "The ref provided for the element does not exist locally" 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_ref_not_in_track_warn_error(cli, tmpdir, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
++
 +    # Add fatal-warnings ref-not-in-track to project.conf
 +    project_template = {
 +        "name": "foo",
 +        "fatal-warnings": [CoreWarnings.REF_NOT_IN_TRACK]
 +    }
++
 +    _yaml.dump(project_template, os.path.join(project, 'project.conf'))
++
 +    # Create the repo from 'repofiles', create a branch without latest commit
 +    repo = create_repo('git', str(tmpdir))
 +    ref = repo.create(os.path.join(project, 'repofiles'))
++
 +    gitsource = repo.source_config(ref=ref)
++
 +    # Overwrite the track value to the added branch
 +    gitsource['track'] = 'foo'
++
 +    # Write out our test target
 +    element = {
 +        'kind': 'import',
 +        'sources': [
 +            gitsource
 +        ]
 +    }
 +    _yaml.dump(element, os.path.join(project, 'target.bst'))
++
 +    # Assert that build raises a warning here that is captured
 +    # as plugin error, due to the fatal warning being set
 +    result = cli.run(project=project, args=['build', 'target.bst'])
 +    result.assert_main_error(ErrorDomain.STREAM, None)
 +    result.assert_task_error(ErrorDomain.PLUGIN, CoreWarnings.REF_NOT_IN_TRACK)

tests/sources/tar.py

@@ -67,7 +67,7 @@ def test_fetch_bad_url(cli, tmpdir, datafiles):
      result = cli.run(project=project, args=[
          'fetch', 'target.bst'
      ])
 -    assert "Try #" in result.stderr
 +    assert "FAILURE Try #" in result.stderr
      result.assert_main_error(ErrorDomain.STREAM, None)
      result.assert_task_error(ErrorDomain.SOURCE, None)

tests/sources/zip.py

@@ -53,7 +53,7 @@ def test_fetch_bad_url(cli, tmpdir, datafiles):
      result = cli.run(project=project, args=[
          'fetch', 'target.bst'
      ])
 -    assert "Try #" in result.stderr
 +    assert "FAILURE Try #" in result.stderr
      result.assert_main_error(ErrorDomain.STREAM, None)
      result.assert_task_error(ErrorDomain.SOURCE, None)

tests/testutils/repo/bzr.py

@@ -2,6 +2,7 @@ import os
  import subprocess
  import pytest
 +from buildstream import utils
  from .repo import Repo
  from ..site import HAVE_BZR
@@ -16,15 +17,16 @@ class Bzr(Repo):
          if not HAVE_BZR:
              pytest.skip("bzr is not available")
          super(Bzr, self).__init__(directory, subdir)
 +        self.bzr = utils.get_host_tool('bzr')
      def create(self, directory):
          branch_dir = os.path.join(self.repo, 'trunk')
 -        subprocess.call(['bzr', 'init-repo', self.repo], env=BZR_ENV)
 -        subprocess.call(['bzr', 'init', branch_dir], env=BZR_ENV)
 +        subprocess.call([self.bzr, 'init-repo', self.repo], env=BZR_ENV)
 +        subprocess.call([self.bzr, 'init', branch_dir], env=BZR_ENV)
          self.copy_directory(directory, branch_dir)
 -        subprocess.call(['bzr', 'add', '.'], env=BZR_ENV, cwd=branch_dir)
 -        subprocess.call(['bzr', 'commit', '--message="Initial commit"'],
 +        subprocess.call([self.bzr, 'add', '.'], env=BZR_ENV, cwd=branch_dir)
 +        subprocess.call([self.bzr, 'commit', '--message="Initial commit"'],
                          env=BZR_ENV, cwd=branch_dir)
          return self.latest_commit()
@@ -42,7 +44,7 @@ class Bzr(Repo):
      def latest_commit(self):
          output = subprocess.check_output([
 -            'bzr', 'version-info',
 +            self.bzr, 'version-info',
              '--custom', '--template={revno}',
              os.path.join(self.repo, 'trunk')
          ], env=BZR_ENV)

[Notes] [Git][BuildStream/buildstream][jmac/remote_execution_client] 30 commits: Replacing string 'bzr' with value from host tools

Martin Blanchard pushed to branch jmac/remote_execution_client at BuildStream / buildstream

Commits:

22 changed files:

Changes: