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

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

Commits:

df60258a

by Jürg Billeter at 2018-11-27T12:11:24Z

source.py: Add optional _get_local_path() method

42809933
by Jürg Billeter at 2018-11-27T12:11:24Z
```
local.py: Implement _get_local_path()
```

eb5b10f7

by Jürg Billeter at 2018-11-27T12:11:24Z

_loader.py: Use Source._get_local_path() for subprojects

This eliminates unneeded copies of the project directory for junctions
with a single local source.

b2ea208c

by Jürg Billeter at 2018-11-27T12:50:46Z

Merge branch 'juerg/local-junctions' into 'master'

Optimization for local junctions

See merge request BuildStream/buildstream!290

223c0073

by Jürg Billeter at 2018-11-27T13:41:09Z

tests/integration/manual.py: Add test for command logging

38d16c62

by Jürg Billeter at 2018-11-27T13:41:09Z

sandbox: Deduplicate code to process cwd, env, and command arguments

85f5b574

by Jürg Billeter at 2018-11-27T13:41:09Z

sandbox/sandbox.py: Add SandboxFlags.NONE

This improves readability.

bdd6925f

by Jürg Billeter at 2018-11-27T13:41:09Z

element.py: Use SandboxFlags.NONE instead of 0

69005c76

by Jürg Billeter at 2018-11-27T13:41:09Z

scriptelement.py: Use SandboxFlags.NONE instead of 0

f1767de2

by Jürg Billeter at 2018-11-27T13:41:09Z

sandbox/sandbox.py: Add command batching API

This adds the batch() context manager.

87f7c99a

by Jürg Billeter at 2018-11-27T13:41:09Z

element.py: Pass element object to sandbox

This allows the sandbox to output messages using the plugin id.

024d584d

by Jürg Billeter at 2018-11-27T13:41:09Z

element.py: Cache artifact also on SandboxCommandError

bb7c3213

by Jürg Billeter at 2018-11-27T13:41:09Z

element.py: Defer workspace.prepared setting in case of pending commands

04f05b1a

by Jürg Billeter at 2018-11-27T13:41:09Z

element.py: Support command batching for integration commands

7827ac50

by Jürg Billeter at 2018-11-27T13:44:49Z

element.py: Add batch_prepare_assemble() method

This allows batching of all sandbox commands aross prepare() and
assemble().

63db90b2

by Jürg Billeter at 2018-11-27T13:44:49Z

buildelement.py: Support batching for integration and build commands

73dec86e

by Jürg Billeter at 2018-11-27T13:44:49Z

scriptelement.py: Support batching for integration and build commands

79442e0a

by Jürg Billeter at 2018-11-27T13:44:49Z

compose.py: Support batching for integration commands

6018b3bd

by Jürg Billeter at 2018-11-27T13:44:49Z

sandbox/_sandboxremote.py: Implement command batching

Execute batched commands in a single shell script.

ec3d4486
by Jürg Billeter at 2018-11-27T13:44:49Z
```
NEWS: Add entry for command batching
```

127d332f

by Jürg Billeter at 2018-11-27T14:16:50Z

Merge branch 'juerg/command-batching' into 'master'

Command batching

Closes #675

See merge request BuildStream/buildstream!915

d5b5538d

by Jim MacArthur at 2018-11-27T15:32:19Z

artifactcache: Move ArtifactCacheSpec code into CASRemoteSpec.

There is nothing in ArtifactCacheSpec that's actually specific to
artifacts, so I've made it a CAS class so we can use the same
spec for remote execution.

f224d4cb

by Jim MacArthur at 2018-11-27T15:32:19Z

artifactcache.py, _context.py: Move CASCache object into context

Since the artifact cache and remote execution share the same
local CAS store, they should share the same CASCache object. Moving
this into context allows us to do this.

926c1446

by Jim MacArthur at 2018-11-27T15:33:34Z

_sandboxremote: Remove references to artifact cache.

Uses the new config options introduced earlier to specify remote
exec storage and execution service.

d7cdb03e

by Jim MacArthur at 2018-11-27T15:33:36Z

Documentation: Update to show new config options

95d8244f

by Jim MacArthur at 2018-11-27T15:33:36Z

tests/sandboxes/remote-exec-config.py: New test.

Tests that we get a sensible error message when we specify incomplete
or wrong information in the configuration for remote execution.

7f2f62eb

by Jim MacArthur at 2018-11-27T15:33:36Z

artifactcache.py: Remove unused verify_digest_pushed

c55ca3a5

by Jim MacArthur at 2018-11-27T15:33:36Z

Docs: Correct URLs in artifact cache example

The port number must come after the hostname and before the path.
The path is ignored by CASCache anyway, so including it is misleading.

27 changed files:

NEWS
buildstream/__init__.py
buildstream/_artifactcache/artifactcache.py
buildstream/_artifactcache/cascache.py
buildstream/_context.py
buildstream/_loader/loader.py
buildstream/_project.py
buildstream/buildelement.py
buildstream/data/projectconfig.yaml
buildstream/element.py
buildstream/plugins/elements/compose.py
buildstream/plugins/sources/local.py
buildstream/sandbox/__init__.py
buildstream/sandbox/_sandboxbwrap.py
buildstream/sandbox/_sandboxchroot.py
buildstream/sandbox/_sandboxdummy.py
buildstream/sandbox/_sandboxremote.py
buildstream/sandbox/sandbox.py
buildstream/scriptelement.py
buildstream/source.py
doc/source/format_project.rst
tests/integration/manual.py
tests/integration/sandbox-bwrap.py
+ tests/sandboxes/remote-exec-config.py
+ tests/sandboxes/remote-exec-config/missing-certs/certificates/client.crt
+ tests/sandboxes/remote-exec-config/missing-certs/certificates/client.key
+ tests/sandboxes/remote-exec-config/missing-certs/element.bst

Changes:

NEWS

@@ -67,6 +67,9 @@ buildstream 1.3.1
      allows the user to set a default location for their creation. This has meant
      that the new CLI is no longer backwards compatible with buildstream 1.2.
 +  o Add sandbox API for command batching and use it for build, script, and
 +    compose elements.
++
  =================
  buildstream 1.1.5

buildstream/__init__.py

@@ -27,7 +27,7 @@ if "_BST_COMPLETION" not in os.environ:
      del get_versions
      from .utils import UtilError, ProgramNotFoundError
 -    from .sandbox import Sandbox, SandboxFlags
 +    from .sandbox import Sandbox, SandboxFlags, SandboxCommandError
      from .types import Scope, Consistency
      from .plugin import Plugin
      from .source import Source, SourceError, SourceFetcher

buildstream/_artifactcache/artifactcache.py

@@ -21,7 +21,6 @@ import multiprocessing
  import os
  import signal
  import string
 -from collections import namedtuple
  from collections.abc import Mapping
  from ..types import _KeyStrength
@@ -31,7 +30,7 @@ from .. import _signals
  from .. import utils
  from .. import _yaml
 -from .cascache import CASCache, CASRemote
 +from .cascache import CASRemote, CASRemoteSpec
  CACHE_SIZE_FILE = "cache_size"
@@ -45,48 +44,8 @@ CACHE_SIZE_FILE = "cache_size"
  #     push (bool): Whether we should attempt to push artifacts to this cache,
  #                  in addition to pulling from it.
+ #
 -class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push server_cert client_key client_cert')):
+-
 -    # _new_from_config_node
 -    #
 -    # Creates an ArtifactCacheSpec() from a YAML loaded node
 -    #
 -    @staticmethod
 -    def _new_from_config_node(spec_node, basedir=None):
 -        _yaml.node_validate(spec_node, ['url', 'push', 'server-cert', 'client-key', 'client-cert'])
 -        url = _yaml.node_get(spec_node, str, 'url')
 -        push = _yaml.node_get(spec_node, bool, 'push', default_value=False)
 -        if not url:
 -            provenance = _yaml.node_get_provenance(spec_node, 'url')
 -            raise LoadError(LoadErrorReason.INVALID_DATA,
 -                            "{}: empty artifact cache URL".format(provenance))
+-
 -        server_cert = _yaml.node_get(spec_node, str, 'server-cert', default_value=None)
 -        if server_cert and basedir:
 -            server_cert = os.path.join(basedir, server_cert)
+-
 -        client_key = _yaml.node_get(spec_node, str, 'client-key', default_value=None)
 -        if client_key and basedir:
 -            client_key = os.path.join(basedir, client_key)
+-
 -        client_cert = _yaml.node_get(spec_node, str, 'client-cert', default_value=None)
 -        if client_cert and basedir:
 -            client_cert = os.path.join(basedir, client_cert)
+-
 -        if client_key and not client_cert:
 -            provenance = _yaml.node_get_provenance(spec_node, 'client-key')
 -            raise LoadError(LoadErrorReason.INVALID_DATA,
 -                            "{}: 'client-key' was specified without 'client-cert'".format(provenance))
+-
 -        if client_cert and not client_key:
 -            provenance = _yaml.node_get_provenance(spec_node, 'client-cert')
 -            raise LoadError(LoadErrorReason.INVALID_DATA,
 -                            "{}: 'client-cert' was specified without 'client-key'".format(provenance))
+-
 -        return ArtifactCacheSpec(url, push, server_cert, client_key, client_cert)
+-
+-
 -ArtifactCacheSpec.__new__.__defaults__ = (None, None, None)
 +class ArtifactCacheSpec(CASRemoteSpec):
 +    pass
  # An ArtifactCache manages artifacts.
@@ -99,7 +58,7 @@ class ArtifactCache():
          self.context = context
          self.extractdir = os.path.join(context.artifactdir, 'extract')
 -        self.cas = CASCache(context.artifactdir)
 +        self.cas = context.get_cascache()
          self.global_remote_specs = []
          self.project_remote_specs = {}
@@ -792,34 +751,6 @@ class ArtifactCache():
          return message_digest
 -    # verify_digest_pushed():
 -    #
 -    # Check whether the object is already on the server in which case
 -    # there is no need to upload it.
 -    #
 -    # Args:
 -    #     project (Project): The current project
 -    #     digest (Digest): The object digest.
 -    #
 -    def verify_digest_pushed(self, project, digest):
+-
 -        if self._has_push_remotes:
 -            push_remotes = [r for r in self._remotes[project] if r.spec.push]
 -        else:
 -            push_remotes = []
+-
 -        if not push_remotes:
 -            raise ArtifactError("verify_digest_pushed was called, but no remote artifact " +
 -                                "servers are configured as push remotes.")
+-
 -        pushed = False
+-
 -        for remote in push_remotes:
 -            if self.cas.verify_digest_on_remote(remote, digest):
 -                pushed = True
+-
 -        return pushed
+-
      # link_key():
+     #
      # Add a key for an existing artifact.

buildstream/_artifactcache/cascache.py

@@ -17,6 +17,7 @@
  #  Authors:
  #        Jürg Billeter <juerg billeter codethink co uk>
 +from collections import namedtuple
  import hashlib
  import itertools
  import io
@@ -34,7 +35,8 @@ from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remo
  from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
  from .. import utils
 -from .._exceptions import CASError
 +from .._exceptions import CASError, LoadError, LoadErrorReason
 +from .. import _yaml
  # The default limit for gRPC messages is 4 MiB.
@@ -42,6 +44,50 @@ from .._exceptions import CASError
  _MAX_PAYLOAD_BYTES = 1024 * 1024
 +class CASRemoteSpec(namedtuple('CASRemoteSpec', 'url push server_cert client_key client_cert')):
++
 +    # _new_from_config_node
 +    #
 +    # Creates an CASRemoteSpec() from a YAML loaded node
 +    #
 +    @staticmethod
 +    def _new_from_config_node(spec_node, basedir=None):
 +        _yaml.node_validate(spec_node, ['url', 'push', 'server-cert', 'client-key', 'client-cert'])
 +        url = _yaml.node_get(spec_node, str, 'url')
 +        push = _yaml.node_get(spec_node, bool, 'push', default_value=False)
 +        if not url:
 +            provenance = _yaml.node_get_provenance(spec_node, 'url')
 +            raise LoadError(LoadErrorReason.INVALID_DATA,
 +                            "{}: empty artifact cache URL".format(provenance))
++
 +        server_cert = _yaml.node_get(spec_node, str, 'server-cert', default_value=None)
 +        if server_cert and basedir:
 +            server_cert = os.path.join(basedir, server_cert)
++
 +        client_key = _yaml.node_get(spec_node, str, 'client-key', default_value=None)
 +        if client_key and basedir:
 +            client_key = os.path.join(basedir, client_key)
++
 +        client_cert = _yaml.node_get(spec_node, str, 'client-cert', default_value=None)
 +        if client_cert and basedir:
 +            client_cert = os.path.join(basedir, client_cert)
++
 +        if client_key and not client_cert:
 +            provenance = _yaml.node_get_provenance(spec_node, 'client-key')
 +            raise LoadError(LoadErrorReason.INVALID_DATA,
 +                            "{}: 'client-key' was specified without 'client-cert'".format(provenance))
++
 +        if client_cert and not client_key:
 +            provenance = _yaml.node_get_provenance(spec_node, 'client-cert')
 +            raise LoadError(LoadErrorReason.INVALID_DATA,
 +                            "{}: 'client-cert' was specified without 'client-key'".format(provenance))
++
 +        return CASRemoteSpec(url, push, server_cert, client_key, client_cert)
++
++
 +CASRemoteSpec.__new__.__defaults__ = (None, None, None)
++
++
  # A CASCache manages a CAS repository as specified in the Remote Execution API.
+ #
  # Args:

buildstream/_context.py

@@ -31,6 +31,7 @@ from ._exceptions import LoadError, LoadErrorReason, BstError
  from ._message import Message, MessageType
  from ._profile import Topics, profile_start, profile_end
  from ._artifactcache import ArtifactCache
 +from ._artifactcache.cascache import CASCache
  from ._workspaces import Workspaces
  from .plugin import _plugin_lookup
@@ -141,6 +142,7 @@ class Context():
          self._workspaces = None
          self._log_handle = None
          self._log_filename = None
 +        self._cascache = None
      # load()
+     #
@@ -620,6 +622,11 @@ class Context():
          if not os.environ.get('XDG_DATA_HOME'):
              os.environ['XDG_DATA_HOME'] = os.path.expanduser('~/.local/share')
 +    def get_cascache(self):
 +        if self._cascache is None:
 +            self._cascache = CASCache(self.artifactdir)
 +        return self._cascache
++
  # _node_get_option_str()
+ #

buildstream/_loader/loader.py

@@ -563,17 +563,23 @@ class Loader():
                                  "Subproject has no ref for junction: {}".format(filename),
                                  detail=detail)
 -        # Stage sources
 -        os.makedirs(self._context.builddir, exist_ok=True)
 -        basedir = tempfile.mkdtemp(prefix="{}-".format(element.normal_name), dir=self._context.builddir)
 -        element._stage_sources_at(basedir, mount_workspaces=False)
 +        if len(sources) == 1 and sources[0]._get_local_path():
 +            # Optimization for junctions with a single local source
 +            basedir = sources[0]._get_local_path()
 +            tempdir = None
 +        else:
 +            # Stage sources
 +            os.makedirs(self._context.builddir, exist_ok=True)
 +            basedir = tempfile.mkdtemp(prefix="{}-".format(element.normal_name), dir=self._context.builddir)
 +            element._stage_sources_at(basedir, mount_workspaces=False)
 +            tempdir = basedir
          # Load the project
          project_dir = os.path.join(basedir, element.path)
          try:
              from .._project import Project
              project = Project(project_dir, self._context, junction=element,
 -                              parent_loader=self, tempdir=basedir)
 +                              parent_loader=self, tempdir=tempdir)
          except LoadError as e:
              if e.reason == LoadErrorReason.MISSING_PROJECT_CONF:
                  raise LoadError(reason=LoadErrorReason.INVALID_JUNCTION,

buildstream/_project.py

@@ -30,6 +30,7 @@ from ._profile import Topics, profile_start, profile_end
  from ._exceptions import LoadError, LoadErrorReason
  from ._options import OptionPool
  from ._artifactcache import ArtifactCache
 +from .sandbox import SandboxRemote
  from ._elementfactory import ElementFactory
  from ._sourcefactory import SourceFactory
  from .plugin import CoreWarnings
@@ -130,7 +131,7 @@ class Project():
          self._shell_host_files = []   # A list of HostMount objects
          self.artifact_cache_specs = None
 -        self.remote_execution_url = None
 +        self.remote_execution_specs = None
          self._sandbox = None
          self._splits = None
@@ -493,9 +494,7 @@ class Project():
          self.artifact_cache_specs = ArtifactCache.specs_from_config_node(config, self.directory)
          # Load remote-execution configuration for this project
 -        remote_execution = _yaml.node_get(config, Mapping, 'remote-execution')
 -        _yaml.node_validate(remote_execution, ['url'])
 -        self.remote_execution_url = _yaml.node_get(remote_execution, str, 'url')
 +        self.remote_execution_specs = SandboxRemote.specs_from_config_node(config, self.directory)
          # Load sandbox environment variables
          self.base_environment = _yaml.node_get(config, Mapping, 'environment')

buildstream/buildelement.py

@@ -127,7 +127,7 @@ artifact collection purposes.
  """
  import os
 -from . import Element, Scope, ElementError
 +from . import Element, Scope
  from . import SandboxFlags
@@ -207,6 +207,10 @@ class BuildElement(Element):
          # Setup environment
          sandbox.set_environment(self.get_environment())
 +        # Enable command batching across prepare() and assemble()
 +        self.batch_prepare_assemble(SandboxFlags.ROOT_READ_ONLY,
 +                                    collect=self.get_variable('install-root'))
++
      def stage(self, sandbox):
          # Stage deps in the sandbox root
@@ -215,7 +219,7 @@ class BuildElement(Element):
          # Run any integration commands provided by the dependencies
          # once they are all staged and ready
 -        with self.timed_activity("Integrating sandbox"):
 +        with sandbox.batch(SandboxFlags.NONE, label="Integrating sandbox"):
              for dep in self.dependencies(Scope.BUILD):
                  dep.integrate(sandbox)
@@ -223,14 +227,13 @@ class BuildElement(Element):
          self.stage_sources(sandbox, self.get_variable('build-root'))
      def assemble(self, sandbox):
+-
          # Run commands
          for command_name in _command_steps:
              commands = self.__commands[command_name]
              if not commands or command_name == 'configure-commands':
                  continue
 -            with self.timed_activity("Running {}".format(command_name)):
 +            with sandbox.batch(SandboxFlags.ROOT_READ_ONLY, label="Running {}".format(command_name)):
                  for cmd in commands:
                      self.__run_command(sandbox, cmd, command_name)
@@ -254,7 +257,7 @@ class BuildElement(Element):
      def prepare(self, sandbox):
          commands = self.__commands['configure-commands']
          if commands:
 -            with self.timed_activity("Running configure-commands"):
 +            with sandbox.batch(SandboxFlags.ROOT_READ_ONLY, label="Running configure-commands"):
                  for cmd in commands:
                      self.__run_command(sandbox, cmd, 'configure-commands')
@@ -282,13 +285,9 @@ class BuildElement(Element):
          return commands
      def __run_command(self, sandbox, cmd, cmd_name):
 -        self.status("Running {}".format(cmd_name), detail=cmd)
+-
          # Note the -e switch to 'sh' means to exit with an error
          # if any untested command fails.
+         #
 -        exitcode = sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 -                               SandboxFlags.ROOT_READ_ONLY)
 -        if exitcode != 0:
 -            raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode),
 -                               collect=self.get_variable('install-root'))
 +        sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 +                    SandboxFlags.ROOT_READ_ONLY,
 +                    label=cmd)

buildstream/data/projectconfig.yaml

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

buildstream/element.py

@@ -78,6 +78,7 @@ import stat
  import copy
  from collections import OrderedDict
  from collections.abc import Mapping
 +import contextlib
  from contextlib import contextmanager
  import tempfile
  import shutil
@@ -89,7 +90,7 @@ from ._exceptions import BstError, LoadError, LoadErrorReason, ImplError, \
      ErrorDomain
  from .utils import UtilError
  from . import Plugin, Consistency, Scope
 -from . import SandboxFlags
 +from . import SandboxFlags, SandboxCommandError
  from . import utils
  from . import _cachekey
  from . import _signals
@@ -217,6 +218,10 @@ class Element(Plugin):
          self.__build_result = None              # The result of assembling this Element (success, description, detail)
          self._build_log_path = None            # The path of the build log for this Element
 +        self.__batch_prepare_assemble = False         # Whether batching across prepare()/assemble() is configured
 +        self.__batch_prepare_assemble_flags = 0       # Sandbox flags for batching across prepare()/assemble()
 +        self.__batch_prepare_assemble_collect = None  # Collect dir for batching across prepare()/assemble()
++
          # hash tables of loaded artifact metadata, hashed by key
          self.__metadata_keys = {}                     # Strong and weak keys for this key
          self.__metadata_dependencies = {}             # Dictionary of dependency strong keys
@@ -250,9 +255,9 @@ class Element(Plugin):
          # Extract remote execution URL
          if not self.__is_junction:
 -            self.__remote_execution_url = project.remote_execution_url
 +            self.__remote_execution_specs = project.remote_execution_specs
          else:
 -            self.__remote_execution_url = None
 +            self.__remote_execution_specs = None
          # Extract Sandbox config
          self.__sandbox_config = self.__extract_sandbox_config(meta)
@@ -770,13 +775,13 @@ class Element(Plugin):
          environment = self.get_environment()
          if bstdata is not None:
 -            commands = self.node_get_member(bstdata, list, 'integration-commands', [])
 -            for i in range(len(commands)):
 -                cmd = self.node_subst_list_element(bstdata, 'integration-commands', [i])
 -                self.status("Running integration command", detail=cmd)
 -                exitcode = sandbox.run(['sh', '-e', '-c', cmd], 0, env=environment, cwd='/')
 -                if exitcode != 0:
 -                    raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode))
 +            with sandbox.batch(SandboxFlags.NONE):
 +                commands = self.node_get_member(bstdata, list, 'integration-commands', [])
 +                for i in range(len(commands)):
 +                    cmd = self.node_subst_list_element(bstdata, 'integration-commands', [i])
++
 +                    sandbox.run(['sh', '-e', '-c', cmd], 0, env=environment, cwd='/',
 +                                label=cmd)
      def stage_sources(self, sandbox, directory):
          """Stage this element's sources to a directory in the sandbox
@@ -863,6 +868,24 @@ class Element(Plugin):
          return None
 +    def batch_prepare_assemble(self, flags, *, collect=None):
 +        """ Configure command batching across prepare() and assemble()
++
 +        Args:
 +           flags (:class:`.SandboxFlags`): The sandbox flags for the command batch
 +           collect (str): An optional directory containing partial install contents
 +                          on command failure.
++
 +        This may be called in :func:`Element.configure_sandbox() <buildstream.element.Element.configure_sandbox>`
 +        to enable batching of all sandbox commands issued in prepare() and assemble().
 +        """
 +        if self.__batch_prepare_assemble:
 +            raise ElementError("{}: Command batching for prepare/assemble is already configured".format(self))
++
 +        self.__batch_prepare_assemble = True
 +        self.__batch_prepare_assemble_flags = flags
 +        self.__batch_prepare_assemble_collect = collect
++
      #############################################################
      #            Private Methods used in BuildStream            #
      #############################################################
@@ -1323,7 +1346,7 @@ class Element(Plugin):
                              bare_directory=bare_directory) as sandbox:
              # Configure always comes first, and we need it.
 -            self.configure_sandbox(sandbox)
 +            self.__configure_sandbox(sandbox)
              # Stage something if we need it
              if not directory:
@@ -1556,15 +1579,24 @@ class Element(Plugin):
                  # Call the abstract plugin methods
                  try:
                      # Step 1 - Configure
 -                    self.configure_sandbox(sandbox)
 +                    self.__configure_sandbox(sandbox)
                      # Step 2 - Stage
                      self.stage(sandbox)
 -                    # Step 3 - Prepare
 -                    self.__prepare(sandbox)
 -                    # Step 4 - Assemble
 -                    collect = self.assemble(sandbox)  # pylint: disable=assignment-from-no-return
++
 +                    if self.__batch_prepare_assemble:
 +                        cm = sandbox.batch(self.__batch_prepare_assemble_flags,
 +                                           collect=self.__batch_prepare_assemble_collect)
 +                    else:
 +                        cm = contextlib.suppress()
++
 +                    with cm:
 +                        # Step 3 - Prepare
 +                        self.__prepare(sandbox)
 +                        # Step 4 - Assemble
 +                        collect = self.assemble(sandbox)  # pylint: disable=assignment-from-no-return
++
                      self.__set_build_result(success=True, description="succeeded")
 -                except ElementError as e:
 +                except (ElementError, SandboxCommandError) as e:
                      # Shelling into a sandbox is useful to debug this error
                      e.sandbox = True
@@ -2059,6 +2091,15 @@ class Element(Plugin):
      def __can_build_incrementally(self):
          return bool(self._get_workspace())
 +    # __configure_sandbox():
 +    #
 +    # Internal method for calling public abstract configure_sandbox() method.
 +    #
 +    def __configure_sandbox(self, sandbox):
 +        self.__batch_prepare_assemble = False
++
 +        self.configure_sandbox(sandbox)
++
      # __prepare():
+     #
      # Internal method for calling public abstract prepare() method.
@@ -2074,7 +2115,12 @@ class Element(Plugin):
              self.prepare(sandbox)
              if workspace:
 -                workspace.prepared = True
 +                def mark_workspace_prepared():
 +                    workspace.prepared = True
++
 +                # Defer workspace.prepared setting until pending batch commands
 +                # have been executed.
 +                sandbox._callback(mark_workspace_prepared)
      def __is_cached(self, keystrength):
          if keystrength is None:
@@ -2125,7 +2171,7 @@ class Element(Plugin):
      # supports it.
+     #
      def __use_remote_execution(self):
 -        return self.__remote_execution_url and self.BST_VIRTUAL_DIRECTORY
 +        return self.__remote_execution_specs and self.BST_VIRTUAL_DIRECTORY
      # __sandbox():
+     #
@@ -2157,16 +2203,17 @@ class Element(Plugin):
              sandbox = SandboxRemote(context, project,
                                      directory,
 +                                    plugin=self,
                                      stdout=stdout,
                                      stderr=stderr,
                                      config=config,
 -                                    server_url=self.__remote_execution_url,
 +                                    specs=self.__remote_execution_specs,
                                      bare_directory=bare_directory,
                                      allow_real_directory=False)
              yield sandbox
          elif directory is not None and os.path.exists(directory):
 -            if allow_remote and self.__remote_execution_url:
 +            if allow_remote and self.__remote_execution_specs:
                  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")
@@ -2175,6 +2222,7 @@ class Element(Plugin):
              sandbox = platform.create_sandbox(context, project,
                                                directory,
 +                                              plugin=self,
                                                stdout=stdout,
                                                stderr=stderr,
                                                config=config,

buildstream/plugins/elements/compose.py

@@ -122,8 +122,9 @@ class ComposeElement(Element):
                      snapshot = set(vbasedir.list_relative_paths())
                      vbasedir.mark_unmodified()
 -                for dep in self.dependencies(Scope.BUILD):
 -                    dep.integrate(sandbox)
 +                with sandbox.batch(0):
 +                    for dep in self.dependencies(Scope.BUILD):
 +                        dep.integrate(sandbox)
                  if require_split:
                      # Calculate added, modified and removed files

buildstream/plugins/sources/local.py

@@ -124,6 +124,9 @@ class LocalSource(Source):
                      else:
                          os.chmod(path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
 +    def _get_local_path(self):
 +        return self.fullpath
++
  # Create a unique key for a file
  def unique_key(filename):

buildstream/sandbox/__init__.py

@@ -17,6 +17,6 @@
  #  Authors:
  #        Tristan Maat <tristan maat codethink co uk>
 -from .sandbox import Sandbox, SandboxFlags
 +from .sandbox import Sandbox, SandboxFlags, SandboxCommandError
  from ._sandboxremote import SandboxRemote
  from ._sandboxdummy import SandboxDummy

buildstream/sandbox/_sandboxbwrap.py

@@ -58,22 +58,12 @@ class SandboxBwrap(Sandbox):
          self.die_with_parent_available = kwargs['die_with_parent_available']
          self.json_status_available = kwargs['json_status_available']
 -    def run(self, command, flags, *, cwd=None, env=None):
 +    def _run(self, command, flags, *, cwd, env):
          stdout, stderr = self._get_output()
          # Allowable access to underlying storage as we're part of the sandbox
          root_directory = self.get_virtual_directory()._get_underlying_directory()
 -        # Fallback to the sandbox default settings for
 -        # the cwd and env.
 -        #
 -        cwd = self._get_work_directory(cwd=cwd)
 -        env = self._get_environment(cwd=cwd, env=env)
+-
 -        # Convert single-string argument to a list
 -        if isinstance(command, str):
 -            command = [command]
+-
          if not self._has_command(command[0], env):
              raise SandboxError("Staged artifacts do not provide command "
                                 "'{}'".format(command[0]),

buildstream/sandbox/_sandboxchroot.py

@@ -49,17 +49,7 @@ class SandboxChroot(Sandbox):
          self.mount_map = None
 -    def run(self, command, flags, *, cwd=None, env=None):
+-
 -        # Fallback to the sandbox default settings for
 -        # the cwd and env.
 -        #
 -        cwd = self._get_work_directory(cwd=cwd)
 -        env = self._get_environment(cwd=cwd, env=env)
+-
 -        # Convert single-string argument to a list
 -        if isinstance(command, str):
 -            command = [command]
 +    def _run(self, command, flags, *, cwd, env):
          if not self._has_command(command[0], env):
              raise SandboxError("Staged artifacts do not provide command "

buildstream/sandbox/_sandboxdummy.py

@@ -25,17 +25,7 @@ class SandboxDummy(Sandbox):
          super().__init__(*args, **kwargs)
          self._reason = kwargs.get("dummy_reason", "no reason given")
 -    def run(self, command, flags, *, cwd=None, env=None):
+-
 -        # Fallback to the sandbox default settings for
 -        # the cwd and env.
 -        #
 -        cwd = self._get_work_directory(cwd=cwd)
 -        env = self._get_environment(cwd=cwd, env=env)
+-
 -        # Convert single-string argument to a list
 -        if isinstance(command, str):
 -            command = [command]
 +    def _run(self, command, flags, *, cwd, env):
          if not self._has_command(command[0], env):
              raise SandboxError("Staged artifacts do not provide command "

buildstream/sandbox/_sandboxremote.py

@@ -19,19 +19,28 @@
  #        Jim MacArthur <jim macarthur codethink co uk>
  import os
 +import shlex
 +from collections import namedtuple
  from urllib.parse import urlparse
  from functools import partial
  import grpc
 -from . import Sandbox
 +from . import Sandbox, SandboxCommandError
 +from .sandbox import _SandboxBatch
  from ..storage._filebaseddirectory import FileBasedDirectory
  from ..storage._casbaseddirectory import CasBasedDirectory
  from .. import _signals
  from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
  from .._protos.google.rpc import code_pb2
  from .._exceptions import SandboxError
 +from .. import _yaml
  from .._protos.google.longrunning import operations_pb2, operations_pb2_grpc
 +from .._artifactcache.cascache import CASRemote, CASRemoteSpec
++
++
 +class RemoteExecutionSpec(namedtuple('RemoteExecutionSpec', 'exec_service storage_service')):
 +    pass
  # SandboxRemote()
@@ -44,18 +53,70 @@ class SandboxRemote(Sandbox):
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
 -        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(kwargs['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).")
 +        config = kwargs['specs']  # This should be a RemoteExecutionSpec
 +        if config is None:
 +            return
++
 +        self.storage_url = config.storage_service['url']
 +        self.exec_url = config.exec_service['url']
 -        self.server_url = '{}:{}'.format(url.hostname, url.port)
 +        self.storage_remote_spec = CASRemoteSpec(self.storage_url, push=True,
 +                                                 server_cert=config.storage_service['server-cert'],
 +                                                 client_key=config.storage_service['client-key'],
 +                                                 client_cert=config.storage_service['client-cert'])
          self.operation_name = None
 +    @staticmethod
 +    def specs_from_config_node(config_node, basedir):
++
 +        def require_node(config, keyname):
 +            val = config.get(keyname)
 +            if val is None:
 +                provenance = _yaml.node_get_provenance(remote_config, key=keyname)
 +                raise _yaml.LoadError(_yaml.LoadErrorReason.INVALID_DATA,
 +                                      "{}: '{}' was not present in the remote "
 +                                      "execution configuration (remote-execution). "
 +                                      .format(str(provenance), keyname))
 +            return val
++
 +        remote_config = config_node.get("remote-execution", None)
 +        if remote_config is None:
 +            return None
++
 +        # Maintain some backwards compatibility with older configs, in which 'url' was the only valid key for
 +        # remote-execution.
++
 +        tls_keys = ['client-key', 'client-cert', 'server-cert']
++
 +        _yaml.node_validate(remote_config, ['execution-service', 'storage-service', 'url'])
 +        remote_exec_service_config = require_node(remote_config, 'execution-service')
 +        remote_exec_storage_config = require_node(remote_config, 'storage-service')
++
 +        _yaml.node_validate(remote_exec_service_config, ['url'])
 +        _yaml.node_validate(remote_exec_storage_config, ['url'] + tls_keys)
++
 +        if 'url' in remote_config:
 +            if 'execution-service' not in remote_config:
 +                remote_config['execution-service'] = {'url': remote_config['url']}
 +            else:
 +                provenance = _yaml.node_get_provenance(remote_config, key='url')
 +                raise _yaml.LoadError(_yaml.LoadErrorReason.INVALID_DATA,
 +                                      "{}: 'url' and 'execution-service' keys were found in the remote "
 +                                      "execution configuration (remote-execution). "
 +                                      "You can only specify one of these."
 +                                      .format(str(provenance)))
++
 +        for key in tls_keys:
 +            if key not in remote_exec_storage_config:
 +                provenance = _yaml.node_get_provenance(remote_config, key='storage-service')
 +                raise _yaml.LoadError(_yaml.LoadErrorReason.INVALID_DATA,
 +                                      "{}: The keys {} are necessary for the storage-service section of "
 +                                      "remote-execution configuration. Your config is missing '{}'."
 +                                      .format(str(provenance), tls_keys, key))
++
 +        spec = RemoteExecutionSpec(remote_config['execution-service'], remote_config['storage-service'])
 +        return spec
++
      def run_remote_command(self, command, input_root_digest, working_directory, environment):
          # Sends an execution request to the remote execution server.
+         #
@@ -73,12 +134,13 @@ class SandboxRemote(Sandbox):
                                                        output_directories=[self._output_directory],
                                                        platform=None)
          context = self._get_context()
 -        cascache = context.artifactcache
 +        cascache = context.get_cascache()
 +        casremote = CASRemote(self.storage_remote_spec)
++
          # 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_digest_pushed(self._get_project(), command_digest):
 +        command_digest = cascache.push_message(casremote, remote_command)
 +        if not command_digest or not cascache.verify_digest_on_remote(casremote, command_digest):
              raise SandboxError("Failed pushing build command to remote CAS.")
+-
          # Create and send the action.
          action = remote_execution_pb2.Action(command_digest=command_digest,
                                               input_root_digest=input_root_digest,
@@ -86,12 +148,21 @@ class SandboxRemote(Sandbox):
                                               do_not_cache=False)
          # 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_digest_pushed(self._get_project(), action_digest):
 +        action_digest = cascache.push_message(casremote, action)
 +        if not action_digest or not cascache.verify_digest_on_remote(casremote, action_digest):
              raise SandboxError("Failed pushing build action to remote CAS.")
          # Next, try to create a communication channel to the BuildGrid server.
 -        channel = grpc.insecure_channel(self.server_url)
 +        url = urlparse(self.exec_url)
 +        if not url.port:
 +            raise SandboxError("You must supply a protocol and port number in the execution-service url, "
 +                               "for example: http://buildservice:50051.")
 +        if url.scheme == 'http':
 +            channel = grpc.insecure_channel('{}:{}'.format(url.hostname, url.port))
 +        else:
 +            raise SandboxError("Remote execution currently only supports the 'http' protocol "
 +                               "and '{}' was supplied.".format(url.scheme))
++
          stub = remote_execution_pb2_grpc.ExecutionStub(channel)
          request = remote_execution_pb2.ExecuteRequest(action_digest=action_digest,
                                                        skip_cache_lookup=False)
@@ -117,7 +188,7 @@ class SandboxRemote(Sandbox):
                  status_code = e.code()
                  if status_code == grpc.StatusCode.UNAVAILABLE:
                      raise SandboxError("Failed contacting remote execution server at {}."
 -                                       .format(self.server_url))
 +                                       .format(self.exec_url))
                  elif status_code in (grpc.StatusCode.INVALID_ARGUMENT,
                                       grpc.StatusCode.FAILED_PRECONDITION,
@@ -188,9 +259,11 @@ class SandboxRemote(Sandbox):
              raise SandboxError("Output directory structure had no digest attached.")
          context = self._get_context()
 -        cascache = context.artifactcache
 +        cascache = context.get_cascache()
 +        casremote = CASRemote(self.storage_remote_spec)
++
          # Now do a pull to ensure we have the necessary parts.
 -        dir_digest = cascache.pull_tree(self._get_project(), tree_digest)
 +        dir_digest = cascache.pull_tree(casremote, 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.")
@@ -212,33 +285,28 @@ class SandboxRemote(Sandbox):
          new_dir = CasBasedDirectory(self._get_context().artifactcache.cas, ref=dir_digest)
          self._set_virtual_directory(new_dir)
 -    def run(self, command, flags, *, cwd=None, env=None):
 +    def _run(self, command, flags, *, cwd, env):
          # Upload sources
          upload_vdir = self.get_virtual_directory()
 +        cascache = self._get_context().get_cascache()
          if isinstance(upload_vdir, FileBasedDirectory):
              # Make a new temporary directory to put source in
 -            upload_vdir = CasBasedDirectory(self._get_context().artifactcache.cas, ref=None)
 +            upload_vdir = CasBasedDirectory(cascache, ref=None)
              upload_vdir.import_files(self.get_virtual_directory()._get_underlying_directory())
          upload_vdir.recalculate_hash()
 -        context = self._get_context()
 -        cascache = context.artifactcache
 +        casremote = CASRemote(self.storage_remote_spec)
          # Now, push that key (without necessarily needing a ref) to the remote.
 -        cascache.push_directory(self._get_project(), upload_vdir)
 -        if not cascache.verify_digest_pushed(self._get_project(), upload_vdir.ref):
 -            raise SandboxError("Failed to verify that source has been pushed to the remote artifact cache.")
 -        # Fallback to the sandbox default settings for
 -        # the cwd and env.
 -        #
 -        cwd = self._get_work_directory(cwd=cwd)
 -        env = self._get_environment(cwd=cwd, env=env)
 +        try:
 +            cascache.push_directory(casremote, upload_vdir)
 +        except grpc.RpcError as e:
 +            raise SandboxError("Failed to push source directory to remote: {}".format(e)) from e
 -        # We want command args as a list of strings
 -        if isinstance(command, str):
 -            command = [command]
 +        if not cascache.verify_digest_on_remote(casremote, upload_vdir.ref):
 +            raise SandboxError("Failed to verify that source has been pushed to the remote artifact cache.")
          # Now transmit the command to execute
          operation = self.run_remote_command(command, upload_vdir.ref, cwd, env)
@@ -275,3 +343,69 @@ class SandboxRemote(Sandbox):
          self.process_job_output(action_result.output_directories, action_result.output_files)
          return 0
++
 +    def _create_batch(self, main_group, flags, *, collect=None):
 +        return _SandboxRemoteBatch(self, main_group, flags, collect=collect)
++
++
 +# _SandboxRemoteBatch()
 +#
 +# Command batching by shell script generation.
 +#
 +class _SandboxRemoteBatch(_SandboxBatch):
++
 +    def __init__(self, sandbox, main_group, flags, *, collect=None):
 +        super().__init__(sandbox, main_group, flags, collect=collect)
++
 +        self.script = None
 +        self.first_command = None
 +        self.cwd = None
 +        self.env = None
++
 +    def execute(self):
 +        self.script = ""
++
 +        self.main_group.execute(self)
++
 +        first = self.first_command
 +        if first and self.sandbox.run(['sh', '-c', '-e', self.script], self.flags, cwd=first.cwd, env=first.env) != 0:
 +            raise SandboxCommandError("Command execution failed", collect=self.collect)
++
 +    def execute_group(self, group):
 +        group.execute_children(self)
++
 +    def execute_command(self, command):
 +        if self.first_command is None:
 +            # First command in batch
 +            # Initial working directory and environment of script already matches
 +            # the command configuration.
 +            self.first_command = command
 +        else:
 +            # Change working directory for this command
 +            if command.cwd != self.cwd:
 +                self.script += "mkdir -p {}\n".format(command.cwd)
 +                self.script += "cd {}\n".format(command.cwd)
++
 +            # Update environment for this command
 +            for key in self.env.keys():
 +                if key not in command.env:
 +                    self.script += "unset {}\n".format(key)
 +            for key, value in command.env.items():
 +                if key not in self.env or self.env[key] != value:
 +                    self.script += "export {}={}\n".format(key, shlex.quote(value))
++
 +        # Keep track of current working directory and environment
 +        self.cwd = command.cwd
 +        self.env = command.env
++
 +        # Actual command execution
 +        cmdline = ' '.join(shlex.quote(cmd) for cmd in command.command)
 +        self.script += "(set -ex; {})".format(cmdline)
++
 +        # Error handling
 +        label = command.label or cmdline
 +        quoted_label = shlex.quote("'{}'".format(label))
 +        self.script += " || (echo Command {} failed with exitcode $? >&2 ; exit 1)\n".format(quoted_label)
++
 +    def execute_call(self, call):
 +        raise SandboxError("SandboxRemote does not support callbacks in command batches")

buildstream/sandbox/sandbox.py

+ #
  #  Copyright (C) 2017 Codethink Limited
 +#  Copyright (C) 2018 Bloomberg Finance LP
+ #
  #  This program is free software; you can redistribute it and/or
  #  modify it under the terms of the GNU Lesser General Public
@@ -29,7 +30,12 @@ See also: :ref:`sandboxing`.
  """
  import os
 -from .._exceptions import ImplError, BstError
 +import shlex
 +import contextlib
 +from contextlib import contextmanager
++
 +from .._exceptions import ImplError, BstError, SandboxError
 +from .._message import Message, MessageType
  from ..storage._filebaseddirectory import FileBasedDirectory
  from ..storage._casbaseddirectory import CasBasedDirectory
@@ -38,6 +44,10 @@ class SandboxFlags():
      """Flags indicating how the sandbox should be run.
      """
 +    NONE = 0
 +    """Use default sandbox configuration.
 +    """
++
      ROOT_READ_ONLY = 0x01
      """The root filesystem is read only.
@@ -71,6 +81,19 @@ class SandboxFlags():
      """
 +class SandboxCommandError(SandboxError):
 +    """Raised by :class:`.Sandbox` implementations when a command fails.
++
 +    Args:
 +       message (str): The error message to report to the user
 +       collect (str): An optional directory containing partial install contents
 +    """
 +    def __init__(self, message, *, collect=None):
 +        super().__init__(message, reason='command-failed')
++
 +        self.collect = collect
++
++
  class Sandbox():
      """Sandbox()
@@ -94,6 +117,13 @@ class Sandbox():
          self.__mount_sources = {}
          self.__allow_real_directory = kwargs['allow_real_directory']
 +        # Plugin ID for logging
 +        plugin = kwargs.get('plugin', None)
 +        if plugin:
 +            self.__plugin_id = plugin._get_unique_id()
 +        else:
 +            self.__plugin_id = None
++
          # Configuration from kwargs common to all subclasses
          self.__config = kwargs['config']
          self.__stdout = kwargs['stdout']
@@ -121,6 +151,9 @@ class Sandbox():
          # directory via get_directory.
          self._never_cache_vdirs = False
 +        # Pending command batch
 +        self.__batch = None
++
      def get_directory(self):
          """Fetches the sandbox root directory
@@ -209,9 +242,16 @@ class Sandbox():
              'artifact': artifact
          })
 -    def run(self, command, flags, *, cwd=None, env=None):
 +    def run(self, command, flags, *, cwd=None, env=None, label=None):
          """Run a command in the sandbox.
 +        If this is called outside a batch context, the command is immediately
 +        executed.
++
 +        If this is called in a batch context, the command is added to the batch
 +        for later execution. If the command fails, later commands will not be
 +        executed. Command flags must match batch flags.
++
          Args:
              command (list): The command to run in the sandboxed environment, as a list
                              of strings starting with the binary to run.
@@ -219,9 +259,10 @@ class Sandbox():
              cwd (str): The sandbox relative working directory in which to run the command.
              env (dict): A dictionary of string key, value pairs to set as environment
                          variables inside the sandbox environment.
 +            label (str): An optional label for the command, used for logging. (*Since: 1.4*)
          Returns:
 -            (int): The program exit code.
 +            (int|None): The program exit code, or None if running in batch context.
          Raises:
              (:class:`.ProgramNotFoundError`): If a host tool which the given sandbox
@@ -234,9 +275,115 @@ class Sandbox():
             function must make sure the directory will be created if it does
             not exist yet, even if a workspace is being used.
          """
 -        raise ImplError("Sandbox of type '{}' does not implement run()"
++
 +        # Fallback to the sandbox default settings for
 +        # the cwd and env.
 +        #
 +        cwd = self._get_work_directory(cwd=cwd)
 +        env = self._get_environment(cwd=cwd, env=env)
++
 +        # Convert single-string argument to a list
 +        if isinstance(command, str):
 +            command = [command]
++
 +        if self.__batch:
 +            if flags != self.__batch.flags:
 +                raise SandboxError("Inconsistent sandbox flags in single command batch")
++
 +            batch_command = _SandboxBatchCommand(command, cwd=cwd, env=env, label=label)
++
 +            current_group = self.__batch.current_group
 +            current_group.append(batch_command)
 +            return None
 +        else:
 +            return self._run(command, flags, cwd=cwd, env=env)
++
 +    @contextmanager
 +    def batch(self, flags, *, label=None, collect=None):
 +        """Context manager for command batching
++
 +        This provides a batch context that defers execution of commands until
 +        the end of the context. If a command fails, the batch will be aborted
 +        and subsequent commands will not be executed.
++
 +        Command batches may be nested. Execution will start only when the top
 +        level batch context ends.
++
 +        Args:
 +            flags (:class:`.SandboxFlags`): The flags for this command batch.
 +            label (str): An optional label for the batch group, used for logging.
 +            collect (str): An optional directory containing partial install contents
 +                           on command failure.
++
 +        Raises:
 +            (:class:`.SandboxCommandError`): If a command fails.
++
 +        *Since: 1.4*
 +        """
++
 +        group = _SandboxBatchGroup(label=label)
++
 +        if self.__batch:
 +            # Nested batch
 +            if flags != self.__batch.flags:
 +                raise SandboxError("Inconsistent sandbox flags in single command batch")
++
 +            parent_group = self.__batch.current_group
 +            parent_group.append(group)
 +            self.__batch.current_group = group
 +            try:
 +                yield
 +            finally:
 +                self.__batch.current_group = parent_group
 +        else:
 +            # Top-level batch
 +            batch = self._create_batch(group, flags, collect=collect)
++
 +            self.__batch = batch
 +            try:
 +                yield
 +            finally:
 +                self.__batch = None
++
 +            batch.execute()
++
 +    #####################################################
 +    #    Abstract Methods for Sandbox implementations   #
 +    #####################################################
++
 +    # _run()
 +    #
 +    # Abstract method for running a single command
 +    #
 +    # Args:
 +    #    command (list): The command to run in the sandboxed environment, as a list
 +    #                    of strings starting with the binary to run.
 +    #    flags (:class:`.SandboxFlags`): The flags for running this command.
 +    #    cwd (str): The sandbox relative working directory in which to run the command.
 +    #    env (dict): A dictionary of string key, value pairs to set as environment
 +    #                variables inside the sandbox environment.
 +    #
 +    # Returns:
 +    #    (int): The program exit code.
 +    #
 +    def _run(self, command, flags, *, cwd, env):
 +        raise ImplError("Sandbox of type '{}' does not implement _run()"
                          .format(type(self).__name__))
 +    # _create_batch()
 +    #
 +    # Abstract method for creating a batch object. Subclasses can override
 +    # this method to instantiate a subclass of _SandboxBatch.
 +    #
 +    # Args:
 +    #    main_group (:class:`_SandboxBatchGroup`): The top level batch group.
 +    #    flags (:class:`.SandboxFlags`): The flags for commands in this batch.
 +    #    collect (str): An optional directory containing partial install contents
 +    #                   on command failure.
 +    #
 +    def _create_batch(self, main_group, flags, *, collect=None):
 +        return _SandboxBatch(self, main_group, flags, collect=collect)
++
      ################################################
      #               Private methods                #
      ################################################
@@ -385,3 +532,138 @@ class Sandbox():
                  return True
          return False
++
 +    # _get_plugin_id()
 +    #
 +    # Get the plugin's unique identifier
 +    #
 +    def _get_plugin_id(self):
 +        return self.__plugin_id
++
 +    # _callback()
 +    #
 +    # If this is called outside a batch context, the specified function is
 +    # invoked immediately.
 +    #
 +    # If this is called in a batch context, the function is added to the batch
 +    # for later invocation.
 +    #
 +    # Args:
 +    #    callback (callable): The function to invoke
 +    #
 +    def _callback(self, callback):
 +        if self.__batch:
 +            batch_call = _SandboxBatchCall(callback)
++
 +            current_group = self.__batch.current_group
 +            current_group.append(batch_call)
 +        else:
 +            callback()
++
++
 +# _SandboxBatch()
 +#
 +# A batch of sandbox commands.
 +#
 +class _SandboxBatch():
++
 +    def __init__(self, sandbox, main_group, flags, *, collect=None):
 +        self.sandbox = sandbox
 +        self.main_group = main_group
 +        self.current_group = main_group
 +        self.flags = flags
 +        self.collect = collect
++
 +    def execute(self):
 +        self.main_group.execute(self)
++
 +    def execute_group(self, group):
 +        if group.label:
 +            context = self.sandbox._get_context()
 +            cm = context.timed_activity(group.label, unique_id=self.sandbox._get_plugin_id())
 +        else:
 +            cm = contextlib.suppress()
++
 +        with cm:
 +            group.execute_children(self)
++
 +    def execute_command(self, command):
 +        if command.label:
 +            context = self.sandbox._get_context()
 +            message = Message(self.sandbox._get_plugin_id(), MessageType.STATUS,
 +                              'Running {}'.format(command.label))
 +            context.message(message)
++
 +        exitcode = self.sandbox._run(command.command, self.flags, cwd=command.cwd, env=command.env)
 +        if exitcode != 0:
 +            cmdline = ' '.join(shlex.quote(cmd) for cmd in command.command)
 +            label = command.label or cmdline
 +            raise SandboxCommandError("Command '{}' failed with exitcode {}".format(label, exitcode),
 +                                      collect=self.collect)
++
 +    def execute_call(self, call):
 +        call.callback()
++
++
 +# _SandboxBatchItem()
 +#
 +# An item in a command batch.
 +#
 +class _SandboxBatchItem():
++
 +    def __init__(self, *, label=None):
 +        self.label = label
++
++
 +# _SandboxBatchCommand()
 +#
 +# A command item in a command batch.
 +#
 +class _SandboxBatchCommand(_SandboxBatchItem):
++
 +    def __init__(self, command, *, cwd, env, label=None):
 +        super().__init__(label=label)
++
 +        self.command = command
 +        self.cwd = cwd
 +        self.env = env
++
 +    def execute(self, batch):
 +        batch.execute_command(self)
++
++
 +# _SandboxBatchGroup()
 +#
 +# A group in a command batch.
 +#
 +class _SandboxBatchGroup(_SandboxBatchItem):
++
 +    def __init__(self, *, label=None):
 +        super().__init__(label=label)
++
 +        self.children = []
++
 +    def append(self, item):
 +        self.children.append(item)
++
 +    def execute(self, batch):
 +        batch.execute_group(self)
++
 +    def execute_children(self, batch):
 +        for item in self.children:
 +            item.execute(batch)
++
++
 +# _SandboxBatchCall()
 +#
 +# A call item in a command batch.
 +#
 +class _SandboxBatchCall(_SandboxBatchItem):
++
 +    def __init__(self, callback):
 +        super().__init__()
++
 +        self.callback = callback
++
 +    def execute(self, batch):
 +        batch.execute_call(self)

buildstream/scriptelement.py

@@ -226,10 +226,11 @@ class ScriptElement(Element):
                                           .format(build_dep.name), silent_nested=True):
                      build_dep.stage_dependency_artifacts(sandbox, Scope.RUN, path="/")
 -            for build_dep in self.dependencies(Scope.BUILD, recurse=False):
 -                with self.timed_activity("Integrating {}".format(build_dep.name), silent_nested=True):
 -                    for dep in build_dep.dependencies(Scope.RUN):
 -                        dep.integrate(sandbox)
 +            with sandbox.batch(SandboxFlags.NONE):
 +                for build_dep in self.dependencies(Scope.BUILD, recurse=False):
 +                    with self.timed_activity("Integrating {}".format(build_dep.name), silent_nested=True):
 +                        for dep in build_dep.dependencies(Scope.RUN):
 +                            dep.integrate(sandbox)
          else:
              # If layout, follow its rules.
              for item in self.__layout:
@@ -251,37 +252,40 @@ class ScriptElement(Element):
                          virtual_dstdir.descend(item['destination'].lstrip(os.sep).split(os.sep), create=True)
                          element.stage_dependency_artifacts(sandbox, Scope.RUN, path=item['destination'])
 -            for item in self.__layout:
 +            with sandbox.batch(SandboxFlags.NONE):
 +                for item in self.__layout:
 -                # Skip layout members which dont stage an element
 -                if not item['element']:
 -                    continue
 +                    # Skip layout members which dont stage an element
 +                    if not item['element']:
 +                        continue
 -                element = self.search(Scope.BUILD, item['element'])
 +                    element = self.search(Scope.BUILD, item['element'])
 -                # Integration commands can only be run for elements staged to /
 -                if item['destination'] == '/':
 -                    with self.timed_activity("Integrating {}".format(element.name),
 -                                             silent_nested=True):
 -                        for dep in element.dependencies(Scope.RUN):
 -                            dep.integrate(sandbox)
 +                    # Integration commands can only be run for elements staged to /
 +                    if item['destination'] == '/':
 +                        with self.timed_activity("Integrating {}".format(element.name),
 +                                                 silent_nested=True):
 +                            for dep in element.dependencies(Scope.RUN):
 +                                dep.integrate(sandbox)
          install_root_path_components = self.__install_root.lstrip(os.sep).split(os.sep)
          sandbox.get_virtual_directory().descend(install_root_path_components, create=True)
      def assemble(self, sandbox):
 -        for groupname, commands in self.__commands.items():
 -            with self.timed_activity("Running '{}'".format(groupname)):
 -                for cmd in commands:
 -                    self.status("Running command", detail=cmd)
 -                    # Note the -e switch to 'sh' means to exit with an error
 -                    # if any untested command fails.
 -                    exitcode = sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 -                                           SandboxFlags.ROOT_READ_ONLY if self.__root_read_only else 0)
 -                    if exitcode != 0:
 -                        raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode),
 -                                           collect=self.__install_root)
 +        flags = SandboxFlags.NONE
 +        if self.__root_read_only:
 +            flags |= SandboxFlags.ROOT_READ_ONLY
++
 +        with sandbox.batch(flags, collect=self.__install_root):
 +            for groupname, commands in self.__commands.items():
 +                with sandbox.batch(flags, label="Running '{}'".format(groupname)):
 +                    for cmd in commands:
 +                        # Note the -e switch to 'sh' means to exit with an error
 +                        # if any untested command fails.
 +                        sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 +                                    flags,
 +                                    label=cmd)
          # Return where the result can be collected from
          return self.__install_root

buildstream/source.py

@@ -615,6 +615,23 @@ class Source(Plugin):
          with utils._tempdir(dir=mirrordir) as tempdir:
              yield tempdir
 +    #############################################################
 +    #       Private Abstract Methods used in BuildStream        #
 +    #############################################################
++
 +    # Returns the local path to the source
 +    #
 +    # If the source is locally available, this method returns the absolute
 +    # path. Otherwise, the return value is None.
 +    #
 +    # This is an optimization for local sources and optional to implement.
 +    #
 +    # Returns:
 +    #    (str): The local absolute path, or None
 +    #
 +    def _get_local_path(self):
 +        return None
++
      #############################################################
      #            Private Methods used in BuildStream            #
      #############################################################

doc/source/format_project.rst

@@ -201,10 +201,10 @@ with an artifact share.
+   #
    artifacts:
      # A remote cache from which to download prebuilt artifacts
 -    - url: https://foo.com/artifacts:11001
 +    - url: https://foo.com:11001
        server.cert: server.crt
      # A remote cache from which to upload/download built/prebuilt artifacts
 -    - url: https://foo.com/artifacts:11002
 +    - url: https://foo.com:11002
        server-cert: server.crt
        client-cert: client.crt
        client-key: client.key
@@ -231,10 +231,24 @@ using the `remote-execution` option:
    remote-execution:
      # A url defining a remote execution server
 -    url: http://buildserver.example.com:50051
 +    execution-service:
 +      url: http://buildserver.example.com:50051
 +    storage-service:
 +    - url: https://foo.com:11002/
 +      server-cert: server.crt
 +      client-cert: client.crt
 +      client-key: client.key
++
 +The execution-service part of remote execution does not support encrypted
 +connections yet, so the protocol must always be http.
++
 +storage-service specifies a remote CAS store and the parameters are the
 +same as those used to specify an :ref:`artifact server <artifacts>`.
 -The url should contain a hostname and port separated by ':'. Only plain HTTP is
 -currently suported (no HTTPS).
 +The storage service may be the same endpoint used for artifact
 +caching. Remote execution cannot work without push access to the
 +storage endpoint, so you must specify a client certificate and key,
 +and a server certificate.
  The Remote Execution API can be found via https://github.com/bazelbuild/remote-apis.

tests/integration/manual.py

@@ -128,3 +128,28 @@ def test_manual_element_noparallel(cli, tmpdir, datafiles):
      assert text == """-j1 -Wall
  """
++
++
 +@pytest.mark.datafiles(DATA_DIR)
 +@pytest.mark.skipif(IS_LINUX and not HAVE_BWRAP, reason='Only available with bubblewrap on Linux')
 +def test_manual_element_logging(cli, tmpdir, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
 +    checkout = os.path.join(cli.directory, 'checkout')
 +    element_path = os.path.join(project, 'elements')
 +    element_name = 'import/import.bst'
++
 +    create_manual_element(element_name, element_path, {
 +        'configure-commands': ["echo configure"],
 +        'build-commands': ["echo build"],
 +        'install-commands': ["echo install"],
 +        'strip-commands': ["echo strip"]
 +    }, {}, {})
++
 +    res = cli.run(project=project, args=['build', element_name])
 +    assert res.exit_code == 0
++
 +    # Verify that individual commands are logged
 +    assert "echo configure" in res.stderr
 +    assert "echo build" in res.stderr
 +    assert "echo install" in res.stderr
 +    assert "echo strip" in res.stderr

tests/integration/sandbox-bwrap.py

@@ -58,5 +58,5 @@ def test_sandbox_bwrap_return_subprocess(cli, tmpdir, datafiles):
      })
      result = cli.run(project=project, args=['build', element_name])
 -    result.assert_task_error(error_domain=ErrorDomain.ELEMENT, error_reason=None)
 +    result.assert_task_error(error_domain=ErrorDomain.SANDBOX, error_reason="command-failed")
      assert "sandbox-bwrap/command-exit-42.bst|Command 'exit 42' failed with exitcode 42" in result.stderr

tests/sandboxes/remote-exec-config.py

 +import pytest
++
 +import itertools
 +import os
++
 +from buildstream import _yaml
 +from buildstream._exceptions import ErrorDomain, LoadErrorReason
++
 +from tests.testutils.runcli import cli
++
 +DATA_DIR = os.path.join(
 +    os.path.dirname(os.path.realpath(__file__)),
 +    "remote-exec-config"
 +)
++
 +# Tests that we get a useful error message when supplying invalid
 +# remote execution configurations.
++
++
 +# Assert that if both 'url' (the old style) and 'execution-service' (the new style)
 +# are used at once, a LoadError results.
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_old_and_new_configs(cli, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename, 'missing-certs')
++
 +    project_conf = {
 +        'name': 'test',
++
 +        'remote-execution': {
 +            'url': 'https://cache.example.com:12345',
 +            'execution-service': {
 +                'url': 'http://localhost:8088'
 +            },
 +            'storage-service': {
 +                'url': 'http://charactron:11001',
 +            }
 +        }
 +    }
 +    project_conf_file = os.path.join(project, 'project.conf')
 +    _yaml.dump(project_conf, project_conf_file)
++
 +    # Use `pull` here to ensure we try to initialize the remotes, triggering the error
 +    #
 +    # This does not happen for a simple `bst show`.
 +    result = cli.run(project=project, args=['pull', 'element.bst'])
 +    result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.INVALID_DATA, "specify one")
++
++
 +# Assert that if either the client key or client cert is specified
 +# without specifying its counterpart, we get a comprehensive LoadError
 +# instead of an unhandled exception.
 +@pytest.mark.datafiles(DATA_DIR)
 +@pytest.mark.parametrize('config_key, config_value', [
 +    ('client-cert', 'client.crt'),
 +    ('client-key', 'client.key')
 +])
 +def test_missing_certs(cli, datafiles, config_key, config_value):
 +    project = os.path.join(datafiles.dirname, datafiles.basename, 'missing-certs')
++
 +    project_conf = {
 +        'name': 'test',
++
 +        'remote-execution': {
 +            'execution-service': {
 +                'url': 'http://localhost:8088'
 +            },
 +            'storage-service': {
 +                'url': 'http://charactron:11001',
 +                config_key: config_value,
 +            }
 +        }
 +    }
 +    project_conf_file = os.path.join(project, 'project.conf')
 +    _yaml.dump(project_conf, project_conf_file)
++
 +    # Use `pull` here to ensure we try to initialize the remotes, triggering the error
 +    #
 +    # This does not happen for a simple `bst show`.
 +    result = cli.run(project=project, args=['show', 'element.bst'])
 +    result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.INVALID_DATA, "Your config is missing")
++
++
 +# Assert that if incomplete information is supplied we get a sensible error message.
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_empty_config(cli, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename, 'missing-certs')
++
 +    project_conf = {
 +        'name': 'test',
++
 +        'remote-execution': {
 +        }
 +    }
 +    project_conf_file = os.path.join(project, 'project.conf')
 +    _yaml.dump(project_conf, project_conf_file)
++
 +    # Use `pull` here to ensure we try to initialize the remotes, triggering the error
 +    #
 +    # This does not happen for a simple `bst show`.
 +    result = cli.run(project=project, args=['pull', 'element.bst'])
 +    result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.INVALID_DATA, "specify one")

tests/sandboxes/remote-exec-config/missing-certs/certificates/client.crt

tests/sandboxes/remote-exec-config/missing-certs/certificates/client.key

tests/sandboxes/remote-exec-config/missing-certs/element.bst

+kind: autotools

[Notes] [Git][BuildStream/buildstream][jmac/remote_execution_split] 28 commits: source.py: Add optional _get_local_path() method

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

Commits:

27 changed files:

Changes: