Repository: allenai/tango (language: python)

Issue:
Run in beaker
Goal: Run either some steps or maybe all steps in Beaker. Steps need a way to specify requirements for the number of CPUs, number of GPUs, amount of memory. 

Not a goal: Run multi-node jobs

This is not trivial, not least because we have to find a way to package up a project and stick it into a Docker image to run.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index bf4cbfe10..1f828dc74 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -23,6 +23,7 @@ env:
   BEAKER_WORKSPACE: ai2/tango-testing
   BEAKER_DEFAULT_CLUSTER: ai2/tango-gpu-tests
   BEAKER_IMAGE: petew/tango-testing
+  GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
 jobs:
   checks:
@@ -155,7 +156,17 @@ jobs:
             python: '3.10'
 
     steps:
-      - uses: actions/checkout@v3
+      - name: Checkout
+        if: github.event_name != 'pull_request'
+        uses: actions/checkout@v3
+
+      # For pull requests we need to checkout the HEAD commit instead of the merge
+      # commit since some tests depend on having an existing commit.
+      - name: Checkout (pull request)
+        if: github.event_name == 'pull_request'
+        uses: actions/checkout@v3
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
 
       - name: Setup Python
         uses: actions/setup-python@v4
@@ -320,8 +331,6 @@ jobs:
 
       - name: Publish GitHub release
         uses: softprops/action-gh-release@v1
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
           body_path: ${{ github.workspace }}-RELEASE_NOTES.md
           prerelease: ${{ contains(env.TAG, 'rc') }}
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f9b903a2f..185328477 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,9 +7,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- **Step resources:**
+  - Added a `step_resources` parameter to the `Step` class which should be used to describe the computational resources required to run a step.
+    `Executor` implementations can use this information. For example, if your step needs 2 GPUs, you should set
+    `step_resources=StepResources(gpu_count=2)` (`"step_resources": {"gpu_count": 2}` in the configuration language).
+  - Added a `Step.resources()` property method. By default this returns the value specified by the `step_resources` parameter.
+    If your step implementation always requires the same resources, you can just override this method so you don't have to provide
+    the `step_resources` parameter.
+- **Step execution:**
+  - Added an `executor` field to the `tango.yml` settings. You can use this to define the executor you want to use by default.
+  - Added a Beaker `Executor` to the Beaker integration, registered as an `Executor` with the name "beaker".
+    To use this executor, add these lines to your `tango.yml` file:
+    ```yaml
+    executor:
+      type: beaker
+      beaker_workspace: ai2/my-workspace
+      clusters:
+        - ai2/general-cirrascale
+    ```
+    See the docs for the `BeakerExecutor` for more information on the input parameters.
+
+### Changed
+
+- **CLI:**
+  - The `tango run` command will throw an error if you have uncommitted changes in your repository, unless
+    you use the `--allow-dirty` flag.
+  - The `tango run` command will use the lightweight base executor (single process) by default.
+    To use the multi-process executor, set `-j/--parallelism` to 1 or higher or -1 to use all available CPU cores.
+
 ## [v0.11.0](https://github.com/allenai/tango/releases/tag/v0.11.0) - 2022-08-04
 
 ### Added
+
 - Added a [Flax](https://flax.readthedocs.io/en/latest/) integration along with an example config.
 
 ## [v0.10.1](https://github.com/allenai/tango/releases/tag/v0.10.1) - 2022-07-26
diff --git a/docs/source/api/components/step.rst b/docs/source/api/components/step.rst
index cd830c428..e63f9e649 100644
--- a/docs/source/api/components/step.rst
+++ b/docs/source/api/components/step.rst
@@ -12,6 +12,9 @@ Base class
 .. autoclass:: tango.step.WithUnresolvedSteps
    :members:
 
+.. autoclass:: tango.step.StepResources
+   :members:
+
 Implementations
 ---------------
 
diff --git a/docs/source/api/integrations/beaker.rst b/docs/source/api/integrations/beaker.rst
index 0f5f915ac..094b36221 100644
--- a/docs/source/api/integrations/beaker.rst
+++ b/docs/source/api/integrations/beaker.rst
@@ -9,3 +9,6 @@ Reference
 .. autoclass:: tango.integrations.beaker.BeakerWorkspace
 
 .. autoclass:: tango.integrations.beaker.BeakerStepCache
+
+.. autoclass:: tango.integrations.beaker.BeakerExecutor
+   :members: DEFAULT_BEAKER_IMAGE
diff --git a/docs/source/api/integrations/fairscale.rst b/docs/source/api/integrations/fairscale.rst
index 2d98cb91f..d7890f909 100644
--- a/docs/source/api/integrations/fairscale.rst
+++ b/docs/source/api/integrations/fairscale.rst
@@ -1,5 +1,5 @@
-FairScale
-=========
+🔥 FairScale
+============
 
 .. automodule:: tango.integrations.fairscale
 
diff --git a/docs/source/api/integrations/torch.rst b/docs/source/api/integrations/torch.rst
index b54ea62ff..5996bc793 100644
--- a/docs/source/api/integrations/torch.rst
+++ b/docs/source/api/integrations/torch.rst
@@ -1,5 +1,5 @@
-PyTorch
-=======
+🔥 PyTorch
+==========
 
 .. automodule:: tango.integrations.torch
 
diff --git a/docs/source/api/integrations/wandb.rst b/docs/source/api/integrations/wandb.rst
index c894e2bb6..4f123bd3c 100644
--- a/docs/source/api/integrations/wandb.rst
+++ b/docs/source/api/integrations/wandb.rst
@@ -1,6 +1,6 @@
-Weights & Biases
-================
-
+⚖️ Weights & Biases
+===================
+ 
 .. automodule:: tango.integrations.wandb
 
 Reference
diff --git a/docs/source/api/settings.rst b/docs/source/api/settings.rst
index 6ab0e6178..25e21a70e 100644
--- a/docs/source/api/settings.rst
+++ b/docs/source/api/settings.rst
@@ -4,8 +4,9 @@ Global settings
 Some command-line options can set globally in a ``tango.yml`` or ``tango.yaml`` settings file.
 Tango will check the current directory and ``~/.config/``, in that order.
 
-The full spec of this config is defined by :class:`TangoGlobalSettings`.
+The full spec of this config is defined by the :class:`~tango.settings.TangoGlobalSettings` class.
 
 .. autoclass:: tango.settings.TangoGlobalSettings
    :members:
    :exclude-members: path,find_or_default
+   :member-order: bysource
diff --git a/requirements.txt b/requirements.txt
index 1059fa27b..606f49260 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,6 +3,7 @@
 ################################
 cached-path>=1.0,<1.2
 jsonnet>=0.10 ; sys.platform != 'win32'
+GitPython>=3.0,<4.0
 PyYAML>=5.4.1,<6.1
 dill
 base58
@@ -17,6 +18,8 @@ sqlitedict
 glob2>=0.7
 petname>=2.6,<3.0
 pytz
+jsonpickle>=2.1.0,<2.2.0
+tblib>=1.7.0,<1.8.0
 
 # Protobuf is a dependency of wandb and tensorboard, but they are missing this pin.
 protobuf<4.22.0
@@ -42,4 +45,4 @@ sacremoses                   # needed by: transformers
 jax>=0.3.13                  # needed by: flax
 flax>=0.5.0                  # needed by: flax
 optax>=0.1.2                 # needed by: flax
-tensorflow-cpu>=2.9.1        # needed by: flax
\ No newline at end of file
+tensorflow-cpu>=2.9.1        # needed by: flax
diff --git a/setup.py b/setup.py
index c6b927846..987ccf502 100644
--- a/setup.py
+++ b/setup.py
@@ -99,7 +99,11 @@ def fix_url_dependencies(req: str) -> str:
             "scripts.*",
         ],
     ),
-    package_data={"tango.server": ["*.svg", "*.css", "*.js", "*.html"], "tango": ["py.typed"]},
+    package_data={
+        "tango.server": ["*.svg", "*.css", "*.js", "*.html"],
+        "tango.integrations.beaker": ["*.sh"],
+        "tango": ["py.typed"],
+    },
     entry_points={"console_scripts": ["tango=tango.__main__:main"]},
     install_requires=install_requirements,
     extras_require=extras,
diff --git a/tango/__init__.py b/tango/__init__.py
index b52a61843..93bc32986 100644
--- a/tango/__init__.py
+++ b/tango/__init__.py
@@ -11,6 +11,7 @@
     "JsonFormatIterator",
     "SqliteDictFormat",
     "Step",
+    "StepResources",
     "StepCache",
     "Workspace",
 ]
@@ -24,7 +25,7 @@
     JsonFormatIterator,
     SqliteDictFormat,
 )
-from tango.step import Step
+from tango.step import Step, StepResources
 from tango.step_cache import StepCache
 from tango.step_graph import StepGraph
 from tango.workspace import Workspace
diff --git a/tango/__main__.py b/tango/__main__.py
index 948e700bd..8eaf9f85e 100644
--- a/tango/__main__.py
+++ b/tango/__main__.py
@@ -267,12 +267,10 @@ def cleanup(*args, **kwargs):
     "-j",
     "--parallelism",
     type=int,
-    help="""The maximum number of steps to run in parallel (if possible).
-    By default there is no parallelism, but each step is ran in its own subprocess (same as setting '-j=1').
-    A value of 0 or less means each step is ran in the main process using the default executor.
-    A value greater than 1 means at most 'j' steps will be ran in parallel,
-    each within its own subprocess, using the multicore executor.
-    This option is ignored if you've configured your executor in a 'tango.yml' settings file.""",
+    help="""The maximum number of steps to run in parallel (for executors that support this).
+    The exact behavior depends on the executor. If you're using the default executors,
+    a value of 0 (or left unspecified) means each step is run in the main process using the default executor,
+    otherwise the multicore executor is used.""",
 )
 @click.option(
     "-s",
@@ -287,6 +285,11 @@ def cleanup(*args, **kwargs):
     type=str,
     help="""Specify the name for this run.""",
 )
+@click.option(
+    "--allow-dirty",
+    is_flag=True,
+    help="""Allow running the experiment with a dirty working directory (uncommitted code changes).""",
+)
 @click.pass_obj
 def run(
     settings: TangoGlobalSettings,
@@ -299,6 +302,7 @@ def run(
     parallelism: Optional[int] = None,
     step_name: Optional[str] = None,
     name: Optional[str] = None,
+    allow_dirty: bool = False,
 ):
     """
     Run a tango experiment.
@@ -331,9 +335,64 @@ def run(
         step_name=step_name,
         name=name,
         called_by_executor=_CALLED_BY_EXECUTOR,
+        allow_dirty=allow_dirty,
     )
 
 
+@main.command(hidden=True)
+@click.argument(
+    "experiment",
+    type=click.Path(exists=True, dir_okay=False, resolve_path=True),
+)
+@click.argument(
+    "step_name",
+    type=str,
+)
+@click.argument(
+    "workspace_url",
+    type=str,
+)
+@click.option(
+    "-i",
+    "--include-package",
+    type=str,
+    help="Python packages or modules to import for tango components.",
+    multiple=True,
+)
+def beaker_executor_run(
+    experiment: str,
+    step_name: str,
+    workspace_url: str,
+    include_package: Optional[Sequence[str]] = None,
+):
+    """
+    This command is only used internally by the BeakerExecutor.
+    """
+    from tango.common.logging import do_json_logging
+    from tango.executor import Executor
+
+    if include_package:
+        for package_name in include_package:
+            import_extra_module(package_name)
+
+    # Load step graph and step.
+    step_graph = StepGraph.from_file(experiment)
+    step = step_graph[step_name]
+
+    # Initialize workspace and executor.
+    # NOTE: We use the default executor here because we're just running the step
+    # locally in the main process.
+    workspace = Workspace.from_url(workspace_url)
+    executor = Executor(workspace=workspace, include_package=include_package)
+
+    # Initialize logging.
+    initialize_logging(log_level="debug", enable_cli_logs=True)
+    do_json_logging(f"step {step.name}")
+
+    # Run step.
+    executor.execute_step(step)
+
+
 @main.command(**_CLICK_COMMAND_DEFAULTS)
 @click.option(
     "-w",
@@ -650,12 +709,24 @@ def _run(
     multicore: Optional[bool] = None,
     name: Optional[str] = None,
     called_by_executor: bool = False,
+    allow_dirty: bool = False,
 ) -> str:
+    from git import InvalidGitRepositoryError, Repo
+
     from tango.executor import Executor
     from tango.executors import MulticoreExecutor
     from tango.server.workspace_server import WorkspaceServer
     from tango.workspaces import MemoryWorkspace, default_workspace
 
+    if not called_by_executor and not allow_dirty:
+        # Make sure repository is clean, if we're in one.
+        try:
+            repo = Repo(".")
+            if repo.is_dirty():
+                raise CliRunError("You have uncommitted changes! Use --allow-dirty to force.")
+        except InvalidGitRepositoryError:
+            pass
+
     # Read params.
     params = Params.from_file(experiment, params_overrides=overrides or "")
 
@@ -685,13 +756,11 @@ def _run(
                 "Ignoring argument 'multicore' since executor is defined in %s",
                 settings.path or "setting",
             )
-        if parallelism is not None:
-            logger.warning(
-                "Ignoring parallelism ('-j/--parallelism') since executor is defined in %s",
-                settings.path or "setting",
-            )
         executor = Executor.from_params(
-            settings.executor, workspace=workspace, include_package=include_package
+            settings.executor,
+            workspace=workspace,
+            include_package=include_package,
+            parallelism=parallelism,
         )
     else:
         # Determine if we can use the multicore executor.
@@ -703,28 +772,31 @@ def _run(
                 # Pydevd doesn't reliably follow child processes, so we disable multicore under the debugger.
                 logger.warning("Debugger detected, disabling multicore.")
                 multicore = False
-            elif parallelism is not None and parallelism <= 0:
+            elif parallelism is None or parallelism == 0:
                 multicore = False
             else:
                 multicore = True
 
         if multicore:
             executor = MulticoreExecutor(
-                workspace=workspace, include_package=include_package, parallelism=parallelism or 1
+                workspace=workspace, include_package=include_package, parallelism=parallelism
             )
         else:
             executor = Executor(workspace=workspace, include_package=include_package)
 
-    # Initialize step graph and register run.
+    # Initialize step graph.
     step_graph = StepGraph.from_params(params.pop("steps", keep_as_dict=True))
+    sub_graph: Optional[StepGraph] = None
     params.assert_empty("'tango run'")
 
+    # Register run.
+    run: "Run"
     if step_name is not None:
         assert step_name in step_graph, (
             f"You want to run a step called '{step_name}', but it cannot be found in the experiment config. "
             f"The config contains: {list(step_graph.keys())}."
         )
-
+        sub_graph = step_graph.sub_graph(step_name)
         if called_by_executor and name is not None:
             try:
                 run = workspace.registered_run(name)
@@ -734,7 +806,6 @@ def _run(
                     f"'{name}' is not already registered as a run. This should never happen!"
                 )
         else:
-            sub_graph = step_graph.sub_graph(step_name)
             run = workspace.register_run((step for step in sub_graph.values()), name)
     else:
         run = workspace.register_run((step for step in step_graph.values()), name)
@@ -752,14 +823,14 @@ def log_and_execute_run():
             )
 
         if step_name is not None:
-            step = step_graph[step_name]
-            executor.execute_step(step)
+            assert sub_graph is not None
+            step = sub_graph[step_name]
+            if step.cache_results and step in workspace.step_cache:
+                step.log_cache_hit()
+            else:
+                executor.execute_sub_graph_for_step(sub_graph, step_name, run_name=run.name)
             if not called_by_executor:
-                cli_logger.info(
-                    "[green]\N{check mark} Finished run for step [bold]%s[/] (%s)[/]",
-                    step_name,
-                    run.name,
-                )
+                step.log_finished(run.name)
         else:
             executor_output = executor.execute_step_graph(step_graph, run_name=run.name)
             if executor_output.failed:
@@ -779,7 +850,7 @@ def log_and_execute_run():
         # We set this environment variable so that any steps that contain multiprocessing
         # and call `initialize_worker_logging` also log the messages with the `step_name` prefix.
         os.environ[EnvVarNames.LOGGING_PREFIX.value] = f"step {step_name}"
-        initialize_prefix_logging(f"step {step_name}", main_process=False)
+        initialize_prefix_logging(prefix=f"step {step_name}", main_process=False)
         log_and_execute_run()
     else:
         # Capture logs to file.
diff --git a/tango/common/exceptions.py b/tango/common/exceptions.py
index a433a1183..9da365bf4 100644
--- a/tango/common/exceptions.py
+++ b/tango/common/exceptions.py
@@ -35,12 +35,26 @@ class RegistryKeyError(ConfigurationError):
     """
 
 
-class SigTermReceived(TangoError):
+class CancellationError(TangoError):
+    """
+    Base class for errors raised due to manual cancellation of a run or step.
+    """
+
+
+class SigTermReceived(CancellationError):
     """
     Raised when a SIGTERM is caught.
     """
 
 
+class StepCancelled(CancellationError):
+    pass
+
+
+class RunCancelled(CancellationError):
+    pass
+
+
 class CliRunError(TangoError):
     """
     Raised when `tango run` command fails.
@@ -79,3 +93,15 @@ def __init__(
         if context is not None:
             msg = msg + " " + context
         super().__init__(msg)
+
+
+class DirtyRepoError(TangoError):
+    """
+    Raised when a repository is in a dirty state.
+    """
+
+
+class ExecutorError(TangoError):
+    """
+    A base class for executor-specific errors.
+    """
diff --git a/tango/common/from_params.py b/tango/common/from_params.py
index 4304a56f1..0e5f6a4c8 100644
--- a/tango/common/from_params.py
+++ b/tango/common/from_params.py
@@ -130,7 +130,13 @@ def infer_method_params(
 
     has_kwargs = False
     var_positional_key = None
-    for param_name in parameters.keys():
+    for param_name in list(parameters.keys()):
+        # Ignore special private parameters.
+        # This is necessary to make `FromParams` work with Pydantic, for example.
+        if param_name.startswith("__"):
+            del parameters[param_name]
+            continue
+
         param = parameters[param_name]
         if param.kind == param.VAR_KEYWORD:
             has_kwargs = True
diff --git a/tango/common/logging.py b/tango/common/logging.py
index 401d12760..22c833254 100644
--- a/tango/common/logging.py
+++ b/tango/common/logging.py
@@ -89,7 +89,9 @@ def _worker_function(worker_id: int):
 from contextlib import contextmanager
 from typing import ContextManager, Generator, List, Optional
 
+import jsonpickle
 import rich
+import tblib
 from rich.console import Console
 from rich.logging import RichHandler as _RichHandler
 from rich.padding import Padding
@@ -97,7 +99,7 @@ def _worker_function(worker_id: int):
 from rich.table import Table
 
 from .aliases import EnvVarNames, PathOrStr
-from .exceptions import CliRunError, SigTermReceived
+from .exceptions import CancellationError, CliRunError, SigTermReceived
 from .util import _parse_bool, _parse_optional_int
 
 FILE_FRIENDLY_LOGGING: bool = _parse_bool(
@@ -353,28 +355,34 @@ def excepthook(exctype, value, traceback):
     """
     Used to patch `sys.excepthook` in order to log exceptions.
     """
+    log_exc_info(exctype, value, traceback)
+
+
+def log_exception(exc: Optional[BaseException] = None, logger: Optional[logging.Logger] = None):
+    if exc is None:
+        et, ev, tb = sys.exc_info()
+        log_exc_info(et, ev, tb, logger=logger)
+    else:
+        log_exc_info(exc.__class__, exc, exc.__traceback__, logger=logger)
+
+
+def log_exc_info(exctype, value, traceback, logger: Optional[logging.Logger] = None):
     global _EXCEPTIONS_LOGGED
-    # Ignore `CliRunError` because we don't need a traceback for those.
-    if issubclass(exctype, (CliRunError,)):
-        return
-    # For interruptions, call the original exception handler.
-    if issubclass(
-        exctype,
-        (
-            KeyboardInterrupt,
-            SigTermReceived,
-        ),
-    ):
-        sys.__excepthook__(exctype, value, traceback)
-        return
     if value not in _EXCEPTIONS_LOGGED:
         _EXCEPTIONS_LOGGED.append(value)
-        root_logger = logging.getLogger()
-        root_logger.error(
-            "Uncaught exception",
-            exc_info=(exctype, value, traceback),
-            extra={"highlighter": rich.highlighter.ReprHighlighter()},
-        )
+        logger = logger or logging.getLogger()
+        if isinstance(value, CliRunError):
+            msg = str(value)
+            if msg:
+                cli_logger.error(msg)
+        elif isinstance(value, (KeyboardInterrupt, CancellationError)):
+            logger.error("%s: %s", exctype.__name__, value)
+        else:
+            logger.error(
+                "Uncaught exception",
+                exc_info=(exctype, value, traceback),
+                extra={"highlighter": rich.highlighter.ReprHighlighter()},
+            )
 
 
 def initialize_logging(
@@ -439,16 +447,21 @@ def initialize_worker_logging(worker_rank: Optional[int] = None):
     return initialize_prefix_logging(prefix=prefix, main_process=False)
 
 
-def initialize_prefix_logging(prefix: Optional[str] = None, main_process: bool = False):
+def initialize_prefix_logging(
+    *, log_level: Optional[str] = None, prefix: Optional[str] = None, main_process: bool = False
+):
     """
     Initialize logging with a prefix.
 
+    :param log_level:
+        Can be one of "debug", "info", "warning", "error". Defaults to the value
+        of :data:`TANGO_LOG_LEVEL`, if set, or "error".
     :param prefix:
         The string prefix to add to the log message.
     :param main_process:
         Whether it is for the main/worker process.
     """
-    return _initialize_logging(prefix=prefix, main_process=main_process)
+    return _initialize_logging(log_level=log_level, prefix=prefix, main_process=main_process)
 
 
 def _initialize_logging(
@@ -470,6 +483,10 @@ def _initialize_logging(
         file_friendly_logging = FILE_FRIENDLY_LOGGING
     if enable_cli_logs is None:
         enable_cli_logs = TANGO_CLI_LOGGER_ENABLED
+    if prefix:
+        prefix = _LOGGING_PREFIX + " " + prefix if _LOGGING_PREFIX else prefix
+    else:
+        prefix = _LOGGING_PREFIX
 
     level = logging._nameToLevel[log_level.upper()]
 
@@ -528,6 +545,12 @@ def _initialize_logging(
                 cli_handler.addFilter(LevelFilter(handler_level))
                 cli_logger.addHandler(cli_handler)
 
+        # Add prefix.
+        if prefix:
+            for logger in (root_logger, cli_logger, tqdm_logger):
+                for handler in logger.handlers:
+                    handler.addFilter(PrefixLogFilter(prefix))
+
         # Main process: set formatter and handlers, initialize logging socket and server.
         # Set up logging socket to emit log records from worker processes/threads.
         # Inspired by:
@@ -548,10 +571,6 @@ def _initialize_logging(
                 "did you forget to call 'initialize_logging()' from the main process?"
             )
         socket_handler = logging.handlers.SocketHandler(_LOGGING_HOST, _LOGGING_PORT)
-        if prefix:
-            prefix = _LOGGING_PREFIX + " " + prefix if _LOGGING_PREFIX else prefix
-        else:
-            prefix = _LOGGING_PREFIX
         if prefix:
             socket_handler.addFilter(PrefixLogFilter(prefix))
 
@@ -604,9 +623,10 @@ def insert_handlers(*handlers: logging.Handler) -> Generator[None, None, None]:
     try:
         yield None
     except BaseException as e:
-        # We don't log `CliRunError` because we don't need a traceback for those.
-        if not isinstance(e, CliRunError):
-            root_logger.exception(e)
+        if not isinstance(
+            e, (CliRunError, KeyboardInterrupt, SigTermReceived)
+        ):  # don't need tracebacks for these
+            log_exception(e)
             _EXCEPTIONS_LOGGED.append(e)
         raise
     finally:
@@ -658,3 +678,38 @@ def file_handler(filepath: PathOrStr) -> ContextManager[None]:
         handler.addFilter(CliFilter(filter_out=not is_cli_handler))
         handlers.append(handler)
     return insert_handlers(*handlers)
+
+
+def log_record_to_json(record: logging.LogRecord) -> str:
+    attrs = record.__dict__
+    if attrs.get("exc_info") is not None:
+        # Tracebacks cannot be pickled directly, so we need help from tblib.
+        et, ev, tb = attrs["exc_info"]
+        tb_dict = tblib.Traceback(tb).to_dict()
+        attrs["exc_info"] = (et, ev, tb_dict)
+    return jsonpickle.dumps(attrs)
+
+
+def log_record_from_json(json_record: str) -> logging.LogRecord:
+    attrs = jsonpickle.loads(json_record)
+    if attrs.get("exc_info") is not None:
+        et, ev, tb_dict = attrs["exc_info"]
+        tb = tblib.Traceback.from_dict(tb_dict)
+        attrs["exc_info"] = (et, ev, tb)
+    return logging.makeLogRecord(attrs)
+
+
+class JsonHandler(logging.Handler):
+    def emit(self, record: logging.LogRecord):
+        print(log_record_to_json(record))
+
+
+def do_json_logging(prefix: str):
+    from tango.common.tqdm import logger as tqdm_logger
+
+    root_logger = logging.getLogger()
+    for logger in (root_logger, cli_logger, tqdm_logger):
+        logger.handlers.clear()
+        handler = JsonHandler()
+        handler.addFilter(PrefixLogFilter(prefix))
+        logger.addHandler(handler)
diff --git a/tango/executor.py b/tango/executor.py
index 0ba92fc2b..4af7764bf 100644
--- a/tango/executor.py
+++ b/tango/executor.py
@@ -1,8 +1,9 @@
 import logging
-import traceback
+import warnings
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, List, Optional, Sequence, Set, TypeVar
+from typing import TYPE_CHECKING, Optional, Sequence, Set, TypeVar
 
+from .common.logging import log_exception
 from .common.registrable import Registrable
 from .common.util import import_extra_module
 from .step_graph import StepGraph
@@ -38,6 +39,11 @@ class Executor(Registrable):
     An ``Executor`` is a class that is responsible for running steps and caching their results.
 
     This is the base class and default implementation, registered as "default".
+
+    .. note::
+        The ``parallelism`` parameter has no effect with this default :class:`Executor`,
+        but is part of the API because most subclass implementations allow configuring
+        parallelism.
     """
 
     default_implementation = "default"
@@ -46,9 +52,11 @@ def __init__(
         self,
         workspace: Workspace,
         include_package: Optional[Sequence[str]] = None,
+        parallelism: Optional[int] = None,
     ) -> None:
         self.workspace = workspace
         self.include_package = include_package
+        self.parallelism = parallelism
 
     def execute_step(self, step: "Step") -> None:
         # Import included packages to find registered components.
@@ -65,16 +73,21 @@ def execute_step_graph(
         self, step_graph: StepGraph, run_name: Optional[str] = None
     ) -> ExecutorOutput:
         """
-        Execute a :class:`tango.step_graph.StepGraph`. This attempts to execute
+        Execute a :class:`~tango.step_graph.StepGraph`. This attempts to execute
         every step in order. If a step fails, its dependent steps are not run,
         but unrelated steps are still executed. Step failures will be logged, but
         no exceptions will be raised.
         """
+        if self.parallelism is not None:
+            warnings.warn(
+                "The 'parallelism' parameter has no effect with the default Executor. "
+                "If you want to run steps in parallel, consider using the MulticoreExecutor.",
+                UserWarning,
+            )
 
         successful: Set[str] = set()
         failed: Set[str] = set()
         not_run: Set[str] = set()
-        error_tracebacks: List[str] = []
         uncacheable_leaf_steps = step_graph.uncacheable_leaf_steps()
 
         for step in step_graph.values():
@@ -88,14 +101,26 @@ def execute_step_graph(
                 try:
                     self.execute_step(step)
                     successful.add(step.name)
-                except Exception:
+                except Exception as exc:
                     failed.add(step.name)
-                    error_tracebacks.append(traceback.format_exc())
-
-        for stacktrace in error_tracebacks:
-            logger.error(stacktrace)
+                    log_exception(exc, logger)
 
         return ExecutorOutput(successful=successful, failed=failed, not_run=not_run)
 
+    # NOTE: The reason for having this method instead of just using `execute_step()` to run
+    # a single step is that the certain executors, such as the BeakerExecutor, need to
+    # serialize steps somehow, and the easiest way to serialize a step is by serializing the
+    # whole step config (which can be accessed via the step graph).
+
+    def execute_sub_graph_for_step(
+        self, step_graph: StepGraph, step_name: str, run_name: Optional[str] = None
+    ) -> None:
+        """
+        Execute the sub-graph associated with a particular step in a
+        :class:`~tango.step_graph.StepGraph`.
+        """
+        step = step_graph[step_name]
+        self.execute_step(step)
+
 
 Executor.register("default")(Executor)
diff --git a/tango/executors/multicore_executor.py b/tango/executors/multicore_executor.py
index 2ac83152e..d88a8cf57 100644
--- a/tango/executors/multicore_executor.py
+++ b/tango/executors/multicore_executor.py
@@ -5,7 +5,6 @@
 from tempfile import NamedTemporaryFile
 from typing import Dict, List, Optional, OrderedDict, Sequence, Set, TypeVar
 
-from tango.common.logging import cli_logger
 from tango.executor import Executor, ExecutorOutput
 from tango.step import Step
 from tango.step_graph import StepGraph
@@ -27,12 +26,14 @@ def __init__(
         self,
         workspace: Workspace,
         include_package: Optional[Sequence[str]] = None,
-        parallelism: int = 1,
+        parallelism: Optional[int] = 1,
         num_tries_to_sync_states: int = 3,
         wait_seconds_to_sync_states: int = 3,
     ) -> None:
-        super().__init__(workspace, include_package=include_package)
-        self.parallelism = parallelism
+        super().__init__(workspace, include_package=include_package, parallelism=parallelism or 1)
+        assert self.parallelism is not None
+        if self.parallelism < 0:
+            self.parallelism = min(32, os.cpu_count() or 1)
 
         # Perhaps there's a better way to do this without these being passed as args.
         self._num_tries_to_sync_states = num_tries_to_sync_states
@@ -217,7 +218,7 @@ def _try_to_execute_next_step(config_path: str, run_name: Optional[str] = None)
             if len(_queued_steps) == 0:
                 logger.debug("No steps in queue!")
                 return
-            if len(_running) < self.parallelism:
+            if len(_running) < (self.parallelism or 1):
                 step_name = _queued_steps.pop(0)
                 command: List[str] = [
                     "tango",
@@ -239,7 +240,7 @@ def _try_to_execute_next_step(config_path: str, run_name: Optional[str] = None)
                 _running[step_name] = process
             else:
                 logger.debug(
-                    f"{self.parallelism} steps are already running. Will attempt to execute later."
+                    f"{self.parallelism or 1} steps are already running. Will attempt to execute later."
                 )
 
         # Creates a temporary file in which to store the config. This is passed as a command line
@@ -263,7 +264,7 @@ def _try_to_execute_next_step(config_path: str, run_name: Optional[str] = None)
                     _queue_step(step_name)
 
                 # Begin processes for any queued steps (if not enough processes are already running).
-                while len(_queued_steps) > 0 and len(_running) < self.parallelism:
+                while len(_queued_steps) > 0 and len(_running) < (self.parallelism or 1):
                     _try_to_execute_next_step(config_path=file_ref.name, run_name=run_name)
 
                 # Re-sync the StepState info.
@@ -287,10 +288,7 @@ def _try_to_execute_next_step(config_path: str, run_name: Optional[str] = None)
                 # NOTE: since neither `Step.result()` nor `Step.ensure_result()` will have been
                 # called, we invoke the CLI logger here to let users know that we didn't run this
                 # step because we found it in the cache.
-                cli_logger.info(
-                    '[green]\N{check mark} Found output for step [bold]"%s"[/] in cache...[/]',
-                    step_name,
-                )
+                step.log_cache_hit()
                 _successful.add(step_name)
             else:
                 # step wasn't executed because parents failed, or
diff --git a/tango/integrations/beaker/__init__.py b/tango/integrations/beaker/__init__.py
index 77626a83a..ebef3f1aa 100644
--- a/tango/integrations/beaker/__init__.py
+++ b/tango/integrations/beaker/__init__.py
@@ -14,7 +14,8 @@
 except ModuleNotFoundError:
     raise IntegrationMissingError("beaker", dependencies={"beaker-py"})
 
+from .executor import BeakerExecutor
 from .step_cache import BeakerStepCache
 from .workspace import BeakerWorkspace
 
-__all__ = ["BeakerStepCache", "BeakerWorkspace"]
+__all__ = ["BeakerStepCache", "BeakerWorkspace", "BeakerExecutor"]
diff --git a/tango/integrations/beaker/common.py b/tango/integrations/beaker/common.py
index d172b8ee0..5e507495f 100644
--- a/tango/integrations/beaker/common.py
+++ b/tango/integrations/beaker/common.py
@@ -17,6 +17,9 @@ class Constants:
     STEP_DATASET_PREFIX = "tango-step-"
     STEP_INFO_FNAME = "step_info.json"
     STEP_RESULT_DIR = "result"
+    ENTRYPOINT_DATASET_PREFIX = "tango-entrypoint-"
+    STEP_GRAPH_DATASET_PREFIX = "tango-step-graph-"
+    STEP_EXPERIMENT_PREFIX = "tango-step-"
 
 
 def step_dataset_name(step: Union[str, StepInfo, Step]) -> str:
diff --git a/tango/integrations/beaker/entrypoint.sh b/tango/integrations/beaker/entrypoint.sh
new file mode 100644
index 000000000..30c17de30
--- /dev/null
+++ b/tango/integrations/beaker/entrypoint.sh
@@ -0,0 +1,117 @@
+#!/bin/bash
+#
+# This is the entrypoint script that the Beaker Executor uses when it runs a step
+# on Beaker.
+# It will work on any Docker image that has bash and conda / miniconda installed.
+
+set -eo pipefail
+
+# Ensure we have all the environment variables we need.
+for env_var in "$GITHUB_TOKEN" "$GITHUB_REPO" "$GIT_REF"; do
+    if [[ -z "$env_var" ]]; then
+        echo >&2 "error: required environment variable is empty"
+        exit 1
+    fi
+done
+
+# Initialize conda for bash.
+# See https://stackoverflow.com/a/58081608/4151392
+eval "$(command conda 'shell.bash' 'hook' 2> /dev/null)"
+
+echo "
+[TANGO] [1/3] Installing prerequisites...
+"
+
+# Install GitHub CLI.
+conda install gh --channel conda-forge
+
+# Configure git to use GitHub CLI as a credential helper so that we can clone private repos.
+gh auth setup-git
+
+echo "
+[TANGO] [2/3] Cloning source code...
+"
+
+# Clone the repo and checkout the target commit.
+gh repo clone "$GITHUB_REPO" .
+git checkout "$GIT_REF"
+
+echo "
+[TANGO] [3/3] Reconstructing Python env...
+"
+
+if [[ -z "$VENV_NAME" ]]; then
+    VENV_NAME=venv
+fi
+if [[ -z "$CONDA_ENV_FILE" ]]; then
+    # shellcheck disable=SC2296
+    CONDA_ENV_FILE="environment.yml"
+fi
+if [[ -z "$PIP_REQUIREMENTS_FILE" ]]; then
+    # shellcheck disable=SC2296
+    PIP_REQUIREMENTS_FILE="requirements.txt"
+fi
+
+if conda activate $VENV_NAME &>/dev/null; then
+    echo "[TANGO] Using existing conda environment '$VENV_NAME'"
+    # The virtual environment already exists. Possibly update it based on an environment file.
+    if [[ -f "$CONDA_ENV_FILE" ]]; then
+        echo "[TANGO] Updating environment from conda env file '$CONDA_ENV_FILE'..."
+        conda env update -f "$CONDA_ENV_FILE"
+    fi
+else
+    # The virtual environment doesn't exist yet. Create it.
+    if [[ -f "$CONDA_ENV_FILE" ]]; then
+        # Create from the environment file.
+        echo "[TANGO] Initializing environment from conda env file '$CONDA_ENV_FILE'..."
+        conda env create -n "$VENV_NAME" -f "$CONDA_ENV_FILE" 
+    elif [[ -z "$PYTHON_VERSION" ]]; then
+        # Create a new empty environment with the whatever the default Python version is.
+        echo "[TANGO] Initializing environment with default Python version..."
+        conda create -n "$VENV_NAME" pip
+    else
+        # Create a new empty environment with the specific Python version.
+        echo "[TANGO] Initializing environment with Python $PYTHON_VERSION..."
+        conda create -n "$VENV_NAME" "python=$PYTHON_VERSION" pip
+    fi
+    conda activate "$VENV_NAME"
+fi
+
+if [[ -z "$INSTALL_CMD" ]]; then
+    # Check for a 'requirements.txt' and/or 'setup.py' file.
+    if [[ -f 'setup.py' ]] && [[ -f "$PIP_REQUIREMENTS_FILE" ]]; then
+        echo "[TANGO] Installing packages from 'setup.py' and '$PIP_REQUIREMENTS_FILE'..."
+        pip install . -r "$PIP_REQUIREMENTS_FILE"
+    elif [[ -f 'setup.py' ]]; then
+        echo "[TANGO] Installing packages from 'setup.py'..."
+        pip install .
+    elif [[ -f "$PIP_REQUIREMENTS_FILE" ]]; then
+        echo "[TANGO] Installing dependencies from '$PIP_REQUIREMENTS_FILE'..."
+        pip install -r "$PIP_REQUIREMENTS_FILE"
+    fi
+else
+    echo "[TANGO] Installing packages with given command: $INSTALL_CMD"
+    eval "$INSTALL_CMD"
+fi
+
+PYTHONPATH="$(pwd)"
+export PYTHONPATH
+
+echo "
+Environment info:
+"
+
+echo "Using $(python --version) from $(which python)"
+echo "Packages:"
+if which sed >/dev/null; then
+    pip freeze | sed 's/^/- /'
+else
+    pip freeze
+fi
+
+echo "
+[TANGO] Setup complete ✓
+"
+
+# Execute the arguments to this script as commands themselves.
+exec "$@"
diff --git a/tango/integrations/beaker/executor.py b/tango/integrations/beaker/executor.py
new file mode 100644
index 000000000..4ab6c9365
--- /dev/null
+++ b/tango/integrations/beaker/executor.py
@@ -0,0 +1,688 @@
+import logging
+import os
+import tempfile
+import threading
+import time
+import uuid
+import warnings
+from json import JSONDecodeError
+from pathlib import Path
+from typing import List, Optional, Sequence, Set, Tuple
+
+from beaker import (
+    Beaker,
+    DataMount,
+    Dataset,
+    DatasetConflict,
+    DatasetNotFound,
+    Digest,
+    EnvVar,
+    ExperimentSpec,
+    JobFailedError,
+    TaskResources,
+    TaskSpec,
+)
+
+from tango.common.exceptions import (
+    CancellationError,
+    ConfigurationError,
+    ExecutorError,
+    RunCancelled,
+)
+from tango.common.logging import cli_logger, log_exception, log_record_from_json
+from tango.executor import Executor, ExecutorOutput
+from tango.step import Step
+from tango.step_graph import StepGraph
+from tango.version import VERSION
+from tango.workspace import Workspace
+
+from .common import Constants
+
+logger = logging.getLogger(__name__)
+
+
+@Executor.register("beaker")
+class BeakerExecutor(Executor):
+    """
+    This is a :class:`~tango.executor.Executor` that runs steps on `Beaker`_.
+    Each step is run as its own Beaker experiment.
+
+    .. tip::
+        Registered as an :class:`~tango.executor.Executor` under the name "beaker".
+
+    .. important::
+        The :class:`BeakerExecutor` requires that you run Tango within a GitHub repository and you push
+        all of your changes prior to each ``tango run`` call. It also requires that you have
+        a `GitHub personal access token <https://github.com/settings/tokens/new>`_
+        with at least the "repo" scope set to the environment variable ``GITHUB_TOKEN``
+        (you can also set it using the ``github_token`` parameter, see below).
+
+        This is because :class:`BeakerExecutor` has to be able to clone your code from Beaker.
+
+    .. important::
+        The :class:`BeakerExecutor` will try to recreate your Python environment on Beaker
+        every time a step is run, so it's important that you specify all of your dependencies
+        in a PIP ``requirements.txt`` file, ``setup.py`` file, or a conda ``environment.yml`` file.
+        Alternatively you could provide the ``install_cmd`` argument.
+
+    .. important::
+        The :class:`BeakerExecutor` takes no responsibility for saving the results of steps that
+        it runs on Beaker. That's the job of your workspace. So make sure your using the
+        right type of workspace or your results will be lost.
+
+        For example, any "remote" workspace (like the :class:`BeakerWorkspace`) would work,
+        or in some cases you could use a :class:`~tango.workspaces.LocalWorkspace` on an NFS drive.
+
+    .. important::
+        If you're running a step that requires special hardware, e.g. a GPU, you should
+        specify that in the ``step_resources`` parameter to the step, or by overriding
+        the step's :meth:`.resources() <tango.step.Step.resources>` property method.
+
+    :param workspace: The :class:`~tango.workspace.Workspace` to use.
+    :param clusters: A list of Beaker clusters that the executor may use to run steps.
+    :param include_package: A list of Python packages to import before running steps.
+    :param beaker_workspace: The name or ID of the Beaker workspace to use.
+    :param github_token: You can use this parameter to set a GitHub personal access token instead of using
+        the ``GITHUB_TOKEN`` environment variable.
+    :param beaker_image: The name or ID of a Beaker image to use for running steps on Beaker.
+        The image must come with bash and `conda <https://docs.conda.io/en/latest/index.html>`_
+        installed (Miniconda is okay).
+        This is mutually exclusive with the ``docker_image`` parameter. If neither ``beaker_image``
+        nor ``docker_image`` is specified, the :data:`DEFAULT_BEAKER_IMAGE` will be used.
+    :param docker_image: The name of a publicly-available Docker image to use for running
+        steps on Beaker. The image must come with bash and `conda <https://docs.conda.io/en/latest/index.html>`_
+        installed (Miniconda is okay).
+        This is mutually exclusive with the ``beaker_image`` parameter.
+    :param datasets: External data sources to mount into the Beaker job for each step. You could use
+        this to mount an NFS drive, for example.
+    :param env_vars: Environment variables to set in the Beaker job for each step.
+    :param venv_name: The name of the conda virtual environment to use or create on the image.
+        If you're using your own image that already has a conda environment you want to use,
+        you should set this variable to the name of that environment.
+        You can also set this to "base" to use the base environment.
+    :param parallelism: Control the maximum number of steps run in parallel on Beaker.
+    :param install_cmd: Override the command used to install your code and its dependencies
+        in each Beaker job.
+        For example, you could set ``install_cmd="pip install .[dev]"``.
+    :param kwargs: Additional keyword arguments passed to :meth:`Beaker.from_env() <beaker.Beaker.from_env()>`.
+
+    .. attention::
+        Certain parameters should not be included in the :data:`~tango.settings.TangoGlobalSettings.executor`
+        part of your ``tango.yml`` file, namely ``workspace`` and ``include_package``.
+        Instead use the top-level :data:`~tango.settings.TangoGlobalSettings.workspace`
+        and :data:`~tango.settings.TangoGlobalSettings.include_package` fields, respectively.
+
+    :examples:
+
+    **Minimal tango.yaml file**
+
+    You can use this executor by specifying it in your ``tango.yml`` settings file:
+
+    .. code:: yaml
+
+        executor:
+          type: beaker
+          beaker_workspace: ai2/my-workspace
+          clusters:
+            - ai2/general-cirrascale
+
+    **Using GPUs**
+
+    If you have a step that requires a GPU, there are two things you need to do:
+
+    1. First, you'll need to ensure that the :class:`BeakerExecutor` can install your dependencies the right way
+    to support the GPU hardware. There are usually two ways to do this: use a Docker image that comes
+    with a proper installation of your hardware-specific dependencies (e.g. PyTorch), or add a conda
+    ``environment.yml`` file to your project that specifies the proper version of those dependencies.
+
+    If you go with first option you don't necessarily need to build your own Docker image.
+    If PyTorch is the only hardware-specific dependency you have, you could just use
+    one of AI2's pre-built PyTorch images. Just add these lines to your ``tango.yml`` file:
+
+    .. code:: diff
+
+         executor:
+           type: beaker
+           beaker_workspace: ai2/my-workspace
+        +  docker_image: ghcr.io/allenai/pytorch:1.12.0-cuda11.3-python3.9
+        +  venv_name: base
+           clusters:
+             - ai2/general-cirrascale
+
+    The ``venv_name: base`` line tells the :class:`BeakerExecutor` to use the existing
+    conda environment called "base" on the image instead of creating a new one.
+
+    Alternatively, you could use the :data:`default image <DEFAULT_BEAKER_IMAGE>`
+    and just add a conda ``environment.yml`` file to the root of your project
+    that looks like this:
+
+    .. code:: yaml
+
+        name: torch-env
+        channels:
+          - pytorch
+        dependencies:
+          - python=3.9
+          - cudatoolkit=11.3
+          - numpy
+          - pytorch
+          - ...
+
+    2. And second, you'll need to specify the GPUs required by each step in the config for that step under
+    the :class:`step_resources <tango.step.StepResources>` parameter. For example,
+
+    .. code:: json
+
+        "steps": {
+            "train": {
+                "type": "torch::train",
+                "step_resources": {
+                    "gpu_count": 1
+                }
+            }
+        }
+
+    """
+
+    GITHUB_TOKEN_SECRET_NAME: str = "TANGO_GITHUB_TOKEN"
+
+    BEAKER_TOKEN_SECRET_NAME: str = "BEAKER_TOKEN"
+
+    RESULTS_DIR: str = "/tango/output"
+
+    ENTRYPOINT_DIR: str = "/tango/entrypoint"
+
+    ENTRYPOINT_FILENAME: str = "entrypoint.sh"
+
+    INPUT_DIR: str = "/tango/input"
+
+    STEP_GRAPH_FILENAME: str = "config.json"
+
+    DEFAULT_BEAKER_IMAGE: str = "ai2/conda"
+    """
+    The default image. Used if neither ``beaker_image`` nor ``docker_image`` are set.
+    """
+
+    DEFAULT_NFS_DRIVE = "/net/nfs.cirrascale"
+
+    def __init__(
+        self,
+        workspace: Workspace,
+        clusters: List[str],
+        include_package: Optional[Sequence[str]] = None,
+        beaker_workspace: Optional[str] = None,
+        github_token: Optional[str] = None,
+        beaker_image: Optional[str] = None,
+        docker_image: Optional[str] = None,
+        datasets: Optional[List[DataMount]] = None,
+        env_vars: Optional[List[EnvVar]] = None,
+        venv_name: Optional[str] = None,
+        parallelism: Optional[int] = -1,
+        install_cmd: Optional[str] = None,
+        **kwargs,
+    ):
+        # Pre-validate arguments.
+        if beaker_image is None and docker_image is None:
+            beaker_image = self.DEFAULT_BEAKER_IMAGE
+        elif (beaker_image is None) == (docker_image is None):
+            raise ConfigurationError(
+                "Either 'beaker_image' or 'docker_image' must be specified for BeakerExecutor, but not both."
+            )
+
+        from tango.workspaces import LocalWorkspace, MemoryWorkspace
+
+        if isinstance(workspace, MemoryWorkspace):
+            raise ConfigurationError(
+                "You cannot use the `MemoryWorkspace` with the `BeakerExecutor`! "
+                "Please specify a different workspace."
+            )
+        elif isinstance(workspace, LocalWorkspace):
+            if str(workspace.dir).startswith(self.DEFAULT_NFS_DRIVE):
+                # Mount the NFS drive if not mount already.
+                datasets = datasets or []
+                if not datasets or not any(
+                    [
+                        dm.source.host_path is not None
+                        and dm.source.host_path.startswith(self.DEFAULT_NFS_DRIVE)
+                        for dm in datasets
+                    ]
+                ):
+                    nfs_mount = DataMount.new(
+                        self.DEFAULT_NFS_DRIVE, host_path=self.DEFAULT_NFS_DRIVE
+                    )
+                    datasets.append(nfs_mount)
+            else:
+                warnings.warn(
+                    "It appears that you're using a `LocalWorkspace` on a directory that is not an NFS drive. "
+                    "If the `BeakerExecutor` cannot access this directory from Beaker, your results will be lost.",
+                    UserWarning,
+                )
+
+        super().__init__(workspace, include_package=include_package, parallelism=parallelism)
+        self.beaker = Beaker.from_env(default_workspace=beaker_workspace, session=True, **kwargs)
+        self.beaker_image = beaker_image
+        self.docker_image = docker_image
+        self.datasets = datasets
+        self.env_vars = env_vars
+        self.clusters = clusters
+        self.venv_name = venv_name
+        self.install_cmd = install_cmd
+        self._is_cancelled = threading.Event()
+
+        try:
+            self.github_token: str = github_token or os.environ["GITHUB_TOKEN"]
+        except KeyError:
+            raise ConfigurationError(
+                "A GitHub personal access token with the 'repo' scope is required. "
+                "This can be set with the 'github_token' argument to the BeakerExecutor, "
+                "or as the environment variable 'GITHUB_TOKEN'."
+            )
+
+        # Ensure entrypoint dataset exists.
+        self._ensure_entrypoint_dataset()
+
+    def execute_step(self, step: Step) -> None:
+        raise NotImplementedError
+
+    def execute_step_graph(
+        self, step_graph: StepGraph, run_name: Optional[str] = None
+    ) -> ExecutorOutput:
+        import concurrent.futures
+
+        self._is_cancelled.clear()
+
+        # These will hold the final results which we'll update along the way.
+        successful: Set[str] = set()
+        failed: Set[str] = set()
+        not_run: Set[str] = set()
+
+        # Keeps track of steps that are next up to run on Beaker.
+        steps_to_run: Set[str] = set()
+        # These are steps that have been submitted to Beaker but haven't completed yet.
+        submitted_steps: Set[str] = set()
+        # Futures for tracking the Beaker runs for each step.
+        step_futures: List[concurrent.futures.Future] = []
+
+        uncacheable_leaf_steps = step_graph.uncacheable_leaf_steps()
+
+        def update_steps_to_run():
+            nonlocal steps_to_run, not_run
+            for step_name, step in step_graph.items():
+                if (
+                    step_name in submitted_steps
+                    or step_name in successful
+                    or step_name in failed
+                    or step_name in not_run
+                ):
+                    # Make sure step is no longer in queue.
+                    steps_to_run.discard(step_name)  # This does NOT raise KeyError if not found
+                else:
+                    # Check dependencies.
+                    for dependency in step.dependencies:
+                        if dependency.name not in successful and dependency.cache_results:
+                            if dependency.name in failed or dependency.name in not_run:
+                                # A dependency failed or can't be run, so this step can't be run.
+                                not_run.add(step_name)
+                                steps_to_run.discard(step_name)
+                            break
+                    else:
+                        # Dependencies are OK, so we can run this step now.
+                        if step.cache_results or step in uncacheable_leaf_steps:
+                            steps_to_run.add(step_name)
+
+        def make_future_done_callback(step_name: str):
+            def future_done_callback(future: concurrent.futures.Future):
+                nonlocal successful, failed
+                try:
+                    exc = future.exception()
+                    if exc is None:
+                        successful.add(step_name)
+                    else:
+                        log_exception(exc, logger)
+                        failed.add(step_name)
+                except concurrent.futures.TimeoutError as exc:
+                    log_exception(exc, logger)
+                    failed.add(step_name)
+
+            return future_done_callback
+
+        update_steps_to_run()
+
+        try:
+            with concurrent.futures.ThreadPoolExecutor(
+                max_workers=self.parallelism or None
+            ) as pool:
+                while steps_to_run or step_futures:
+                    # Submit steps left to run.
+                    for step_name in steps_to_run:
+                        future = pool.submit(
+                            self._execute_sub_graph_for_step, step_graph, step_name, run_name, True
+                        )
+                        future.add_done_callback(make_future_done_callback(step_name))
+                        step_futures.append(future)
+                        submitted_steps.add(step_name)
+
+                    # Wait for something to happen.
+                    _, not_done = concurrent.futures.wait(
+                        step_futures, return_when=concurrent.futures.FIRST_COMPLETED
+                    )
+
+                    # Update the list of running futures.
+                    step_futures.clear()
+                    step_futures = list(not_done)
+
+                    # Update the step queue.
+                    update_steps_to_run()
+        except (KeyboardInterrupt, CancellationError):
+            if step_futures:
+                cli_logger.warning("Received interrupt, canceling steps...")
+                self._is_cancelled.set()
+                concurrent.futures.wait(step_futures)
+            raise
+        finally:
+            self._is_cancelled.clear()
+
+        # NOTE: The 'done callback' added to each future is executed in a thread,
+        # and so might not complete before the last 'update_steps_to_run()' is called
+        # in the loop above. Therefore we have to call 'update_steps_to_run()'
+        # one last time here to ensure the 'not_run' set is up-to-date.
+        update_steps_to_run()
+
+        return ExecutorOutput(successful=successful, failed=failed, not_run=not_run)
+
+    def execute_sub_graph_for_step(
+        self, step_graph: StepGraph, step_name: str, run_name: Optional[str] = None
+    ) -> None:
+        self._execute_sub_graph_for_step(step_graph, step_name, run_name)
+
+    def _check_if_cancelled(self):
+        if self._is_cancelled.is_set():
+            raise RunCancelled
+
+    def _execute_sub_graph_for_step(
+        self,
+        step_graph: StepGraph,
+        step_name: str,
+        run_name: Optional[str],
+        in_thread: bool = False,
+    ) -> None:
+        if not in_thread:
+            self._is_cancelled.clear()
+
+        step = step_graph[step_name]
+
+        if step.cache_results and step in self.workspace.step_cache:
+            cli_logger.info(
+                '[green]\N{check mark} Found output for step [bold]"%s"[/] in cache...[/]',
+                step_name,
+            )
+            return
+
+        # Initialize experiment and task spec.
+        spec = self._build_experiment_spec(step_graph, step_name)
+        self._check_if_cancelled()
+
+        # Create experiment.
+        experiment_name = (
+            f"{Constants.STEP_EXPERIMENT_PREFIX}{step.unique_id}-{str(uuid.uuid4())[:8]}"
+        )
+        experiment = self.beaker.experiment.create(experiment_name, spec)
+        cli_logger.info(
+            'Submitted Beaker experiment [cyan]%s[/] for step [b]"%s"[/] (%s)',
+            self.beaker.experiment.url(experiment),
+            step_name,
+            step.unique_id,
+        )
+
+        # Follow the experiment and stream the logs until it completes.
+        setup_stage: bool = True
+        try:
+            for line in self.beaker.experiment.follow(experiment, strict=True):
+                self._check_if_cancelled()
+                # Every log line from Beaker starts with an RFC 3339 UTC timestamp
+                # (e.g. '2021-12-07T19:30:24.637600011Z'). We don't want to print
+                # the timestamps so we split them off like this:
+                line = line[line.find(b"Z ") + 2 :]
+                line_str = line.decode(errors="ignore").rstrip()
+                if not line_str:
+                    continue
+
+                # Try parsing a JSON log record from the line.
+                try:
+                    log_record = log_record_from_json(line_str)
+                    setup_stage = False
+                    logging.getLogger(log_record.name).handle(log_record)
+                except JSONDecodeError:
+                    if setup_stage:
+                        if line_str.startswith("[TANGO] "):
+                            logger.info(
+                                "[step %s] [setup] %s", step_name, line_str[len("[TANGO] ") :]
+                            )
+                        else:
+                            logger.debug("[step %s] [setup] %s", step_name, line_str)
+                    else:
+                        logger.info("[step %s] %s", step_name, line_str)
+        except JobFailedError:
+            raise ExecutorError(
+                f"Beaker job for step '{step_name}' failed. "
+                f"You can check the logs at {self.beaker.experiment.url(experiment)}"
+            )
+        except (KeyboardInterrupt, CancellationError):
+            cli_logger.warning(
+                'Stopping Beaker experiment [cyan]%s[/] for step [b]"%s"[/] (%s)',
+                self.beaker.experiment.url(experiment),
+                step_name,
+                step.unique_id,
+            )
+            self.beaker.experiment.stop(experiment)
+            raise
+
+    @staticmethod
+    def _parse_git_remote(url: str) -> Tuple[str, str]:
+        """
+        Parse a git remote URL into a GitHub (account, repo) pair.
+        """
+        account, repo = (
+            url.split("https://github.com/")[-1]
+            .split("git@github.com:")[-1]
+            .split(".git")[0]
+            .split("/")
+        )
+        return account, repo
+
+    def _ensure_entrypoint_dataset(self) -> Dataset:
+        import hashlib
+        from importlib.resources import read_binary
+
+        import tango.integrations.beaker
+
+        workspace_id = self.beaker.workspace.get().id
+
+        # Get hash of the local entrypoint source file.
+        sha256_hash = hashlib.sha256()
+        contents = read_binary(tango.integrations.beaker, "entrypoint.sh")
+        sha256_hash.update(contents)
+
+        entrypoint_dataset_name = (
+            f"{Constants.ENTRYPOINT_DATASET_PREFIX}{workspace_id}-{sha256_hash.hexdigest()[:6]}"
+        )
+        tmp_entrypoint_dataset_name = (
+            f"{Constants.ENTRYPOINT_DATASET_PREFIX}{str(uuid.uuid4())}-tmp"
+        )
+
+        # Ensure entrypoint dataset exists.
+        entrypoint_dataset: Dataset
+        try:
+            entrypoint_dataset = self.beaker.dataset.get(entrypoint_dataset_name)
+        except DatasetNotFound:
+            # Create it.
+            logger.debug(f"Creating entrypoint dataset '{entrypoint_dataset_name}'")
+            try:
+                with tempfile.TemporaryDirectory() as tmpdirname:
+                    tmpdir = Path(tmpdirname)
+                    entrypoint_path = tmpdir / "entrypoint.sh"
+                    with open(entrypoint_path, "wb") as entrypoint_file:
+                        entrypoint_file.write(contents)
+                    tmp_entrypoint_dataset = self.beaker.dataset.create(
+                        tmp_entrypoint_dataset_name, entrypoint_path, quiet=True
+                    )
+                    entrypoint_dataset = self.beaker.dataset.rename(
+                        tmp_entrypoint_dataset, entrypoint_dataset_name
+                    )
+            except DatasetConflict:  # could be in a race with another `tango` process.
+                time.sleep(1.0)
+                entrypoint_dataset = self.beaker.dataset.get(entrypoint_dataset_name)
+
+        # Verify contents.
+        err_msg = (
+            f"Checksum failed for entrypoint dataset {self.beaker.dataset.url(entrypoint_dataset)}\n"
+            f"This could be a bug, or it could mean someone has tampered with the dataset.\n"
+            f"If you're sure no one has tampered with it, you can delete the dataset from "
+            f"the Beaker dashboard and try again."
+        )
+        ds_files = list(self.beaker.dataset.ls(entrypoint_dataset))
+        if len(ds_files) != 1:
+            raise ExecutorError(err_msg)
+        if ds_files[0].digest != Digest(sha256_hash.digest()):
+            raise ExecutorError(err_msg)
+
+        return entrypoint_dataset
+
+    def _ensure_step_graph_dataset(self, step_graph: StepGraph) -> Dataset:
+        step_graph_dataset_name = f"{Constants.STEP_GRAPH_DATASET_PREFIX}{str(uuid.uuid4())}"
+        try:
+            with tempfile.TemporaryDirectory() as tmpdirname:
+                tmpdir = Path(tmpdirname)
+                path = tmpdir / self.STEP_GRAPH_FILENAME
+                step_graph.to_file(path, include_unique_id=True)
+                dataset = self.beaker.dataset.create(step_graph_dataset_name, path, quiet=True)
+        except DatasetConflict:  # could be in a race with another `tango` process.
+            time.sleep(1.0)
+            dataset = self.beaker.dataset.get(step_graph_dataset_name)
+        return dataset
+
+    def _ensure_cluster(self, task_resources: TaskResources) -> str:
+        cluster_to_use: str
+        if not self.clusters:
+            raise ConfigurationError("At least one cluster is required in 'clusters'")
+        elif len(self.clusters) == 1:
+            cluster_to_use = self.clusters[0]
+        else:
+            available_clusters = sorted(
+                self.beaker.cluster.filter_available(task_resources, *self.clusters),
+                key=lambda x: x.queued_jobs,
+            )
+            if available_clusters:
+                cluster_to_use = available_clusters[0].cluster.full_name
+                logger.debug(f"Using cluster '{cluster_to_use}'")
+            else:
+                cluster_to_use = self.clusters[0]
+                logger.debug(
+                    "No clusters currently have enough free resources available. "
+                    "Will use '%' anyway.",
+                    cluster_to_use,
+                )
+        return cluster_to_use
+
+    def _build_experiment_spec(self, step_graph: StepGraph, step_name: str) -> ExperimentSpec:
+        step = step_graph[step_name]
+        sub_graph = step_graph.sub_graph(step_name)
+        step_info = self.workspace.step_info(step)
+
+        step_resources = step.resources
+        task_resources = TaskResources(
+            cpu_count=step_resources.cpu_count,
+            gpu_count=step_resources.gpu_count,
+            memory=step_resources.memory,
+            shared_memory=step_resources.shared_memory,
+        )
+
+        # Ensure we're working in a GitHub repository.
+        if (
+            step_info.environment.git is None
+            or step_info.environment.git.commit is None
+            or step_info.environment.git.remote is None
+            or "github.com" not in step_info.environment.git.remote
+        ):
+            raise ExecutorError(
+                f"Missing git data for step '{step.unique_id}'. "
+                f"BeakerExecutor requires a git repository with a GitHub remote."
+            )
+        try:
+            github_account, github_repo = self._parse_git_remote(step_info.environment.git.remote)
+        except ValueError:
+            raise ExecutorError("BeakerExecutor requires a git repository with a GitHub remote.")
+        git_ref = step_info.environment.git.commit
+        self._check_if_cancelled()
+
+        # Ensure dataset with the entrypoint script exists and get it.
+        entrypoint_dataset = self._ensure_entrypoint_dataset()
+        self._check_if_cancelled()
+
+        # Create dataset for step graph.
+        step_graph_dataset = self._ensure_step_graph_dataset(sub_graph)
+        self._check_if_cancelled()
+
+        # Write the GitHub token secret.
+        self.beaker.secret.write(self.GITHUB_TOKEN_SECRET_NAME, self.github_token)
+        self._check_if_cancelled()
+
+        # Write the Beaker token secret.
+        self.beaker.secret.write(self.BEAKER_TOKEN_SECRET_NAME, self.beaker.config.user_token)
+        self._check_if_cancelled()
+
+        # Build Tango command to run.
+        command = [
+            "tango",
+            "--log-level",
+            "debug",
+            "--called-by-executor",
+            "beaker-executor-run",
+            self.INPUT_DIR + "/" + self.STEP_GRAPH_FILENAME,
+            step.name,
+            self.workspace.url,
+        ]
+        if self.include_package is not None:
+            for package in self.include_package:
+                command += ["-i", package]
+
+        # Get cluster to use.
+        cluster = self._ensure_cluster(task_resources)
+        self._check_if_cancelled()
+
+        # Ignore the patch version.
+        # E.g. '3.9.7' -> '3.9'
+        python_version = step_info.environment.python
+        python_version = python_version[: python_version.find(".", python_version.find(".") + 1)]
+
+        # Build task spec.
+        task_spec = (
+            TaskSpec.new(
+                step.unique_id,
+                cluster,
+                beaker_image=self.beaker_image,
+                docker_image=self.docker_image,
+                result_path=self.RESULTS_DIR,
+                command=["bash", self.ENTRYPOINT_DIR + "/" + self.ENTRYPOINT_FILENAME],
+                arguments=command,
+                resources=task_resources,
+                datasets=self.datasets,
+                env_vars=self.env_vars,
+            )
+            .with_env_var(name="TANGO_VERSION", value=VERSION)
+            .with_env_var(name="GITHUB_TOKEN", secret=self.GITHUB_TOKEN_SECRET_NAME)
+            .with_env_var(name="BEAKER_TOKEN", secret=self.BEAKER_TOKEN_SECRET_NAME)
+            .with_env_var(name="GITHUB_REPO", value=f"{github_account}/{github_repo}")
+            .with_env_var(name="GIT_REF", value=git_ref)
+            .with_env_var(name="PYTHON_VERSION", value=python_version)
+            .with_dataset(self.ENTRYPOINT_DIR, beaker=entrypoint_dataset.id)
+            .with_dataset(self.INPUT_DIR, beaker=step_graph_dataset.id)
+        )
+
+        if self.venv_name is not None:
+            task_spec = task_spec.with_env_var(name="VENV_NAME", value=self.venv_name)
+
+        if self.install_cmd is not None:
+            task_spec = task_spec.with_env_var(name="INSTALL_CMD", value=self.install_cmd)
+
+        return ExperimentSpec(tasks=[task_spec])
diff --git a/tango/integrations/beaker/step_cache.py b/tango/integrations/beaker/step_cache.py
index 5cfd9b2a6..e072ec7e9 100644
--- a/tango/integrations/beaker/step_cache.py
+++ b/tango/integrations/beaker/step_cache.py
@@ -29,11 +29,11 @@ class BeakerStepCache(LocalStepCache):
     It also keeps a limited in-memory cache as well as a local backup on disk, so fetching a
     step's resulting subsequent times should be fast.
 
+    .. tip::
+        Registered as a :class:`~tango.step_cache.StepCache` under the name "beaker".
+
     :param workspace: The name or ID of the Beaker workspace to use.
     :param beaker: The Beaker client to use.
-
-    .. tip::
-        Registered as :class:`~tango.step_cache.StepCache` under the name "beaker".
     """
 
     def __init__(self, beaker_workspace: Optional[str] = None, beaker: Optional[Beaker] = None):
diff --git a/tango/integrations/beaker/workspace.py b/tango/integrations/beaker/workspace.py
index e543818df..816127995 100644
--- a/tango/integrations/beaker/workspace.py
+++ b/tango/integrations/beaker/workspace.py
@@ -35,11 +35,11 @@ class BeakerWorkspace(Workspace):
     """
     This is a :class:`~tango.workspace.Workspace` that stores step artifacts on `Beaker`_.
 
-    :param workspace: The name or ID of the Beaker workspace to use.
-    :param kwargs: Additional keyword arguments passed to :meth:`Beaker.from_env() <beaker.Beaker.from_env()>`.
-
     .. tip::
-        Registered as :class:`~tango.workspace.Workspace` under the name "beaker".
+        Registered as a :class:`~tango.workspace.Workspace` under the name "beaker".
+
+    :param beaker_workspace: The name or ID of the Beaker workspace to use.
+    :param kwargs: Additional keyword arguments passed to :meth:`Beaker.from_env() <beaker.Beaker.from_env()>`.
     """
 
     STEP_INFO_CACHE_SIZE = 512
@@ -203,7 +203,7 @@ def register_run(self, targets: Iterable[Step], name: Optional[str] = None) -> R
                     break
         else:
             try:
-                run_dataset = self.beaker.dataset.create(name, commit=False)
+                run_dataset = self.beaker.dataset.create(run_dataset_name(name), commit=False)
             except DatasetConflict:
                 raise ValueError("Run name '{name}' is already in use")
 
@@ -287,7 +287,7 @@ def _get_run_from_dataset(self, dataset: Dataset) -> Optional[Run]:
             run_name = dataset.name[len(Constants.RUN_DATASET_PREFIX) :]
             steps: Dict[str, StepInfo] = {}
             steps_info_bytes = b"".join(
-                self.beaker.dataset.stream_file(dataset, Constants.RUN_DATA_FNAME, quiet=True)
+                list(self.beaker.dataset.stream_file(dataset, Constants.RUN_DATA_FNAME, quiet=True))
             )
             steps_info = json.loads(steps_info_bytes)
         except (DatasetNotFound, FileNotFoundError):
@@ -302,7 +302,9 @@ def _get_run_from_dataset(self, dataset: Dataset) -> Optional[Run]:
             for step_name, unique_id in steps_info.items():
                 step_info_futures.append(executor.submit(self.step_info, unique_id))
             for future in concurrent.futures.as_completed(step_info_futures):
-                steps[step_name] = future.result()
+                step_info = future.result()
+                assert step_info.step_name is not None
+                steps[step_info.step_name] = step_info
 
         return Run(name=run_name, start_date=dataset.created, steps=steps)
 
diff --git a/tango/integrations/torch/eval.py b/tango/integrations/torch/eval.py
index 2a4990335..ecc0842a5 100644
--- a/tango/integrations/torch/eval.py
+++ b/tango/integrations/torch/eval.py
@@ -9,7 +9,7 @@
 from tango.common.lazy import Lazy
 from tango.common.tqdm import Tqdm
 from tango.format import Format, JsonFormat
-from tango.step import Step
+from tango.step import Step, StepResources
 
 from .data import DataLoader
 from .eval_callback import EvalCallback
@@ -52,6 +52,10 @@ class TorchEvalStep(Step):
     FORMAT: Format = JsonFormat()
     SKIP_ID_ARGUMENTS = {"log_every"}
 
+    @property
+    def resources(self) -> StepResources:
+        return self.step_resources or StepResources(gpu_count=1)
+
     def run(  # type: ignore[override]
         self,
         model: Model,
diff --git a/tango/integrations/torch/train.py b/tango/integrations/torch/train.py
index fbffab76e..014b09bac 100644
--- a/tango/integrations/torch/train.py
+++ b/tango/integrations/torch/train.py
@@ -16,7 +16,7 @@
 from tango.common.tqdm import Tqdm
 from tango.common.util import get_extra_imported_modules, import_extra_module
 from tango.format import Format
-from tango.step import Step
+from tango.step import Step, StepResources
 from tango.workspace import Workspace
 
 from .data import DataLoader
@@ -72,6 +72,10 @@ class TorchTrainStep(Step):
     FORMAT: Format = TorchFormat()
     SKIP_ID_ARGUMENTS = {"distributed_port", "log_every"}
 
+    @property
+    def resources(self) -> StepResources:
+        return self.step_resources or StepResources(gpu_count=self.kwargs["device_count"])
+
     def run(  # type: ignore[override]
         self,
         model: Union[Lazy[Model], Model],  # Lazy has to come first
diff --git a/tango/step.py b/tango/step.py
index 3409f267c..9e267e75f 100644
--- a/tango/step.py
+++ b/tango/step.py
@@ -6,6 +6,7 @@
 import warnings
 from abc import abstractmethod
 from copy import deepcopy
+from dataclasses import dataclass
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import (
@@ -26,12 +27,13 @@
 from tango.common.det_hash import CustomDetHash, det_hash
 from tango.common.exceptions import ConfigurationError, StepStateError
 from tango.common.from_params import (
+    FromParams,
     infer_constructor_params,
     infer_method_params,
     pop_and_construct_arg,
 )
 from tango.common.lazy import Lazy
-from tango.common.logging import cli_logger
+from tango.common.logging import cli_logger, log_exception
 from tango.common.params import Params
 from tango.common.registrable import Registrable
 from tango.format import DillFormat, Format
@@ -58,6 +60,40 @@ def get_args(tp):  # type: ignore
 _random_for_step_names = random.Random()
 
 
+@dataclass
+class StepResources(FromParams):
+    """
+    TaskResources describe minimum external hardware requirements which must be available for a
+    step to run.
+    """
+
+    cpu_count: Optional[float] = None
+    """
+    Minimum number of logical CPU cores. It may be fractional.
+
+    Examples: ``4``, ``0.5``.
+    """
+
+    gpu_count: Optional[float] = None
+    """
+    Minimum number of GPUs. It must be non-negative.
+    """
+
+    memory: Optional[str] = None
+    """
+    Minimum available system memory as a number with unit suffix.
+
+    Examples: ``2.5GiB``, ``1024m``.
+    """
+
+    shared_memory: Optional[str] = None
+    """
+    Size of ``/dev/shm`` as a number with unit suffix.
+
+    Examples: ``2.5GiB``, ``1024m``.
+    """
+
+
 class Step(Registrable, Generic[T]):
     """
     This class defines one step in your experiment. To write your own step, derive from this class
@@ -79,6 +115,8 @@ class Step(Registrable, Generic[T]):
       This can be accessed via the :attr:`config` property within each step's :meth:`run()` method.
     :param step_unique_id_override: overrides the construction of the step's unique id using the hash
       of inputs.
+    :param step_resources: gives you a way to set the minimum compute resources required
+      to run this step. Certain executors require this information.
 
     .. important::
         Overriding the unique id means that the step will always map to this value, regardless of the inputs,
@@ -129,6 +167,7 @@ def __init__(
         step_format: Optional[Format] = None,
         step_config: Optional[Dict[str, Any]] = None,
         step_unique_id_override: Optional[str] = None,
+        step_resources: Optional[StepResources] = None,
         **kwargs,
     ):
         if self.VERSION is not None:
@@ -201,8 +240,8 @@ def __init__(
         self.work_dir_for_run: Optional[
             Path
         ] = None  # This is set only while the run() method runs.
-
         self._config = step_config
+        self.step_resources = step_resources
 
     @classmethod
     def massage_kwargs(cls, kwargs: Dict[str, Any]) -> Dict[str, Any]:
@@ -360,29 +399,18 @@ def _run_with_work_dir(self, workspace: "Workspace", needed_by: Optional["Step"]
 
         try:
             kwargs = self._replace_steps_with_results(self.kwargs, workspace)
-
-            if needed_by:
-                cli_logger.info(
-                    '[blue]\N{black circle} Starting step [bold]"%s"[/] (needed by "%s")...[/]',
-                    self.name,
-                    needed_by.name,
-                )
-            else:
-                cli_logger.info(
-                    '[blue]\N{black circle} Starting step [bold]"%s"[/]...[/]',
-                    self.name,
-                )
-
+            self.log_starting(needed_by=needed_by)
             workspace.step_starting(self)
 
             try:
                 result = self.run(**kwargs)
             except BaseException as e:
+                self.log_failure(e)
                 workspace.step_failed(self, e)
                 raise
 
             result = workspace.step_finished(self, result)
-            cli_logger.info(f'[green]\N{check mark} Finished step [bold]"{self.name}"[/][/]')
+            self.log_finished()
             return result
         finally:
             self._workspace = None
@@ -435,6 +463,17 @@ def config(self) -> Dict[str, Any]:
     def det_hash_object(self) -> Any:
         return self.unique_id
 
+    @property
+    def resources(self) -> StepResources:
+        """
+        Defines the minimum compute resources required to run this step.
+        Certain executors require this information in order to allocate resources for each step.
+
+        You can set this with the ``step_resources`` argument to :class:`Step`
+        or you can override this method to automatically define the required resources.
+        """
+        return self.step_resources or StepResources()
+
     @property
     def unique_id(self) -> str:
         """Returns the unique ID for this step.
@@ -539,19 +578,7 @@ def result(
                 # Step has been completed (and cached) by a different process, so we
                 # do nothing here and pull the result from the cache below.
 
-        if needed_by:
-            cli_logger.info(
-                '[green]\N{check mark} Found output for step [bold]"%s"[/bold] in cache '
-                '(needed by "%s")...[/green]',
-                self.name,
-                needed_by.name,
-            )
-        else:
-            cli_logger.info(
-                '[green]\N{check mark} Found output for step [bold]"%s"[/] in cache...[/]',
-                self.name,
-            )
-
+        self.log_cache_hit(needed_by=needed_by)
         return workspace.step_cache[self]
 
     def ensure_result(
@@ -571,10 +598,7 @@ def ensure_result(
             workspace = default_workspace
 
         if self in workspace.step_cache:
-            cli_logger.info(
-                '[green]\N{check mark} Found output for step [bold]"%s"[/] in cache...[/]',
-                self.name,
-            )
+            self.log_cache_hit()
         else:
             self.result(workspace)
 
@@ -620,6 +644,50 @@ def recursive_dependencies(self) -> Set["Step"]:
             steps.extend(step.dependencies)
         return seen
 
+    def log_cache_hit(self, needed_by: Optional["Step"] = None) -> None:
+        if needed_by is not None:
+            cli_logger.info(
+                '[green]\N{check mark} Found output for step [bold]"%s"[/bold] in cache '
+                '(needed by "%s")...[/green]',
+                self.name,
+                needed_by.name,
+            )
+        else:
+            cli_logger.info(
+                '[green]\N{check mark} Found output for step [bold]"%s"[/] in cache...[/]',
+                self.name,
+            )
+
+    def log_starting(self, needed_by: Optional["Step"] = None) -> None:
+        if needed_by is not None:
+            cli_logger.info(
+                '[blue]\N{black circle} Starting step [bold]"%s"[/] (needed by "%s")...[/]',
+                self.name,
+                needed_by.name,
+            )
+        else:
+            cli_logger.info(
+                '[blue]\N{black circle} Starting step [bold]"%s"[/]...[/]',
+                self.name,
+            )
+
+    def log_finished(self, run_name: Optional[str] = None) -> None:
+        if run_name is not None:
+            cli_logger.info(
+                '[green]\N{check mark} Finished run for step [bold]"%s"[/] (%s)[/]',
+                self.name,
+                run_name,
+            )
+        else:
+            cli_logger.info(
+                '[green]\N{check mark} Finished step [bold]"%s"[/][/]',
+                self.name,
+            )
+
+    def log_failure(self, exception: BaseException) -> None:
+        log_exception(exception, logger=self.logger)
+        cli_logger.error('[red]\N{ballot x} Step [bold]"%s"[/] failed[/]', self.name)
+
 
 class WithUnresolvedSteps(CustomDetHash):
     """
diff --git a/tango/step_graph.py b/tango/step_graph.py
index 55850a4d7..22b9b2ffc 100644
--- a/tango/step_graph.py
+++ b/tango/step_graph.py
@@ -4,7 +4,6 @@
 from tango.common import PathOrStr
 from tango.common.exceptions import ConfigurationError
 from tango.common.params import Params
-from tango.common.util import import_extra_module
 from tango.step import Step
 
 logger = logging.getLogger(__name__)
@@ -228,8 +227,6 @@ def uncacheable_leaf_steps(self) -> Set[Step]:
     @classmethod
     def from_file(cls, filename: PathOrStr) -> "StepGraph":
         params = Params.from_file(filename)
-        for package_name in params.pop("include_package", []):
-            import_extra_module(package_name)
         return StepGraph.from_params(params.pop("steps", keep_as_dict=True))
 
     def to_config(self, include_unique_id: bool = False) -> Dict[str, Dict]:
diff --git a/tango/step_info.py b/tango/step_info.py
index db66b856d..e0da5edc3 100644
--- a/tango/step_info.py
+++ b/tango/step_info.py
@@ -13,6 +13,7 @@
 import pytz
 
 from .common.from_params import FromParams
+from .common.logging import log_exception
 from .common.util import jsonify, local_timezone, replace_steps_with_unique_id
 from .step import Step
 from .version import VERSION
@@ -31,7 +32,8 @@ def get_pip_packages() -> Optional[List[Tuple[str, str]]]:
 
         return sorted([(d.key, d.version) for d in iter(pkg_resources.working_set)])
     except Exception as exc:
-        logger.exception("Error saving pip packages: %s", exc)
+        logger.error("Error saving pip packages")
+        log_exception(exc)
     return None
 
 
@@ -69,34 +71,15 @@ class GitMetadata(FromParams):
 
     @classmethod
     def check_for_repo(cls) -> Optional["GitMetadata"]:
-        import subprocess
+        from git import InvalidGitRepositoryError, Repo
 
         try:
-            commit = (
-                subprocess.check_output("git rev-parse HEAD".split(" "), stderr=subprocess.DEVNULL)
-                .decode("ascii")
-                .strip()
-            )
-            remote: Optional[str] = None
-            for line in (
-                subprocess.check_output("git remote -v".split(" "))
-                .decode("ascii")
-                .strip()
-                .split("\n")
-            ):
-                remotes: Dict[str, str] = {}
-                if line.endswith("(fetch)"):
-                    name, info = line.split("\t")
-                    url = info.split(" ")[0]
-                    remotes[name] = url
-                if "origin" in remotes:
-                    remote = remotes["origin"]
-                elif remotes:
-                    remote = list(remotes.values())[0]
-            return cls(commit=commit, remote=remote)
-        except (subprocess.CalledProcessError, FileNotFoundError):
+            repo = Repo(".")
+        except InvalidGitRepositoryError:
             return None
 
+        return cls(commit=str(repo.commit()), remote=repo.remote().url)
+
 
 @dataclass
 class TangoMetadata(FromParams):


Test patch:
diff --git a/tango/common/testing/__init__.py b/tango/common/testing/__init__.py
index a4aafcb8d..1aaadbd9b 100644
--- a/tango/common/testing/__init__.py
+++ b/tango/common/testing/__init__.py
@@ -9,6 +9,7 @@
 from tango.common.aliases import EnvVarNames, PathOrStr
 from tango.common.logging import initialize_logging, teardown_logging
 from tango.common.params import Params
+from tango.settings import TangoGlobalSettings
 
 
 class TangoTestCase:
@@ -84,11 +85,12 @@ def run(
         include_package: Optional[List[str]] = None,
         workspace_url: Optional[str] = None,
         step_name: Optional[str] = None,
-        parallelism: int = 1,
-        multicore: bool = False,
+        parallelism: Optional[int] = 1,
+        multicore: Optional[bool] = False,
         name: Optional[str] = None,
+        settings: Optional[TangoGlobalSettings] = None,
     ) -> Path:
-        from tango.__main__ import TangoGlobalSettings, _run
+        from tango.__main__ import _run
 
         if isinstance(config, (dict, Params)):
             params = config if isinstance(config, Params) else Params(config)
@@ -101,7 +103,7 @@ def run(
             overrides = json.dumps(overrides)
 
         run_name = _run(
-            TangoGlobalSettings(),
+            settings or TangoGlobalSettings(),
             str(config),
             workspace_url=workspace_url or "local://" + str(self.TEST_DIR / "workspace"),
             overrides=overrides,
@@ -111,6 +113,7 @@ def run(
             parallelism=parallelism,
             multicore=multicore,
             name=name,
+            allow_dirty=True,
         )
 
         return self.TEST_DIR / "workspace" / "runs" / run_name
@@ -122,9 +125,11 @@ def run_experiment(
     overrides: Optional[Union[Dict[str, Any], str]] = None,
     file_friendly_logging: bool = True,
     include_package: Optional[List[str]] = None,
-    parallelism: int = 1,
-    multicore: bool = False,
+    workspace_url: Optional[str] = None,
+    parallelism: Optional[int] = 1,
+    multicore: Optional[bool] = False,
     name: Optional[str] = None,
+    settings: Optional[TangoGlobalSettings] = None,
 ):
     """
     A context manager to make testing experiments easier. On ``__enter__`` it runs
@@ -139,9 +144,11 @@ def run_experiment(
             config,
             overrides=overrides,
             include_package=include_package,
+            workspace_url=workspace_url,
             parallelism=parallelism,
             multicore=multicore,
             name=name,
+            settings=settings,
         )
     finally:
         test_case.teardown_method()
diff --git a/tests/integrations/beaker/conftest.py b/tests/integrations/beaker/conftest.py
index 7165f51c9..a29d0cbd1 100644
--- a/tests/integrations/beaker/conftest.py
+++ b/tests/integrations/beaker/conftest.py
@@ -29,16 +29,26 @@ def patched_constants_prefix(monkeypatch) -> str:
     PREFIX = os.environ.get("GITHUB_SHA", "A")[:6] + "-" + str(uuid.uuid1())[:6] + "-"
     monkeypatch.setattr(Constants, "STEP_DATASET_PREFIX", "tango-step-" + PREFIX)
     monkeypatch.setattr(Constants, "RUN_DATASET_PREFIX", "tango-run-" + PREFIX)
+    monkeypatch.setattr(Constants, "ENTRYPOINT_DATASET_PREFIX", "tango-entrypoint-" + PREFIX)
+    monkeypatch.setattr(Constants, "STEP_GRAPH_DATASET_PREFIX", "tango-step-graph-" + PREFIX)
+    monkeypatch.setattr(Constants, "STEP_EXPERIMENT_PREFIX", "tango-step-" + PREFIX)
     return PREFIX
 
 
+@pytest.fixture
+def beaker_workspace_name() -> str:
+    return "ai2/tango-beaker-testing"
+
+
 @pytest.fixture
 def beaker_workspace(
-    patched_unique_id_suffix: str, patched_constants_prefix: str
+    beaker_workspace_name: str, patched_unique_id_suffix: str, patched_constants_prefix: str
 ) -> Generator[str, None, None]:
-    workspace_name = "ai2/tango-beaker-testing"
-    beaker = Beaker.from_env(default_workspace=workspace_name)
-    yield workspace_name
+    beaker = Beaker.from_env(default_workspace=beaker_workspace_name)
+    yield beaker_workspace_name
+    # Remove experiments.
+    #  for experiment in beaker.workspace.experiments(match=patched_constants_prefix):
+    #      beaker.experiment.delete(experiment)
     # Remove datasets.
     for dataset in beaker.workspace.datasets(match=patched_unique_id_suffix):
         beaker.dataset.delete(dataset)
diff --git a/tests/integrations/beaker/executor_test.py b/tests/integrations/beaker/executor_test.py
new file mode 100644
index 000000000..255606691
--- /dev/null
+++ b/tests/integrations/beaker/executor_test.py
@@ -0,0 +1,69 @@
+import petname
+import pytest
+from beaker import DataMount
+
+from tango.common.exceptions import ConfigurationError
+from tango.common.testing import run_experiment
+from tango.executor import Executor
+from tango.integrations.beaker.executor import BeakerExecutor
+from tango.integrations.beaker.workspace import BeakerWorkspace
+from tango.settings import TangoGlobalSettings
+from tango.workspaces import default_workspace
+
+
+def test_from_params(beaker_workspace_name: str):
+    executor = Executor.from_params(
+        dict(
+            type="beaker",
+            beaker_workspace=beaker_workspace_name,
+            beaker_image="ai2/conda",
+            github_token="FAKE_TOKEN",
+            datasets=[{"source": {"beaker": "some-dataset"}, "mount_path": "/input"}],
+        ),
+        workspace=BeakerWorkspace(beaker_workspace=beaker_workspace_name),
+        clusters=["fake-cluster"],
+    )
+    assert isinstance(executor, BeakerExecutor)
+    assert executor.datasets is not None
+    assert len(executor.datasets) == 1
+    assert isinstance(executor.datasets[0], DataMount)
+    assert executor.datasets[0].source.beaker == "some-dataset"
+
+
+def test_init_with_mem_workspace(beaker_workspace_name: str):
+    with pytest.raises(ConfigurationError, match="MemoryWorkspace"):
+        BeakerExecutor(
+            workspace=default_workspace,
+            beaker_workspace=beaker_workspace_name,
+            beaker_image="ai2/conda",
+            github_token="FAKE_TOKEN",
+            clusters=["fake-cluster"],
+        )
+
+
+@pytest.fixture
+def settings(beaker_workspace_name: str) -> TangoGlobalSettings:
+    return TangoGlobalSettings(
+        workspace={"type": "beaker", "beaker_workspace": beaker_workspace_name},
+        executor={
+            "type": "beaker",
+            "beaker_workspace": beaker_workspace_name,
+            "install_cmd": "pip install .[beaker]",
+            "clusters": ["ai2/allennlp-cirrascale", "ai2/general-cirrascale"],
+        },
+    )
+
+
+def test_beaker_executor(
+    settings: TangoGlobalSettings, beaker_workspace_name: str, patched_unique_id_suffix: str
+):
+    run_name = petname.generate()
+    with run_experiment(
+        {"steps": {"hello": {"type": "string", "result": "Hello, World!"}}},
+        settings=settings,
+        workspace_url=f"beaker://{beaker_workspace_name}",
+        name=run_name,
+        multicore=None,
+    ):
+        workspace = BeakerWorkspace(beaker_workspace=beaker_workspace_name)
+        assert "hello" in workspace.registered_run(run_name).steps
diff --git a/tests/main_test.py b/tests/main_test.py
index 639162009..3998ca217 100644
--- a/tests/main_test.py
+++ b/tests/main_test.py
@@ -85,32 +85,29 @@ def test_logging_all_levels(self, log_level: str, raise_error):
         _, cleaned_log_lines = self.check_logs(run_dir, result)
 
         # Debug messages.
-        assert cleaned_log_lines.count("[step noisy_step] debug message from cli_logger") == 1
-        assert cleaned_log_lines.count("[step noisy_step] debug message") == (
-            1 if log_level == "debug" else 0
-        )
+        assert cleaned_log_lines.count("debug message from cli_logger") == 1
+        assert cleaned_log_lines.count("debug message") == (1 if log_level == "debug" else 0)
 
         # Info messages.
-        assert cleaned_log_lines.count("[step noisy_step] info message from cli_logger") == 1
-        assert cleaned_log_lines.count("[step noisy_step] info message") == (
+        assert cleaned_log_lines.count("info message from cli_logger") == 1
+        assert cleaned_log_lines.count("info message") == (
             1 if log_level in {"debug", "info"} else 0
         )
 
         # Warning messages.
-        assert cleaned_log_lines.count("[step noisy_step] warning message from cli_logger") == 1
-        assert cleaned_log_lines.count("[step noisy_step] warning message") == (
+        assert cleaned_log_lines.count("warning message from cli_logger") == 1
+        assert cleaned_log_lines.count("warning message") == (
             1 if log_level in {"debug", "info", "warning"} else 0
         )
 
         # Error messages.
-        assert cleaned_log_lines.count("[step noisy_step] error message from cli_logger") == 1
-        assert cleaned_log_lines.count("[step noisy_step] error message") == (
+        assert cleaned_log_lines.count("error message from cli_logger") == 1
+        assert cleaned_log_lines.count("error message") == (
             1 if log_level in {"debug", "info", "warning", "error"} else 0
         )
 
         # Traceback.
         if raise_error:
-            assert "[step noisy_step] Uncaught exception" in cleaned_log_lines
             assert "Traceback (most recent call last):" in cleaned_log_lines
             assert "ValueError: Oh no!" in cleaned_log_lines