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

Command batching

Closes #675

See merge request BuildStream/buildstream!915

14 files changed, 498 insertions, 108 deletions

diff --git a/NEWS b/NEWS
index f52ccb4b4..16759e1d1 100644
--- a/NEWS
+++ b/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
diff --git a/buildstream/__init__.py b/buildstream/__init__.py
index af2122ef7..fc09f2812 100644
--- a/buildstream/__init__.py
+++ b/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
diff --git a/buildstream/buildelement.py b/buildstream/buildelement.py
index 48b7f2423..f1b13e796 100644
--- a/buildstream/buildelement.py
+++ b/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)
diff --git a/buildstream/element.py b/buildstream/element.py
index e0d8ce77d..a1a084ecf 100644
--- a/buildstream/element.py
+++ b/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
@@ -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:
@@ -2157,6 +2203,7 @@ class Element(Plugin):
 
             sandbox = SandboxRemote(context, project,
                                     directory,
+                                    plugin=self,
                                     stdout=stdout,
                                     stderr=stderr,
                                     config=config,
@@ -2175,6 +2222,7 @@ class Element(Plugin):
 
             sandbox = platform.create_sandbox(context, project,
                                               directory,
+                                              plugin=self,
                                               stdout=stdout,
                                               stderr=stderr,
                                               config=config,
diff --git a/buildstream/plugins/elements/compose.py b/buildstream/plugins/elements/compose.py
index 6b99947d5..d61a324cc 100644
--- a/buildstream/plugins/elements/compose.py
+++ b/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
diff --git a/buildstream/sandbox/__init__.py b/buildstream/sandbox/__init__.py
index 5999aba7a..5966d194f 100644
--- a/buildstream/sandbox/__init__.py
+++ b/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
diff --git a/buildstream/sandbox/_sandboxbwrap.py b/buildstream/sandbox/_sandboxbwrap.py
index 839780f95..f7b11326a 100644
--- a/buildstream/sandbox/_sandboxbwrap.py
+++ b/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]),
diff --git a/buildstream/sandbox/_sandboxchroot.py b/buildstream/sandbox/_sandboxchroot.py
index e63a4f237..71422476a 100644
--- a/buildstream/sandbox/_sandboxchroot.py
+++ b/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 "
diff --git a/buildstream/sandbox/_sandboxdummy.py b/buildstream/sandbox/_sandboxdummy.py
index 0e3754c1b..4cc3aae9c 100644
--- a/buildstream/sandbox/_sandboxdummy.py
+++ b/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 "
diff --git a/buildstream/sandbox/_sandboxremote.py b/buildstream/sandbox/_sandboxremote.py
index 60a053e0c..758160219 100644
--- a/buildstream/sandbox/_sandboxremote.py
+++ b/buildstream/sandbox/_sandboxremote.py
@@ -19,12 +19,14 @@
 #        Jim MacArthur <jim.macarthur@codethink.co.uk>
 
 import os
+import shlex
 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
@@ -212,7 +214,7 @@ 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()
 
@@ -230,16 +232,6 @@ class SandboxRemote(Sandbox):
         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)
-
-        # We want command args as a list of strings
-        if isinstance(command, str):
-            command = [command]
-
         # Now transmit the command to execute
         operation = self.run_remote_command(command, upload_vdir.ref, cwd, env)
 
@@ -275,3 +267,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")
diff --git a/buildstream/sandbox/sandbox.py b/buildstream/sandbox/sandbox.py
index a76b2d9d2..b21b350ff 100644
--- a/buildstream/sandbox/sandbox.py
+++ b/buildstream/sandbox/sandbox.py
@@ -1,5 +1,6 @@
 #
 #  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)
diff --git a/buildstream/scriptelement.py b/buildstream/scriptelement.py
index d2165ce32..697cd2822 100644
--- a/buildstream/scriptelement.py
+++ b/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
diff --git a/tests/integration/manual.py b/tests/integration/manual.py
index 4789d552b..c6a905ddc 100644
--- a/tests/integration/manual.py
+++ b/tests/integration/manual.py
@@ -128,3 +128,28 @@ def test_manual_element_noparallel(cli, tmpdir, datafiles):
     assert text == """-j1 -Wall
 2
 """
+
+
+@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
diff --git a/tests/integration/sandbox-bwrap.py b/tests/integration/sandbox-bwrap.py
index d2484bc17..b77709c35 100644
--- a/tests/integration/sandbox-bwrap.py
+++ b/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