SERVER-47274: Refactor task generation in evergreen

(cherry picked from commit 4d82d10588dbeca498e46d51a36b6efdf8379af1)

17 files changed, 823 insertions, 951 deletions

diff --git a/buildscripts/burn_in_tags.py b/buildscripts/burn_in_tags.py
index 6a79a67445f..6ed3d1d5bec 100644
--- a/buildscripts/burn_in_tags.py
+++ b/buildscripts/burn_in_tags.py
@@ -1,35 +1,35 @@
 #!/usr/bin/env python3
 """Generate burn in tests to run on certain build variants."""
 
-import sys
-import os
-
 from collections import namedtuple
+import os
+import sys
 from typing import Any, Dict, Iterable
+
 import click
 
-from evergreen.api import RetryingEvergreenApi
+from evergreen.api import RetryingEvergreenApi, EvergreenApi
 from git import Repo
-from shrub.config import Configuration
-from shrub.variant import TaskSpec
+from shrub.v2 import ShrubProject, BuildVariant, ExistingTask
 
 # Get relative imports to work when the package is not installed on the PYTHONPATH.
 if __name__ == "__main__" and __package__ is None:
     sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 # pylint: disable=wrong-import-position
+from buildscripts.util.fileops import write_file_to_dir
 import buildscripts.util.read_config as read_config
 from buildscripts.ciconfig import evergreen
-from buildscripts.ciconfig.evergreen import EvergreenProjectConfig
+from buildscripts.ciconfig.evergreen import EvergreenProjectConfig, Variant
 from buildscripts.burn_in_tests import create_generate_tasks_config, create_tests_by_task, \
     GenerateConfig, RepeatConfig, DEFAULT_REPO_LOCATIONS
-
 # pylint: enable=wrong-import-position
 
 CONFIG_DIRECTORY = "generated_burn_in_tags_config"
 CONFIG_FILE = "burn_in_tags_gen.json"
 EVERGREEN_FILE = "etc/evergreen.yml"
 EVG_CONFIG_FILE = ".evergreen.yml"
+COMPILE_TASK = "compile_without_package_TG"
 
 ConfigOptions = namedtuple("ConfigOptions", [
     "build_variant",
@@ -94,86 +94,72 @@ def _create_evg_build_variant_map(expansions_file_data, evergreen_conf):
     return {}
 
 
-def _generate_evg_build_variant(shrub_config, build_variant, run_build_variant,
-                                burn_in_tags_gen_variant, evg_conf):
+def _generate_evg_build_variant(
+        source_build_variant: Variant,
+        run_build_variant: str,
+        bypass_build_variant: str,
+) -> BuildVariant:
     """
-    Generate buildvariants for a given shrub config.
+    Generate a shrub build variant for the given run build variant.
 
-    :param shrub_config: Shrub config object that the generated buildvariant will be built upon.
-    :param build_variant: The base variant that the generated run_buildvariant will be based on.
-    :param run_build_variant: The generated buildvariant.
-    :param burn_in_tags_gen_variant: The buildvariant on which the burn_in_tags_gen task runs.
+    :param source_build_variant: The build variant to base configuration on.
+    :param run_build_variant: The build variant to generate.
+    :param bypass_build_variant: The build variant to get compile artifacts from.
+    :return: Shrub build variant configuration.
     """
-    base_variant_config = evg_conf.get_variant(build_variant)
-
-    new_variant_display_name = f"! {base_variant_config.display_name}"
-    new_variant_run_on = base_variant_config.run_on[0]
-
-    task_spec = TaskSpec("compile_without_package_TG")
-
-    new_variant = shrub_config.variant(run_build_variant).expansion("burn_in_bypass",
-                                                                    burn_in_tags_gen_variant)
-    new_variant.display_name(new_variant_display_name)
-    new_variant.run_on(new_variant_run_on)
-    new_variant.task(task_spec)
+    display_name = f"! {source_build_variant.display_name}"
+    run_on = source_build_variant.run_on
+    modules = source_build_variant.modules
 
-    base_variant_expansions = base_variant_config.expansions
-    new_variant.expansions(base_variant_expansions)
+    expansions = source_build_variant.expansions
+    expansions["burn_in_bypass"] = bypass_build_variant
 
-    modules = base_variant_config.modules
-    new_variant.modules(modules)
+    build_variant = BuildVariant(run_build_variant, display_name, expansions=expansions,
+                                 modules=modules, run_on=run_on)
+    build_variant.add_existing_task(ExistingTask(COMPILE_TASK))
+    return build_variant
 
 
 # pylint: disable=too-many-arguments
-def _generate_evg_tasks(evergreen_api, shrub_config, expansions_file_data, build_variant_map, repos,
-                        evg_conf):
+def _generate_evg_tasks(evergreen_api: EvergreenApi, shrub_project: ShrubProject,
+                        task_expansions: Dict[str, Any], build_variant_map: Dict[str, str],
+                        repos: Iterable[Repo], evg_conf: EvergreenProjectConfig) -> None:
     """
-    Generate burn in tests tasks for a given shrub config and group of buildvariants.
+    Generate burn in tests tasks for a given shrub config and group of build variants.
 
     :param evergreen_api: Evergreen.py object.
-    :param shrub_config: Shrub config object that the build variants will be built upon.
-    :param expansions_file_data: Config data file to use.
+    :param shrub_project: Shrub config object that the build variants will be built upon.
+    :param task_expansions: Dictionary of expansions for the running task.
     :param build_variant_map: Map of base buildvariants to their generated buildvariant.
     :param repos: Git repositories.
     """
     for build_variant, run_build_variant in build_variant_map.items():
-        config_options = _get_config_options(expansions_file_data, build_variant, run_build_variant)
+        config_options = _get_config_options(task_expansions, build_variant, run_build_variant)
         tests_by_task = create_tests_by_task(build_variant, repos, evg_conf)
         if tests_by_task:
-            _generate_evg_build_variant(shrub_config, build_variant, run_build_variant,
-                                        expansions_file_data["build_variant"], evg_conf)
+            shrub_build_variant = _generate_evg_build_variant(
+                evg_conf.get_variant(build_variant), run_build_variant,
+                task_expansions["build_variant"])
             gen_config = GenerateConfig(build_variant, config_options.project, run_build_variant,
                                         config_options.distro).validate(evg_conf)
             repeat_config = RepeatConfig(repeat_tests_min=config_options.repeat_tests_min,
                                          repeat_tests_max=config_options.repeat_tests_max,
                                          repeat_tests_secs=config_options.repeat_tests_secs)
 
-            create_generate_tasks_config(shrub_config, tests_by_task, gen_config, repeat_config,
-                                         evergreen_api, include_gen_task=False)
-
+            create_generate_tasks_config(shrub_build_variant, tests_by_task, gen_config,
+                                         repeat_config, evergreen_api, include_gen_task=False)
+            shrub_project.add_build_variant(shrub_build_variant)
 
-def _write_to_file(shrub_config):
-    """
-    Save shrub config to file.
-
-    :param shrub_config: Shrub config object.
-    """
-    if not os.path.exists(CONFIG_DIRECTORY):
-        os.makedirs(CONFIG_DIRECTORY)
 
-    with open(os.path.join(CONFIG_DIRECTORY, CONFIG_FILE), "w") as file_handle:
-        file_handle.write(shrub_config.to_json())
-
-
-def burn_in(expansions_file_data: Dict[str, Any], evg_conf: EvergreenProjectConfig,
+def burn_in(task_expansions: Dict[str, Any], evg_conf: EvergreenProjectConfig,
             evergreen_api: RetryingEvergreenApi, repos: Iterable[Repo]):
     """Execute Main program."""
-
-    shrub_config = Configuration()
-    build_variant_map = _create_evg_build_variant_map(expansions_file_data, evg_conf)
-    _generate_evg_tasks(evergreen_api, shrub_config, expansions_file_data, build_variant_map, repos,
+    shrub_project = ShrubProject.empty()
+    build_variant_map = _create_evg_build_variant_map(task_expansions, evg_conf)
+    _generate_evg_tasks(evergreen_api, shrub_project, task_expansions, build_variant_map, repos,
                         evg_conf)
-    _write_to_file(shrub_config)
+
+    write_file_to_dir(CONFIG_DIRECTORY, CONFIG_FILE, shrub_project.json())
 
 
 @click.command()
diff --git a/buildscripts/burn_in_tests.py b/buildscripts/burn_in_tests.py
index a475b8f74e5..80cea58fa65 100644
--- a/buildscripts/burn_in_tests.py
+++ b/buildscripts/burn_in_tests.py
@@ -1,9 +1,7 @@
 #!/usr/bin/env python3
 """Command line utility for determining what jstests have been added or modified."""
-
 import copy
 import datetime
-import json
 import logging
 import os.path
 import shlex
@@ -18,10 +16,11 @@ import requests
 import yaml
 from evergreen.api import RetryingEvergreenApi, EvergreenApi
 from git import Repo
-from shrub.config import Configuration
 import structlog
 from structlog.stdlib import LoggerFactory
 
+from shrub.v2 import Task, TaskDependency, BuildVariant, ShrubProject, ExistingTask
+
 # Get relative imports to work when the package is not installed on the PYTHONPATH.
 if __name__ == "__main__" and __package__ is None:
     sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -33,10 +32,11 @@ from buildscripts.resmokelib.suitesconfig import create_test_membership_map, get
 from buildscripts.resmokelib.utils import default_if_none, globstar
 from buildscripts.ciconfig.evergreen import parse_evergreen_file, ResmokeArgs, \
     EvergreenProjectConfig, VariantTask
+from buildscripts.util.fileops import write_file
 from buildscripts.util.teststats import TestStats
 from buildscripts.util.taskname import name_generated_task
-from buildscripts.patch_builds.task_generation import resmoke_commands, TimeoutInfo, TaskList
-
+from buildscripts.patch_builds.task_generation import (resmoke_commands, TimeoutInfo,
+                                                       validate_task_generation_limit)
 # pylint: enable=wrong-import-position
 
 structlog.configure(logger_factory=LoggerFactory())
@@ -55,7 +55,6 @@ DEFAULT_PROJECT = "mongodb-mongo-master"
 DEFAULT_REPO_LOCATIONS = [".", "./src/mongo/db/modules/enterprise"]
 REPEAT_SUITES = 2
 EVERGREEN_FILE = "etc/evergreen.yml"
-MAX_TASKS_TO_CREATE = 1000
 MIN_AVG_TEST_OVERFLOW_SEC = float(60)
 MIN_AVG_TEST_TIME_SEC = 5 * 60
 # The executor_file and suite_files defaults are required to make the suite resolver work
@@ -396,13 +395,6 @@ def create_task_list(evergreen_conf: EvergreenProjectConfig, build_variant: str,
     return task_list
 
 
-def _write_json_file(json_data, pathname):
-    """Write out a JSON file."""
-
-    with open(pathname, "w") as fstream:
-        json.dump(json_data, fstream, indent=4)
-
-
 def _set_resmoke_cmd(repeat_config: RepeatConfig, resmoke_args: [str]) -> [str]:
     """Build the resmoke command, if a resmoke.py command wasn't passed in."""
     new_args = [sys.executable, "buildscripts/resmoke.py"]
@@ -521,54 +513,93 @@ def _get_task_runtime_history(evg_api: Optional[EvergreenApi], project: str, tas
             raise
 
 
-def create_generate_tasks_config(evg_config: Configuration, tests_by_task: Dict,
+def _create_task(index: int, test_count: int, test: str, task_data: Dict,
+                 task_runtime_stats: List[TestStats], generate_config: GenerateConfig,
+                 repeat_config: RepeatConfig, task_prefix: str) -> Task:
+    # pylint: disable=too-many-arguments,too-many-locals
+    """
+    Create the described shrub sub task.
+
+    :param index: Index of task being created.
+    :param test_count: Total number of testing being created.
+    :param test: Test task is being generated for.
+    :param task_data: Data about task to create.
+    :param task_runtime_stats: Historical runtime of test.
+    :param generate_config: Configuration of how to generate the task.
+    :param repeat_config: Configuration of how the task should be repeated.
+    :param task_prefix: String to prefix generated task with.
+    :return: Shrub task for given configuration.
+    """
+    # TODO: Extract multiversion related code into separate tooling - SERVER-47137
+    multiversion_path = task_data.get("use_multiversion")
+    display_task_name = task_data["display_task_name"]
+    resmoke_args = task_data["resmoke_args"]
+    sub_task_name = name_generated_task(f"{task_prefix}:{display_task_name}", index, test_count,
+                                        generate_config.run_build_variant)
+    LOGGER.debug("Generating sub-task", sub_task=sub_task_name)
+
+    test_unix_style = test.replace('\\', '/')
+    run_tests_vars = {
+        "resmoke_args":
+            f"{resmoke_args} {repeat_config.generate_resmoke_options()} {test_unix_style}"
+    }
+    if multiversion_path:
+        run_tests_vars["task_path_suffix"] = multiversion_path
+    timeout = _generate_timeouts(repeat_config, test, task_runtime_stats)
+    commands = resmoke_commands("run tests", run_tests_vars, timeout, multiversion_path)
+    dependencies = {TaskDependency("compile")}
+
+    return Task(sub_task_name, commands, dependencies)
+
+
+def create_generated_tasks(tests_by_task: Dict, task_prefix: str, generate_config: GenerateConfig,
+                           repeat_config: RepeatConfig, evg_api: EvergreenApi) -> Set[Task]:
+    """
+    Create the set of tasks to run the given tests_by_task.
+
+    :param tests_by_task: Dictionary of tests to generate tasks for.
+    :param task_prefix: Prefix all task names with this.
+    :param generate_config: Configuration of what to generate.
+    :param repeat_config: Configuration of how to repeat tests.
+    :param evg_api: Evergreen API.
+    :return: Set of shrub tasks to run tests_by_task.
+    """
+    tasks: Set[Task] = set()
+    for task in sorted(tests_by_task):
+        task_info = tests_by_task[task]
+        test_list = task_info["tests"]
+        task_runtime_stats = _get_task_runtime_history(evg_api, generate_config.project,
+                                                       task_info["display_task_name"],
+                                                       generate_config.build_variant)
+        test_count = len(test_list)
+        for index, test in enumerate(test_list):
+            tasks.add(
+                _create_task(index, test_count, test, task_info, task_runtime_stats,
+                             generate_config, repeat_config, task_prefix))
+
+    return tasks
+
+
+def create_generate_tasks_config(build_variant: BuildVariant, tests_by_task: Dict,
                                  generate_config: GenerateConfig, repeat_config: RepeatConfig,
                                  evg_api: Optional[EvergreenApi], include_gen_task: bool = True,
-                                 task_prefix: str = "burn_in") -> Configuration:
+                                 task_prefix: str = "burn_in") -> None:
     # pylint: disable=too-many-arguments,too-many-locals
     """
     Create the config for the Evergreen generate.tasks file.
 
-    :param evg_config: Shrub configuration to add to.
+    :param build_variant: Shrub configuration to add to.
     :param tests_by_task: Dictionary of tests to generate tasks for.
     :param generate_config: Configuration of what to generate.
     :param repeat_config: Configuration of how to repeat tests.
     :param evg_api: Evergreen API.
     :param include_gen_task: Should generating task be include in display task.
     :param task_prefix: Prefix all task names with this.
-    :return: Shrub configuration with added tasks.
     """
-    task_list = TaskList(evg_config)
-    resmoke_options = repeat_config.generate_resmoke_options()
-    for task in sorted(tests_by_task):
-        test_list = tests_by_task[task]["tests"]
-        for index, test in enumerate(test_list):
-            # TODO: Extract multiversion related code into separate tooling - SERVER-47137
-            multiversion_path = tests_by_task[task].get("use_multiversion")
-            display_task_name = tests_by_task[task]["display_task_name"]
-            task_runtime_stats = _get_task_runtime_history(
-                evg_api, generate_config.project, display_task_name, generate_config.build_variant)
-            resmoke_args = tests_by_task[task]["resmoke_args"]
-            distro = tests_by_task[task].get("distro", generate_config.distro)
-            # Evergreen always uses a unix shell, even on Windows, so instead of using os.path.join
-            # here, just use the forward slash; otherwise the path separator will be treated as
-            # the escape character on Windows.
-            sub_task_name = name_generated_task(f"{task_prefix}:{display_task_name}", index,
-                                                len(test_list), generate_config.run_build_variant)
-            LOGGER.debug("Generating sub-task", sub_task=sub_task_name)
-
-            test_unix_style = test.replace('\\', '/')
-            run_tests_vars = {"resmoke_args": f"{resmoke_args} {resmoke_options} {test_unix_style}"}
-            if multiversion_path:
-                run_tests_vars["task_path_suffix"] = multiversion_path
-            timeout = _generate_timeouts(repeat_config, test, task_runtime_stats)
-            commands = resmoke_commands("run tests", run_tests_vars, timeout, multiversion_path)
-
-            task_list.add_task(sub_task_name, commands, ["compile"], distro)
-
-    existing_tasks = [BURN_IN_TESTS_GEN_TASK] if include_gen_task else None
-    task_list.add_to_variant(generate_config.run_build_variant, BURN_IN_TESTS_TASK, existing_tasks)
-    return evg_config
+    tasks = create_generated_tasks(tests_by_task, task_prefix, generate_config, repeat_config,
+                                   evg_api)
+    existing_tasks = {ExistingTask(BURN_IN_TESTS_GEN_TASK)} if include_gen_task else None
+    build_variant.display_task(BURN_IN_TESTS_TASK, tasks, execution_existing_tasks=existing_tasks)
 
 
 def create_task_list_for_tests(
@@ -625,7 +656,7 @@ def create_tests_by_task(build_variant: str, repos: Iterable[Repo],
 # pylint: disable=too-many-arguments
 def create_generate_tasks_file(tests_by_task: Dict, generate_config: GenerateConfig,
                                repeat_config: RepeatConfig, evg_api: Optional[EvergreenApi],
-                               task_prefix: str = 'burn_in', include_gen_task: bool = True) -> Dict:
+                               task_prefix: str = 'burn_in', include_gen_task: bool = True) -> str:
     """
     Create an Evergreen generate.tasks file to run the given tasks and tests.
 
@@ -637,18 +668,18 @@ def create_generate_tasks_file(tests_by_task: Dict, generate_config: GenerateCon
     :param include_gen_task: Should the generating task be included in the display task.
     :returns: Configuration to pass to 'generate.tasks'.
     """
-    evg_config = Configuration()
-    evg_config = create_generate_tasks_config(
-        evg_config, tests_by_task, generate_config, repeat_config, evg_api,
-        include_gen_task=include_gen_task, task_prefix=task_prefix)
+    build_variant = BuildVariant(generate_config.run_build_variant)
+    create_generate_tasks_config(build_variant, tests_by_task, generate_config, repeat_config,
+                                 evg_api, include_gen_task=include_gen_task,
+                                 task_prefix=task_prefix)
+
+    shrub_project = ShrubProject.empty()
+    shrub_project.add_build_variant(build_variant)
 
-    json_config = evg_config.to_map()
-    tasks_to_create = len(json_config.get('tasks', []))
-    if tasks_to_create > MAX_TASKS_TO_CREATE:
-        LOGGER.warning("Attempting to create more tasks than max, aborting", tasks=tasks_to_create,
-                       max=MAX_TASKS_TO_CREATE)
+    if not validate_task_generation_limit(shrub_project):
         sys.exit(1)
-    return json_config
+
+    return shrub_project.json()
 
 
 def run_tests(tests_by_task: Dict, resmoke_cmd: [str]):
@@ -704,7 +735,7 @@ def _get_evg_api(evg_api_config: str, local_mode: bool) -> Optional[EvergreenApi
 
 def burn_in(repeat_config: RepeatConfig, generate_config: GenerateConfig, resmoke_args: str,
             generate_tasks_file: str, no_exec: bool, evg_conf: EvergreenProjectConfig,
-            repos: Iterable[Repo], evg_api: EvergreenApi):
+            repos: Iterable[Repo], evg_api: EvergreenApi) -> None:
     """
     Run burn_in_tests with the given configuration.
 
@@ -724,9 +755,9 @@ def burn_in(repeat_config: RepeatConfig, generate_config: GenerateConfig, resmok
     LOGGER.debug("tests and tasks found", tests_by_task=tests_by_task)
 
     if generate_tasks_file:
-        json_config = create_generate_tasks_file(tests_by_task, generate_config, repeat_config,
-                                                 evg_api)
-        _write_json_file(json_config, generate_tasks_file)
+        json_text = create_generate_tasks_file(tests_by_task, generate_config, repeat_config,
+                                               evg_api)
+        write_file(generate_tasks_file, json_text)
     elif not no_exec:
         run_tests(tests_by_task, resmoke_cmd)
     else:
diff --git a/buildscripts/burn_in_tests_multiversion.py b/buildscripts/burn_in_tests_multiversion.py
index 4388ca07710..519be44f831 100644
--- a/buildscripts/burn_in_tests_multiversion.py
+++ b/buildscripts/burn_in_tests_multiversion.py
@@ -8,8 +8,7 @@ from typing import Dict
 import click
 from evergreen.api import EvergreenApi
 from git import Repo
-from shrub.config import Configuration
-from shrub.variant import DisplayTaskDefinition
+from shrub.v2 import BuildVariant, ExistingTask, ShrubProject
 import structlog
 from structlog.stdlib import LoggerFactory
 
@@ -17,9 +16,11 @@ import buildscripts.evergreen_gen_multiversion_tests as gen_multiversion
 import buildscripts.evergreen_generate_resmoke_tasks as gen_resmoke
 from buildscripts.burn_in_tests import GenerateConfig, DEFAULT_PROJECT, CONFIG_FILE, _configure_logging, RepeatConfig, \
     _get_evg_api, EVERGREEN_FILE, DEFAULT_REPO_LOCATIONS, _set_resmoke_cmd, create_tests_by_task, \
-    _write_json_file, run_tests, MAX_TASKS_TO_CREATE
+    run_tests
 from buildscripts.ciconfig.evergreen import parse_evergreen_file
+from buildscripts.patch_builds.task_generation import validate_task_generation_limit
 from buildscripts.resmokelib.suitesconfig import get_named_suites_with_root_level_key
+from buildscripts.util.fileops import write_file
 
 structlog.configure(logger_factory=LoggerFactory())
 LOGGER = structlog.getLogger(__name__)
@@ -31,21 +32,18 @@ BURN_IN_MULTIVERSION_TASK = gen_multiversion.BURN_IN_TASK
 TASK_PATH_SUFFIX = "/data/multiversion"
 
 
-def create_multiversion_generate_tasks_config(evg_config: Configuration, tests_by_task: Dict,
-                                              evg_api: EvergreenApi,
-                                              generate_config: GenerateConfig) -> Configuration:
+def create_multiversion_generate_tasks_config(tests_by_task: Dict, evg_api: EvergreenApi,
+                                              generate_config: GenerateConfig) -> BuildVariant:
     """
     Create the multiversion config for the Evergreen generate.tasks file.
 
-    :param evg_config: Shrub configuration to add to.
     :param tests_by_task: Dictionary of tests to generate tasks for.
     :param evg_api: Evergreen API.
     :param generate_config: Configuration of what to generate.
     :return: Shrub configuration with added tasks.
     """
-
-    dt = DisplayTaskDefinition(BURN_IN_MULTIVERSION_TASK)
-
+    build_variant = BuildVariant(generate_config.build_variant)
+    tasks = set()
     if tests_by_task:
         # Get the multiversion suites that will run in as part of burn_in_multiversion.
         multiversion_suites = get_named_suites_with_root_level_key(MULTIVERSION_CONFIG_KEY)
@@ -72,8 +70,8 @@ def create_multiversion_generate_tasks_config(evg_config: Configuration, tests_b
             }
             config_options.update(gen_resmoke.DEFAULT_CONFIG_VALUES)
 
-            config_generator = gen_multiversion.EvergreenConfigGenerator(
-                evg_api, evg_config, gen_resmoke.ConfigOptions(config_options))
+            config_generator = gen_multiversion.EvergreenMultiversionConfigGenerator(
+                evg_api, gen_resmoke.ConfigOptions(config_options))
             test_list = tests_by_task[suite["origin"]]["tests"]
             for test in test_list:
                 # Exclude files that should be blacklisted from multiversion testing.
@@ -83,14 +81,14 @@ def create_multiversion_generate_tasks_config(evg_config: Configuration, tests_b
                              suite=suite["multiversion_name"])
                 if test not in files_to_exclude:
                     # Generate the multiversion tasks for each test.
-                    config_generator.generate_evg_tasks(test, idx)
+                    sub_tasks = config_generator.get_burn_in_tasks(test, idx)
+                    tasks = tasks.union(sub_tasks)
                     idx += 1
-            dt.execution_tasks(config_generator.task_names)
-            evg_config.variant(generate_config.build_variant).tasks(config_generator.task_specs)
 
-    dt.execution_task(f"{BURN_IN_MULTIVERSION_TASK}_gen")
-    evg_config.variant(generate_config.build_variant).display_task(dt)
-    return evg_config
+    existing_tasks = {ExistingTask(f"{BURN_IN_MULTIVERSION_TASK}_gen")}
+    build_variant.display_task(BURN_IN_MULTIVERSION_TASK, tasks,
+                               execution_existing_tasks=existing_tasks)
+    return build_variant
 
 
 @click.command()
@@ -180,17 +178,16 @@ def main(build_variant, run_build_variant, distro, project, generate_tasks_file,
         # MULTIVERSION_CONFIG_KEY as a root level key and must be set to true.
         multiversion_suites = get_named_suites_with_root_level_key(MULTIVERSION_CONFIG_KEY)
         assert len(multiversion_tasks) == len(multiversion_suites)
-        evg_config = Configuration()
-        evg_config = create_multiversion_generate_tasks_config(evg_config, tests_by_task, evg_api,
-                                                               generate_config)
-
-        json_config = evg_config.to_map()
-        tasks_to_create = len(json_config.get('tasks', []))
-        if tasks_to_create > MAX_TASKS_TO_CREATE:
-            LOGGER.warning("Attempting to create more tasks than max, aborting",
-                           tasks=tasks_to_create, max=MAX_TASKS_TO_CREATE)
+
+        build_variant = create_multiversion_generate_tasks_config(tests_by_task, evg_api,
+                                                                  generate_config)
+        shrub_project = ShrubProject.empty()
+        shrub_project.add_build_variant(build_variant)
+
+        if not validate_task_generation_limit(shrub_project):
             sys.exit(1)
-        _write_json_file(json_config, generate_tasks_file)
+
+        write_file(generate_tasks_file, shrub_project.json())
     elif not no_exec:
         run_tests(tests_by_task, resmoke_cmd)
     else:
diff --git a/buildscripts/ciconfig/evergreen.py b/buildscripts/ciconfig/evergreen.py
index 2791b62206e..4731525290d 100644
--- a/buildscripts/ciconfig/evergreen.py
+++ b/buildscripts/ciconfig/evergreen.py
@@ -3,6 +3,7 @@
 The API also provides methods to access specific fields present in the mongodb/mongo
 configuration file.
 """
+from __future__ import annotations
 
 import datetime
 import distutils.spawn  # pylint: disable=no-name-in-module
@@ -79,7 +80,7 @@ class EvergreenProjectConfig(object):  # pylint: disable=too-many-instance-attri
         """Get the list of build variant names."""
         return list(self._variants_by_name.keys())
 
-    def get_variant(self, variant_name):
+    def get_variant(self, variant_name: str) -> Variant:
         """Return the variant with the given name as a Variant instance."""
         return self._variants_by_name.get(variant_name)
 
diff --git a/buildscripts/evergreen_gen_fuzzer_tests.py b/buildscripts/evergreen_gen_fuzzer_tests.py
index 276cff5623d..eb750149348 100755
--- a/buildscripts/evergreen_gen_fuzzer_tests.py
+++ b/buildscripts/evergreen_gen_fuzzer_tests.py
@@ -1,19 +1,12 @@
 #!/usr/bin/env python3
 """Generate fuzzer tests to run in evergreen in parallel."""
-
 import argparse
-import math
-import os
-
 from collections import namedtuple
+from typing import Set
 
-from shrub.config import Configuration
-from shrub.command import CommandDefinition
-from shrub.task import TaskDependency
-from shrub.variant import DisplayTaskDefinition
-from shrub.variant import TaskSpec
+from shrub.v2 import ShrubProject, FunctionCall, Task, TaskDependency, BuildVariant, ExistingTask
 
-import buildscripts.evergreen_generate_resmoke_tasks as generate_resmoke
+from buildscripts.util.fileops import write_file_to_dir
 import buildscripts.util.read_config as read_config
 import buildscripts.util.taskname as taskname
 
@@ -81,76 +74,74 @@ def _get_config_options(cmd_line_options, config_file):  # pylint: disable=too-m
                          timeout_secs, use_multiversion, suite)
 
 
-def _name_task(parent_name, task_index, total_tasks):
+def build_fuzzer_sub_task(task_name: str, task_index: int, options: ConfigOptions) -> Task:
     """
-    Create a zero-padded sub-task name.
+    Build a shrub task to run the fuzzer.
 
-    :param parent_name: Name of the parent task.
-    :param task_index: Index of this sub-task.
-    :param total_tasks: Total number of sub-tasks being generated.
-    :return: Zero-padded name of sub-task.
+    :param task_name: Parent name of task.
+    :param task_index: Index of sub task being generated.
+    :param options: Options to use for task.
+    :return: Shrub task to run the fuzzer.
     """
-    index_width = int(math.ceil(math.log10(total_tasks)))
-    return "{0}_{1}".format(parent_name, str(task_index).zfill(index_width))
+    sub_task_name = taskname.name_generated_task(task_name, task_index, options.num_tasks,
+                                                 options.variant)
+
+    run_jstestfuzz_vars = {
+        "jstestfuzz_vars":
+            "--numGeneratedFiles {0} {1}".format(options.num_files, options.jstestfuzz_vars),
+        "npm_command":
+            options.npm_command,
+    }
+    suite_arg = f"--suites={options.suite}"
+    run_tests_vars = {
+        "continue_on_failure": options.continue_on_failure,
+        "resmoke_args": f"{suite_arg} {options.resmoke_args}",
+        "resmoke_jobs_max": options.resmoke_jobs_max,
+        "should_shuffle": options.should_shuffle,
+        "task_path_suffix": options.use_multiversion,
+        "timeout_secs": options.timeout_secs,
+        "task": options.name
+    }  # yapf: disable
+
+    commands = [
+        FunctionCall("do setup"),
+        FunctionCall("do multiversion setup") if options.use_multiversion else None,
+        FunctionCall("setup jstestfuzz"),
+        FunctionCall("run jstestfuzz", run_jstestfuzz_vars),
+        FunctionCall("run generated tests", run_tests_vars)
+    ]
+    commands = [command for command in commands if command is not None]
+
+    return Task(sub_task_name, commands, {TaskDependency("compile")})
+
+
+def generate_fuzzer_sub_tasks(task_name: str, options: ConfigOptions) -> Set[Task]:
+    """
+    Generate evergreen tasks for fuzzers based on the options given.
+
+    :param task_name: Parent name for tasks being generated.
+    :param options: task options.
+    :return: Set of shrub tasks.
+    """
+    sub_tasks = {
+        build_fuzzer_sub_task(task_name, index, options)
+        for index in range(options.num_tasks)
+    }
+    return sub_tasks
 
 
-def generate_evg_tasks(options, evg_config, task_name_suffix=None, display_task=None):
+def create_fuzzer_task(options: ConfigOptions, build_variant: BuildVariant) -> None:
     """
-    Generate an evergreen configuration for fuzzers based on the options given.
+    Generate an evergreen configuration for fuzzers and add it to the given build variant.
 
     :param options: task options.
-    :param evg_config: evergreen configuration.
-    :param task_name_suffix: suffix to be appended to each task name.
-    :param display_task: an existing display task definition to append to.
-    :return: An evergreen configuration.
+    :param build_variant: Build variant to add tasks to.
     """
-    task_names = []
-    task_specs = []
-
-    for task_index in range(options.num_tasks):
-        task_name = options.name if not task_name_suffix else f"{options.name}_{task_name_suffix}"
-        name = taskname.name_generated_task(task_name, task_index, options.num_tasks,
-                                            options.variant)
-        task_names.append(name)
-        task_specs.append(TaskSpec(name))
-        task = evg_config.task(name)
-
-        commands = [CommandDefinition().function("do setup")]
-        if options.use_multiversion:
-            commands.append(CommandDefinition().function("do multiversion setup"))
-
-        commands.append(CommandDefinition().function("setup jstestfuzz"))
-        commands.append(CommandDefinition().function("run jstestfuzz").vars({
-            "jstestfuzz_vars":
-                "--numGeneratedFiles {0} {1}".format(options.num_files, options.jstestfuzz_vars),
-            "npm_command":
-                options.npm_command
-        }))
-        # Unix path separators are used because Evergreen only runs this script in unix shells,
-        # even on Windows.
-        suite_arg = f"--suites={options.suite}"
-        run_tests_vars = {
-            "continue_on_failure": options.continue_on_failure,
-            "resmoke_args": f"{suite_arg} {options.resmoke_args}",
-            "resmoke_jobs_max": options.resmoke_jobs_max,
-            "should_shuffle": options.should_shuffle,
-            "task_path_suffix": options.use_multiversion,
-            "timeout_secs": options.timeout_secs,
-            "task": options.name
-        }  # yapf: disable
-
-        commands.append(CommandDefinition().function("run generated tests").vars(run_tests_vars))
-        task.dependency(TaskDependency("compile")).commands(commands)
-
-    # Create a new DisplayTaskDefinition or append to the one passed in.
-    dt = DisplayTaskDefinition(task_name) if not display_task else display_task
-    dt.execution_tasks(task_names)
-    evg_config.variant(options.variant).tasks(task_specs)
-    if not display_task:
-        dt.execution_task("{0}_gen".format(options.name))
-        evg_config.variant(options.variant).display_task(dt)
-
-    return evg_config
+    task_name = options.name
+    sub_tasks = generate_fuzzer_sub_tasks(task_name, options)
+
+    build_variant.display_task(task_name, sub_tasks,
+                               execution_existing_tasks={ExistingTask(f"{options.name}_gen")})
 
 
 def main():
@@ -184,14 +175,13 @@ def main():
     options = parser.parse_args()
 
     config_options = _get_config_options(options, options.expansion_file)
-    evg_config = Configuration()
-    generate_evg_tasks(config_options, evg_config)
+    build_variant = BuildVariant(config_options.variant)
+    create_fuzzer_task(config_options, build_variant)
 
-    if not os.path.exists(CONFIG_DIRECTORY):
-        os.makedirs(CONFIG_DIRECTORY)
+    shrub_project = ShrubProject.empty()
+    shrub_project.add_build_variant(build_variant)
 
-    with open(os.path.join(CONFIG_DIRECTORY, config_options.name + ".json"), "w") as file_handle:
-        file_handle.write(evg_config.to_json())
+    write_file_to_dir(CONFIG_DIRECTORY, f"{config_options.name}.json", shrub_project.json())
 
 
 if __name__ == '__main__':
diff --git a/buildscripts/evergreen_gen_multiversion_tests.py b/buildscripts/evergreen_gen_multiversion_tests.py
index cd077662ce4..952488e1584 100755
--- a/buildscripts/evergreen_gen_multiversion_tests.py
+++ b/buildscripts/evergreen_gen_multiversion_tests.py
@@ -2,35 +2,29 @@
 """Generate multiversion tests to run in evergreen in parallel."""
 
 import datetime
-from datetime import timedelta
 import logging
 import os
 import sys
 import tempfile
+from typing import Optional, List, Set
 
-from collections import namedtuple
 from subprocess import check_output
 
 import requests
-import yaml
 import click
 import structlog
 
-from evergreen.api import RetryingEvergreenApi
-from git import Repo
-from shrub.config import Configuration
-from shrub.command import CommandDefinition
-from shrub.task import TaskDependency
-from shrub.variant import DisplayTaskDefinition
-from shrub.variant import TaskSpec
+from evergreen.api import RetryingEvergreenApi, EvergreenApi
+from shrub.v2 import ShrubProject, FunctionCall, Task, TaskDependency, BuildVariant, ExistingTask
 
 from buildscripts.resmokelib import config as _config
 from buildscripts.resmokelib.multiversionconstants import (LAST_STABLE_MONGO_BINARY,
                                                            REQUIRES_FCV_TAG)
 import buildscripts.resmokelib.parser
-import buildscripts.util.read_config as read_config
 import buildscripts.util.taskname as taskname
+from buildscripts.util.fileops import write_file_to_dir
 import buildscripts.evergreen_generate_resmoke_tasks as generate_resmoke
+from buildscripts.evergreen_generate_resmoke_tasks import Suite, ConfigOptions
 import buildscripts.evergreen_gen_fuzzer_tests as gen_fuzzer
 
 LOGGER = structlog.getLogger(__name__)
@@ -71,26 +65,27 @@ def enable_logging():
     structlog.configure(logger_factory=structlog.stdlib.LoggerFactory())
 
 
-def prepare_directory_for_suite(directory):
-    """Ensure that directory exists."""
-    if not os.path.exists(directory):
-        os.makedirs(directory)
-
-
-def is_suite_sharded(suite_dir, suite_name):
+def is_suite_sharded(suite_dir: str, suite_name: str) -> bool:
     """Return true if a suite uses ShardedClusterFixture."""
     source_config = generate_resmoke.read_yaml(suite_dir, suite_name + ".yml")
     return source_config["executor"]["fixture"]["class"] == "ShardedClusterFixture"
 
 
-def get_multiversion_resmoke_args(is_sharded):
+def get_version_configs(is_sharded: bool) -> List[str]:
+    """Get the version configurations to use."""
+    if is_sharded:
+        return SHARDED_MIXED_VERSION_CONFIGS
+    return REPL_MIXED_VERSION_CONFIGS
+
+
+def get_multiversion_resmoke_args(is_sharded: bool) -> str:
     """Return resmoke args used to configure a cluster for multiversion testing."""
-    args_for_sharded_cluster = "--numShards=2 --numReplSetNodes=2 "
-    args_for_replset = "--numReplSetNodes=3 --linearChain=on "
-    return args_for_sharded_cluster if is_sharded else args_for_replset
+    if is_sharded:
+        return "--numShards=2 --numReplSetNodes=2 "
+    return "--numReplSetNodes=3 --linearChain=on "
 
 
-def get_backports_required_last_stable_hash(task_path_suffix):
+def get_backports_required_last_stable_hash(task_path_suffix: str):
     """Parse the last-stable shell binary to get the commit hash."""
     last_stable_shell_exec = os.path.join(task_path_suffix, LAST_STABLE_MONGO_BINARY)
     shell_version = check_output([last_stable_shell_exec, "--version"]).decode('utf-8')
@@ -150,86 +145,99 @@ def get_exclude_files(suite_name, task_path_suffix):
             elem["test_file"] for elem in latest_suite_yaml if elem not in last_stable_suite_yaml)
 
 
-class EvergreenConfigGenerator(object):
+def _generate_resmoke_args(suite_file: str, mixed_version_config: str, is_sharded: bool, options,
+                           burn_in_test: Optional[str]) -> str:
+    return (f"{options.resmoke_args} --suite={suite_file} --mixedBinVersions={mixed_version_config}"
+            f" --excludeWithAnyTags={EXCLUDE_TAGS} --originSuite={options.suite} "
+            f" {get_multiversion_resmoke_args(is_sharded)} {burn_in_test if burn_in_test else ''}")
+
+
+class EvergreenMultiversionConfigGenerator(object):
     """Generate evergreen configurations for multiversion tests."""
 
-    def __init__(self, evg_api, evg_config, options):
-        """Create new EvergreenConfigGenerator object."""
+    def __init__(self, evg_api: EvergreenApi, options):
+        """Create new EvergreenMultiversionConfigGenerator object."""
         self.evg_api = evg_api
-        self.evg_config = evg_config
         self.options = options
-        self.task_names = []
-        self.task_specs = []
         # Strip the "_gen" suffix appended to the name of tasks generated by evergreen.
         self.task = generate_resmoke.remove_gen_suffix(self.options.task)
 
-    def _generate_sub_task(self, mixed_version_config, task, task_index, suite, num_suites,
-                           is_sharded, burn_in_test=None):
+    def _generate_sub_task(self, mixed_version_config: str, task: str, task_index: int, suite: str,
+                           num_suites: int, is_sharded: bool,
+                           burn_in_test: Optional[str] = None) -> Task:
         # pylint: disable=too-many-arguments
-        """Generate a sub task to be run with the provided suite and  mixed version config."""
-
+        """
+        Generate a sub task to be run with the provided suite and  mixed version config.
+
+        :param mixed_version_config: mixed version configuration.
+        :param task: Name of task.
+        :param task_index: Index of task to generate.
+        :param suite: Name of suite being generated.
+        :param num_suites: Number os suites being generated.
+        :param is_sharded: If this is being generated for a sharded configuration.
+        :param burn_in_test: If generation is for burn_in, burn_in options to use.
+        :return: Shrub configuration for task specified.
+        """
         # Create a sub task name appended with the task_index and build variant name.
-        task_name = "{0}_{1}".format(task, mixed_version_config)
+        task_name = f"{task}_{mixed_version_config}"
         sub_task_name = taskname.name_generated_task(task_name, task_index, num_suites,
                                                      self.options.variant)
-        self.task_names.append(sub_task_name)
-        self.task_specs.append(TaskSpec(sub_task_name))
-        task = self.evg_config.task(sub_task_name)
-
         gen_task_name = BURN_IN_TASK if burn_in_test is not None else self.task
 
-        commands = [
-            CommandDefinition().function("do setup"),
-            # Fetch and download the proper mongod binaries before running multiversion tests.
-            CommandDefinition().function("do multiversion setup")
-        ]
         run_tests_vars = {
             "resmoke_args":
-                "{0} --suite={1} --mixedBinVersions={2} --excludeWithAnyTags={3} --originSuite={4} "
-                .format(self.options.resmoke_args, suite, mixed_version_config, EXCLUDE_TAGS,
-                        self.options.suite),
+                _generate_resmoke_args(suite, mixed_version_config, is_sharded, self.options,
+                                       burn_in_test),
             "task":
                 gen_task_name,
         }
-        # Update the resmoke args to configure the cluster for multiversion testing.
-        run_tests_vars["resmoke_args"] += get_multiversion_resmoke_args(is_sharded)
-
-        if burn_in_test is not None:
-            run_tests_vars["resmoke_args"] += burn_in_test
-
-        commands.append(CommandDefinition().function("run generated tests").vars(run_tests_vars))
-        task.dependency(TaskDependency("compile")).commands(commands)
 
-    def _write_evergreen_config_to_file(self, task_name):
-        """Save evergreen config to file."""
-        if not os.path.exists(CONFIG_DIR):
-            os.makedirs(CONFIG_DIR)
-
-        with open(os.path.join(CONFIG_DIR, task_name + ".json"), "w") as file_handle:
-            file_handle.write(self.evg_config.to_json())
+        commands = [
+            FunctionCall("do setup"),
+            # Fetch and download the proper mongod binaries before running multiversion tests.
+            FunctionCall("do multiversion setup"),
+            FunctionCall("run generated tests", run_tests_vars),
+        ]
 
-    def create_display_task(self, task_name, task_specs, task_list):
-        """Create the display task definition for the MultiversionConfig object."""
-        dt = DisplayTaskDefinition(task_name).execution_tasks(task_list)\
-            .execution_task("{0}_gen".format(task_name))
-        self.evg_config.variant(self.options.variant).tasks(task_specs).display_task(dt)
+        return Task(sub_task_name, commands, {TaskDependency("compile")})
 
-    def _generate_burn_in_execution_tasks(self, version_configs, suites, burn_in_test, burn_in_idx,
-                                          is_sharded):
+    def _generate_burn_in_execution_tasks(self, version_configs: List[str], suites: List[Suite],
+                                          burn_in_test: str, burn_in_idx: int,
+                                          is_sharded: bool) -> Set[Task]:
+        """
+        Generate shrub tasks for burn_in executions.
+
+        :param version_configs: Version configs to generate for.
+        :param suites: Suites to generate.
+        :param burn_in_test: burn_in_test configuration.
+        :param burn_in_idx: Index of burn_in task being generated.
+        :param is_sharded: If configuration should be generated for sharding tests.
+        :return: Set of generated shrub tasks.
+        """
         # pylint: disable=too-many-arguments
         burn_in_prefix = "burn_in_multiversion"
-        task = "{0}:{1}".format(burn_in_prefix, self.task)
+        task = f"{burn_in_prefix}:{self.task}"
 
-        for version_config in version_configs:
-            # For burn in tasks, it doesn't matter which generated suite yml to use as all the
-            # yaml configurations are the same.
-            source_suite = os.path.join(CONFIG_DIR, suites[0].name + ".yml")
+        # For burn in tasks, it doesn't matter which generated suite yml to use as all the
+        # yaml configurations are the same.
+        source_suite = os.path.join(CONFIG_DIR, suites[0].name + ".yml")
+        tasks = {
             self._generate_sub_task(version_config, task, burn_in_idx, source_suite, 1, is_sharded,
                                     burn_in_test)
-        return self.evg_config
+            for version_config in version_configs
+        }
 
-    def _get_fuzzer_options(self, version_config, is_sharded):
-        fuzzer_config = generate_resmoke.ConfigOptions(self.options.config)
+        return tasks
+
+    def _get_fuzzer_options(self, version_config: str, is_sharded: bool) -> ConfigOptions:
+        """
+        Get options to generate fuzzer tasks.
+
+        :param version_config: Version configuration to generate for.
+        :param is_sharded: If configuration is for sharded tests.
+        :return: Configuration options to generate fuzzer tasks.
+        """
+        fuzzer_config = ConfigOptions(self.options.config)
         fuzzer_config.name = f"{self.options.suite}_multiversion"
         fuzzer_config.num_files = int(self.options.num_files)
         fuzzer_config.num_tasks = int(self.options.num_tasks)
@@ -238,37 +246,27 @@ class EvergreenConfigGenerator(object):
             f"--mixedBinVersions={version_config} {add_resmoke_args}"
         return fuzzer_config
 
-    def _generate_fuzzer_tasks(self, version_configs, is_sharded):
-        dt = DisplayTaskDefinition(self.task)
-
-        for version_config in version_configs:
-            fuzzer_config = generate_resmoke.ConfigOptions(self.options.config)
-            fuzzer_config = self._get_fuzzer_options(version_config, is_sharded)
-            gen_fuzzer.generate_evg_tasks(fuzzer_config, self.evg_config,
-                                          task_name_suffix=version_config, display_task=dt)
-        dt.execution_task(f"{fuzzer_config.name}_gen")
-        self.evg_config.variant(self.options.variant).display_task(dt)
-        return self.evg_config
-
-    def generate_evg_tasks(self, burn_in_test=None, burn_in_idx=0):
-        # pylint: disable=too-many-locals
+    def _generate_fuzzer_tasks(self, build_variant: BuildVariant, version_configs: List[str],
+                               is_sharded: bool) -> None:
         """
-        Generate evergreen tasks for multiversion tests.
-
-        The number of tasks generated equals
-        (the number of version configs) * (the number of generated suites).
+        Generate fuzzer tasks and add them to the given build variant.
 
-        :param burn_in_test: The test to be run as part of the burn in multiversion suite.
+        :param build_variant: Build variant to add tasks to.
+        :param version_configs: Version configurations to generate.
+        :param is_sharded: Should configuration be generated for sharding.
         """
-        is_sharded = is_suite_sharded(TEST_SUITE_DIR, self.options.suite)
-        if is_sharded:
-            version_configs = SHARDED_MIXED_VERSION_CONFIGS
-        else:
-            version_configs = REPL_MIXED_VERSION_CONFIGS
+        tasks = set()
+        for version_config in version_configs:
+            fuzzer_config = self._get_fuzzer_options(version_config, is_sharded)
+            task_name = f"{fuzzer_config.name}_{version_config}"
+            sub_tasks = gen_fuzzer.generate_fuzzer_sub_tasks(task_name, fuzzer_config)
+            tasks = tasks.union(sub_tasks)
 
-        if self.options.is_jstestfuzz:
-            return self._generate_fuzzer_tasks(version_configs, is_sharded)
+        existing_tasks = {ExistingTask(f"{self.options.suite}_multiversion_gen")}
+        build_variant.display_task(self.task, tasks, execution_existing_tasks=existing_tasks)
 
+    def generate_resmoke_suites(self) -> List[Suite]:
+        """Generate the resmoke configuration files for this generator."""
         # Divide tests into suites based on run-time statistics for the last
         # LOOKBACK_DURATION_DAYS. Tests without enough run-time statistics will be placed
         # in the misc suite.
@@ -282,39 +280,77 @@ class EvergreenConfigGenerator(object):
                                                                self.options.create_misc_suite)
         generate_resmoke.write_file_dict(CONFIG_DIR, config_file_dict)
 
-        if burn_in_test is not None:
-            # Generate the subtasks to run burn_in_test against the appropriate mixed version
-            # configurations. The display task is defined later as part of generating the burn
-            # in tests.
-            self._generate_burn_in_execution_tasks(version_configs, suites, burn_in_test,
-                                                   burn_in_idx, is_sharded)
-            return self.evg_config
+        return suites
+
+    def get_burn_in_tasks(self, burn_in_test: str, burn_in_idx: int) -> Set[Task]:
+        """
+        Get the burn_in tasks being generated.
+
+        :param burn_in_test: Burn in test configuration.
+        :param burn_in_idx: Index of burn_in configuration being generated.
+        :return: Set of shrub tasks for the specified burn_in.
+        """
+        is_sharded = is_suite_sharded(TEST_SUITE_DIR, self.options.suite)
+        version_configs = get_version_configs(is_sharded)
+        suites = self.generate_resmoke_suites()
+
+        # Generate the subtasks to run burn_in_test against the appropriate mixed version
+        # configurations. The display task is defined later as part of generating the burn
+        # in tests.
+        tasks = self._generate_burn_in_execution_tasks(version_configs, suites, burn_in_test,
+                                                       burn_in_idx, is_sharded)
+        return tasks
+
+    def generate_evg_tasks(self, build_variant: BuildVariant) -> None:
+        # pylint: disable=too-many-locals
+        """
+        Generate evergreen tasks for multiversion tests.
+
+        The number of tasks generated equals
+        (the number of version configs) * (the number of generated suites).
 
+        :param build_variant: Build variant to add generated configuration to.
+        """
+        is_sharded = is_suite_sharded(TEST_SUITE_DIR, self.options.suite)
+        version_configs = get_version_configs(is_sharded)
+
+        if self.options.is_jstestfuzz:
+            self._generate_fuzzer_tasks(build_variant, version_configs, is_sharded)
+
+        suites = self.generate_resmoke_suites()
+        sub_tasks = set()
         for version_config in version_configs:
             idx = 0
             for suite in suites:
                 # Generate the newly divided test suites
                 source_suite = os.path.join(CONFIG_DIR, suite.name + ".yml")
-                self._generate_sub_task(version_config, self.task, idx, source_suite, len(suites),
-                                        is_sharded)
+                sub_tasks.add(
+                    self._generate_sub_task(version_config, self.task, idx, source_suite,
+                                            len(suites), is_sharded))
                 idx += 1
 
             # Also generate the misc task.
             misc_suite_name = "{0}_misc".format(self.options.suite)
             misc_suite = os.path.join(CONFIG_DIR, misc_suite_name + ".yml")
-            self._generate_sub_task(version_config, self.task, idx, misc_suite, 1, is_sharded)
+            sub_tasks.add(
+                self._generate_sub_task(version_config, self.task, idx, misc_suite, 1, is_sharded))
             idx += 1
-        self.create_display_task(self.task, self.task_specs, self.task_names)
-        return self.evg_config
 
-    def run(self):
-        """Generate and run multiversion suites that run within a specified target execution time."""
+        build_variant.display_task(self.task, sub_tasks,
+                                   execution_existing_tasks={ExistingTask(f"{self.task}_gen")})
+
+    def run(self) -> None:
+        """Generate multiversion suites that run within a specified target execution time."""
         if not generate_resmoke.should_tasks_be_generated(self.evg_api, self.options.task_id):
             LOGGER.info("Not generating configuration due to previous successful generation.")
             return
 
-        self.generate_evg_tasks()
-        self._write_evergreen_config_to_file(self.task)
+        build_variant = BuildVariant(self.options.variant)
+        self.generate_evg_tasks(build_variant)
+
+        shrub_project = ShrubProject.empty()
+        shrub_project.add_build_variant(build_variant)
+        write_file_to_dir(CONFIG_DIR, f"{self.task}.json", shrub_project.json())
 
 
 @click.group()
@@ -328,7 +364,7 @@ def main():
               help="Location of expansions file generated by evergreen.")
 @click.option("--evergreen-config", type=str, default=CONFIG_FILE,
               help="Location of evergreen configuration file.")
-def run_generate_tasks(expansion_file, evergreen_config=None):
+def run_generate_tasks(expansion_file: str, evergreen_config: Optional[str] = None):
     """
     Create a configuration for generate tasks to create sub suites for the specified resmoke suite.
 
@@ -343,11 +379,9 @@ def run_generate_tasks(expansion_file, evergreen_config=None):
     :param evergreen_config: Evergreen configuration file.
     """
     evg_api = RetryingEvergreenApi.get_api(config_file=evergreen_config)
-    prepare_directory_for_suite(CONFIG_DIR)
-    evg_config = Configuration()
     config_options = generate_resmoke.ConfigOptions.from_file(
         expansion_file, REQUIRED_CONFIG_KEYS, DEFAULT_CONFIG_VALUES, CONFIG_FORMAT_FN)
-    config_generator = EvergreenConfigGenerator(evg_api, evg_config, config_options)
+    config_generator = EvergreenMultiversionConfigGenerator(evg_api, config_options)
     config_generator.run()
 
 
@@ -358,7 +392,7 @@ def run_generate_tasks(expansion_file, evergreen_config=None):
               help="The directory in which multiversion binaries are stored.")
 @click.option("--is-generated-suite", type=bool, required=True,
               help="Indicates whether the suite yaml to update is generated or static.")
-def generate_exclude_yaml(suite, task_path_suffix, is_generated_suite):
+def generate_exclude_yaml(suite: str, task_path_suffix: str, is_generated_suite: bool) -> None:
     # pylint: disable=too-many-locals
     """
     Update the given multiversion suite configuration yaml to exclude tests.
@@ -392,8 +426,8 @@ def generate_exclude_yaml(suite, task_path_suffix, is_generated_suite):
     else:
         # We expect the generated suites to already have been generated in the generated config
         # directory.
-        for file_name in os.listdir(CONFIG_DIR):
-            suites_dir = CONFIG_DIR
+        suites_dir = CONFIG_DIR
+        for file_name in os.listdir(suites_dir):
             # Update the 'exclude_files' for each of the appropriate generated suites.
             if file_name.endswith('misc.yml'):
                 # New tests will be run as part of misc.yml. We want to make sure to properly
diff --git a/buildscripts/evergreen_generate_resmoke_tasks.py b/buildscripts/evergreen_generate_resmoke_tasks.py
index 0f9996de01f..84e19c9448b 100755
--- a/buildscripts/evergreen_generate_resmoke_tasks.py
+++ b/buildscripts/evergreen_generate_resmoke_tasks.py
@@ -15,7 +15,7 @@ import os
 import re
 import sys
 from distutils.util import strtobool  # pylint: disable=no-name-in-module
-from typing import Dict, List, Set, Tuple
+from typing import Dict, List, Set, Sequence, Optional, Any, Match
 
 import click
 import requests
@@ -23,22 +23,21 @@ import structlog
 import yaml
 
 from evergreen.api import EvergreenApi, RetryingEvergreenApi
-from shrub.config import Configuration
-from shrub.task import TaskDependency
-from shrub.variant import DisplayTaskDefinition
-from shrub.variant import TaskSpec
+from evergreen.stats import TestStats
+
+from shrub.v2 import Task, TaskDependency, BuildVariant, ExistingTask, ShrubProject
 
 # Get relative imports to work when the package is not installed on the PYTHONPATH.
 if __name__ == "__main__" and __package__ is None:
     sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-import buildscripts.resmokelib.parser as _parser  # pylint: disable=wrong-import-position
-import buildscripts.resmokelib.suitesconfig as suitesconfig  # pylint: disable=wrong-import-position
-import buildscripts.util.read_config as read_config  # pylint: disable=wrong-import-position
-import buildscripts.util.taskname as taskname  # pylint: disable=wrong-import-position
-import buildscripts.util.teststats as teststats  # pylint: disable=wrong-import-position
-
 # pylint: disable=wrong-import-position
+import buildscripts.resmokelib.parser as _parser
+import buildscripts.resmokelib.suitesconfig as suitesconfig
+from buildscripts.util.fileops import write_file_to_dir
+import buildscripts.util.read_config as read_config
+import buildscripts.util.taskname as taskname
+import buildscripts.util.teststats as teststats
 from buildscripts.patch_builds.task_generation import TimeoutInfo, resmoke_commands
 # pylint: enable=wrong-import-position
 
@@ -165,6 +164,16 @@ class ConfigOptions(object):
         return True
 
     @property
+    def display_task_name(self):
+        """Return the name to use as the display task."""
+        return self.task
+
+    @property
+    def gen_task_set(self):
+        """Return the set of tasks used to generate this configuration."""
+        return {self.task_name}
+
+    @property
     def variant(self):
         """Return build variant is being run on."""
         return self.build_variant
@@ -180,17 +189,6 @@ class ConfigOptions(object):
 
         return config.get(item, None)
 
-    def generate_display_task(self, task_names: List[str]) -> DisplayTaskDefinition:
-        """
-        Generate a display task with execution tasks.
-
-        :param task_names: The names of the execution tasks to include under the display task.
-        :return: Display task definition for the generated display task.
-        """
-        return DisplayTaskDefinition(self.task) \
-                .execution_tasks(task_names) \
-                .execution_task("{0}_gen".format(self.task))
-
     def __getattr__(self, item):
         """Determine the value of the given attribute."""
         return self._lookup(self.config, item)
@@ -213,19 +211,7 @@ def enable_logging(verbose):
     structlog.configure(logger_factory=structlog.stdlib.LoggerFactory())
 
 
-def write_file(directory: str, filename: str, contents: str):
-    """
-    Write the given contents to the specified file.
-
-    :param directory: Directory to write file into.
-    :param filename: Name of file to write to.
-    :param contents: Data to write to file.
-    """
-    with open(os.path.join(directory, filename), "w") as fileh:
-        fileh.write(contents)
-
-
-def write_file_dict(directory: str, file_dict: Dict[str, str]):
+def write_file_dict(directory: str, file_dict: Dict[str, str]) -> None:
     """
     Write files in the given dictionary to disk.
 
@@ -237,11 +223,8 @@ def write_file_dict(directory: str, file_dict: Dict[str, str]):
     :param directory: Directory to write files to.
     :param file_dict: Dictionary of files to write.
     """
-    if not os.path.exists(directory):
-        os.makedirs(directory)
-
     for name, contents in file_dict.items():
-        write_file(directory, name, contents)
+        write_file_to_dir(directory, name, contents)
 
 
 def read_yaml(directory: str, filename: str) -> Dict:
@@ -412,7 +395,7 @@ def generate_resmoke_suite_config(source_config, source_file, roots=None, exclud
 
 
 def render_suite_files(suites: List, suite_name: str, test_list: List[str], suite_dir,
-                       create_misc_suite: bool):
+                       create_misc_suite: bool) -> Dict:
     """
     Render the given list of suites.
 
@@ -552,12 +535,10 @@ class Suite(object):
 class EvergreenConfigGenerator(object):
     """Generate evergreen configurations."""
 
-    def __init__(self, shrub_config: Configuration, suites: List[Suite], options: ConfigOptions,
-                 evg_api: EvergreenApi):
+    def __init__(self, suites: List[Suite], options: ConfigOptions, evg_api: EvergreenApi):
         """
         Create new EvergreenConfigGenerator object.
 
-        :param shrub_config: Shrub configuration the generated Evergreen config will be added to.
         :param suites: The suite the Evergreen config will be generated for.
         :param options: The ConfigOptions object containing the config file values.
         :param evg_api: Evergreen API object.
@@ -565,25 +546,38 @@ class EvergreenConfigGenerator(object):
         self.suites = suites
         self.options = options
         self.evg_api = evg_api
-        self.evg_config = shrub_config
         self.task_specs = []
         self.task_names = []
         self.build_tasks = None
 
-    def _set_task_distro(self, task_spec):
+    def _get_distro(self) -> Optional[Sequence[str]]:
+        """Get the distros that the tasks should be run on."""
         if self.options.use_large_distro and self.options.large_distro_name:
-            task_spec.distro(self.options.large_distro_name)
+            return [self.options.large_distro_name]
+        return None
 
-    def _generate_resmoke_args(self, suite_file):
-        resmoke_args = "--suite={0}.yml --originSuite={1} {2}".format(
-            suite_file, self.options.suite, self.options.resmoke_args)
+    def _generate_resmoke_args(self, suite_file: str) -> str:
+        """
+        Generate the resmoke args for the given suite.
+
+        :param suite_file: File containing configuration for test suite.
+        :return: arguments to pass to resmoke.
+        """
+        resmoke_args = (f"--suite={suite_file}.yml --originSuite={self.options.suite} "
+                        f" {self.options.resmoke_args}")
         if self.options.repeat_suites and not string_contains_any_of_args(
                 resmoke_args, ["repeatSuites", "repeat"]):
-            resmoke_args += " --repeatSuites={0} ".format(self.options.repeat_suites)
+            resmoke_args += f" --repeatSuites={self.options.repeat_suites} "
 
         return resmoke_args
 
-    def _get_run_tests_vars(self, suite_file):
+    def _get_run_tests_vars(self, suite_file: str) -> Dict[str, Any]:
+        """
+        Generate a dictionary of the variables to pass to the task.
+
+        :param suite_file: Suite being run.
+        :return: Dictionary containing variables and value to pass to generated task.
+        """
         variables = {
             "resmoke_args": self._generate_resmoke_args(suite_file),
             "run_multiple_jobs": self.options.run_multiple_jobs,
@@ -600,7 +594,8 @@ class EvergreenConfigGenerator(object):
 
         return variables
 
-    def _get_timeout_command(self, max_test_runtime, expected_suite_runtime, use_default):
+    def _get_timeout_command(self, max_test_runtime: int, expected_suite_runtime: int,
+                             use_default: bool) -> TimeoutInfo:
         """
         Add an evergreen command to override the default timeouts to the list of commands.
 
@@ -638,38 +633,56 @@ class EvergreenConfigGenerator(object):
         return TimeoutInfo.default_timeout()
 
     @staticmethod
-    def _is_task_dependency(task, possible_dependency):
-        return re.match("{0}_(\\d|misc)".format(task), possible_dependency)
+    def _is_task_dependency(task: str, possible_dependency: str) -> Optional[Match[str]]:
+        """
+        Determine if the given possible_dependency belongs to the given task.
 
-    def _get_tasks_for_depends_on(self, dependent_task):
+        :param task: Name of dependency being checked.
+        :param possible_dependency: Task to check if dependency.
+        :return: None is task is not a dependency.
+        """
+        return re.match(f"{task}_(\\d|misc)", possible_dependency)
+
+    def _get_tasks_for_depends_on(self, dependent_task: str) -> List[str]:
+        """
+        Get a list of tasks that belong to the given dependency.
+
+        :param dependent_task: Dependency to check.
+        :return: List of tasks that are a part of the given dependency.
+        """
         return [
             str(task.display_name) for task in self.build_tasks
             if self._is_task_dependency(dependent_task, str(task.display_name))
         ]
 
-    def _add_dependencies(self, task):
-        task.dependency(TaskDependency("compile"))
+    def _get_dependencies(self) -> Set[TaskDependency]:
+        """Get the set of dependency tasks for these suites."""
+        dependencies = {TaskDependency("compile")}
         if not self.options.is_patch:
             # Don"t worry about task dependencies in patch builds, only mainline.
             if self.options.depends_on:
                 for dep in self.options.depends_on:
                     depends_on_tasks = self._get_tasks_for_depends_on(dep)
                     for dependency in depends_on_tasks:
-                        task.dependency(TaskDependency(dependency))
+                        dependencies.add(TaskDependency(dependency))
 
-        return task
+        return dependencies
 
-    def _generate_task(self, sub_suite_name, sub_task_name, target_dir, max_test_runtime=None,
-                       expected_suite_runtime=None):
-        """Generate evergreen config for a resmoke task."""
+    def _generate_task(self, sub_suite_name: str, sub_task_name: str, target_dir: str,
+                       max_test_runtime: Optional[int] = None,
+                       expected_suite_runtime: Optional[int] = None) -> Task:
+        """
+        Generate a shrub evergreen config for a resmoke task.
+
+        :param sub_suite_name: Name of suite being generated.
+        :param sub_task_name: Name of task to generate.
+        :param target_dir: Directory containing generated suite files.
+        :param max_test_runtime: Runtime of the longest test in this sub suite.
+        :param expected_suite_runtime: Expected total runtime of this suite.
+        :return: Shrub configuration for the described task.
+        """
         # pylint: disable=too-many-arguments
         LOGGER.debug("Generating task", sub_suite=sub_suite_name)
-        spec = TaskSpec(sub_task_name)
-        self._set_task_distro(spec)
-        self.task_specs.append(spec)
-
-        self.task_names.append(sub_task_name)
-        task = self.evg_config.task(sub_task_name)
 
         # Evergreen always uses a unix shell, even on Windows, so instead of using os.path.join
         # here, just use the forward slash; otherwise the path separator will be treated as
@@ -683,45 +696,65 @@ class EvergreenConfigGenerator(object):
         commands = resmoke_commands("run generated tests", run_tests_vars, timeout_info,
                                     use_multiversion)
 
-        self._add_dependencies(task).commands(commands)
+        return Task(sub_task_name, commands, self._get_dependencies())
+
+    def _create_sub_task(self, idx: int, suite: Suite) -> Task:
+        """
+        Create the sub task for the given suite.
 
-    def _generate_all_tasks(self):
-        for idx, suite in enumerate(self.suites):
-            sub_task_name = taskname.name_generated_task(self.options.task, idx, len(self.suites),
-                                                         self.options.variant)
-            max_runtime = None
-            total_runtime = None
-            if suite.should_overwrite_timeout():
-                max_runtime = suite.max_runtime
-                total_runtime = suite.get_runtime()
-            self._generate_task(suite.name, sub_task_name, self.options.generated_config_dir,
-                                max_runtime, total_runtime)
+        :param idx: Index of suite to created.
+        :param suite: Suite to create.
+        :return: Shrub configuration for the suite.
+        """
+        sub_task_name = taskname.name_generated_task(self.options.task, idx, len(self.suites),
+                                                     self.options.variant)
+        max_runtime = None
+        total_runtime = None
+        if suite.should_overwrite_timeout():
+            max_runtime = suite.max_runtime
+            total_runtime = suite.get_runtime()
+        return self._generate_task(suite.name, sub_task_name, self.options.generated_config_dir,
+                                   max_runtime, total_runtime)
+
+    def _generate_all_tasks(self) -> Set[Task]:
+        """Get a set of shrub task for all the sub tasks."""
+        tasks = {self._create_sub_task(idx, suite) for idx, suite in enumerate(self.suites)}
 
         if self.options.create_misc_suite:
             # Add the misc suite
             misc_suite_name = f"{os.path.basename(self.options.suite)}_misc"
             misc_task_name = f"{self.options.task}_misc_{self.options.variant}"
-            self._generate_task(misc_suite_name, misc_task_name, self.options.generated_config_dir)
+            tasks.add(
+                self._generate_task(misc_suite_name, misc_task_name,
+                                    self.options.generated_config_dir))
 
-    def _generate_variant(self):
-        self._generate_all_tasks()
+        return tasks
 
-        self.evg_config.variant(self.options.variant)\
-            .tasks(self.task_specs)\
-            .display_task(self.options.generate_display_task(self.task_names))
+    def generate_config(self, build_variant: BuildVariant) -> None:
+        """
+        Generate evergreen configuration.
 
-    def generate_config(self):
-        """Generate evergreen configuration."""
+        :param build_variant: Build variant to add generated configuration to.
+        """
         self.build_tasks = self.evg_api.tasks_by_build(self.options.build_id)
-        self._generate_variant()
-        return self.evg_config
+
+        tasks = self._generate_all_tasks()
+        generating_task = {ExistingTask(task_name) for task_name in self.options.gen_task_set}
+        distros = self._get_distro()
+        build_variant.display_task(self.options.display_task_name, execution_tasks=tasks,
+                                   execution_existing_tasks=generating_task, distros=distros)
 
 
 class GenerateSubSuites(object):
     """Orchestrate the execution of generate_resmoke_suites."""
 
-    def __init__(self, evergreen_api, config_options):
-        """Initialize the object."""
+    def __init__(self, evergreen_api: EvergreenApi, config_options: ConfigOptions):
+        """
+        Initialize the object.
+
+        :param evergreen_api: Evergreen API client.
+        :param config_options: Generation configuration options.
+        """
         self.evergreen_api = evergreen_api
         self.config_options = config_options
         self.test_list = []
@@ -729,8 +762,14 @@ class GenerateSubSuites(object):
         # Populate config values for methods like list_tests()
         _parser.set_options()
 
-    def calculate_suites(self, start_date, end_date):
-        """Divide tests into suites based on statistics for the provided period."""
+    def calculate_suites(self, start_date: datetime, end_date: datetime) -> List[Suite]:
+        """
+        Divide tests into suites based on statistics for the provided period.
+
+        :param start_date: Time to start historical analysis.
+        :param end_date: Time to end historical analysis.
+        :return: List of sub suites to be generated.
+        """
         try:
             evg_stats = self.get_evg_stats(self.config_options.project, start_date, end_date,
                                            self.config_options.task, self.config_options.variant)
@@ -751,8 +790,18 @@ class GenerateSubSuites(object):
             else:
                 raise
 
-    def get_evg_stats(self, project, start_date, end_date, task, variant):
-        """Collect test execution statistics data from Evergreen."""
+    def get_evg_stats(self, project: str, start_date: datetime, end_date: datetime, task: str,
+                      variant: str) -> List[TestStats]:
+        """
+        Collect test execution statistics data from Evergreen.
+
+        :param project: Evergreen project to query.
+        :param start_date: Time to start historical analysis.
+        :param end_date: Time to end historical analysis.
+        :param task: Task to query.
+        :param variant: Build variant to query.
+        :return: List of test stats for specified task.
+        """
         # pylint: disable=too-many-arguments
 
         days = (end_date - start_date).days
@@ -761,8 +810,15 @@ class GenerateSubSuites(object):
             before_date=end_date.strftime("%Y-%m-%d"), tasks=[task], variants=[variant],
             group_by="test", group_num_days=days)
 
-    def calculate_suites_from_evg_stats(self, data, execution_time_secs):
-        """Divide tests into suites that can be run in less than the specified execution time."""
+    def calculate_suites_from_evg_stats(self, data: List[TestStats],
+                                        execution_time_secs: int) -> List[Suite]:
+        """
+        Divide tests into suites that can be run in less than the specified execution time.
+
+        :param data: Historical test results for task being split.
+        :param execution_time_secs: Target execution time of each suite (in seconds).
+        :return: List of sub suites calculated.
+        """
         test_stats = teststats.TestStats(data)
         tests_runtimes = self.filter_tests(test_stats.get_tests_runtimes())
         if not tests_runtimes:
@@ -787,7 +843,8 @@ class GenerateSubSuites(object):
                                                     tests_runtimes)
         return tests_runtimes
 
-    def filter_existing_tests(self, tests_runtimes):
+    def filter_existing_tests(self, tests_runtimes: List[teststats.TestRuntime]) \
+            -> List[teststats.TestRuntime]:
         """Filter out tests that do not exist in the filesystem."""
         all_tests = [teststats.normalize_test_name(test) for test in self.list_tests()]
         return [
@@ -795,7 +852,7 @@ class GenerateSubSuites(object):
             if os.path.exists(info.test_name) and info.test_name in all_tests
         ]
 
-    def calculate_fallback_suites(self):
+    def calculate_fallback_suites(self) -> List[Suite]:
         """Divide tests into a fixed number of suites."""
         LOGGER.debug("Splitting tasks based on fallback",
                      fallback=self.config_options.fallback_num_sub_suites)
@@ -806,21 +863,31 @@ class GenerateSubSuites(object):
             suites[idx % num_suites].add_test(test_file, 0)
         return suites
 
-    def list_tests(self):
+    def list_tests(self) -> List[Dict]:
         """List the test files that are part of the suite being split."""
         return suitesconfig.get_suite(self.config_options.suite).tests
 
-    def generate_task_config(self, shrub_config: Configuration, suites: List[Suite]):
+    def add_suites_to_build_variant(self, suites: List[Suite], build_variant: BuildVariant) -> None:
+        """
+        Add the given suites to the build variant specified.
+
+        :param suites: Suites to add.
+        :param build_variant: Build variant to add suite to.
+        """
+        EvergreenConfigGenerator(suites, self.config_options, self.evergreen_api) \
+            .generate_config(build_variant)
+
+    def generate_task_config(self, suites: List[Suite]) -> BuildVariant:
         """
         Generate the evergreen configuration for the new suite.
 
-        :param shrub_config: Shrub configuration the generated Evergreen config will be added to.
         :param suites: The suite the generated Evergreen config will be generated for.
         """
-        EvergreenConfigGenerator(shrub_config, suites, self.config_options,
-                                 self.evergreen_api).generate_config()
+        build_variant = BuildVariant(self.config_options.variant)
+        self.add_suites_to_build_variant(suites, build_variant)
+        return build_variant
 
-    def generate_suites_config(self, suites: List[Suite]) -> Tuple[dict, str]:
+    def generate_suites_config(self, suites: List[Suite]) -> Dict:
         """
         Generate the suites files and evergreen configuration for the generated task.
 
@@ -853,10 +920,10 @@ class GenerateSubSuites(object):
 
         config_dict_of_suites = self.generate_suites_config(suites)
 
-        shrub_config = Configuration()
-        self.generate_task_config(shrub_config, suites)
+        shrub_config = ShrubProject.empty()
+        shrub_config.add_build_variant(self.generate_task_config(suites))
 
-        config_dict_of_suites[self.config_options.task + ".json"] = shrub_config.to_json()
+        config_dict_of_suites[self.config_options.task + ".json"] = shrub_config.json()
         write_file_dict(self.config_options.generated_config_dir, config_dict_of_suites)
 
 
diff --git a/buildscripts/patch_builds/task_generation.py b/buildscripts/patch_builds/task_generation.py
index 576ac95b15a..2ca8a7e184f 100644
--- a/buildscripts/patch_builds/task_generation.py
+++ b/buildscripts/patch_builds/task_generation.py
@@ -1,24 +1,19 @@
 """Utilities to help generate evergreen tasks."""
-from typing import Optional, List
+from __future__ import annotations
 
-from shrub.command import CommandDefinition
-from shrub.config import Configuration
-from shrub.operations import CmdTimeoutUpdate
-from shrub.task import TaskDependency
-from shrub.variant import TaskSpec, DisplayTaskDefinition
+from typing import Any, Dict, Optional, List
 
+from shrub.v2 import FunctionCall, ShrubProject
+from shrub.v2.command import timeout_update, ShrubCommand
+from structlog import get_logger
 
-def _cmd_by_name(cmd_name):
-    """
-    Create a command definition of a function with the given name.
-
-    :param cmd_name: Name of function.
-    :return: Command Definition for function.
-    """
-    return CommandDefinition().function(cmd_name)
+LOGGER = get_logger(__name__)
+MAX_SHRUB_TASKS_FOR_SINGLE_TASK = 1000
 
 
-def resmoke_commands(run_tests_fn_name, run_tests_vars, timeout_info, use_multiversion=None):
+def resmoke_commands(run_tests_fn_name: str, run_tests_vars: Dict[str, Any],
+                     timeout_info: TimeoutInfo,
+                     use_multiversion: Optional[str] = None) -> List[ShrubCommand]:
     """
     Create a list of commands to run a resmoke task.
 
@@ -30,9 +25,9 @@ def resmoke_commands(run_tests_fn_name, run_tests_vars, timeout_info, use_multiv
     """
     commands = [
         timeout_info.cmd,
-        _cmd_by_name("do setup"),
-        _cmd_by_name("do multiversion setup") if use_multiversion else None,
-        _cmd_by_name(run_tests_fn_name).vars(run_tests_vars),
+        FunctionCall("do setup"),
+        FunctionCall("do multiversion setup") if use_multiversion else None,
+        FunctionCall(run_tests_fn_name, run_tests_vars),
     ]
 
     return [cmd for cmd in commands if cmd]
@@ -75,13 +70,7 @@ class TimeoutInfo(object):
     def cmd(self):
         """Create a command that sets timeouts as specified."""
         if not self.use_defaults:
-            timeout_cmd = CmdTimeoutUpdate()
-            if self.timeout:
-                timeout_cmd.timeout(self.timeout)
-
-            if self.exec_timeout:
-                timeout_cmd.exec_timeout(self.exec_timeout)
-            return timeout_cmd.validate().resolve()
+            return timeout_update(exec_timeout_secs=self.exec_timeout, timeout_secs=self.timeout)
 
         return None
 
@@ -92,70 +81,16 @@ class TimeoutInfo(object):
         return f"<exec_timeout={self.exec_timeout}, timeout={self.timeout}>"
 
 
-class TaskList(object):
-    """A list of evergreen tasks to be generated together."""
-
-    def __init__(self, evg_config: Configuration):
-        """
-        Create a list of evergreen tasks to create.
-
-        :param evg_config: Evergreen configuration to add tasks to.
-        """
-        self.evg_config = evg_config
-        self.task_specs = []
-        self.task_names = []
-
-    def add_task(self, name: str, commands: [CommandDefinition],
-                 depends_on: Optional[List[str]] = None, distro: Optional[str] = None):
-        """
-        Add a new task to the task list.
-
-        :param name: Name of task to add.
-        :param commands: List of commands comprising task.
-        :param depends_on: Any dependencies for the task.
-        :param distro: Distro task should be run on.
-        """
-        task = self.evg_config.task(name)
-        task.commands(commands)
-
-        if depends_on:
-            for dep in depends_on:
-                task.dependency(TaskDependency(dep))
-
-        task_spec = TaskSpec(name)
-        if distro:
-            task_spec.distro(distro)
-        self.task_specs.append(task_spec)
-        self.task_names.append(name)
-
-    def display_task(self, display_name: str, existing_tasks: Optional[List[str]] = None) \
-            -> DisplayTaskDefinition:
-        """
-        Create a display task for the list of tasks.
-
-        Note: This function should be called after all calls to `add_task` have been done.
-
-        :param display_name: Name of display tasks.
-        :param existing_tasks: Any existing tasks that should be part of the display task.
-        :return: Display task object.
-        """
-        execution_tasks = self.task_names
-        if existing_tasks:
-            execution_tasks.extend(existing_tasks)
-
-        display_task = DisplayTaskDefinition(display_name).execution_tasks(execution_tasks)
-        return display_task
-
-    def add_to_variant(self, variant_name: str, display_name: Optional[str] = None,
-                       existing_tasks: Optional[List[str]] = None):
-        """
-        Add this task list to a build variant.
+def validate_task_generation_limit(shrub_project: ShrubProject) -> bool:
+    """
+    Determine if this shrub configuration generates less than the limit.
 
-        :param variant_name: Variant to add to.
-        :param display_name: Display name to add tasks under.
-        :param existing_tasks: Any existing tasks that should be added to the display group.
-        """
-        variant = self.evg_config.variant(variant_name)
-        variant.tasks(self.task_specs)
-        if display_name:
-            variant.display_task(self.display_task(display_name, existing_tasks))
+    :param shrub_project: Shrub configuration to validate.
+    :return: True if the configuration is under the limit.
+    """
+    tasks_to_create = len(shrub_project.all_tasks())
+    if tasks_to_create > MAX_SHRUB_TASKS_FOR_SINGLE_TASK:
+        LOGGER.warning("Attempting to create more tasks than max, aborting", tasks=tasks_to_create,
+                       max=MAX_SHRUB_TASKS_FOR_SINGLE_TASK)
+        return False
+    return True
diff --git a/buildscripts/selected_tests.py b/buildscripts/selected_tests.py
index b6191050a64..1c6f9f8c4d4 100644
--- a/buildscripts/selected_tests.py
+++ b/buildscripts/selected_tests.py
@@ -1,19 +1,18 @@
 #!/usr/bin/env python3
 """Command line utility for determining what jstests should run for the given changed files."""
-
 import logging
 import os
 import re
 import sys
-from typing import Any, Dict, List, Optional, Set, Tuple
+from typing import Any, Dict, List, Set
 
 import click
 import structlog
 from structlog.stdlib import LoggerFactory
 from evergreen.api import EvergreenApi, RetryingEvergreenApi
 from git import Repo
-from shrub.config import Configuration
-from shrub.variant import DisplayTaskDefinition, Variant
+
+from shrub.v2 import ShrubProject, BuildVariant
 
 # Get relative imports to work when the package is not installed on the PYTHONPATH.
 if __name__ == "__main__" and __package__ is None:
@@ -29,6 +28,7 @@ from buildscripts.ciconfig.evergreen import (
     ResmokeArgs,
     Task,
     parse_evergreen_file,
+    Variant,
 )
 from buildscripts.evergreen_generate_resmoke_tasks import (
     CONFIG_FORMAT_FN,
@@ -127,14 +127,15 @@ class SelectedTestsConfigOptions(ConfigOptions):
         """Whether or not a _misc suite file should be created."""
         return not self.selected_tests_to_run
 
-    def generate_display_task(self, task_names: List[str]) -> DisplayTaskDefinition:
-        """
-        Generate a display task with execution tasks.
+    @property
+    def display_task_name(self):
+        """Return the name to use as the display task."""
+        return f"{self.task}_{self.variant}"
 
-        :param task_names: The names of the execution tasks to include under the display task.
-        :return: Display task definition for the generated display task.
-        """
-        return DisplayTaskDefinition(f"{self.task}_{self.variant}").execution_tasks(task_names)
+    @property
+    def gen_task_set(self):
+        """Return the set of tasks used to generate this configuration."""
+        return set()
 
 
 def _configure_logging(verbose: bool):
@@ -264,14 +265,15 @@ def _get_evg_task_config(
     }
 
 
-def _update_config_with_task(evg_api: EvergreenApi, shrub_config: Configuration,
+def _update_config_with_task(evg_api: EvergreenApi, build_variant: BuildVariant,
                              config_options: SelectedTestsConfigOptions,
-                             config_dict_of_suites_and_tasks: Dict[str, str]):
+                             config_dict_of_suites_and_tasks: Dict[str, str]) -> None:
     """
     Generate the suites config and the task shrub config for a given task config.
 
     :param evg_api: Evergreen API object.
-    :param shrub_config: Shrub configuration for task.
+    :param build_variant: Build variant to add tasks to.
+    :param shrub_project: Shrub configuration for task.
     :param config_options: Task configuration options.
     :param config_dict_of_suites_and_tasks: Dict of shrub configs and suite file contents.
     """
@@ -281,7 +283,7 @@ def _update_config_with_task(evg_api: EvergreenApi, shrub_config: Configuration,
     config_dict_of_suites = task_generator.generate_suites_config(suites)
     config_dict_of_suites_and_tasks.update(config_dict_of_suites)
 
-    task_generator.generate_task_config(shrub_config, suites)
+    task_generator.add_suites_to_build_variant(suites, build_variant)
 
 
 def _get_task_configs_for_test_mappings(selected_tests_variant_expansions: Dict[str, str],
@@ -390,7 +392,8 @@ def _get_task_configs(evg_conf: EvergreenProjectConfig,
 def run(evg_api: EvergreenApi, evg_conf: EvergreenProjectConfig,
         selected_tests_service: SelectedTestsService,
         selected_tests_variant_expansions: Dict[str, str], repos: List[Repo],
-        origin_build_variants: List[str]) -> Dict[str, dict]:
+        origin_build_variants: List[str]) -> Dict[str, str]:
+    # pylint: disable=too-many-locals
     """
     Run code to select tasks to run based on test and task mappings for each of the build variants.
 
@@ -402,14 +405,15 @@ def run(evg_api: EvergreenApi, evg_conf: EvergreenProjectConfig,
     :param origin_build_variants: Build variants to collect task info from.
     :return: Dict of files and file contents for generated tasks.
     """
-    shrub_config = Configuration()
     config_dict_of_suites_and_tasks = {}
 
     changed_files = find_changed_files_in_repos(repos)
     changed_files = {_remove_repo_path_prefix(file_path) for file_path in changed_files}
     LOGGER.debug("Found changed files", files=changed_files)
 
+    shrub_project = ShrubProject()
     for build_variant in origin_build_variants:
+        shrub_build_variant = BuildVariant(build_variant)
         build_variant_config = evg_conf.get_variant(build_variant)
         origin_variant_expansions = build_variant_config.expansions
 
@@ -426,10 +430,12 @@ def run(evg_api: EvergreenApi, evg_conf: EvergreenProjectConfig,
                 DEFAULT_CONFIG_VALUES,
                 CONFIG_FORMAT_FN,
             )
-            _update_config_with_task(evg_api, shrub_config, config_options,
+            _update_config_with_task(evg_api, shrub_build_variant, config_options,
                                      config_dict_of_suites_and_tasks)
 
-    config_dict_of_suites_and_tasks["selected_tests_config.json"] = shrub_config.to_json()
+        shrub_project.add_build_variant(shrub_build_variant)
+
+    config_dict_of_suites_and_tasks["selected_tests_config.json"] = shrub_project.json()
     return config_dict_of_suites_and_tasks
 
 
@@ -479,13 +485,11 @@ def main(
 
     buildscripts.resmokelib.parser.set_options()
 
-    selected_tests_variant_expansions = read_config.read_config_file(expansion_file)
-    origin_build_variants = selected_tests_variant_expansions["selected_tests_buildvariants"].split(
-        " ")
+    task_expansions = read_config.read_config_file(expansion_file)
+    origin_build_variants = task_expansions["selected_tests_buildvariants"].split(" ")
 
     config_dict_of_suites_and_tasks = run(evg_api, evg_conf, selected_tests_service,
-                                          selected_tests_variant_expansions, repos,
-                                          origin_build_variants)
+                                          task_expansions, repos, origin_build_variants)
     write_file_dict(SELECTED_TESTS_CONFIG_DIR, config_dict_of_suites_and_tasks)
 
 
diff --git a/buildscripts/tests/patch_builds/test_task_generation.py b/buildscripts/tests/patch_builds/test_task_generation.py
index 8c015e12406..336c86d9c90 100644
--- a/buildscripts/tests/patch_builds/test_task_generation.py
+++ b/buildscripts/tests/patch_builds/test_task_generation.py
@@ -1,8 +1,6 @@
 """Unittests for buildscripts.patch_builds.task_generation.py"""
 import unittest
 
-from shrub.config import Configuration
-
 import buildscripts.patch_builds.task_generation as under_test
 
 # pylint: disable=missing-docstring,protected-access,too-many-lines,no-self-use
@@ -63,7 +61,7 @@ class TestTimeoutInfo(unittest.TestCase):
         timeout = 5
         timeout_info = under_test.TimeoutInfo.overridden(timeout=timeout)
 
-        cmd = timeout_info.cmd.to_map()
+        cmd = timeout_info.cmd.as_dict()
 
         self.assertEqual("timeout.update", cmd["command"])
         self.assertEqual(timeout, cmd["params"]["timeout_secs"])
@@ -73,7 +71,7 @@ class TestTimeoutInfo(unittest.TestCase):
         exec_timeout = 5
         timeout_info = under_test.TimeoutInfo.overridden(exec_timeout=exec_timeout)
 
-        cmd = timeout_info.cmd.to_map()
+        cmd = timeout_info.cmd.as_dict()
 
         self.assertEqual("timeout.update", cmd["command"])
         self.assertEqual(exec_timeout, cmd["params"]["exec_timeout_secs"])
@@ -84,7 +82,7 @@ class TestTimeoutInfo(unittest.TestCase):
         exec_timeout = 5
         timeout_info = under_test.TimeoutInfo.overridden(exec_timeout=exec_timeout, timeout=timeout)
 
-        cmd = timeout_info.cmd.to_map()
+        cmd = timeout_info.cmd.as_dict()
 
         self.assertEqual("timeout.update", cmd["command"])
         self.assertEqual(exec_timeout, cmd["params"]["exec_timeout_secs"])
@@ -93,131 +91,3 @@ class TestTimeoutInfo(unittest.TestCase):
     def test_override_with_no_values(self):
         with self.assertRaises(ValueError):
             under_test.TimeoutInfo.overridden()
-
-
-class TestTaskList(unittest.TestCase):
-    def test_adding_a_task(self):
-        config = Configuration()
-        task_list = under_test.TaskList(config)
-
-        func = "test"
-        task = "task 1"
-        variant = "variant 1"
-        task_list.add_task(task, [under_test._cmd_by_name(func)])
-        task_list.add_to_variant(variant)
-
-        cfg_dict = config.to_map()
-
-        cmd_dict = cfg_dict["tasks"][0]
-        self.assertEqual(task, cmd_dict["name"])
-        self.assertEqual(func, cmd_dict["commands"][0]["func"])
-
-        self.assertEqual(task, cfg_dict["buildvariants"][0]["tasks"][0]["name"])
-
-    def test_adding_a_task_with_distro(self):
-        config = Configuration()
-        task_list = under_test.TaskList(config)
-
-        func = "test"
-        task = "task 1"
-        variant = "variant 1"
-        distro = "distro 1"
-        task_list.add_task(task, [under_test._cmd_by_name(func)], distro=distro)
-        task_list.add_to_variant(variant)
-
-        cfg_dict = config.to_map()
-
-        cmd_dict = cfg_dict["tasks"][0]
-        self.assertEqual(task, cmd_dict["name"])
-        self.assertEqual(func, cmd_dict["commands"][0]["func"])
-
-        self.assertEqual(task, cfg_dict["buildvariants"][0]["tasks"][0]["name"])
-        self.assertIn(distro, cfg_dict["buildvariants"][0]["tasks"][0]["distros"])
-
-    def test_adding_a_task_with_dependecies(self):
-        config = Configuration()
-        task_list = under_test.TaskList(config)
-
-        func = "test"
-        task = "task 1"
-        variant = "variant 1"
-        dependencies = ["dep task 1", "dep task 2"]
-        task_list.add_task(task, [under_test._cmd_by_name(func)], depends_on=dependencies)
-        task_list.add_to_variant(variant)
-
-        cfg_dict = config.to_map()
-
-        cmd_dict = cfg_dict["tasks"][0]
-        self.assertEqual(task, cmd_dict["name"])
-        self.assertEqual(func, cmd_dict["commands"][0]["func"])
-        for dep in dependencies:
-            self.assertIn(dep, {d["name"] for d in cmd_dict["depends_on"]})
-
-        task_dict = cfg_dict["buildvariants"][0]["tasks"][0]
-        self.assertEqual(task, task_dict["name"])
-
-    def test_adding_multiple_tasks(self):
-        config = Configuration()
-        task_list = under_test.TaskList(config)
-
-        func = "test"
-        variant = "variant 1"
-        tasks = ["task 1", "task 2"]
-        for task in tasks:
-            task_list.add_task(task, [under_test._cmd_by_name(func)])
-
-        task_list.add_to_variant(variant)
-
-        cfg_dict = config.to_map()
-
-        self.assertEqual(len(tasks), len(cfg_dict["tasks"]))
-        self.assertEqual(len(tasks), len(cfg_dict["buildvariants"][0]["tasks"]))
-
-    def test_using_display_task(self):
-        config = Configuration()
-        task_list = under_test.TaskList(config)
-
-        func = "test"
-        variant = "variant 1"
-        tasks = ["task 1", "task 2"]
-        for task in tasks:
-            task_list.add_task(task, [under_test._cmd_by_name(func)])
-
-        display_task = "display_task"
-        task_list.add_to_variant(variant, display_task)
-
-        cfg_dict = config.to_map()
-
-        self.assertEqual(len(tasks), len(cfg_dict["tasks"]))
-        variant_dict = cfg_dict["buildvariants"][0]
-        self.assertEqual(len(tasks), len(variant_dict["tasks"]))
-        dt_dict = variant_dict["display_tasks"][0]
-        self.assertEqual(display_task, dt_dict["name"])
-        for task in tasks:
-            self.assertIn(task, dt_dict["execution_tasks"])
-
-    def test_using_display_task_with_existing_tasks(self):
-        config = Configuration()
-        task_list = under_test.TaskList(config)
-
-        func = "test"
-        variant = "variant 1"
-        tasks = ["task 1", "task 2"]
-        for task in tasks:
-            task_list.add_task(task, [under_test._cmd_by_name(func)])
-
-        display_task = "display_task"
-        existing_tasks = ["other task 1", "other task 2"]
-        task_list.add_to_variant(variant, display_task, existing_tasks)
-
-        cfg_dict = config.to_map()
-
-        self.assertEqual(len(tasks), len(cfg_dict["tasks"]))
-        variant_dict = cfg_dict["buildvariants"][0]
-        self.assertEqual(len(tasks), len(variant_dict["tasks"]))
-        dt_dict = variant_dict["display_tasks"][0]
-        self.assertEqual(display_task, dt_dict["name"])
-        for task in tasks:
-            self.assertIn(task, dt_dict["execution_tasks"])
-        for task in existing_tasks:
-            self.assertIn(task, dt_dict["execution_tasks"])
diff --git a/buildscripts/tests/test_burn_in_tags.py b/buildscripts/tests/test_burn_in_tags.py
index 6adaeb43773..c9e9d46843b 100644
--- a/buildscripts/tests/test_burn_in_tags.py
+++ b/buildscripts/tests/test_burn_in_tags.py
@@ -1,23 +1,29 @@
 """Unit tests for the burn_in_tags.py script."""
-import sys
 from collections import defaultdict
+import json
 import os
+import sys
 import unittest
 from unittest.mock import MagicMock, patch
 
-from shrub.config import Configuration
-
-import buildscripts.burn_in_tags as under_test
+from shrub.v2 import ShrubProject
 
 import buildscripts.ciconfig.evergreen as _evergreen
-
 from buildscripts.tests.test_burn_in_tests import ns as burn_in_tests_ns
+from buildscripts.ciconfig.evergreen import EvergreenProjectConfig
+
+import buildscripts.burn_in_tags as under_test
 
 # pylint: disable=missing-docstring,invalid-name,unused-argument,no-self-use,protected-access
 
-NS = "buildscripts.burn_in_tags"
+EMPTY_PROJECT = {
+    "buildvariants": [],
+    "tasks": [],
+}
 TEST_FILE_PATH = os.path.join(os.path.dirname(__file__), "test_burn_in_tags_evergreen.yml")
 
+NS = "buildscripts.burn_in_tags"
+
 
 def ns(relative_name):  # pylint: disable-invalid-name
     """Return a full name from a name relative to the test module"s name space."""
@@ -40,7 +46,7 @@ def get_expansions_data():
     }  # yapf: disable
 
 
-def get_evergreen_config():
+def get_evergreen_config() -> EvergreenProjectConfig:
     return _evergreen.parse_evergreen_file(TEST_FILE_PATH, evergreen_binary=None)
 
 
@@ -77,21 +83,18 @@ class TestGenerateEvgBuildVariants(unittest.TestCase):
         base_variant = "enterprise-rhel-62-64-bit-inmem"
         generated_variant = "enterprise-rhel-62-64-bit-inmem-required"
         burn_in_tags_gen_variant = "enterprise-rhel-62-64-bit"
-        shrub_config = Configuration()
+        variant = evg_conf_mock.get_variant(base_variant)
 
-        under_test._generate_evg_build_variant(shrub_config, base_variant, generated_variant,
-                                               burn_in_tags_gen_variant, evg_conf_mock)
+        build_variant = under_test._generate_evg_build_variant(variant, generated_variant,
+                                                               burn_in_tags_gen_variant)
 
-        expected_variant_data = get_evergreen_config().get_variant(base_variant)
-        generated_buildvariants = shrub_config.to_map()["buildvariants"]
-        self.assertEqual(len(generated_buildvariants), 1)
-        generated_build_variant = generated_buildvariants[0]
+        generated_build_variant = build_variant.as_dict()
         self.assertEqual(generated_build_variant["name"], generated_variant)
-        self.assertEqual(generated_build_variant["modules"], expected_variant_data.modules)
+        self.assertEqual(generated_build_variant["modules"], variant.modules)
         generated_expansions = generated_build_variant["expansions"]
         burn_in_bypass_expansion_value = generated_expansions.pop("burn_in_bypass")
         self.assertEqual(burn_in_bypass_expansion_value, burn_in_tags_gen_variant)
-        self.assertEqual(generated_expansions, expected_variant_data.expansions)
+        self.assertEqual(generated_expansions, variant.expansions)
 
 
 class TestGenerateEvgTasks(unittest.TestCase):
@@ -105,13 +108,13 @@ class TestGenerateEvgTasks(unittest.TestCase):
             "enterprise-rhel-62-64-bit-majority-read-concern-off":
                 "enterprise-rhel-62-64-bit-majority-read-concern-off-required",
         }  # yapf: disable
-        shrub_config = Configuration()
+        shrub_config = ShrubProject()
         evergreen_api = MagicMock()
         repo = MagicMock()
         under_test._generate_evg_tasks(evergreen_api, shrub_config, expansions_file_data,
                                        buildvariant_map, [repo], evg_conf_mock)
 
-        self.assertEqual(shrub_config.to_map(), {})
+        self.assertEqual(shrub_config.as_dict(), EMPTY_PROJECT)
 
     @patch(ns("create_tests_by_task"))
     def test_generate_evg_tasks_one_test_changed(self, create_tests_by_task_mock):
@@ -131,7 +134,7 @@ class TestGenerateEvgTasks(unittest.TestCase):
             "enterprise-rhel-62-64-bit-majority-read-concern-off":
                 "enterprise-rhel-62-64-bit-majority-read-concern-off-required",
         }  # yapf: disable
-        shrub_config = Configuration()
+        shrub_config = ShrubProject.empty()
         evergreen_api = MagicMock()
         repo = MagicMock()
         evergreen_api.test_stats_by_project.return_value = [
@@ -140,13 +143,14 @@ class TestGenerateEvgTasks(unittest.TestCase):
         under_test._generate_evg_tasks(evergreen_api, shrub_config, expansions_file_data,
                                        buildvariant_map, [repo], evg_conf_mock)
 
-        generated_config = shrub_config.to_map()
+        generated_config = shrub_config.as_dict()
         self.assertEqual(len(generated_config["buildvariants"]), 2)
         first_generated_build_variant = generated_config["buildvariants"][0]
+        self.assertIn(first_generated_build_variant["name"], buildvariant_map.values())
         self.assertEqual(first_generated_build_variant["display_tasks"][0]["name"], "burn_in_tests")
         self.assertEqual(
             first_generated_build_variant["display_tasks"][0]["execution_tasks"][0],
-            "burn_in:aggregation_mongos_passthrough_0_enterprise-rhel-62-64-bit-inmem-required")
+            f"burn_in:aggregation_mongos_passthrough_0_{first_generated_build_variant['name']}")
 
 
 EXPANSIONS_FILE_DATA = {
@@ -196,7 +200,7 @@ CREATE_TEST_MEMBERSHIP_MAP = {
 
 
 class TestAcceptance(unittest.TestCase):
-    @patch(ns("_write_to_file"))
+    @patch(ns("write_file_to_dir"))
     @patch(ns("_create_evg_build_variant_map"))
     @patch(burn_in_tests_ns("find_changed_tests"))
     def test_no_tests_run_if_none_changed(self, find_changed_tests_mock,
@@ -215,11 +219,11 @@ class TestAcceptance(unittest.TestCase):
         under_test.burn_in(EXPANSIONS_FILE_DATA, evg_conf_mock, None, repos)
 
         write_to_file_mock.assert_called_once()
-        shrub_config = write_to_file_mock.call_args[0][0]
-        self.assertEqual('{}', shrub_config.to_json())
+        shrub_config = write_to_file_mock.call_args[0][2]
+        self.assertEqual(EMPTY_PROJECT, json.loads(shrub_config))
 
     @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
-    @patch(ns("_write_to_file"))
+    @patch(ns("write_file_to_dir"))
     @patch(ns("_create_evg_build_variant_map"))
     @patch(burn_in_tests_ns("find_changed_tests"))
     @patch(burn_in_tests_ns("create_test_membership_map"))
@@ -244,8 +248,8 @@ class TestAcceptance(unittest.TestCase):
         under_test.burn_in(EXPANSIONS_FILE_DATA, evg_conf, None, repos)
 
         write_to_file_mock.assert_called_once()
-        written_config = write_to_file_mock.call_args[0][0]
-        written_config_map = written_config.to_map()
+        written_config = write_to_file_mock.call_args[0][2]
+        written_config_map = json.loads(written_config)
 
         n_tasks = len(written_config_map["tasks"])
         # Ensure we are generating at least one task for the test.
diff --git a/buildscripts/tests/test_burn_in_tests.py b/buildscripts/tests/test_burn_in_tests.py
index fae23d10dba..18ba3f1b210 100644
--- a/buildscripts/tests/test_burn_in_tests.py
+++ b/buildscripts/tests/test_burn_in_tests.py
@@ -4,6 +4,7 @@ from __future__ import absolute_import
 
 import collections
 import datetime
+import json
 import os
 import sys
 import subprocess
@@ -14,7 +15,7 @@ from mock import Mock, patch, MagicMock
 
 import requests
 
-from shrub.config import Configuration
+from shrub.v2 import ShrubProject, BuildVariant
 
 import buildscripts.burn_in_tests as under_test
 from buildscripts.ciconfig.evergreen import parse_evergreen_file
@@ -78,7 +79,7 @@ class TestAcceptance(unittest.TestCase):
     def tearDown(self):
         _parser.set_options()
 
-    @patch(ns("_write_json_file"))
+    @patch(ns("write_file"))
     def test_no_tests_run_if_none_changed(self, write_json_mock):
         """
         Given a git repository with no changes,
@@ -98,13 +99,13 @@ class TestAcceptance(unittest.TestCase):
         under_test.burn_in(repeat_config, gen_config, "", "testfile.json", False, None, repos, None)
 
         write_json_mock.assert_called_once()
-        written_config = write_json_mock.call_args[0][0]
+        written_config = json.loads(write_json_mock.call_args[0][1])
         display_task = written_config["buildvariants"][0]["display_tasks"][0]
         self.assertEqual(1, len(display_task["execution_tasks"]))
         self.assertEqual(under_test.BURN_IN_TESTS_GEN_TASK, display_task["execution_tasks"][0])
 
     @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
-    @patch(ns("_write_json_file"))
+    @patch(ns("write_file"))
     def test_tests_generated_if_a_file_changed(self, write_json_mock):
         """
         Given a git repository with changes,
@@ -129,7 +130,7 @@ class TestAcceptance(unittest.TestCase):
                            None)
 
         write_json_mock.assert_called_once()
-        written_config = write_json_mock.call_args[0][0]
+        written_config = json.loads(write_json_mock.call_args[0][1])
         n_tasks = len(written_config["tasks"])
         # Ensure we are generating at least one task for the test.
         self.assertGreaterEqual(n_tasks, 1)
@@ -467,31 +468,31 @@ TESTS_BY_TASK = {
 class TestCreateGenerateTasksConfig(unittest.TestCase):
     @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
     def test_no_tasks_given(self):
-        evg_config = Configuration()
+        build_variant = BuildVariant("build variant")
         gen_config = MagicMock(run_build_variant="variant")
         repeat_config = MagicMock()
 
-        evg_config = under_test.create_generate_tasks_config(evg_config, {}, gen_config,
-                                                             repeat_config, None)
+        under_test.create_generate_tasks_config(build_variant, {}, gen_config, repeat_config, None)
 
-        evg_config_dict = evg_config.to_map()
-        self.assertNotIn("tasks", evg_config_dict)
+        evg_config_dict = build_variant.as_dict()
+        self.assertEqual(0, len(evg_config_dict["tasks"]))
 
     @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
     def test_one_task_one_test(self):
         n_tasks = 1
         n_tests = 1
         resmoke_options = "options for resmoke"
-        evg_config = Configuration()
+        build_variant = BuildVariant("build variant")
         gen_config = MagicMock(run_build_variant="variant", distro=None)
         repeat_config = MagicMock()
         repeat_config.generate_resmoke_options.return_value = resmoke_options
         tests_by_task = create_tests_by_task_mock(n_tasks, n_tests)
 
-        evg_config = under_test.create_generate_tasks_config(evg_config, tests_by_task, gen_config,
-                                                             repeat_config, None)
+        under_test.create_generate_tasks_config(build_variant, tests_by_task, gen_config,
+                                                repeat_config, None)
 
-        evg_config_dict = evg_config.to_map()
+        shrub_config = ShrubProject.empty().add_build_variant(build_variant)
+        evg_config_dict = shrub_config.as_dict()
         tasks = evg_config_dict["tasks"]
         self.assertEqual(n_tasks * n_tests, len(tasks))
         cmd = tasks[0]["commands"]
@@ -503,58 +504,30 @@ class TestCreateGenerateTasksConfig(unittest.TestCase):
     def test_n_task_m_test(self):
         n_tasks = 3
         n_tests = 5
-        evg_config = Configuration()
+        build_variant = BuildVariant("build variant")
         gen_config = MagicMock(run_build_variant="variant", distro=None)
         repeat_config = MagicMock()
         tests_by_task = create_tests_by_task_mock(n_tasks, n_tests)
 
-        evg_config = under_test.create_generate_tasks_config(evg_config, tests_by_task, gen_config,
-                                                             repeat_config, None)
+        under_test.create_generate_tasks_config(build_variant, tests_by_task, gen_config,
+                                                repeat_config, None)
 
-        evg_config_dict = evg_config.to_map()
+        evg_config_dict = build_variant.as_dict()
         self.assertEqual(n_tasks * n_tests, len(evg_config_dict["tasks"]))
 
 
 class TestCreateGenerateTasksFile(unittest.TestCase):
     @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
-    @patch("buildscripts.burn_in_tests.create_generate_tasks_config")
-    def test_gen_tasks_configuration_is_returned(self, gen_tasks_config_mock):
+    @patch(ns("sys.exit"))
+    @patch(ns("create_generate_tasks_config"))
+    @patch(ns("validate_task_generation_limit"))
+    def test_cap_on_task_generate(self, validate_mock, _, exit_mock):
         evg_api = MagicMock()
         gen_config = MagicMock(use_multiversion=False)
         repeat_config = MagicMock()
         tests_by_task = MagicMock()
 
-        task_list = [f"task_{i}" for i in range(10)]
-
-        evg_config = MagicMock()
-        evg_config.to_map.return_value = {
-            "tasks": task_list,
-        }
-
-        gen_tasks_config_mock.return_value = evg_config
-
-        config = under_test.create_generate_tasks_file(tests_by_task, gen_config, repeat_config,
-                                                       evg_api)
-
-        self.assertEqual(config, evg_config.to_map.return_value)
-
-    @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
-    @patch("buildscripts.burn_in_tests.sys.exit")
-    @patch("buildscripts.burn_in_tests.create_generate_tasks_config")
-    def test_cap_on_task_generate(self, gen_tasks_config_mock, exit_mock):
-        evg_api = MagicMock()
-        gen_config = MagicMock(use_multiversion=False)
-        repeat_config = MagicMock()
-        tests_by_task = MagicMock()
-
-        task_list = [f"task_{i}" for i in range(1005)]
-
-        evg_config = MagicMock()
-        evg_config.to_map.return_value = {
-            "tasks": task_list,
-        }
-
-        gen_tasks_config_mock.return_value = evg_config
+        validate_mock.return_value = False
 
         exit_mock.side_effect = ValueError("exiting")
         with self.assertRaises(ValueError):
diff --git a/buildscripts/tests/test_burn_in_tests_multiversion.py b/buildscripts/tests/test_burn_in_tests_multiversion.py
index fbd5d422344..04c935cff2e 100644
--- a/buildscripts/tests/test_burn_in_tests_multiversion.py
+++ b/buildscripts/tests/test_burn_in_tests_multiversion.py
@@ -9,10 +9,10 @@ import unittest
 
 from mock import MagicMock, patch
 
-from shrub.config import Configuration
+from shrub.v2 import BuildVariant, ShrubProject
 
 import buildscripts.burn_in_tests_multiversion as under_test
-from buildscripts.burn_in_tests import create_generate_tasks_file, _gather_task_info, create_generate_tasks_config
+from buildscripts.burn_in_tests import _gather_task_info, create_generate_tasks_config
 from buildscripts.ciconfig.evergreen import parse_evergreen_file
 import buildscripts.resmokelib.parser as _parser
 import buildscripts.evergreen_gen_multiversion_tests as gen_multiversion
@@ -102,20 +102,18 @@ def create_variant_task_mock(task_name, suite_name, distro="distro"):
 
 class TestCreateMultiversionGenerateTasksConfig(unittest.TestCase):
     def tests_no_tasks_given(self):
-        evg_config = Configuration()
         gen_config = MagicMock(run_build_variant="variant", fallback_num_sub_suites=1,
                                project="project", build_variant="build_variant", task_id="task_id",
                                target_resmoke_time=60)
         evg_api = MagicMock()
-        evg_config = under_test.create_multiversion_generate_tasks_config(
-            evg_config, {}, evg_api, gen_config)
-        evg_config_dict = evg_config.to_map()
-        self.assertNotIn("tasks", evg_config_dict)
+        build_variant = under_test.create_multiversion_generate_tasks_config({}, evg_api,
+                                                                             gen_config)
+        evg_config_dict = build_variant.as_dict()
+        self.assertEqual(0, len(evg_config_dict["tasks"]))
 
     def test_tasks_not_in_multiversion_suites(self):
         n_tasks = 1
         n_tests = 1
-        evg_config = Configuration()
         gen_config = MagicMock(run_build_variant="variant", fallback_num_sub_suites=1,
                                project="project", build_variant="build_variant", task_id="task_id",
                                target_resmoke_time=60)
@@ -123,28 +121,27 @@ class TestCreateMultiversionGenerateTasksConfig(unittest.TestCase):
 
         # Create a tests_by_tasks dict that doesn't contain any multiversion suites.
         tests_by_task = create_tests_by_task_mock(n_tasks, n_tests)
-        evg_config = under_test.create_multiversion_generate_tasks_config(
-            evg_config, tests_by_task, evg_api, gen_config)
-        evg_config_dict = evg_config.to_map()
+        build_variant = under_test.create_multiversion_generate_tasks_config(
+            tests_by_task, evg_api, gen_config)
+        evg_config_dict = build_variant.as_dict()
 
         # We should not generate any tasks that are not part of the burn_in_multiversion suite.
-        self.assertNotIn("tasks", evg_config_dict)
+        self.assertEqual(0, len(evg_config_dict["tasks"]))
 
     @patch("buildscripts.evergreen_gen_multiversion_tests.get_backports_required_last_stable_hash")
     def test_one_task_one_test(self, mock_hash):
         mock_hash.return_value = MONGO_4_2_HASH
         n_tasks = 1
         n_tests = 1
-        evg_config = Configuration()
         gen_config = MagicMock(run_build_variant="variant", fallback_num_sub_suites=1,
                                project="project", build_variant="build_variant", task_id="task_id",
                                target_resmoke_time=60)
         evg_api = MagicMock()
 
         tests_by_task = create_multiversion_tests_by_task_mock(n_tasks, n_tests)
-        evg_config = under_test.create_multiversion_generate_tasks_config(
-            evg_config, tests_by_task, evg_api, gen_config)
-        evg_config_dict = evg_config.to_map()
+        build_variant = under_test.create_multiversion_generate_tasks_config(
+            tests_by_task, evg_api, gen_config)
+        evg_config_dict = build_variant.as_dict()
         tasks = evg_config_dict["tasks"]
         self.assertEqual(len(tasks), NUM_REPL_MIXED_VERSION_CONFIGS * n_tests)
 
@@ -153,16 +150,15 @@ class TestCreateMultiversionGenerateTasksConfig(unittest.TestCase):
         mock_hash.return_value = MONGO_4_2_HASH
         n_tasks = 2
         n_tests = 1
-        evg_config = Configuration()
         gen_config = MagicMock(run_build_variant="variant", fallback_num_sub_suites=1,
                                project="project", build_variant="build_variant", task_id="task_id",
                                target_resmoke_time=60)
         evg_api = MagicMock()
 
         tests_by_task = create_multiversion_tests_by_task_mock(n_tasks, n_tests)
-        evg_config = under_test.create_multiversion_generate_tasks_config(
-            evg_config, tests_by_task, evg_api, gen_config)
-        evg_config_dict = evg_config.to_map()
+        build_variant = under_test.create_multiversion_generate_tasks_config(
+            tests_by_task, evg_api, gen_config)
+        evg_config_dict = build_variant.as_dict()
         tasks = evg_config_dict["tasks"]
         self.assertEqual(
             len(tasks),
@@ -173,16 +169,15 @@ class TestCreateMultiversionGenerateTasksConfig(unittest.TestCase):
         mock_hash.return_value = MONGO_4_2_HASH
         n_tasks = 1
         n_tests = 2
-        evg_config = Configuration()
         gen_config = MagicMock(run_build_variant="variant", fallback_num_sub_suites=1,
                                project="project", build_variant="build_variant", task_id="task_id",
                                target_resmoke_time=60)
         evg_api = MagicMock()
 
         tests_by_task = create_multiversion_tests_by_task_mock(n_tasks, n_tests)
-        evg_config = under_test.create_multiversion_generate_tasks_config(
-            evg_config, tests_by_task, evg_api, gen_config)
-        evg_config_dict = evg_config.to_map()
+        build_variant = under_test.create_multiversion_generate_tasks_config(
+            tests_by_task, evg_api, gen_config)
+        evg_config_dict = build_variant.as_dict()
         tasks = evg_config_dict["tasks"]
         self.assertEqual(len(tasks), NUM_REPL_MIXED_VERSION_CONFIGS * n_tests)
 
@@ -191,16 +186,15 @@ class TestCreateMultiversionGenerateTasksConfig(unittest.TestCase):
         mock_hash.return_value = MONGO_4_2_HASH
         n_tasks = 2
         n_tests = 3
-        evg_config = Configuration()
         gen_config = MagicMock(run_build_variant="variant", fallback_num_sub_suites=1,
                                project="project", build_variant="build_variant", task_id="task_id",
                                target_resmoke_time=60)
         evg_api = MagicMock()
 
         tests_by_task = create_multiversion_tests_by_task_mock(n_tasks, n_tests)
-        evg_config = under_test.create_multiversion_generate_tasks_config(
-            evg_config, tests_by_task, evg_api, gen_config)
-        evg_config_dict = evg_config.to_map()
+        build_variant = under_test.create_multiversion_generate_tasks_config(
+            tests_by_task, evg_api, gen_config)
+        evg_config_dict = build_variant.as_dict()
         tasks = evg_config_dict["tasks"]
         self.assertEqual(
             len(tasks),
@@ -228,7 +222,7 @@ class TestCreateGenerateTasksConfig(unittest.TestCase):
     def test_multiversion_path_is_used(self):
         n_tasks = 1
         n_tests = 1
-        evg_config = Configuration()
+        build_variant = BuildVariant("variant")
         gen_config = MagicMock(run_build_variant="variant", distro=None)
         repeat_config = MagicMock()
         tests_by_task = create_tests_by_task_mock(n_tasks, n_tests)
@@ -236,46 +230,14 @@ class TestCreateGenerateTasksConfig(unittest.TestCase):
         multiversion_path = "multiversion_path"
         tests_by_task[first_task]["use_multiversion"] = multiversion_path
 
-        evg_config = create_generate_tasks_config(evg_config, tests_by_task, gen_config,
-                                                  repeat_config, None)
+        create_generate_tasks_config(build_variant, tests_by_task, gen_config, repeat_config, None)
 
-        evg_config_dict = evg_config.to_map()
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        evg_config_dict = shrub_project.as_dict()
         tasks = evg_config_dict["tasks"]
         self.assertEqual(n_tasks * n_tests, len(tasks))
         self.assertEqual(multiversion_path, tasks[0]["commands"][2]["vars"]["task_path_suffix"])
 
-    @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
-    @patch(bit_ns("create_generate_tasks_config"))
-    def test_gen_tasks_multiversion_configuration_is_returned(self, gen_tasks_config_mock):  # pylint: disable=invalid-name
-        evg_api = MagicMock()
-        gen_config = MagicMock(run_build_variant="variant", project="project",
-                               build_variant="build_variant", task_id="task_id",
-                               use_multiversion=True)
-        repeat_config = MagicMock()
-        tests_by_task = MagicMock()
-
-        evg_config = MagicMock()
-        evg_config.to_map.return_value = {
-            'buildvariants': [
-                {
-                    'name': 'build_variant',
-                    'display_tasks': [
-                        {
-                            'name': 'burn_in_tests_multiversion',
-                            'execution_tasks': [
-                                'burn_in_tests_multiversion_gen'
-                            ]
-                        }
-                    ]
-                }
-            ]
-        }  # yapf: disable
-
-        gen_tasks_config_mock.return_value = evg_config
-
-        config = create_generate_tasks_file(tests_by_task, gen_config, repeat_config, evg_api)
-        self.assertEqual(config, evg_config.to_map.return_value)
-
 
 class TestGatherTaskInfo(unittest.TestCase):
     def test_multiversion_task(self):
diff --git a/buildscripts/tests/test_evergreen_gen_fuzzer_tests.py b/buildscripts/tests/test_evergreen_gen_fuzzer_tests.py
index 688ecf39719..012b9777194 100644
--- a/buildscripts/tests/test_evergreen_gen_fuzzer_tests.py
+++ b/buildscripts/tests/test_evergreen_gen_fuzzer_tests.py
@@ -3,14 +3,14 @@
 import unittest
 import mock
 
-from shrub.config import Configuration
+from shrub.v2 import BuildVariant, ShrubProject
 
-from buildscripts import evergreen_gen_fuzzer_tests as gft
+from buildscripts import evergreen_gen_fuzzer_tests as under_test
 
 # pylint: disable=missing-docstring,protected-access
 
 
-class TestGenerateEvgTasks(unittest.TestCase):
+class TestCreateFuzzerTask(unittest.TestCase):
     @staticmethod
     def _create_options_mock():
         options = mock.Mock
@@ -31,10 +31,12 @@ class TestGenerateEvgTasks(unittest.TestCase):
         return options
 
     def test_evg_config_is_created_without_multiversion(self):
-        evg_config = Configuration()
+        build_variant = BuildVariant("build variant")
         options = self._create_options_mock()
 
-        config = gft.generate_evg_tasks(options, evg_config).to_map()
+        under_test.create_fuzzer_task(options, build_variant)
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
 
         self.assertEqual(options.num_tasks, len(config["tasks"]))
 
@@ -54,11 +56,13 @@ class TestGenerateEvgTasks(unittest.TestCase):
         self.assertIn(options.name + "_gen", buildvariant["display_tasks"][0]["execution_tasks"])
 
     def test_evg_config_is_created_with_multiversion(self):
-        evg_config = Configuration()
+        build_variant = BuildVariant("build variant")
         options = self._create_options_mock()
         options.use_multiversion = "/data/multiversion"
 
-        config = gft.generate_evg_tasks(options, evg_config).to_map()
+        under_test.create_fuzzer_task(options, build_variant)
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
 
         self.assertEqual("do multiversion setup", config["tasks"][0]["commands"][1]["func"])
         self.assertEqual("/data/multiversion",
diff --git a/buildscripts/tests/test_evergreen_generate_resmoke_tasks.py b/buildscripts/tests/test_evergreen_generate_resmoke_tasks.py
index e0f84f7c06d..cbfe65402e1 100644
--- a/buildscripts/tests/test_evergreen_generate_resmoke_tasks.py
+++ b/buildscripts/tests/test_evergreen_generate_resmoke_tasks.py
@@ -10,7 +10,7 @@ import unittest
 import requests
 import yaml
 from mock import patch, MagicMock
-from shrub.config import Configuration
+from shrub.v2 import BuildVariant, ShrubProject
 from shrub.variant import DisplayTaskDefinition
 
 from buildscripts.util.teststats import TestRuntime
@@ -350,17 +350,6 @@ class TestConfigOptions(unittest.TestCase):
         self.assertEqual(1, config_options.number)
         self.assertIsInstance(config_options.number, int)
 
-    def test_generate_display_task(self):
-        config = {"task_name": "my_task"}
-
-        config_options = under_test.ConfigOptions(config)
-        display_task = config_options.generate_display_task(["task_1", "task_2"])
-
-        self.assertEqual("my_task", display_task._name)
-        self.assertIn(config_options.task + "_gen", display_task.to_map()["execution_tasks"])
-        self.assertIn("task_1", display_task.to_map()["execution_tasks"])
-        self.assertIn("task_2", display_task.to_map()["execution_tasks"])
-
 
 class DivideRemainingTestsAmongSuitesTest(unittest.TestCase):
     @staticmethod
@@ -607,13 +596,15 @@ class EvergreenConfigGeneratorTest(unittest.TestCase):
         return options
 
     def test_evg_config_is_created(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         suites = self.generate_mock_suites(3)
+        build_variant = BuildVariant("variant")
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
 
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
         self.assertEqual(len(config["tasks"]), len(suites) + 1)
         command1 = config["tasks"][0]["commands"][2]
         self.assertIn(options.resmoke_args, command1["vars"]["resmoke_args"])
@@ -622,33 +613,38 @@ class EvergreenConfigGeneratorTest(unittest.TestCase):
         self.assertEqual("run generated tests", command1["func"])
 
     def test_evg_config_is_created_with_diff_task_and_suite(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.task = "task"
+        options.display_task_name = "display task"
         options.generate_display_task.return_value = DisplayTaskDefinition("task")
         suites = self.generate_mock_suites(3)
+        build_variant = BuildVariant("variant")
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
 
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
         self.assertEqual(len(config["tasks"]), len(suites) + 1)
         display_task = config["buildvariants"][0]["display_tasks"][0]
-        self.assertEqual(options.task, display_task["name"])
+        self.assertEqual(options.display_task_name, display_task["name"])
 
         task = config["tasks"][0]
         self.assertIn(options.variant, task["name"])
         self.assertIn(options.suite, task["commands"][2]["vars"]["resmoke_args"])
 
     def test_evg_config_can_use_large_distro(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.use_large_distro = "true"
         options.large_distro_name = "large distro name"
-
         suites = self.generate_mock_suites(3)
+        build_variant = BuildVariant("variant")
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
+
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
 
         self.assertEqual(len(config["tasks"]), len(suites) + 1)
         self.assertEqual(options.large_distro_name,
@@ -665,12 +661,10 @@ class EvergreenConfigGeneratorTest(unittest.TestCase):
         self.assertTrue(is_task_dependency("sharding", "sharding_misc"))
 
     def test_get_tasks_depends_on(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         suites = self.generate_mock_suites(3)
 
-        cfg_generator = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                            MagicMock())
+        cfg_generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
         cfg_generator.build_tasks = [
             MagicMock(display_name="sharding_gen"),
             MagicMock(display_name="sharding_0"),
@@ -688,14 +682,12 @@ class EvergreenConfigGeneratorTest(unittest.TestCase):
         self.assertIn("sharding_misc", dependent_tasks)
 
     def test_specified_dependencies_are_added(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.depends_on = ["sharding"]
         options.is_patch = False
         suites = self.generate_mock_suites(3)
 
-        cfg_generator = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                            MagicMock())
+        cfg_generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
         cfg_generator.build_tasks = [
             MagicMock(display_name="sharding_gen"),
             MagicMock(display_name="sharding_0"),
@@ -706,19 +698,20 @@ class EvergreenConfigGeneratorTest(unittest.TestCase):
             MagicMock(display_name="sharding_misc"),
         ]
 
-        cfg_mock = MagicMock()
-        cfg_generator._add_dependencies(cfg_mock)
-        self.assertEqual(4, cfg_mock.dependency.call_count)
+        dependencies = cfg_generator._get_dependencies()
+        self.assertEqual(4, len(dependencies))
 
     def test_evg_config_has_timeouts_for_repeated_suites(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.repeat_suites = 5
         suites = self.generate_mock_suites(3)
+        build_variant = BuildVariant("variant")
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
 
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
         self.assertEqual(len(config["tasks"]), len(suites) + 1)
         command1 = config["tasks"][0]["commands"][2]
         self.assertIn(" --repeatSuites=5 ", command1["vars"]["resmoke_args"])
@@ -731,74 +724,86 @@ class EvergreenConfigGeneratorTest(unittest.TestCase):
         self.assertEqual(expected_exec_timeout, timeout_cmd["params"]["exec_timeout_secs"])
 
     def test_evg_config_has_fails_if_timeout_too_high(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.repeat_suites = under_test.MAX_EXPECTED_TIMEOUT
         suites = self.generate_mock_suites(3)
 
         with self.assertRaises(ValueError):
-            under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                MagicMock()).generate_config()
+            generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+            generator.generate_config(MagicMock())
 
     def test_evg_config_does_not_fails_if_timeout_too_high_on_mainline(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.is_patch = False
         options.repeat_suites = under_test.MAX_EXPECTED_TIMEOUT
         suites = self.generate_mock_suites(3)
+        build_variant = BuildVariant("variant")
+
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        config = build_variant.as_dict()
         self.assertEqual(len(config["tasks"]), len(suites) + 1)
 
     def test_evg_config_does_not_overwrite_repeatSuites_resmoke_arg_with_repeatSuites_default(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.resmoke_args = "resmoke_args --repeatSuites=5"
         suites = self.generate_mock_suites(1)
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        build_variant = BuildVariant("variant")
+
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
+
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
         command1 = config["tasks"][0]["commands"][2]
         self.assertIn("--repeatSuites=5", command1["vars"]["resmoke_args"])
         self.assertNotIn("--repeatSuites=1", command1["vars"]["resmoke_args"])
 
     def test_evg_config_does_not_overwrite_repeat_resmoke_arg_with_repeatSuites_default(self):
-        shrub_config = Configuration()
         options = self.generate_mock_options()
         options.resmoke_args = "resmoke_args --repeat=5"
         suites = self.generate_mock_suites(1)
+        build_variant = BuildVariant("variant")
+
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
         command1 = config["tasks"][0]["commands"][2]
         self.assertIn("--repeat=5", command1["vars"]["resmoke_args"])
         self.assertNotIn("--repeatSuites=1", command1["vars"]["resmoke_args"])
 
     def test_suites_without_enough_info_should_not_include_timeouts(self):
-        shrub_config = Configuration()
         suite_without_timing_info = 1
         options = self.generate_mock_options()
         suites = self.generate_mock_suites(3)
         suites[suite_without_timing_info].should_overwrite_timeout.return_value = False
+        build_variant = BuildVariant("variant")
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
 
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
         timeout_cmd = config["tasks"][suite_without_timing_info]["commands"][0]
         self.assertNotIn("command", timeout_cmd)
         self.assertEqual("do setup", timeout_cmd["func"])
 
     def test_timeout_info_not_included_if_use_default_timeouts_set(self):
-        shrub_config = Configuration()
         suite_without_timing_info = 1
         options = self.generate_mock_options()
         suites = self.generate_mock_suites(3)
         options.use_default_timeouts = True
+        build_variant = BuildVariant("variant")
 
-        config = under_test.EvergreenConfigGenerator(shrub_config, suites, options,
-                                                     MagicMock()).generate_config().to_map()
+        generator = under_test.EvergreenConfigGenerator(suites, options, MagicMock())
+        generator.generate_config(build_variant)
 
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
+        config = shrub_project.as_dict()
         timeout_cmd = config["tasks"][suite_without_timing_info]["commands"][0]
         self.assertNotIn("command", timeout_cmd)
         self.assertEqual("do setup", timeout_cmd["func"])
diff --git a/buildscripts/tests/test_selected_tests.py b/buildscripts/tests/test_selected_tests.py
index 5d2af9ecba3..7d76e728410 100644
--- a/buildscripts/tests/test_selected_tests.py
+++ b/buildscripts/tests/test_selected_tests.py
@@ -1,10 +1,11 @@
 """Unit tests for the selected_tests script."""
-import os
+import json
 import sys
 import unittest
+from typing import Dict, Any
 
 from mock import MagicMock, patch
-from shrub.config import Configuration
+from shrub.v2 import BuildVariant, ShrubProject
 
 # pylint: disable=wrong-import-position
 import buildscripts.ciconfig.evergreen as _evergreen
@@ -40,6 +41,16 @@ def tests_by_task_stub():
     }
 
 
+def empty_build_variant(variant_name: str) -> Dict[str, Any]:
+    return {
+        "buildvariants": [{
+            "name": variant_name,
+            "tasks": [],
+        }],
+        "tasks": [],
+    }
+
+
 class TestAcceptance(unittest.TestCase):
     """A suite of Acceptance tests for selected_tests."""
 
@@ -70,7 +81,9 @@ class TestAcceptance(unittest.TestCase):
                                      selected_tests_variant_expansions, repos,
                                      origin_build_variants)
 
-        self.assertEqual(config_dict["selected_tests_config.json"], "{}")
+        self.assertEqual(
+            json.loads(config_dict["selected_tests_config.json"]),
+            empty_build_variant(origin_build_variants[0]))
 
     @unittest.skipIf(sys.platform.startswith("win"), "not supported on windows")
     def test_when_test_mappings_are_found_for_changed_files(self):
@@ -211,17 +224,6 @@ class TestSelectedTestsConfigOptions(unittest.TestCase):
 
         self.assertFalse(config_options.create_misc_suite)
 
-    @patch(ns("read_config"))
-    def test_generate_display_task(self, read_config_mock):
-        config_options = under_test.SelectedTestsConfigOptions(
-            {"task_name": "my_task", "build_variant": "my_variant"}, {}, {}, {})
-
-        display_task = config_options.generate_display_task(["task_1", "task_2"])
-
-        self.assertEqual("my_task_my_variant", display_task._name)
-        self.assertIn("task_1", display_task.to_map()["execution_tasks"])
-        self.assertIn("task_2", display_task.to_map()["execution_tasks"])
-
 
 class TestFindSelectedTestFiles(unittest.TestCase):
     @patch(ns("is_file_a_test_file"))
@@ -431,27 +433,6 @@ class TestGetEvgTaskConfig(unittest.TestCase):
 class TestUpdateConfigDictWithTask(unittest.TestCase):
     @patch(ns("SelectedTestsConfigOptions"))
     @patch(ns("GenerateSubSuites"))
-    def test_suites_and_tasks_are_generated(self, generate_subsuites_mock,
-                                            selected_tests_config_options_mock):
-        suites_config_mock = {"my_suite_0.yml": "suite file contents"}
-        generate_subsuites_mock.return_value.generate_suites_config.return_value = suites_config_mock
-
-        def generate_task_config(shrub_config, suites):
-            shrub_config.task("my_fake_task")
-
-        generate_subsuites_mock.return_value.generate_task_config.side_effect = generate_task_config
-
-        shrub_config = Configuration()
-        config_dict_of_suites_and_tasks = {}
-        under_test._update_config_with_task(
-            evg_api=MagicMock(), shrub_config=shrub_config, config_options=MagicMock(),
-            config_dict_of_suites_and_tasks=config_dict_of_suites_and_tasks)
-
-        self.assertEqual(config_dict_of_suites_and_tasks, suites_config_mock)
-        self.assertIn("my_fake_task", shrub_config.to_json())
-
-    @patch(ns("SelectedTestsConfigOptions"))
-    @patch(ns("GenerateSubSuites"))
     def test_no_suites_or_tasks_are_generated(self, generate_subsuites_mock,
                                               selected_tests_config_options_mock):
         generate_subsuites_mock.return_value.generate_suites_config.return_value = {}
@@ -461,14 +442,15 @@ class TestUpdateConfigDictWithTask(unittest.TestCase):
 
         generate_subsuites_mock.return_value.generate_task_config.side_effect = generate_task_config
 
-        shrub_config = Configuration()
+        build_variant = BuildVariant("variant")
         config_dict_of_suites_and_tasks = {}
         under_test._update_config_with_task(
-            evg_api=MagicMock(), shrub_config=shrub_config, config_options=MagicMock(),
+            MagicMock(), build_variant, config_options=MagicMock(),
             config_dict_of_suites_and_tasks=config_dict_of_suites_and_tasks)
 
+        shrub_project = ShrubProject.empty().add_build_variant(build_variant)
         self.assertEqual(config_dict_of_suites_and_tasks, {})
-        self.assertEqual(shrub_config.to_json(), "{}")
+        self.assertEqual(shrub_project.as_dict(), empty_build_variant("variant"))
 
 
 class TestGetTaskConfigsForTestMappings(unittest.TestCase):
diff --git a/buildscripts/util/fileops.py b/buildscripts/util/fileops.py
index e56348d8548..e26e67c6593 100644
--- a/buildscripts/util/fileops.py
+++ b/buildscripts/util/fileops.py
@@ -25,3 +25,30 @@ def get_file_handle(path, append_file=False):
     """Open 'path', truncate it if 'append_file' is False, and return file handle."""
     mode = "a+" if append_file else "w"
     return open(path, mode)
+
+
+def write_file(path: str, contents: str) -> None:
+    """
+    Write the contents provided to the file in the specified path.
+
+    :param path: Path of file to write.
+    :param contents: Contents to write to file.
+    """
+    with open(path, "w") as file_handle:
+        file_handle.write(contents)
+
+
+def write_file_to_dir(directory: str, file: str, contents: str) -> None:
+    """
+    Write the contents provided to the file in the given directory.
+
+    The directory will be created if it does not exist.
+
+    :param directory: Directory to write to.
+    :param file: Name of file to write.
+    :param contents: Contents to write to file.
+    """
+    if not os.path.exists(directory):
+        os.makedirs(directory)
+
+    write_file(os.path.join(directory, file), contents)