.gitlab-ci.yml

@@ -172,11 +172,13 @@ docs:
   - make BST_FORCE_SESSION_REBUILD=1 -C doc
   - cd ../..
   - mv dist/buildstream/doc/build/html public
+  - mv dist/buildstream/doc/source/sessions session-output
   except:
   - schedules
   artifacts:
     paths:
     - public/
+    - session-output
 
 .overnight-tests: &overnight-tests-template
   stage: test

CONTRIBUTING.rst

@@ -1247,12 +1247,21 @@ To build the documentation, just run the following::
 This will give you a ``doc/build/html`` directory with the html docs which
 you can view in your browser locally to test.
 
-
 Regenerating session html
 '''''''''''''''''''''''''
-The documentation build will build the session files if they are missing,
-or if explicitly asked to rebuild. We revision the generated session html files
-in order to reduce the burden on documentation contributors.
+The documentation build will build the session files if the html is out of date
+or missing, or if explicitly asked to rebuild.
+
+You can skip running the session files and have dummy examples generated to allow
+changes to the documentation to be render if a user is unable to regenerate the
+session files them selves.
+
+  make BST_DUMMY_REBUILD=1 -C doc
+
+If a user would like to create the docs with the correct session output but is
+unable to run the session files then they may down load the latest session files
+from the gitlab-ci doc element of the master ci PIPELINE and copy them in to the
+doc/source/sessions folder. This process is not currently automated.
 
 To explicitly rebuild the session snapshot html files, it is recommended that you
 first set the ``BST_SOURCE_CACHE`` environment variable to your source cache, this

MANIFEST.in

@@ -19,6 +19,7 @@ recursive-include doc/source *.html
 recursive-include doc/source *.odg
 recursive-include doc/source *.svg
 recursive-include doc/examples *
+recursive-include doc/sessions *.run
 
 # Tests
 recursive-include tests *

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' ]
-
+  command: [ 'sh', '-i' ]
\ No newline at end of file
-remote-execution:
-  url: ""
\ 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")