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

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

Commits:

c15cb951

by Tristan Van Berkom at 2018-08-20T10:19:03Z

setup.cfg: Add tests/integration/project to norecursedirs

This is causing pytest to try to import the recently added app1.py
from the pip related tests. This fails because it's not meant to be
imported outside of the sandbox environment.

372abed5

by Tristan Van Berkom at 2018-08-20T11:47:53Z

Merge branch 'tristan/exclude-project-py-from-tests' into 'master'

setup.cfg: Add tests/integration/project to norecursedirs

See merge request BuildStream/buildstream!682

3feefbf6

by Jim MacArthur at 2018-08-20T13:18:17Z

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

40ae8e1a

by Jim MacArthur at 2018-08-20T13:18:17Z

cascache: Modifications necessary for remote execution.

e8d7f098

by Jim MacArthur at 2018-08-20T13:18:17Z

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

bb20b6d7

by Jim MacArthur at 2018-08-20T13:18:17Z

Add "remote-execution" configuration option.

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

fada50a5

by Jim MacArthur at 2018-08-20T13:18:17Z

Configuration: Add default remote-execution and allow it in elements

78fc5499
by Jim MacArthur at 2018-08-20T13:18:17Z
```
_sandboxremote.py: New file.
```

6b83bf3b

by Jim MacArthur at 2018-08-20T13:18:17Z

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

803189a7
by Jim MacArthur at 2018-08-20T13:18:17Z
```
sandbox/__init__.py: Add SandboxRemote
```

f0e077d2

by Jim MacArthur at 2018-08-20T13:18:17Z

autotools.py: Mark this as a BST_VIRTUAL_DIRECTORY plugin

13322752

by Jim MacArthur at 2018-08-20T13:18:17Z

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

15 changed files:

buildstream/_artifactcache/cascache.py
buildstream/_loader/loadelement.py
buildstream/_loader/loader.py
buildstream/_loader/metaelement.py
buildstream/_loader/types.py
buildstream/_project.py
buildstream/buildelement.py
buildstream/data/projectconfig.yaml
buildstream/element.py
buildstream/plugins/elements/autotools.py
buildstream/sandbox/__init__.py
+ buildstream/sandbox/_sandboxremote.py
buildstream/sandbox/sandbox.py
buildstream/storage/_casbaseddirectory.py
setup.cfg

Changes:

buildstream/_artifactcache/cascache.py

@@ -215,6 +215,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)
@@ -256,10 +279,94 @@ 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():
 +                    uuid_ = uuid.uuid4()
 +                    resource_name = '/'.join(['uploads', str(uuid_), 'blobs',
 +                                              digest.hash, str(digest.size_bytes)])
 +                    def request_stream():
 +                        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]
@@ -267,97 +374,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():
 -                        uuid_ = uuid.uuid4()
 -                        resource_name = '/'.join(['uploads', str(uuid_), 'blobs',
 -                                                  digest.hash, str(digest.size_bytes)])
+-
 -                        def request_stream():
 -                            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           #
      ################################################
@@ -731,26 +793,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 = '/'.join(['blobs', digest.hash, str(digest.size_bytes)])

buildstream/_loader/loadelement.py

@@ -71,7 +71,7 @@ class LoadElement():
              'kind', 'depends', 'sources', 'sandbox',
              'variables', 'environment', 'environment-nocache',
              'config', 'public', 'description',
 -            'build-depends', 'runtime-depends',
 +            'build-depends', 'runtime-depends', 'remote-execution'
          ])
          # Extract the Dependencies

buildstream/_loader/loader.py

@@ -446,6 +446,7 @@ class Loader():
                                     _yaml.node_get(node, list, Symbol.ENV_NOCACHE, default_value=[]),
                                     _yaml.node_get(node, Mapping, Symbol.PUBLIC, default_value={}),
                                     _yaml.node_get(node, Mapping, Symbol.SANDBOX, default_value={}),
 +                                   _yaml.node_get(node, Mapping, Symbol.REMOTE_EXECUTION, default_value={}),
                                     element_kind == 'junction')
          # Cache it now, make sure it's already there before recursing

buildstream/_loader/metaelement.py

@@ -39,7 +39,7 @@ class MetaElement():
      #    first_pass: The element is to be loaded with first pass configuration (junction)
+     #
      def __init__(self, project, name, kind, provenance, sources, config,
 -                 variables, environment, env_nocache, public, sandbox,
 +                 variables, environment, env_nocache, public, sandbox, remote_execution,
                   first_pass):
          self.project = project
          self.name = name
@@ -52,6 +52,7 @@ class MetaElement():
          self.env_nocache = env_nocache
          self.public = public
          self.sandbox = sandbox
 +        self.remote_execution = remote_execution
          self.build_dependencies = []
          self.dependencies = []
          self.first_pass = first_pass

buildstream/_loader/types.py

@@ -41,6 +41,7 @@ class Symbol():
      DIRECTORY = "directory"
      JUNCTION = "junction"
      SANDBOX = "sandbox"
 +    REMOTE_EXECUTION = "remote-execution"
  # Dependency()

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)
@@ -460,7 +461,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 +479,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/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
@@ -250,6 +251,9 @@ class Element(Plugin):
          # 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
@@ -1545,6 +1549,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
@@ -2117,7 +2123,24 @@ 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):
 +            self.info("Using a local sandbox for artifact {}".format(self.name))
              sandbox = platform.create_sandbox(context, project,
                                                directory,
                                                stdout=stdout,
@@ -2289,6 +2312,24 @@ class Element(Plugin):
          return SandboxConfig(self.node_get_member(sandbox_config, int, 'build-uid'),
                               self.node_get_member(sandbox_config, int, 'build-gid'))
 +    def __extract_remote_execution_config(self, meta):
 +        project = self._get_project()
 +        project.ensure_fully_loaded()
 +        rexec_config = _yaml.node_chain_copy(project._remote_execution)
++
 +        # The default config is already composited with the project overrides
 +        rexec_defaults = _yaml.node_get(self.__defaults, Mapping, 'remote-execution', default_value={})
 +        rexec_defaults = _yaml.node_chain_copy(rexec_defaults)
++
 +        _yaml.composite(rexec_config, rexec_defaults)
 +        _yaml.composite(rexec_config, meta.remote_execution)
 +        _yaml.node_final_assertions(rexec_config)
++
 +        # Rexec config, unlike others, has fixed members so we should validate them
 +        _yaml.node_validate(rexec_config, ['url'])
++
 +        return self.node_get_member(rexec_config, str, 'url')
++
      # This makes a special exception for the split rules, which
      # elements may extend but whos defaults are defined in the project.
+     #

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/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 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 re
 +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.cascache = None
 +        self.server_url = kwargs['server_url']
 +        # Check the format of the url ourselves to save the user from
 +        # whatever error messages grpc will produce
 +        m = re.match('^(.+):(\d+)$', self.server_url)
 +        if m is None:
 +            raise SandboxError("Configured remote URL '{}' does not match the expected layout. " +
 +                               "It should be of the form <protocol>://<domain name>:<port>."
 +                               .format(self.server_url))
++
++
 +    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.
++
 +        channel = grpc.insecure_channel(self.server_url)
 +        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)
 +        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):
 +        # 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:
 +                # 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(response.HasField("error") or response.HasField("response"))
++
 +            if response.HasField("error"):
 +                # A normal error during the build
 +                error_message = response.error.message
 +                # response.error also contains 'details' (iterator of Any) which we ignore at the moment.
 +                return response.error.code
 +            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

@@ -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,30 @@ class Sandbox():
                  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.__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 +181,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

@@ -358,6 +358,20 @@ class CasBasedDirectory(Directory):
                      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):

setup.cfg

@@ -11,7 +11,7 @@ test=pytest
  [tool:pytest]
  addopts = --verbose --basetemp ./tmp --pep8 --pylint --pylint-rcfile=.pylintrc --cov=buildstream --cov-config .coveragerc
 -norecursedirs = integration-cache tmp __pycache__ .eggs
 +norecursedirs = tests/integration/project integration-cache tmp __pycache__ .eggs
  python_files = tests/*/*.py
  pep8maxlinelength = 119
  pep8ignore =

[Notes] [Git][BuildStream/buildstream][jmac/remote_execution_client] 12 commits: setup.cfg: Add tests/integration/project to norecursedirs

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

Commits:

15 changed files:

Changes: