diff --git a/monitoring/benchmarker/README.md b/monitoring/benchmarker/README.md
new file mode 100644
index 0000000000..320c197e4c
--- /dev/null
+++ b/monitoring/benchmarker/README.md
@@ -0,0 +1,15 @@
+# benchmarker
+
+`benchmarker` is a tool for evaluating the performance, throughput, and stability of UTM systems under load.
+
+## Running benchmarker
+
+To execute `benchmarker`, run the following command from the root of the `monitoring` repo:
+
+```bash
+PYTHONPATH=. uv run python monitoring/benchmarker/benchmark.py --config file://monitoring/benchmarker/configurations/interuss/isas_uncontended.jsonnet
+```
+
+## Security
+
+Warning: benchmarker has the capability of running shell commands specified in the configuration (see `actions`).  Treat the configuration as code and review at an appropriate level of scrutiny before executing.
diff --git a/monitoring/benchmarker/benchmark.py b/monitoring/benchmarker/benchmark.py
new file mode 100755
index 0000000000..addd267899
--- /dev/null
+++ b/monitoring/benchmarker/benchmark.py
@@ -0,0 +1,80 @@
+#!env/bin/python3
+
+import argparse
+import json
+import os
+import sys
+
+from implicitdict import ImplicitDict
+from loguru import logger
+
+from monitoring.benchmarker.configurations.configuration import BenchmarkConfiguration
+from monitoring.benchmarker.validation import validate_config
+from monitoring.uss_qualifier.fileio import (
+    load_dict_with_references,
+)  # TODO: factor fileio out of uss_qualifier into monitorlib
+
+
+def parseArgs() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Execute benchmarker")
+
+    parser.add_argument(
+        "--config",
+        help="Configuration string indicating file reference (e.g. file://path/to/config.jsonnet)",
+        required=True,
+    )
+
+    parser.add_argument(
+        "--skip-validation",
+        action="store_true",
+        help="If specified, do not validate the format of the provided configuration.",
+    )
+
+    return parser.parse_args()
+
+
+def run_config(
+    config_name: str,
+    skip_validation: bool,
+) -> int:
+    config_src = load_dict_with_references(config_name)
+
+    if not skip_validation:
+        logger.info("Validating configuration...")
+        validation_errors = validate_config(config_src)
+        if validation_errors:
+            for e in validation_errors:
+                logger.error("[{}]: {}", e.json_path, e.message)
+            raise ValueError(
+                f"{len(validation_errors)} benchmark configuration validation errors indicated above.  Hint: resolve the clearest error first and then rerun validation."
+            )
+
+    whole_config = ImplicitDict.parse(config_src, BenchmarkConfiguration)
+
+    # Print the resolved configuration to console as JSON
+    print(json.dumps(whole_config, indent=2, sort_keys=True))
+
+    return os.EX_OK
+
+
+def main() -> int:
+    args = parseArgs()
+
+    logger.info(
+        f"========== Running benchmarker for configuration {args.config} =========="
+    )
+    exit_code = run_config(
+        args.config,
+        args.skip_validation,
+    )
+    if exit_code != os.EX_OK:
+        return exit_code
+    logger.info(
+        f"========== Completed benchmarker for configuration {args.config} =========="
+    )
+
+    return os.EX_OK
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/monitoring/benchmarker/validation.py b/monitoring/benchmarker/validation.py
new file mode 100644
index 0000000000..90a3c3b251
--- /dev/null
+++ b/monitoring/benchmarker/validation.py
@@ -0,0 +1,26 @@
+from monitoring.benchmarker.configurations.configuration import (
+    BenchmarkConfiguration,
+)
+from monitoring.monitorlib.schema_validation import (
+    ValidationError,
+    validate_implicitdict_object,
+)
+from monitoring.uss_qualifier.validation import validate_resource_declarations
+
+
+def validate_config(config: dict) -> list[ValidationError]:
+    """Validate raw data intended to be used to create a BenchmarkConfiguration.
+
+    Args:
+        config: Raw nested dict object intended to be parsed into a BenchmarkConfiguration.
+
+    Returns: List of validation errors discovered.
+    """
+    result = validate_implicitdict_object(config, BenchmarkConfiguration)
+
+    if not result:
+        result.extend(
+            validate_resource_declarations(config, "$.resources.resource_declarations")
+        )
+
+    return result
diff --git a/monitoring/uss_qualifier/validation.py b/monitoring/uss_qualifier/validation.py
index 3ddc5a6ab8..90ab9ab81d 100644
--- a/monitoring/uss_qualifier/validation.py
+++ b/monitoring/uss_qualifier/validation.py
@@ -11,6 +11,52 @@
 from monitoring.uss_qualifier.resources.resource import get_resource_types
 
 
+def validate_resource_declarations(
+    config: dict, base_path: str
+) -> list[ValidationError]:
+    """Validate resource declarations located at base_path in config against their concrete specification schemas.
+
+    Args:
+        config: Raw nested dictionary containing resource declarations at base_path.
+        base_path: JSON path string indicating where resource_declarations is located (e.g. '$.resources.resource_declarations').
+
+    Returns: List of validation errors discovered across the resource declarations.
+    """
+    result: list[ValidationError] = []
+    resource_declarations = config
+    for child in base_path.split(".")[1:]:
+        resource_declarations = resource_declarations.get(child, {})
+    for resource_id, declaration in resource_declarations.items():
+        declaration = ImplicitDict.parse(declaration, ResourceDeclaration)
+        path = base_path + "." + resource_id
+        try:
+            _, specification_type = get_resource_types(declaration)
+        except ValueError as e:
+            result.append(ValidationError(message=str(e), json_path=path))
+            continue
+        if specification_type is not None:
+            for e in validate_implicitdict_object(
+                declaration.specification, specification_type
+            ):
+                subpath = e.json_path
+                if subpath.startswith("$."):
+                    subpath = subpath[2:]
+                result.append(
+                    ValidationError(
+                        message=e.message,
+                        json_path=path + ".specification." + subpath,
+                    )
+                )
+        elif declaration.specification:
+            result.append(
+                ValidationError(
+                    message=f"Resource type {declaration.resource_type} does not accept a specification, but one was provided",
+                    json_path=path + ".specification",
+                )
+            )
+    return result
+
+
 def validate_config(config: dict) -> list[ValidationError]:
     """Validate raw data intended to be used to create a USSQualifierConfiguration.
 
@@ -22,37 +68,10 @@ def validate_config(config: dict) -> list[ValidationError]:
     result = validate_implicitdict_object(config, USSQualifierConfiguration)
 
     if not result:
-        base_path = "$.v1.test_run.resources.resource_declarations"
-        resource_declarations = config
-        for child in base_path.split(".")[1:]:
-            resource_declarations = resource_declarations.get(child, {})
-        for resource_id, declaration in resource_declarations.items():
-            declaration = ImplicitDict.parse(declaration, ResourceDeclaration)
-            path = base_path + "." + resource_id
-            try:
-                _, specification_type = get_resource_types(declaration)
-            except ValueError as e:
-                result.append(ValidationError(message=str(e), json_path=path))
-                continue
-            if specification_type is not None:
-                for e in validate_implicitdict_object(
-                    declaration.specification, specification_type
-                ):
-                    subpath = e.json_path
-                    if subpath.startswith("$."):
-                        subpath = subpath[2:]
-                    result.append(
-                        ValidationError(
-                            message=e.message,
-                            json_path=path + ".specification." + subpath,
-                        )
-                    )
-            elif declaration.specification:
-                result.append(
-                    ValidationError(
-                        message=f"Resource type {declaration.resource_type} does not accept a specification, but one was provided",
-                        json_path=path + ".specification",
-                    )
-                )
+        result.extend(
+            validate_resource_declarations(
+                config, "$.v1.test_run.resources.resource_declarations"
+            )
+        )
 
     return result