diff --git a/monitoring/benchmarker/__init__.py b/monitoring/benchmarker/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/monitoring/benchmarker/configurations/__init__.py b/monitoring/benchmarker/configurations/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/monitoring/benchmarker/configurations/actions.py b/monitoring/benchmarker/configurations/actions.py
new file mode 100644
index 0000000000..bd37a99bf0
--- /dev/null
+++ b/monitoring/benchmarker/configurations/actions.py
@@ -0,0 +1,28 @@
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+
+class BenchmarkActionName(str):
+    """Unique (within benchmark configuration) name for an action to perform, for instance at setup and/or teardown of one or more scenarios."""
+
+    pass
+
+
+class RunCommandActionSpecification(ImplicitDict):
+    """Shell command to run as a benchmark action."""
+
+    command: str
+    """Shell command to run."""
+
+    path: str
+    """Working folder in which to run the command.  `$REPO_ROOT` will be replaced with the root folder of the repo."""
+
+    env: dict[str, str]
+    """Override each environment variable key with the specified value before running the command."""
+
+
+class BenchmarkActionSpecification(ImplicitDict):
+    name: BenchmarkActionName
+
+    run_command: Optional[RunCommandActionSpecification]
diff --git a/monitoring/benchmarker/configurations/artifacts/__init__.py b/monitoring/benchmarker/configurations/artifacts/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/monitoring/benchmarker/configurations/artifacts/artifact.py b/monitoring/benchmarker/configurations/artifacts/artifact.py
new file mode 100644
index 0000000000..dbad71223d
--- /dev/null
+++ b/monitoring/benchmarker/configurations/artifacts/artifact.py
@@ -0,0 +1,15 @@
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+from monitoring.benchmarker.configurations.artifacts.matplotlib_figure import (
+    MatplotlibFigureSpecification,
+)
+from monitoring.benchmarker.configurations.artifacts.raw_report import (
+    RawReportSpecification,
+)
+
+
+class ArtifactSpecification(ImplicitDict):
+    raw_report: Optional[RawReportSpecification]
+    matplotlib_figure: Optional[MatplotlibFigureSpecification]
diff --git a/monitoring/benchmarker/configurations/artifacts/matplotlib_figure.py b/monitoring/benchmarker/configurations/artifacts/matplotlib_figure.py
new file mode 100644
index 0000000000..f610209bc1
--- /dev/null
+++ b/monitoring/benchmarker/configurations/artifacts/matplotlib_figure.py
@@ -0,0 +1,73 @@
+from enum import StrEnum
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+from monitoring.monitorlib.expressions.types import ASTExpression, SymbolExpression
+
+
+class AxisSpecification(ImplicitDict):
+    label: Optional[str]
+
+
+class XYPlotType(StrEnum):
+    Scatter = "Scatter"
+
+
+class XYPlotSpecification(ImplicitDict):
+    """Specification for a plot (artist) to show on a matplotlib Axes.
+
+    When evaluating expressions, the BenchmarkRunReport will be available as the `report` symbol."""
+
+    type: XYPlotType
+
+    evaluation_context: Optional[list[SymbolExpression]]
+    """Symbols available to other expressions in this plot specification."""
+
+    x_data_expr: Optional[ASTExpression]
+    """List of X data values for XY points.
+    
+    Must have the same number of entries as y_data.
+    Defaults to 1, 2, 3, ..., N for N y_data values."""
+
+    y_data_expr: ASTExpression
+    """List of Y data values for XY points."""
+
+    render_expr: Optional[ASTExpression]
+    """If specified, whether this plot should be rendered (boolean).  Default true."""
+
+
+class SubplotSpecification(ImplicitDict):
+    title: Optional[str]
+
+    evaluation_context: Optional[list[SymbolExpression]]
+    """Symbols available to other expressions in this subplot specification."""
+
+    x_axis: Optional[AxisSpecification]
+    y_axis: Optional[AxisSpecification]
+    xy_plots: list[XYPlotSpecification]
+
+
+class SubfigureSpecification(ImplicitDict):
+    title: Optional[str]
+
+    n_subplot_rows: int = 1
+    n_subplot_cols: int = 1
+
+    evaluation_context: Optional[list[SymbolExpression]]
+    """Symbols available to other expressions in this subfigure specification."""
+
+    subplots: list[SubplotSpecification]
+
+
+class MatplotlibFigureSpecification(ImplicitDict):
+    name: str
+    """Machine-level name for this figure.  Used as the output file name."""
+
+    n_subfigure_rows: int = 1
+    n_subfigure_cols: int = 1
+
+    evaluation_context: Optional[list[SymbolExpression]]
+    """Symbols available to other expressions in this figure specification."""
+
+    subfigures: list[SubfigureSpecification]
diff --git a/monitoring/benchmarker/configurations/artifacts/raw_report.py b/monitoring/benchmarker/configurations/artifacts/raw_report.py
new file mode 100644
index 0000000000..9ece4b4b2d
--- /dev/null
+++ b/monitoring/benchmarker/configurations/artifacts/raw_report.py
@@ -0,0 +1,6 @@
+from implicitdict import ImplicitDict
+
+
+class RawReportSpecification(ImplicitDict):
+    name: str
+    """Machine-level name for this report.  Used as the output file name."""
diff --git a/monitoring/benchmarker/configurations/configuration.py b/monitoring/benchmarker/configurations/configuration.py
new file mode 100644
index 0000000000..052e41fb33
--- /dev/null
+++ b/monitoring/benchmarker/configurations/configuration.py
@@ -0,0 +1,36 @@
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+from monitoring.benchmarker.configurations.actions import (
+    BenchmarkActionSpecification,
+)
+from monitoring.benchmarker.configurations.artifacts.artifact import (
+    ArtifactSpecification,
+)
+from monitoring.benchmarker.configurations.loads import BenchmarkLoadSpecification
+from monitoring.benchmarker.configurations.scenarios import (
+    BenchmarkScenarioSpecification,
+)
+from monitoring.benchmarker.configurations.users import BenchmarkUserSpecification
+from monitoring.uss_qualifier.resources.definitions import ResourceCollection
+
+
+class BenchmarkConfiguration(ImplicitDict):
+    resources: Optional[ResourceCollection]
+    """Pool of uss_qualifier resources available for use by other resources in this benchmark configuration."""
+
+    actions: Optional[list[BenchmarkActionSpecification]]
+    """Actions available to be performed during the benchmarker run."""
+
+    user_types: list[BenchmarkUserSpecification]
+    """Types of users available to load the system under test during the benchmarker run."""
+
+    loads: list[BenchmarkLoadSpecification]
+    """Loads that can be applied to the system under test during the benchmarker run."""
+
+    scenarios: list[BenchmarkScenarioSpecification]
+    """The sequential list of scenarios to perform in the benchmarker run."""
+
+    artifacts: Optional[list[ArtifactSpecification]]
+    """Artifacts to produce from the data collected during the benchmarker run."""
diff --git a/monitoring/benchmarker/configurations/interuss/isas_uncontended.jsonnet b/monitoring/benchmarker/configurations/interuss/isas_uncontended.jsonnet
new file mode 100644
index 0000000000..855848b1d9
--- /dev/null
+++ b/monitoring/benchmarker/configurations/interuss/isas_uncontended.jsonnet
@@ -0,0 +1,266 @@
+local num_uss = 3;
+local num_nodes = 3;
+local latencies_ms = [std.max(1, i * 25) for i in std.range(0, 2)];
+local jitter_frac = 0.05;
+
+local nodeIndex = function(uss, node) std.format('%02d', node + num_nodes * (uss - 1));
+
+{
+  resources: {
+    resource_declarations: {
+      utm_auth: {
+        resource_type: 'resources.communications.AuthAdapterResource',
+        specification: {
+          auth_spec: 'DummyOAuth(http://localhost:8085/token,benchmarker)',
+          scopes_authorized: [
+            'rid.service_provider',
+            'rid.display_provider',
+          ],
+        },
+      },
+    } + {
+      ["dss_pool_%d" % uss]: {
+        resource_type: 'resources.astm.f3411.DSSInstancesResource',
+        dependencies: {
+          auth_adapter: 'utm_auth',
+        },
+        specification: {
+          dss_instances: [
+            {
+              participant_id: 'uss%(uss)d_dss%(node)d' % { uss: uss, node: node },
+              base_url: 'http://localhost:80%s/rid/v2' % nodeIndex(uss, node),
+              rid_version: 'F3411-22a',
+            } for node in std.range(1, num_nodes)
+          ],
+        },
+      } for uss in std.range(1, num_uss)
+    },
+  },
+
+  actions: [
+    {
+      name: 'Bring up DSS pool: %(latency)dms latency' % { latency: latency_ms },
+      run_command: {
+        env: {
+          NUM_USS: std.toString(num_uss),
+          NUM_NODES: std.toString(num_nodes),
+          DSS_IMAGE: 'interuss/dss:v0.22.0',
+          DB_TYPE: 'crdb',
+          INTRA_USS_NETEM_CONF: 'delay 600us 40us 25% distribution normal loss 0.0005%',
+          INTER_USS_NETEM_CONF: 'delay %(latency)sms %(jitter)sms 50%% distribution paretonormal loss 0.25%% 15%%' % { latency: latency_ms, jitter: latency_ms * jitter_frac },
+        },
+        path: '$REPO_ROOT',
+        command: 'make start-locally',
+      },
+    }
+    for latency_ms in latencies_ms
+  ] + [
+    {
+      name: 'Tear down DSS pool',
+      run_command: {
+        env: {
+          NUM_USS: std.toString(num_uss),
+          NUM_NODES: std.toString(num_nodes),
+          DB_TYPE: 'crdb',
+        },
+        path: '$REPO_ROOT',
+        command: 'make clean-locally',
+      },
+    }
+  ],
+
+  user_types: [
+    {
+      name: 'FPU%d' % uss, // Flight planner user, DSS instance/USS i
+      flight_planner: {
+        flight_generation: {
+          independent_time_location_shape: {
+            time: {
+              fixed_spacing: '0s',
+            },
+            location: {
+              fixed_location: {
+                horizontal: {lat: 34, lng: -118},
+                vertical: {value: 300, reference: 'W84', units: 'M'},
+              },
+            },
+            shape: {
+              fixed_volumes: {
+                origin_horizontal: {lat: 0, lng: 0},
+                origin_vertical: {value: 0, reference: 'W84', units: 'M'},
+                origin_time: '2026-01-01T00:00:00Z',
+                volumes: [
+                  {
+                    volume: {
+                      outline_polygon: {
+                        vertices: [
+                          {lat: -0.00001, lng: -0.00001},
+                          {lat: 0.00001, lng: -0.00001},
+                          {lat: 0.00001, lng: 0.00001},
+                          {lat: -0.00001, lng: 0.00001},
+                        ],
+                      },
+                      altitude_lower: {value: 0, reference: 'W84', units: 'M'},
+                      altitude_upper: {value: 20, reference: 'W84', units: 'M'},
+                    },
+                    time_start: '2026-01-01T00:00:00Z',
+                    time_end: '2026-01-01T00:00:05Z',
+                  },
+                ],
+              },
+            },
+          },
+        },
+        astm_netrid_behavior: {
+          rid_version: 'F3411-22a',
+          dss_pool: ['dss_pool_%d' % uss],
+          dss_selection_strategy: 'Random',
+          isa_strategy: {
+            isa_per_flight: {
+              before_flight_start: '0s',
+              after_flight_end: '2s',
+            },
+          },
+        },
+      },
+    } for uss in std.range(1, num_uss)
+  ],
+
+  loads: [
+    {
+      name: 'Flight planner ramp for DSS instance %d' % uss,
+      user_ramp: {
+        user_type: 'FPU%d' % uss,
+        initial_users: 10,
+        additional_users_per_step: 5,
+        throughput_stability_criteria: {
+          each_user_completed_at_least: {
+            count: 1,
+            operations: ['workflow.flight_planner.flight'],
+          },
+        },
+        step_completion_criteria: {
+          any_of: [
+            {
+              sampling_duration_at_least: '30s',
+            },
+            {
+              completed_at_least: {
+                count: 100,
+                operations: ['workflow.flight_planner.flight'],
+              },
+            },
+            {
+              average_duration_more_than: {
+                duration: '20s',
+                operations: ['workflow.flight_planner.flight'],
+              },
+            },
+          ],
+          sampling_duration_at_least: '10s',
+          completed_at_least: {
+            count: 5,
+            operations: ['workflow.flight_planner.flight'],
+          }
+        },
+        load_completion_criteria: {
+          any_of: [
+            {
+              throughput_lower_than_peak: {
+                operations: ['workflow.flight_planner.flight'],
+                fraction_of_peak: 0.7,
+              },
+            },
+            {
+              failures_more_than: {
+                count: 10,
+                operations: ['workflow.flight_planner.flight'],
+              }
+            },
+            {
+              most_recent_step: {
+                average_duration_more_than: {
+                  duration: '20s',
+                  operations: ['workflow.flight_planner.flight'],
+                },
+              },
+            },
+            {
+              most_recent_step: {
+                throughput_stability_took_longer_than: '30s',
+              },
+            },
+          ],
+        },
+      },
+    } for uss in std.range(1, num_uss)
+  ],
+
+  scenarios: std.flattenArrays([
+    [
+      {
+        name: 'Instance %(uss)d %(latency)dms internode latency' % { uss: uss, latency: latency_ms },
+        [if uss == 1 then "setup"]: ['Bring up DSS pool: %dms latency' % latency_ms],
+        load: 'Flight planner ramp for DSS instance %d' % uss,
+        [if uss == num_uss then "teardown"]: ['Tear down DSS pool'],
+        metadata: {
+          latency_ms: latency_ms,
+          dss_instance: uss,
+        }
+      } for uss in std.range(1, num_uss)
+    ] for latency_ms in latencies_ms
+  ]),
+
+  artifacts: [
+    {
+      raw_report: {
+        name: 'report',
+      },
+    },
+    {
+      matplotlib_figure: {
+        local aspect_ratio = 16 / 9,
+        local n_cols = std.ceil(std.sqrt(std.length(latencies_ms) * aspect_ratio)),
+        local n_rows = std.ceil(std.length(latencies_ms) / n_cols),
+
+        name: 'throughput_with_latency',
+        n_subfigure_cols: n_cols,
+        n_subfigure_rows: n_rows,
+        subfigures: [
+          {
+            title: '%dms internode latency' % latency_ms,
+            subplots: [
+              {
+                x_axis: {
+                  label: 'Flight planners',
+                },
+                y_axis: {
+                  label: 'Throughput\n(Flights/s)',
+                },
+                xy_plots: [
+                  {
+                    type: 'Scatter',
+                    evaluation_context: [
+                      {
+                        name: 'scenarios',
+                        value: ('[' +
+                          's for s in report.report.scenarios ' +
+                          'if s.metadata["latency_ms"] == %(latency)d and s.metadata["dss_instance"] == %(uss)d' +
+                        ']') %
+                        {latency: latency_ms, uss: uss},
+                      }
+                    ],
+                    render_expr: 'scenarios',
+                    x_data_expr: '[step.load_factor for step in scenarios[0].steps]',
+                    y_data_expr: '[throughput_of_step(scenarios[0], s, types=["workflow.flight_planner.flight"])' +
+                                 ' for s in range(len(scenarios[0].steps))]',
+                  } for uss in std.range(1, num_uss)
+                ],
+              }
+            ],
+          } for latency_ms in latencies_ms
+        ],
+      },
+    },
+  ],
+}
diff --git a/monitoring/benchmarker/configurations/loads.py b/monitoring/benchmarker/configurations/loads.py
new file mode 100644
index 0000000000..3b2f5822ba
--- /dev/null
+++ b/monitoring/benchmarker/configurations/loads.py
@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+from enum import StrEnum
+from typing import Optional
+
+from implicitdict import ImplicitDict, StringBasedTimeDelta
+
+from monitoring.benchmarker.configurations.users import BenchmarkUserName
+from monitoring.monitorlib.fetch import QueryType
+
+
+class BenchmarkLoadName(str):
+    """Unique (within benchmark configuration) name for a load profile."""
+
+
+class WorkflowType(StrEnum):
+    """Type of operation, other than an HTTP query, providing load to a system being benchmarked."""
+
+    FlightPlannerFlight = "flight_planner.flight"
+    """An operation consisting of managing a flight from end to end, including all associated UTM actions like establishing an operational intent and deleting it."""
+
+
+class OperationType(str):
+    """Type of operation providing load to a system being benchmarked."""
+
+    query_type: QueryType | None = None
+    workflow_type: WorkflowType | None = None
+
+    def __new__(
+        cls, value: str | QueryType | WorkflowType | OperationType | None
+    ) -> OperationType:
+        if isinstance(value, OperationType):
+            obj = str.__new__(cls, value)
+            obj.query_type = value.query_type
+            obj.workflow_type = value.workflow_type
+            return obj
+
+        query_type = None
+        workflow_type = None
+
+        if isinstance(value, QueryType):
+            query_type = value
+            str_val = f"query.{value.value}"
+        elif isinstance(value, WorkflowType):
+            workflow_type = value
+            str_val = f"workflow.{value.value}"
+        elif isinstance(value, str):
+            if value.startswith("query."):
+                raw_enum = value[len("query.") :]
+                query_type = QueryType(raw_enum)
+                str_val = value
+            elif value.startswith("workflow."):
+                raw_enum = value[len("workflow.") :]
+                workflow_type = WorkflowType(raw_enum)
+                str_val = value
+            else:
+                raise ValueError(f"Invalid OperationType string value '{value}'")
+        else:
+            raise ValueError(
+                f"Cannot construct OperationType from {type(value).__name__} '{value}'"
+            )
+
+        obj = str.__new__(cls, str_val)
+        obj.query_type = query_type
+        obj.workflow_type = workflow_type
+        return obj
+
+
+class OperationCount(ImplicitDict):
+    count: int
+    """Number of matching operations."""
+
+    operations: list[OperationType]
+    """Particular operations to look for."""
+
+
+class ThroughputStabilityCriteria(ImplicitDict):
+    """Criteria used to determine when it is valid to start collecting throughput data in a step.
+
+    Any specified field that evaluates to false will cause this criteria to evaluate to false"""
+
+    each_user_completed_at_least: Optional[OperationCount]
+    """Evaluates true when each user has completed at least this many operations since the step started."""
+
+
+class OperationLatency(ImplicitDict):
+    duration: StringBasedTimeDelta
+    """Duration of relevant operations."""
+
+    operations: list[OperationType]
+    """Particular operations to look for."""
+
+
+class StepCompletionCriteria(ImplicitDict):
+    """Completion criteria based on a load step.
+
+    Any specified field that evaluates to false will cause this criteria to evaluate to false."""
+
+    any_of: Optional[list[StepCompletionCriteria]]
+
+    sampling_duration_at_least: Optional[StringBasedTimeDelta]
+    """Evaluates true when the step has been collecting valid throughput data for at least this long."""
+
+    completed_at_least: Optional[OperationCount]
+    """Evaluates true when at least this many operations have completed while the step was collecting valid throughput data."""
+
+    average_duration_more_than: Optional[OperationLatency]
+    """Evaluates true when the average duration of operations completed during the step exceeds the specified value."""
+
+    throughput_stability_took_longer_than: Optional[StringBasedTimeDelta]
+    """Evaluates true when reaching throughput stability took longer than this amount of time since the start of the step."""
+
+
+class ThroughputPastPeak(ImplicitDict):
+    operations: list[OperationType]
+    """Operations for which throughput should be calculated."""
+
+    fraction_of_peak: float
+    """Trigger this threshold when throughput drops below this fraction of the peak throughput measured for any past step. [0, 1]"""
+
+
+class LoadCompletionCriteria(ImplicitDict):
+    """Completion criteria for an entire load.
+
+    Any specified field that evaluates to false will cause this criterion to evaluate to false."""
+
+    any_of: Optional[list[LoadCompletionCriteria]]
+
+    throughput_lower_than_peak: Optional[ThroughputPastPeak]
+    """Evaluates true when the throughput of the specified operations for the most recently-completed step drops below the specified fraction of the maximum throughput of all prior steps."""
+
+    failures_more_than: Optional[OperationCount]
+    """Evaluates true when the number of failures for the specified operations exceeds the specified number."""
+
+    most_recent_step: Optional[StepCompletionCriteria]
+    """Evaluates true when the most recently completed step meets these criteria."""
+
+
+class UserRampLoad(ImplicitDict):
+    """Ramps up users of a specified type, observing resulting throughput."""
+
+    user_type: BenchmarkUserName
+    """Type of user to instantiate."""
+
+    initial_users: int = 1
+    """Number of users to start with."""
+
+    additional_users_per_step: int = 1
+    """Additional users to add at each step along the ramp."""
+
+    throughput_stability_criteria: ThroughputStabilityCriteria
+    """Throughput of the current step is considered stable once these criteria are met."""
+
+    step_completion_criteria: StepCompletionCriteria
+    """The current step is considered complete once these criteria are met."""
+
+    load_completion_criteria: LoadCompletionCriteria
+    """The load is considered complete if these criteria are met."""
+
+
+class BenchmarkLoadSpecification(ImplicitDict):
+    """Specification of how load will be applied."""
+
+    name: BenchmarkLoadName
+
+    user_ramp: Optional[UserRampLoad]
+    """Load will be provided by ramping up the number of virtual users of a particular type/behavior."""
diff --git a/monitoring/benchmarker/configurations/scenarios.py b/monitoring/benchmarker/configurations/scenarios.py
new file mode 100644
index 0000000000..b7f493fe71
--- /dev/null
+++ b/monitoring/benchmarker/configurations/scenarios.py
@@ -0,0 +1,31 @@
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+from monitoring.benchmarker.configurations.actions import BenchmarkActionName
+from monitoring.benchmarker.configurations.loads import BenchmarkLoadName
+
+
+class BenchmarkScenarioName(str):
+    """Unique (within benchmark configuration) name for a scenario."""
+
+    pass
+
+
+class BenchmarkScenarioSpecification(ImplicitDict):
+    name: BenchmarkScenarioName
+
+    setup: Optional[list[BenchmarkActionName]]
+    """Actions to perform before beginning load testing in this scenario."""
+
+    load: BenchmarkLoadName
+    """How and when to generate load in this scenario."""
+
+    teardown: Optional[list[BenchmarkActionName]]
+    """Actions to perform after completing load testing in this scenario."""
+
+    record_query_details: bool = False
+    """When true, include full details in the report for queries made during this scenario."""
+
+    metadata: dict
+    """Arbitrary data that may be relevant to the scenario."""
diff --git a/monitoring/benchmarker/configurations/users.py b/monitoring/benchmarker/configurations/users.py
new file mode 100644
index 0000000000..85a0741210
--- /dev/null
+++ b/monitoring/benchmarker/configurations/users.py
@@ -0,0 +1,111 @@
+from enum import StrEnum
+from typing import Optional
+
+from implicitdict import ImplicitDict, StringBasedDateTime, StringBasedTimeDelta
+
+from monitoring.monitorlib.geo import Altitude, LatLngPoint
+from monitoring.monitorlib.geotemporal import Volume4D
+from monitoring.monitorlib.rid import RIDVersion
+from monitoring.uss_qualifier.resources.definitions import ResourceID
+
+
+class BenchmarkUserName(str):
+    """Unique (within benchmark configuration) name for a means to generate load."""
+
+
+class FixedLocationSpecification(ImplicitDict):
+    horizontal: LatLngPoint
+    vertical: Altitude
+
+
+class FlightTimeGenerationSpecification(ImplicitDict):
+    fixed_spacing: Optional[StringBasedTimeDelta]
+    """Set delay between the end of the previous flight and the start of the next flight to this value."""
+
+
+class FlightLocationGenerationSpecification(ImplicitDict):
+    fixed_location: Optional[FixedLocationSpecification]
+    """Always choose the same flight location."""
+
+
+class FixedVolumesSpecification(ImplicitDict):
+    origin_horizontal: LatLngPoint
+    origin_vertical: Altitude
+    origin_time: StringBasedDateTime
+    volumes: list[Volume4D]
+
+
+class FlightShapeGenerationSpecification(ImplicitDict):
+    fixed_volumes: Optional[FixedVolumesSpecification]
+
+
+class IndependentComponentsFlightGenerationSpecification(ImplicitDict):
+    time: FlightTimeGenerationSpecification
+    """Means to generate the start times of flights."""
+
+    location: FlightLocationGenerationSpecification
+    """Means to generate flight locations."""
+
+    shape: FlightShapeGenerationSpecification
+    """Means to generate flight shapes."""
+
+
+class FlightGenerationSpecification(ImplicitDict):
+    independent_time_location_shape: Optional[
+        IndependentComponentsFlightGenerationSpecification
+    ]
+    """The time, location, and shape of flights are generated independently."""
+
+
+class ASTMNetRIDISAPerFlightStrategySpecification(ImplicitDict):
+    """Create an ISA per flight."""
+
+    before_flight_start: StringBasedTimeDelta
+    """Create the ISA this amount of time before the start of the flight."""
+
+    after_flight_end: Optional[StringBasedTimeDelta]
+    """Delete the ISA this amount of time after the end of this flight."""
+
+
+class ASTMNetRIDISAStrategySpecification(ImplicitDict):
+    """Strategy used to ensure at least one ISA covers every flight."""
+
+    isa_per_flight: Optional[ASTMNetRIDISAPerFlightStrategySpecification]
+
+
+class ASTMDSSSelectionStrategy(StrEnum):
+    First = "First"
+    """Always use the first DSS in the pool list."""
+
+    Random = "Random"
+    """Use a random DSS from the pool list for every operation."""
+
+
+class ASTMNetRIDBehaviorSpecification(ImplicitDict):
+    rid_version: RIDVersion
+
+    dss_pool: list[ResourceID]
+    """Means to interact with the ASTM DSS.
+    
+    Benchmark configuration must contain a `resources.astm.f3411.DSSInstanceResource` resource with each of these names."""
+
+    dss_selection_strategy: Optional[ASTMDSSSelectionStrategy]
+
+    isa_strategy: Optional[ASTMNetRIDISAStrategySpecification]
+
+
+class FlightPlannerSpecification(ImplicitDict):
+    """User planning flights coordinated via UTM standards."""
+
+    flight_generation: FlightGenerationSpecification
+    """How flight planner generates their flights."""
+
+    astm_netrid_behavior: Optional[ASTMNetRIDBehaviorSpecification]
+    """How flight planner provides ASTM NetRID service for their flights."""
+
+
+class BenchmarkUserSpecification(ImplicitDict):
+    name: BenchmarkUserName
+
+    flight_planner: Optional[FlightPlannerSpecification]
+    """User is a USS client that plans flights coordinated with UTM technologies."""
diff --git a/monitoring/benchmarker/reports/analysis.py b/monitoring/benchmarker/reports/analysis.py
new file mode 100644
index 0000000000..edf7257aae
--- /dev/null
+++ b/monitoring/benchmarker/reports/analysis.py
@@ -0,0 +1,113 @@
+from collections.abc import Iterable, Sequence
+from datetime import datetime
+
+from monitoring.benchmarker.configurations.loads import OperationType
+from monitoring.benchmarker.reports.report import (
+    BenchmarkOperation,
+    BenchmarkScenarioReport,
+    OperationsByOrigin,
+    OperationsByOutcome,
+    OperationsByType,
+)
+
+OperationsHierarchyMember = (
+    Sequence[OperationsByType]
+    | OperationsByType
+    | OperationsByOrigin
+    | OperationsByOutcome
+    | BenchmarkOperation
+)
+
+
+def select_operations(
+    operations: OperationsHierarchyMember,
+    types: Sequence[OperationType | str] | None = None,
+    origins: Sequence[str] | None = None,
+    outcomes: Sequence[bool] | None = None,
+    completed_after: datetime | None = None,
+    completed_before: datetime | None = None,
+) -> Iterable[BenchmarkOperation]:
+    def nest(sub_operations):
+        for op in sub_operations:
+            yield from select_operations(
+                op, types, origins, outcomes, completed_after, completed_before
+            )
+
+    if isinstance(operations, Sequence):
+        yield from nest(operations)
+    elif isinstance(operations, OperationsByType):
+        norm_types = {OperationType(t) for t in types} if types is not None else None
+        types_match = (
+            norm_types is None
+            or operations.type in norm_types
+            or str(operations.type) in {str(t) for t in norm_types}
+        )
+        if types_match:
+            yield from nest(operations.origins)
+    elif isinstance(operations, OperationsByOrigin):
+        if origins is None or operations.origin in origins:
+            yield from nest(operations.outcomes)
+    elif isinstance(operations, OperationsByOutcome):
+        if "successful" in operations and operations.successful:
+            if outcomes is None or True in outcomes:
+                yield from nest(operations.successful)
+        if "unsuccessful" in operations and operations.unsuccessful:
+            if outcomes is None or False in outcomes:
+                yield from nest(operations.unsuccessful)
+    elif isinstance(operations, BenchmarkOperation):
+        if completed_after and operations.t1.datetime < completed_after:
+            return
+        if completed_before and operations.t1.datetime > completed_before:
+            return
+        yield operations
+    else:
+        raise ValueError(
+            f"`operations` type '{type(operations).__name__}' is not valid"
+        )
+
+
+def throughput_of_operations(
+    operations: OperationsHierarchyMember,
+    start_time: datetime,
+    end_time: datetime,
+    **kwargs,
+) -> float:
+    """Determine the achieved throughput during a specified time range from a list of completed operations.
+
+    Args:
+      * operations: List of relevant operations over all time.
+      * start_time: Beginning of time window in which to inspect throughput.
+      * end_time: End of time window in which to inspect throughput.
+
+    Returns: Throughput in operations of interest per second.
+
+    Notes:
+      Operation flux must have already been in steady-state at `start_time` for this throughput calculation to be
+      valid.  What happens after `end_time` does not affect this calculation.  "Partial credit" is not given for
+      eventually-successful operations in progress at `end_time` as attempting to do so would require operation
+      flux to remain in steady-state after `end_time` until completion of the last operation started before
+      `end_time` for the throughput calculation to be valid.  Instead, the partial work of operations in progress
+      at `end_time` effectively discarded by this approach should be (statistically) exactly balanced by the
+      partial work included "for free" of operations started before `start_time` that end within the time window.
+    """
+    dur = (end_time - start_time).total_seconds()
+    kwargs["outcomes"] = (True,)
+    kwargs["completed_after"] = start_time
+    kwargs["completed_before"] = end_time
+    return (
+        sum(1 for _ in select_operations(operations, **kwargs)) / dur
+        if dur > 0
+        else 0.0
+    )
+
+
+def throughput_of_step(
+    report: BenchmarkScenarioReport, step_index: int, **kwargs
+) -> float:
+    step = report.steps[step_index]
+    return throughput_of_operations(
+        report.operations,
+        step.throughput_stability_time.datetime,
+        step.end_time.datetime,
+        **kwargs,
+    )
diff --git a/monitoring/benchmarker/reports/report.py b/monitoring/benchmarker/reports/report.py
new file mode 100644
index 0000000000..90d9319b96
--- /dev/null
+++ b/monitoring/benchmarker/reports/report.py
@@ -0,0 +1,94 @@
+from typing import Optional
+
+from implicitdict import ImplicitDict, StringBasedDateTime
+
+from monitoring.benchmarker.configurations.configuration import BenchmarkConfiguration
+from monitoring.benchmarker.configurations.loads import OperationType
+from monitoring.monitorlib.fetch import Query
+
+
+class BenchmarkOperation(ImplicitDict):
+    """Record of an operation of a known type, origin, and outcome executed during a benchmark."""
+
+    t0: StringBasedDateTime
+    """Time this operation was started/initiated."""
+
+    t1: StringBasedDateTime
+    """Time this operation completed, either successsfully or in failure."""
+
+    query: Optional[Query]
+    """The query details for this operation, if this was a query operation and query details are being recorded."""
+
+
+class OperationsByOutcome(ImplicitDict):
+    """Record of operations grouped by whether they were successful."""
+
+    successful: Optional[list[BenchmarkOperation]]
+    """Operations that were successful (yielded successful user journeys)."""
+
+    unsuccessful: Optional[list[BenchmarkOperation]]
+    """Operations that were unsuccessful (did not yield successful user journeys due to errors, timeouts, aborts, etc)"""
+
+
+class OperationsByOrigin(ImplicitDict):
+    """Record of operations from a particular origin."""
+
+    origin: str
+    """Source/originator of the operations in this record; e.g., the user that initiated these operations."""
+
+    outcomes: list[OperationsByOutcome]
+    """Operations originating from this origin."""
+
+
+class OperationsByType(ImplicitDict):
+    """Record of operations of a particular type."""
+
+    type: OperationType
+    """The type of the operations in this record."""
+
+    origins: list[OperationsByOrigin]
+    """Operations of this particular type."""
+
+
+class BenchmarkScenarioStepReport(ImplicitDict):
+    load_factor: float
+    """Load factor (e.g., number of users) present during this step."""
+
+    start_time: StringBasedDateTime
+    """Time this step started."""
+
+    throughput_stability_time: StringBasedDateTime
+    """Time during this step at which activity became sufficiently stable for throughput measurement."""
+
+    end_time: StringBasedDateTime
+    """Time this step ended."""
+
+
+class BenchmarkScenarioReport(ImplicitDict):
+    operations: list[OperationsByType]
+    """All operations that occurred during the benchmark run."""
+
+    steps: list[BenchmarkScenarioStepReport]
+    """Boundaries of steps within this scenario."""
+
+    metadata: dict = {}
+    """Arbitrary metadata copied from the scenario specification."""
+
+
+class BenchmarkReport(ImplicitDict):
+    scenarios: list[BenchmarkScenarioReport]
+    """Results of scenarios, corresponding to the scenarios defined in the configuration."""
+
+
+class BenchmarkRunReport(ImplicitDict):
+    codebase_version: str
+    """Version of codebase used to run benchmarker."""
+
+    commit_hash: str
+    """Full commit hash of codebase used to run benchmarker."""
+
+    configuration: BenchmarkConfiguration
+    """Configuration used to run benchmarker."""
+
+    report: BenchmarkReport
+    """Report produced by benchmarker during this run."""
diff --git a/monitoring/monitorlib/expressions/types.py b/monitoring/monitorlib/expressions/types.py
new file mode 100644
index 0000000000..f48d3a6ef9
--- /dev/null
+++ b/monitoring/monitorlib/expressions/types.py
@@ -0,0 +1,15 @@
+from implicitdict import ImplicitDict
+
+
+class ASTExpression(str):
+    """AST expression with undefined return type.
+
+    See https://lmfit.github.io/asteval/"""
+
+
+class SymbolExpression(ImplicitDict):
+    name: str
+    """Name of symbol that will take on the value."""
+
+    value: ASTExpression
+    """Expression for the untyped value to assign to this symbol."""