From 8812f849f3f2e2eb2ede63fc9ea0555e40e7efaa Mon Sep 17 00:00:00 2001 From: Benjamin Pelletier Date: Mon, 20 Jul 2026 23:12:30 +0000 Subject: [PATCH 1/2] Add SCD behavior to benchmarker --- .../interuss/scd/single_s2_cell.jsonnet | 265 +++++++++ .../benchmarker/configurations/loads.py | 9 + .../benchmarker/configurations/scenarios.py | 2 +- .../configurations/user/__init__.py | 0 .../configurations/user/astm/__init__.py | 0 .../configurations/user/astm/dss.py | 9 + .../configurations/user/astm/scd.py | 78 +++ .../benchmarker/configurations/users.py | 34 +- monitoring/benchmarker/engine/coordination.py | 70 +++ monitoring/benchmarker/engine/engine.py | 6 + .../benchmarker/engine/loads/criteria.py | 23 + monitoring/benchmarker/engine/loads/loads.py | 3 + .../engine/loads/user_ramp/user_ramp.py | 14 +- .../benchmarker/engine/users/creation.py | 26 +- .../users/flight_planner/astm_net_rid.py | 8 +- .../users/flight_planner/flight_generation.py | 81 ++- .../users/flight_planner/flight_planner.py | 43 +- .../engine/users/flight_planner/framework.py | 16 +- .../engine/users/flight_planner/scd.py | 556 ++++++++++++++++++ .../benchmarker/engine/users/framework.py | 10 +- monitoring/benchmarker/reports/analysis.py | 3 +- monitoring/monitorlib/geotemporal.py | 28 + .../loads/LoadCompletionCriteria.json | 7 + .../loads/StepCompletionCriteria.json | 11 + .../configurations/loads/UserRampLoad.json | 7 + .../BenchmarkScenarioSpecification.json | 6 +- .../user/astm/scd/BehaviorSpecification.json | 44 ++ .../user/astm/scd/ImplicitSubscription.json | 12 + .../astm/scd/OpIntentRefCleanupStrategy.json | 28 + .../astm/scd/OpIntentRefCreationStrategy.json | 49 ++ .../user/astm/scd/SingleSubscription.json | 40 ++ .../user/astm/scd/SubscriptionStrategy.json | 34 ++ .../users/FlightExecutionSpecification.json | 28 + .../users/FlightGenerationSpecification.json | 11 + .../users/FlightPlannerSpecification.json | 11 + .../FlightTimeGenerationSpecification.json | 10 +- ...mponentsFlightGenerationSpecification.json | 2 +- .../monitorlib/geo/LatLngBoundingBox.json | 34 ++ 38 files changed, 1562 insertions(+), 56 deletions(-) create mode 100644 monitoring/benchmarker/configurations/interuss/scd/single_s2_cell.jsonnet create mode 100644 monitoring/benchmarker/configurations/user/__init__.py create mode 100644 monitoring/benchmarker/configurations/user/astm/__init__.py create mode 100644 monitoring/benchmarker/configurations/user/astm/dss.py create mode 100644 monitoring/benchmarker/configurations/user/astm/scd.py create mode 100644 monitoring/benchmarker/engine/coordination.py create mode 100644 monitoring/benchmarker/engine/users/flight_planner/scd.py create mode 100644 schemas/monitoring/benchmarker/configurations/user/astm/scd/BehaviorSpecification.json create mode 100644 schemas/monitoring/benchmarker/configurations/user/astm/scd/ImplicitSubscription.json create mode 100644 schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCleanupStrategy.json create mode 100644 schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCreationStrategy.json create mode 100644 schemas/monitoring/benchmarker/configurations/user/astm/scd/SingleSubscription.json create mode 100644 schemas/monitoring/benchmarker/configurations/user/astm/scd/SubscriptionStrategy.json create mode 100644 schemas/monitoring/benchmarker/configurations/users/FlightExecutionSpecification.json create mode 100644 schemas/monitoring/monitorlib/geo/LatLngBoundingBox.json diff --git a/monitoring/benchmarker/configurations/interuss/scd/single_s2_cell.jsonnet b/monitoring/benchmarker/configurations/interuss/scd/single_s2_cell.jsonnet new file mode 100644 index 0000000000..0975947738 --- /dev/null +++ b/monitoring/benchmarker/configurations/interuss/scd/single_s2_cell.jsonnet @@ -0,0 +1,265 @@ +local num_uss = 3; +local num_nodes = 3; + +local nodeIndex = function(uss, node) std.format('%02d', node + num_nodes * (uss - 1)); + +local location = { + horizontal: {lat: 34, lng: -118}, + vertical: {value: 300, reference: 'W84', units: 'M'}, +}; + +local shape = { + 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', + }, + ], +}; + +{ + resources: { + resource_declarations: { + utm_auth: { + resource_type: 'resources.communications.AuthAdapterResource', + specification: { + auth_spec: 'DummyOAuth(http://localhost:8085/token,benchmarker)', + scopes_authorized: [ + 'utm.strategic_coordination', + ], + }, + }, + } + { + ['uss%d_dss_pool' % uss]: { + resource_type: 'resources.astm.f3548.v21.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' % nodeIndex(uss, node), + } for node in std.range(1, num_nodes) + ], + }, + } for uss in std.range(1, num_uss) + }, + }, + + user_types: [ + { + name: 'FPU%d' % uss, // Flight planner user using DSS instance from uss + flight_planner: { + flight_generation: { + independent_time_location_shape: { + time: { + fixed_spacing: '19s', + uniform_random_spacing: '2s', + }, + location: { + fixed_location: location, + }, + shape: { + fixed_volumes: shape, + }, + }, + }, + flight_execution: { + end_flight_after_start: '10s', + }, + scd_behavior: { + dss_pool: ['uss%d_dss_pool' % uss], + dss_selection_strategy: 'Random', + subscription_strategy: { + single_subscription: { + subscription_id: '3bdb0b88-a522-4286-9499-160e56c953bb', + duration: '23h', + area: { + lat_min: 34 - 0.00001, + lng_min: -118 - 0.00001, + lat_max: 34 + 0.00001, + lng_max: -118 + 0.00001, + }, + min_alt: {value: 0, units: 'M', reference: 'W84'}, + max_alt: {value: 3000, units: 'M', reference: 'W84'}, + }, + }, + op_intent_ref_creation_strategy: { + ovn_coordination_group: 'cluster1', + coordinate_requested_ovns: true, + retries: 2, + accept_before_flight_start: '5s', + activate_before_flight_start: null, + }, + op_intent_ref_cleanup_strategy: { + after_actual_flight_end: '1s', + }, + }, + }, + } for uss in std.range(1, num_uss) + ], + + loads: [ + { + name: 'Flight planner ramp for USS %d' % uss, + user_ramp: { + user_type: 'FPU%d' % uss, + initial_users: 2, + additional_users_per_step: 2, + random_seed: 1234, + throughput_stability_criteria: { + each_user_completed_at_least: { + count: 1, + operations: ['workflow.flight_planner.flight'], + }, + }, + step_completion_criteria: { + any_of: [ + { + sampling_duration_at_least: '60s', + }, + { + completed_at_least: { + count: 100, + operations: ['workflow.flight_planner.flight'], + }, + }, + { + average_duration_more_than: { + duration: '60s', + operations: ['workflow.flight_planner.flight'], + }, + }, + { + failures_more_than: { + count: 100, + 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: 400, + operations: ['workflow.flight_planner.flight'], + } + }, + { + most_recent_step: { + average_duration_more_than: { + duration: '60s', + operations: ['workflow.flight_planner.flight'], + }, + }, + }, + ], + }, + }, + } for uss in std.range(1, num_uss) + ], + + scenarios: [ + { + name: 'Single S2 cell for USS %d' % uss, + load: 'Flight planner ramp for USS %d' % uss, + } for uss in std.range(1, num_uss) + ], + + artifacts: [ + { + raw_report: { + name: 'report', + }, + }, + { + matplotlib_figure: { + name: 'single_s2_cell', + n_subfigure_cols: num_uss, + subfigures: [ + { + title: 'Single S2 cell\nDSS instance %d' % uss, + subplots: [ + { + evaluation_context: [ + { + name: 'scenario', + value: 'report.report.scenarios[%d]' % (uss - 1), + }, + { + name: 'scale', + value: '[step.load_factor for step in scenario.steps]', + }, + { + name: 'throughput', + value: '[throughput_of_step(scenario, s, types=["workflow.flight_planner.flight"], outcomes=[True])' + + ' for s in range(len(scenario.steps))]', + }, + { + name: 'failures', + value: '[throughput_of_step(scenario, s, types=["workflow.flight_planner.flight"], outcomes=[False])' + + ' for s in range(len(scenario.steps))]', + }, + ], + x_axis: { + label: 'Flight planners', + }, + y_axis: { + label: 'Throughput\n(Flights/s)', + }, + xy_plots: [ + { + type: 'Scatter', + label_expr: '"Successes"', + x_data_expr: 'scale', + y_data_expr: 'throughput', + }, + { + type: 'Scatter', + label_expr: '"Failures"', + x_data_expr: 'scale', + y_data_expr: 'failures', + }, + ], + legend: { + location: 'upper left', + font_size: 'x-small', + label_spacing: 0.2, + border_padding: 0.2, + }, + }, + ], + } for uss in std.range(1, num_uss) + ], + }, + }, + ], +} diff --git a/monitoring/benchmarker/configurations/loads.py b/monitoring/benchmarker/configurations/loads.py index 3b2f5822ba..9f03cbd5ce 100644 --- a/monitoring/benchmarker/configurations/loads.py +++ b/monitoring/benchmarker/configurations/loads.py @@ -110,6 +110,9 @@ class StepCompletionCriteria(ImplicitDict): 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.""" + failures_more_than: Optional[OperationCount] + """Evaluates true when the number of failures for the specified operations exceeds the specified number during this step.""" + class ThroughputPastPeak(ImplicitDict): operations: list[OperationType] @@ -132,6 +135,9 @@ class LoadCompletionCriteria(ImplicitDict): failures_more_than: Optional[OperationCount] """Evaluates true when the number of failures for the specified operations exceeds the specified number.""" + completed_steps: Optional[int] + """Evaluates true when this many steps have been completed.""" + most_recent_step: Optional[StepCompletionCriteria] """Evaluates true when the most recently completed step meets these criteria.""" @@ -157,6 +163,9 @@ class UserRampLoad(ImplicitDict): load_completion_criteria: LoadCompletionCriteria """The load is considered complete if these criteria are met.""" + random_seed: Optional[int] + """Seed to use to randomly generate seeds for each user.""" + class BenchmarkLoadSpecification(ImplicitDict): """Specification of how load will be applied.""" diff --git a/monitoring/benchmarker/configurations/scenarios.py b/monitoring/benchmarker/configurations/scenarios.py index b7f493fe71..1a8c573ee4 100644 --- a/monitoring/benchmarker/configurations/scenarios.py +++ b/monitoring/benchmarker/configurations/scenarios.py @@ -27,5 +27,5 @@ class BenchmarkScenarioSpecification(ImplicitDict): record_query_details: bool = False """When true, include full details in the report for queries made during this scenario.""" - metadata: dict + metadata: Optional[dict] """Arbitrary data that may be relevant to the scenario.""" diff --git a/monitoring/benchmarker/configurations/user/__init__.py b/monitoring/benchmarker/configurations/user/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/monitoring/benchmarker/configurations/user/astm/__init__.py b/monitoring/benchmarker/configurations/user/astm/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/monitoring/benchmarker/configurations/user/astm/dss.py b/monitoring/benchmarker/configurations/user/astm/dss.py new file mode 100644 index 0000000000..f272092c4b --- /dev/null +++ b/monitoring/benchmarker/configurations/user/astm/dss.py @@ -0,0 +1,9 @@ +from enum import StrEnum + + +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.""" diff --git a/monitoring/benchmarker/configurations/user/astm/scd.py b/monitoring/benchmarker/configurations/user/astm/scd.py new file mode 100644 index 0000000000..941dab20ee --- /dev/null +++ b/monitoring/benchmarker/configurations/user/astm/scd.py @@ -0,0 +1,78 @@ +from typing import Optional + +from implicitdict import ImplicitDict, StringBasedTimeDelta +from uas_standards.astm.f3548.v21.api import SubscriptionID + +from monitoring.benchmarker.configurations.user.astm.dss import ASTMDSSSelectionStrategy +from monitoring.benchmarker.engine.coordination import CoordinationGroupID +from monitoring.monitorlib.geo import Altitude, LatLngBoundingBox +from monitoring.uss_qualifier.resources.definitions import ResourceID + + +class SingleSubscription(ImplicitDict): + subscription_id: SubscriptionID + """ID of the single subscription to create, or ensure exists.""" + + duration: StringBasedTimeDelta + """Duration of the subscription, from the time it is created.""" + + area: LatLngBoundingBox + """Horizontal area this subscription should cover.""" + + min_alt: Altitude + """Altitude below which this subscription should not apply.""" + + max_alt: Altitude + """Altitude above which this subscription should not apply.""" + + +class ImplicitSubscription(ImplicitDict): + pass + + +class SubscriptionStrategy(ImplicitDict): + single_subscription: Optional[SingleSubscription] + """Planner ensures there is a single subscription, established at the start of operations, covering all their flights.""" + + implicit_subscription: Optional[ImplicitSubscription] + """Planner has the DSS establish an implicit subscription for each individual flight.""" + + +class OpIntentRefCreationStrategy(ImplicitDict): + ovn_coordination_group: Optional[CoordinationGroupID] + """If specified, exchange OVNs directly between other virtual users using this coordination group to exchange OVNs.""" + + coordinate_requested_ovns: Optional[bool] + """If specified true, use requested OVNs and share them with the coordination group before requesting.""" + + retries: Optional[int] + """If specified, number of times to retry failed operational intent operations before giving up.""" + + accept_before_flight_start: Optional[StringBasedTimeDelta] + """Create an operational intent in the Accepted state this long before the flight starts (start of first volume).""" + + activate_before_flight_start: Optional[StringBasedTimeDelta] + """Create or update an operational intent to the Activated state this long before the flight starts (start of first volume).""" + + +class OpIntentRefCleanupStrategy(ImplicitDict): + after_actual_flight_end: Optional[StringBasedTimeDelta] + """Delete operational intent reference this amount of time after the actual end of the flight.""" + + after_planned_flight_end: Optional[StringBasedTimeDelta] + """Delete operational intent reference this amount of time after the planned end of the flight (end of last volume).""" + + +class BehaviorSpecification(ImplicitDict): + dss_pool: list[ResourceID] + """Means to interact with the ASTM DSS. + + Benchmark configuration must contain a `resources.astm.f3548.v21.DSSInstanceResource` resource with each of these names.""" + + dss_selection_strategy: Optional[ASTMDSSSelectionStrategy] + + subscription_strategy: SubscriptionStrategy + + op_intent_ref_creation_strategy: OpIntentRefCreationStrategy + + op_intent_ref_cleanup_strategy: OpIntentRefCleanupStrategy diff --git a/monitoring/benchmarker/configurations/users.py b/monitoring/benchmarker/configurations/users.py index c2bff398b7..80402b4296 100644 --- a/monitoring/benchmarker/configurations/users.py +++ b/monitoring/benchmarker/configurations/users.py @@ -1,8 +1,9 @@ -from enum import StrEnum from typing import Optional from implicitdict import ImplicitDict, StringBasedDateTime, StringBasedTimeDelta +from monitoring.benchmarker.configurations.user.astm import scd +from monitoring.benchmarker.configurations.user.astm.dss import ASTMDSSSelectionStrategy from monitoring.monitorlib.geo import Altitude, LatLngPoint from monitoring.monitorlib.geotemporal import Volume4D from monitoring.monitorlib.rid import RIDVersion @@ -20,7 +21,10 @@ class FixedLocationSpecification(ImplicitDict): 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.""" + """Add this fixed delay between the end of the previous flight and the start of the next flight.""" + + uniform_random_spacing: Optional[StringBasedTimeDelta] + """Add a random delay uniformly distributed between 0 and this duration between the end of the previous flight and the start of the next flight.""" class FlightLocationGenerationSpecification(ImplicitDict): @@ -47,7 +51,17 @@ class IndependentComponentsFlightGenerationSpecification(ImplicitDict): """Means to generate flight locations.""" shape: FlightShapeGenerationSpecification - """Means to generate flight shapes.""" + """Means to generate flight shapes. + + Note that these volumes represent planned volumes and the flight may be ended before the end of the volumes.""" + + +class FlightExecutionSpecification(ImplicitDict): + end_flight_before_end: Optional[StringBasedTimeDelta] + """If specified, the operator ends their flight early this amount of time before the flight is scheduled to end.""" + + end_flight_after_start: Optional[StringBasedTimeDelta] + """If specified, the operator ends their flight early this amount of time after the flight began.""" class FlightGenerationSpecification(ImplicitDict): @@ -56,6 +70,9 @@ class FlightGenerationSpecification(ImplicitDict): ] """The time, location, and shape of flights are generated independently.""" + execution: Optional[FlightExecutionSpecification] + """How operator actually executes their planned flights.""" + class ASTMNetRIDISAPerFlightStrategySpecification(ImplicitDict): """Create an ISA per flight.""" @@ -73,14 +90,6 @@ class ASTMNetRIDISAStrategySpecification(ImplicitDict): 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 @@ -103,6 +112,9 @@ class FlightPlannerSpecification(ImplicitDict): astm_netrid_behavior: Optional[ASTMNetRIDBehaviorSpecification] """How flight planner provides ASTM NetRID service for their flights.""" + scd_behavior: Optional[scd.BehaviorSpecification] + """How flight planner provides strategic conflict detection service for their flights.""" + class BenchmarkUserSpecification(ImplicitDict): name: BenchmarkUserName diff --git a/monitoring/benchmarker/engine/coordination.py b/monitoring/benchmarker/engine/coordination.py new file mode 100644 index 0000000000..f571d5eac3 --- /dev/null +++ b/monitoring/benchmarker/engine/coordination.py @@ -0,0 +1,70 @@ +from abc import ABC +from collections.abc import Sequence +from dataclasses import dataclass +from threading import Lock +from typing import Any + + +class CoordinationGroupID(str): + """Identity of a group of coordinating users.""" + + +@dataclass +class CoordinationMessage: + """Message exchanged between members of the specified coordination group in order to coordinate.""" + + group_id: CoordinationGroupID | None + subject: str + content: Any + + +class CoordinationSubscriber(ABC): + """Base class for subscribers wishing to receive CoordinationEvent messages from other members of their coordination group.""" + + def receive_coordination_message(self, msg: CoordinationMessage) -> None: + raise NotImplementedError() + + +class CoordinationGroup: + lock: Lock + subscribers: set[CoordinationSubscriber] + messages: list[CoordinationMessage] + + def __init__(self): + self.lock = Lock() + self.subscribers = set() + self.messages = [] + + +class Coordinator: + """Entity that handles pubsub messages for identified coordination groups.""" + + _groups: dict[CoordinationGroupID, CoordinationGroup] + + def __init__(self, coordination_groups: Sequence[CoordinationGroupID]): + """coordination_groups must identify all coordination groups that will be used.""" + self._groups = {g: CoordinationGroup() for g in coordination_groups} + + def publish( + self, group_id: CoordinationGroupID, subject: str, content: Any + ) -> None: + """group_id must be among the coordination_groups identified when Coordinator was instantiated.""" + msg = CoordinationMessage( + group_id=group_id, + subject=subject, + content=content, + ) + group = self._groups[group_id] + with group.lock: + for subscriber in group.subscribers: + subscriber.receive_coordination_message(msg) + group.messages.append(msg) + + def subscribe( + self, subscriber: CoordinationSubscriber, group_id: CoordinationGroupID + ) -> None: + group = self._groups[group_id] + with group.lock: + group.subscribers.add(subscriber) + for msg in group.messages: + subscriber.receive_coordination_message(msg) diff --git a/monitoring/benchmarker/engine/engine.py b/monitoring/benchmarker/engine/engine.py index 16217a4f9f..976282d1af 100644 --- a/monitoring/benchmarker/engine/engine.py +++ b/monitoring/benchmarker/engine/engine.py @@ -5,11 +5,13 @@ from monitoring.benchmarker.configurations.configuration import BenchmarkConfiguration from monitoring.benchmarker.engine.actions.actions import run_scenario_actions +from monitoring.benchmarker.engine.coordination import Coordinator from monitoring.benchmarker.engine.loads.loads import run_scenario_load from monitoring.benchmarker.engine.operations import ( group_operations, ) from monitoring.benchmarker.engine.resources import instantiate_resources +from monitoring.benchmarker.engine.users.creation import enumerate_coordination_groups from monitoring.benchmarker.reports.report import ( BenchmarkReport, BenchmarkRunReport, @@ -39,6 +41,9 @@ async def _run_benchmark_async( action_list = config.actions if "actions" in config and config.actions else [] action_specs = {action_spec.name: action_spec for action_spec in action_list} + coordination_groups = list(enumerate_coordination_groups(config.user_types)) + coordinator = Coordinator(coordination_groups) + try: for scenario_spec in config.scenarios: logger.info( @@ -60,6 +65,7 @@ async def _run_benchmark_async( user_specs_map, resource_pool, executor, + coordinator, ) # Run teardown actions diff --git a/monitoring/benchmarker/engine/loads/criteria.py b/monitoring/benchmarker/engine/loads/criteria.py index 80615dc521..cb3f25bb55 100644 --- a/monitoring/benchmarker/engine/loads/criteria.py +++ b/monitoring/benchmarker/engine/loads/criteria.py @@ -57,6 +57,7 @@ def check_step_completion_criteria( "throughput_stability_took_longer_than" in criteria and criteria.throughput_stability_took_longer_than ) + has_failures = "failures_more_than" in criteria and criteria.failures_more_than if ( not has_any_of @@ -64,6 +65,7 @@ def check_step_completion_criteria( and not has_completed_ops and not has_min_average and not has_stability_duration + and not has_failures ): raise NotImplementedError("StepCompletionCriteria has no specified conditions") @@ -125,6 +127,21 @@ def check_step_completion_criteria( if stability_time - step_start_time <= required_duration: return False + if has_failures and criteria.failures_more_than: + req_count = criteria.failures_more_than.count + relevant_ops = set(criteria.failures_more_than.operations) + fails = sum( + 1 + for op in operations + if not op.successful + and op.type in relevant_ops + and op.completed_at.datetime >= step_start_time + and op.type in relevant_ops + and op.completed_at.datetime <= now + ) + if fails <= req_count: + return False + return True @@ -138,12 +155,14 @@ def check_load_completion_criteria( "throughput_lower_than_peak" in criteria and criteria.throughput_lower_than_peak ) has_failures = "failures_more_than" in criteria and criteria.failures_more_than + has_step_count = "completed_steps" in criteria and criteria.completed_steps has_most_recent_step = "most_recent_step" in criteria and criteria.most_recent_step if ( not has_any_of and not has_throughput and not has_failures + and not has_step_count and not has_most_recent_step ): raise NotImplementedError( @@ -177,6 +196,10 @@ def check_load_completion_criteria( if fails <= req_count: return False + if has_step_count and criteria.completed_steps: + if len(steps) < criteria.completed_steps: + return False + if has_most_recent_step and criteria.most_recent_step: if not steps: return False diff --git a/monitoring/benchmarker/engine/loads/loads.py b/monitoring/benchmarker/engine/loads/loads.py index 1ff82f9f5d..2527d0b152 100644 --- a/monitoring/benchmarker/engine/loads/loads.py +++ b/monitoring/benchmarker/engine/loads/loads.py @@ -8,6 +8,7 @@ BenchmarkUserName, BenchmarkUserSpecification, ) +from monitoring.benchmarker.engine.coordination import Coordinator from monitoring.benchmarker.engine.loads.user_ramp.user_ramp import run_user_ramp_load from monitoring.benchmarker.engine.operations import ExecutedOperation from monitoring.benchmarker.reports.report import BenchmarkScenarioStepReport @@ -19,6 +20,7 @@ async def run_scenario_load( user_specs_map: dict[BenchmarkUserName, BenchmarkUserSpecification], resource_pool: dict[ResourceID, Any], executor: ThreadPoolExecutor, + coordinator: Coordinator, ) -> tuple[list[ExecutedOperation], list[BenchmarkScenarioStepReport]]: """Execute a scenario load.""" if "user_ramp" in load_spec and load_spec.user_ramp: @@ -27,6 +29,7 @@ async def run_scenario_load( user_specs_map, resource_pool, executor, + coordinator, ) else: raise NotImplementedError( diff --git a/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py b/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py index 518db7285d..47a0dc11a4 100644 --- a/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py +++ b/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py @@ -2,6 +2,7 @@ import time from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime +from random import Random from typing import Any from implicitdict import StringBasedDateTime @@ -14,6 +15,7 @@ BenchmarkUserName, BenchmarkUserSpecification, ) +from monitoring.benchmarker.engine.coordination import Coordinator from monitoring.benchmarker.engine.loads.criteria import ( check_load_completion_criteria, check_stability_criteria, @@ -35,6 +37,7 @@ async def run_user_ramp_load( user_specs_map: dict[BenchmarkUserName, BenchmarkUserSpecification], resource_pool: dict[ResourceID, Any], executor: ThreadPoolExecutor, + coordinator: Coordinator, ) -> tuple[list[ExecutedOperation], list[BenchmarkScenarioStepReport]]: """Apply a load by driving virtual user workflows and monitoring step criteria.""" ramp_user_type = ramp.user_type @@ -44,6 +47,12 @@ async def run_user_ramp_load( ) user_spec = user_specs_map[ramp_user_type] + random = ( + Random(ramp.random_seed) + if "random_seed" in ramp and ramp.random_seed + else Random() + ) + active_tasks: list[asyncio.Task] = [] virtual_users: list[VirtualUser] = [] stop_event = asyncio.Event() @@ -62,7 +71,8 @@ def update_status_time() -> None: last_status_time[0] = time.monotonic() def wrapped_record_op(op: ExecutedOperation) -> None: - update_status_time() + if not op.successful: + update_status_time() record_operation(op, operations) logger.info(f"Starting user_ramp load with initial_users={current_load_factor}") @@ -107,7 +117,9 @@ async def _periodic_summary_logger() -> None: user_spec, resource_pool, executor, + coordinator, wrapped_record_op, + Random(random.randint(0, 2 << 31)), ) virtual_users.append(vu) active_tasks.append(asyncio.create_task(vu.run_workflow(stop_event))) diff --git a/monitoring/benchmarker/engine/users/creation.py b/monitoring/benchmarker/engine/users/creation.py index 610ce3dc78..74e3093289 100644 --- a/monitoring/benchmarker/engine/users/creation.py +++ b/monitoring/benchmarker/engine/users/creation.py @@ -1,10 +1,12 @@ -from collections.abc import Callable +from collections.abc import Callable, Iterable, Sequence from concurrent.futures import ThreadPoolExecutor +from random import Random from typing import Any from monitoring.benchmarker.configurations.users import ( BenchmarkUserSpecification, ) +from monitoring.benchmarker.engine.coordination import CoordinationGroupID, Coordinator from monitoring.benchmarker.engine.operations import ExecutedOperation from monitoring.benchmarker.engine.users.flight_planner.flight_planner import ( FlightPlannerUser, @@ -18,13 +20,33 @@ def create_virtual_user( user_spec: BenchmarkUserSpecification, resource_pool: dict[ResourceID, Any], executor: ThreadPoolExecutor, + coordinator: Coordinator, record_operation: Callable[[ExecutedOperation], None], + random: Random, ) -> VirtualUser: if "flight_planner" in user_spec and user_spec.flight_planner is not None: return FlightPlannerUser( - user_id, user_spec, resource_pool, executor, record_operation + user_id, + user_spec, + resource_pool, + executor, + coordinator, + record_operation, + random, ) else: raise NotImplementedError( f"User type '{user_spec.name}' does not specify any implemented behavior" ) + + +def enumerate_coordination_groups( + users: Sequence[BenchmarkUserSpecification], +) -> Iterable[CoordinationGroupID]: + for user in users: + if "flight_planner" in user and user.flight_planner: + yield from FlightPlannerUser.enumerate_coordination_groups( + user.flight_planner + ) + else: + raise NotImplementedError() diff --git a/monitoring/benchmarker/engine/users/flight_planner/astm_net_rid.py b/monitoring/benchmarker/engine/users/flight_planner/astm_net_rid.py index 936d4752fe..26d5f2fabe 100644 --- a/monitoring/benchmarker/engine/users/flight_planner/astm_net_rid.py +++ b/monitoring/benchmarker/engine/users/flight_planner/astm_net_rid.py @@ -133,7 +133,7 @@ async def create_isa(self, flight: Flight, isa_id: str) -> list[FlightAction]: alt_lo=altitude_lower.to_w84_m(), alt_hi=altitude_upper.to_w84_m(), start_time=flight.start_time, - end_time=flight.end_time, + end_time=flight.planned_end_time, uss_base_url=uss_base_url, isa_id=isa_id, rid_version=dss_instance.rid_version, @@ -150,14 +150,14 @@ async def create_isa(self, flight: Flight, isa_id: str) -> list[FlightAction]: CompletedFlightAction( type=FlightActionType.CreateISA, initiated_at=t0, - success=isa_success, + causes_flight_failure=not isa_success, ) ) if isa_success and self.isa_per_flight.after_flight_end: new_actions.append( FlightAction( - timestamp=flight.end_time + timestamp=flight.actual_end_time + self.isa_per_flight.after_flight_end.timedelta, flight_id=flight.id, start=partial( @@ -196,7 +196,7 @@ async def delete_isa(self, flight: Flight, isa: ISA) -> list[FlightAction]: CompletedFlightAction( type=FlightActionType.DeleteISA, initiated_at=t0, - success=del_success, + causes_flight_failure=not del_success, ) ) diff --git a/monitoring/benchmarker/engine/users/flight_planner/flight_generation.py b/monitoring/benchmarker/engine/users/flight_planner/flight_generation.py index 7fa1f92054..afd39d5672 100644 --- a/monitoring/benchmarker/engine/users/flight_planner/flight_generation.py +++ b/monitoring/benchmarker/engine/users/flight_planner/flight_generation.py @@ -1,7 +1,9 @@ import uuid -from datetime import datetime +from datetime import datetime, timedelta +from random import Random from monitoring.benchmarker.configurations.users import ( + FlightExecutionSpecification, FlightGenerationSpecification, IndependentComponentsFlightGenerationSpecification, ) @@ -21,37 +23,42 @@ class IndependentTimeLocationShape(FlightGenerator): spec: IndependentComponentsFlightGenerationSpecification + execution: FlightExecutionSpecification | None = None + random: Random def __init__( - self, spec: IndependentComponentsFlightGenerationSpecification, user_id: str + self, spec: FlightGenerationSpecification, user_id: str, random: Random ): - self.spec = spec - - if "fixed_spacing" in spec.time and spec.time.fixed_spacing: - pass - else: - raise NotImplementedError( - f"No supported time component is specified in independent_time_location_shape.time for user `{user_id}`" + if ( + "independent_time_location_shape" not in spec + or not spec.independent_time_location_shape + ): + raise ValueError( + "`independent_time_location_shape` not specified for IndependentTimeLocationShape FlightGenerator" ) - if "fixed_location" in spec.location and spec.location.fixed_location: + self.spec = spec.independent_time_location_shape + if "execution" in spec and spec.execution: + self.execution = spec.execution + + if "fixed_location" in self.spec.location and self.spec.location.fixed_location: pass else: raise NotImplementedError( "No supported location component is specified in independent_time_location_shape.location for user `user_id`" ) - if "fixed_volumes" in spec.shape and spec.shape.fixed_volumes: + if "fixed_volumes" in self.spec.shape and self.spec.shape.fixed_volumes: pass else: raise NotImplementedError( "No supported shape component specified in independent_time_location_shape.shape for user `user_id`" ) - if spec.shape.fixed_volumes and spec.location.fixed_location: + if self.spec.shape.fixed_volumes and self.spec.location.fixed_location: # Check unit/reference match between fixed_location and fixed_volumes - fixed_vols = spec.shape.fixed_volumes - fixed_loc = spec.location.fixed_location + fixed_vols = self.spec.shape.fixed_volumes + fixed_loc = self.spec.location.fixed_location origin_vert = fixed_vols.origin_vertical if ( fixed_loc.vertical.reference != origin_vert.reference @@ -61,11 +68,18 @@ def __init__( "Combining vertical location and shape with different reference or units is not supported" ) + self.random = random + def generate_flight(self, previous_flight_end: datetime) -> Flight: - if self.spec.time.fixed_spacing: - t0 = previous_flight_end + self.spec.time.fixed_spacing.timedelta - else: - raise NotImplementedError("Specified time component not implemented") + dt = timedelta(seconds=0) + if "fixed_spacing" in self.spec.time and self.spec.time.fixed_spacing: + dt += self.spec.time.fixed_spacing.timedelta + if ( + "uniform_random_spacing" in self.spec.time + and self.spec.time.uniform_random_spacing + ): + dt += self.spec.time.uniform_random_spacing.timedelta * self.random.random() + t0 = previous_flight_end + dt if self.spec.location.fixed_location: xy = self.spec.location.fixed_location.horizontal @@ -107,19 +121,44 @@ def generate_flight(self, previous_flight_end: datetime) -> Flight: else: raise NotImplementedError("Specified shape component not implemented") - return Flight(id=FlightID(uuid.uuid4()), volumes=volumes) + time_end = volumes.time_end_not_none.datetime + actual_end_time = time_end + if self.execution: + time_start = volumes.time_start_not_none.datetime + if ( + "end_flight_after_start" in self.execution + and self.execution.end_flight_after_start + ): + actual_end_time = min( + actual_end_time, + time_start + self.execution.end_flight_after_start.timedelta, + ) + if ( + "end_flight_before_end" in self.execution + and self.execution.end_flight_before_end + ): + actual_end_time = min( + actual_end_time, + time_end - self.execution.end_flight_before_end.timedelta, + ) + actual_end_time = max(actual_end_time, time_start) + + return Flight( + id=FlightID(uuid.uuid4()), volumes=volumes, actual_end_time=actual_end_time + ) def make_flight_generator( - flight_generation: FlightGenerationSpecification, user_id: str + flight_generation: FlightGenerationSpecification, user_id: str, random: Random ) -> FlightGenerator: if ( "independent_time_location_shape" in flight_generation and flight_generation.independent_time_location_shape ): return IndependentTimeLocationShape( - flight_generation.independent_time_location_shape, + flight_generation, user_id, + random, ) else: raise ValueError( diff --git a/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py b/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py index 4a27f7d4b1..ac575afd3d 100644 --- a/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py +++ b/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py @@ -1,8 +1,9 @@ import asyncio import heapq -from collections.abc import Callable +from collections.abc import Callable, Iterable from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime +from random import Random from typing import Any from implicitdict import StringBasedDateTime @@ -11,7 +12,9 @@ from monitoring.benchmarker.configurations.loads import OperationType, WorkflowType from monitoring.benchmarker.configurations.users import ( BenchmarkUserSpecification, + FlightPlannerSpecification, ) +from monitoring.benchmarker.engine.coordination import CoordinationGroupID, Coordinator from monitoring.benchmarker.engine.operations import ExecutedOperation from monitoring.benchmarker.engine.users.flight_planner.astm_net_rid import ( ASTMNetRIDHandler, @@ -25,6 +28,7 @@ FlightAction, FlightID, ) +from monitoring.benchmarker.engine.users.flight_planner.scd import SCDHandler from monitoring.benchmarker.engine.users.framework import VirtualUser from monitoring.uss_qualifier.resources.definitions import ResourceID @@ -38,6 +42,9 @@ class FlightPlannerUser(VirtualUser): net_rid: ASTMNetRIDHandler | None = None """Means by which this user performs ASTM network remote ID (if any).""" + scd: SCDHandler | None = None + """Means by which this user perform strategic conflict detection (if any).""" + flights: dict[FlightID, Flight] """Flights originating from this user with outstanding actions.""" @@ -50,16 +57,20 @@ def __init__( user_spec: BenchmarkUserSpecification, resource_pool: dict[ResourceID, Any], executor: ThreadPoolExecutor, + coordinator: Coordinator, record_operation: Callable[[ExecutedOperation], None], + random: Random, ): - super().__init__(user_id, user_spec.name, executor, record_operation) + super().__init__( + user_id, user_spec.name, executor, coordinator, record_operation + ) if "flight_planner" not in user_spec or not user_spec.flight_planner: raise ValueError( f"User specification '{user_spec.name}' has no flight_planner definition" ) self.flight_generator = make_flight_generator( - user_spec.flight_planner.flight_generation, user_id + user_spec.flight_planner.flight_generation, user_id, random ) if ( @@ -70,25 +81,33 @@ def __init__( user_spec.flight_planner.astm_netrid_behavior, resource_pool, self ) + if ( + "scd_behavior" in user_spec.flight_planner + and user_spec.flight_planner.scd_behavior + ): + self.scd = SCDHandler( + user_spec.flight_planner.scd_behavior, resource_pool, self + ) + self.flights = {} async def _make_next_flight(self) -> list[FlightAction]: if not self.most_recent_flight: t_prev_flight = datetime.now(UTC) else: - t_prev_flight = self.most_recent_flight.end_time + t_prev_flight = self.most_recent_flight.planned_end_time flight = self.flight_generator.generate_flight(t_prev_flight) new_actions: list[FlightAction] = [ FlightAction( - timestamp=flight.end_time, + timestamp=flight.actual_end_time, flight_id=flight.id, start=flight.complete, run_on_shutdown=False, ), FlightAction( - timestamp=flight.end_time, + timestamp=flight.actual_end_time, start=self._make_next_flight, run_on_shutdown=False, ), @@ -97,6 +116,9 @@ async def _make_next_flight(self) -> list[FlightAction]: if self.net_rid: new_actions.extend(self.net_rid.get_utm_actions(flight)) + if self.scd: + new_actions.extend(self.scd.get_utm_actions(flight)) + self.flights[flight.id] = flight self.most_recent_flight = flight @@ -162,3 +184,12 @@ async def run_custom_workflow(self, stop_event: asyncio.Event) -> None: next_action = heapq.heappop(action_queue) if next_action.run_on_shutdown: await next_action.start() + + @staticmethod + def enumerate_coordination_groups( + flight_planner: FlightPlannerSpecification, + ) -> Iterable[CoordinationGroupID]: + if "scd_behavior" in flight_planner and flight_planner.scd_behavior: + yield from SCDHandler.enumerate_coordination_groups( + flight_planner.scd_behavior + ) diff --git a/monitoring/benchmarker/engine/users/flight_planner/framework.py b/monitoring/benchmarker/engine/users/flight_planner/framework.py index a080249100..9ecbe80597 100644 --- a/monitoring/benchmarker/engine/users/flight_planner/framework.py +++ b/monitoring/benchmarker/engine/users/flight_planner/framework.py @@ -25,18 +25,24 @@ class FlightActionType(StrEnum): CreateISA = "astm_netrid_behavior.create_isa" DeleteISA = "astm_netrid_behavior.delete_isa" + UpsertSCDSubscription = "scd_behavior.upsert_subscription" + DeleteSCDSubscription = "scd_behavior.delete_subscription" + UpsertOpIntent = "scd_behavior.upsert_op_intent" + DeleteOpIntent = "scd_behavior.delete_op_intent" + @dataclass class CompletedFlightAction: type: FlightActionType initiated_at: datetime - success: bool + causes_flight_failure: bool @dataclass class Flight: id: FlightID volumes: Volume4DCollection + actual_end_time: datetime completed_actions: list[CompletedFlightAction] = field(default_factory=lambda: []) @property @@ -53,7 +59,7 @@ def start_time(self) -> datetime: return t.datetime @property - def end_time(self) -> datetime: + def planned_end_time(self) -> datetime: t = self.volumes.time_end if not t: raise ValueError(f"The volumes of flight {id} do not have an end time") @@ -61,14 +67,16 @@ def end_time(self) -> datetime: @property def successful(self) -> bool: - return all(action.success for action in self.completed_actions) + return all( + not action.causes_flight_failure for action in self.completed_actions + ) async def complete(self) -> list[FlightAction]: self.completed_actions.append( CompletedFlightAction( type=FlightActionType.CompleteFlight, initiated_at=self.start_time, - success=True, + causes_flight_failure=False, ) ) return [] diff --git a/monitoring/benchmarker/engine/users/flight_planner/scd.py b/monitoring/benchmarker/engine/users/flight_planner/scd.py new file mode 100644 index 0000000000..4190449cf1 --- /dev/null +++ b/monitoring/benchmarker/engine/users/flight_planner/scd.py @@ -0,0 +1,556 @@ +import uuid +from collections.abc import Iterable +from datetime import UTC, datetime +from functools import partial +from random import Random +from threading import Lock +from typing import Any + +import uuid6 +from loguru import logger +from uas_standards.astm.f3548.v21 import api +from uas_standards.astm.f3548.v21.constants import Scope + +from monitoring.benchmarker.configurations.user.astm.dss import ASTMDSSSelectionStrategy +from monitoring.benchmarker.configurations.user.astm.scd import ( + BehaviorSpecification, + OpIntentRefCleanupStrategy, + OpIntentRefCreationStrategy, + SubscriptionStrategy, +) +from monitoring.benchmarker.engine.coordination import ( + CoordinationGroupID, + CoordinationMessage, + CoordinationSubscriber, +) +from monitoring.benchmarker.engine.users.flight_planner.framework import ( + CompletedFlightAction, + Flight, + FlightAction, + FlightActionType, + FlightID, +) +from monitoring.benchmarker.engine.users.framework import VirtualUser +from monitoring.monitorlib.fetch import QueryError +from monitoring.monitorlib.testing import make_fake_url +from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import ( + DSSInstance, + DSSInstanceResource, + DSSInstancesResource, +) +from monitoring.uss_qualifier.resources.definitions import ResourceID + +COORDINATION_SUBJECT_ADD_OVN = "enging.users.flight_planner.scd.add_ovn" +COORDINATION_SUBJECT_REMOVE_OVN = "enging.users.flight_planner.scd.remove_ovn" + + +class SCDHandler(CoordinationSubscriber): + user: VirtualUser + + dss_instances: list[DSSInstance] + dss_selection_strategy: ASTMDSSSelectionStrategy + + subscription_strategy: SubscriptionStrategy + + op_intent_ref_creation_strategy: OpIntentRefCreationStrategy + """All fields guaranteed to be present.""" + + op_intent_ref_cleanup_strategy: OpIntentRefCleanupStrategy + """All fields guaranteed to be present.""" + + random: Random + + subscription_checked: bool = False + op_intent_refs: dict[FlightID, api.OperationalIntentReference] + + key: list[api.EntityOVN] + """OVNs currently known by this user.""" + + key_lock: Lock + """Guards `key` access, coordinating between all the different users communicating OVNs.""" + + def __init__( + self, + behavior: BehaviorSpecification, + resource_pool: dict[ResourceID, Any], + user: VirtualUser, + ): + self.user = user + + self.key = [] + self.key_lock = Lock() + + self.dss_instances = [] + scopes_required = { + Scope.StrategicCoordination.value: "sharing operational intents for flights to provide strategic conflict detection", + } + for res_id in behavior.dss_pool: + if res_id not in resource_pool: + raise ValueError( + f"Resource '{res_id}' in scd_behavior.dss_pool not found in resource pool" + ) + res = resource_pool[res_id] + if isinstance(res, DSSInstancesResource): + for dss_instance_res in res.dss_instances: + self.dss_instances.append( + dss_instance_res.get_instance(scopes_required) + ) + elif isinstance(res, DSSInstanceResource): + self.dss_instances.append(res.get_instance(scopes_required)) + else: + raise ValueError( + f"Resource '{res_id}' is not a uss_qualifier.resources.astm.f3548.v21.dss.DSSInstanceResource nor uss_qualifier.resources.astm.f3548.v21.dss.DSSInstancesResource" + ) + + if not self.dss_instances: + raise ValueError( + f"No SCD DSS instances resolved from scd_behavior.dss_pool for user '{user.user_id}'" + ) + + if len(self.dss_instances) > 1: + if "dss_selection_strategy" in behavior and behavior.dss_selection_strategy: + self.dss_selection_strategy = behavior.dss_selection_strategy + else: + raise ValueError( + f"User `{user.user_id}` specified {len(self.dss_instances)} DSS instances but no dss_selection strategy" + ) + else: + self.dss_selection_strategy = ASTMDSSSelectionStrategy.First + + if self.dss_selection_strategy not in ( + ASTMDSSSelectionStrategy.First, + ASTMDSSSelectionStrategy.Random, + ): + raise NotImplementedError( + f"DSS selection strategy '{self.dss_selection_strategy}' not implemented" + ) + + self.subscription_strategy = SubscriptionStrategy( + **behavior.subscription_strategy + ) + if "single_subscription" not in self.subscription_strategy: + self.subscription_strategy.single_subscription = None + if "implicit_subscription" not in self.subscription_strategy: + self.subscription_strategy.implicit_subscription = None + + self.op_intent_ref_creation_strategy = behavior.op_intent_ref_creation_strategy + + self.op_intent_ref_cleanup_strategy = behavior.op_intent_ref_cleanup_strategy + + self.random = Random() + + self.op_intent_refs = {} + + if ( + "ovn_coordination_group" in self.op_intent_ref_creation_strategy + and self.op_intent_ref_creation_strategy.ovn_coordination_group + ): + self.user.coordinator.subscribe( + self, self.op_intent_ref_creation_strategy.ovn_coordination_group + ) + + def receive_coordination_message(self, msg: CoordinationMessage) -> None: + if msg.subject == COORDINATION_SUBJECT_ADD_OVN: + if not isinstance(msg.content, api.EntityOVN): + raise RuntimeError( + f"Expected an EntityOVN content in a CoordinationMessage with subject {msg.subject} but instead found a {type(msg.content).__name__}" + ) + with self.key_lock: + self.key.append(msg.content) + elif msg.subject == COORDINATION_SUBJECT_REMOVE_OVN: + if not isinstance(msg.content, api.EntityOVN): + raise RuntimeError( + f"Expected an EntityOVN content in a CoordinationMessage with subject {msg.subject} but instead found a {type(msg.content).__name__}" + ) + with self.key_lock: + self.key.remove(msg.content) + + def select_dss_instance(self) -> DSSInstance: + if self.dss_selection_strategy == ASTMDSSSelectionStrategy.First: + return self.dss_instances[0] + elif self.dss_selection_strategy == ASTMDSSSelectionStrategy.Random: + return self.random.choice(self.dss_instances) + else: + raise NotImplementedError( + f"Unsupported DSS selection strategy `{self.dss_selection_strategy}`" + ) + + def get_utm_actions(self, flight: Flight) -> Iterable[FlightAction]: + if ( + not self.subscription_checked + and "single_subscription" in self.subscription_strategy + and self.subscription_strategy.single_subscription + ): + yield FlightAction( + timestamp=datetime.now(UTC), + start=partial( + self.ensure_subscription_exists, + flight, + self.subscription_strategy.single_subscription.subscription_id, + ), + run_on_shutdown=False, + ) + else: + yield from self.get_create_actions(flight) + + async def ensure_subscription_exists( + self, + flight: Flight, + subscription_id: api.SubscriptionID, + ) -> list[FlightAction]: + t0 = datetime.now(UTC) + if not self.subscription_strategy.single_subscription: + raise RuntimeError( + "ensure_subscription_exists called even though single_subscription was undefined" + ) + + start_time = t0 + end_time = ( + t0 + self.subscription_strategy.single_subscription.duration.timedelta + ) + area = self.subscription_strategy.single_subscription.area.to_latlngrect() + min_alt = self.subscription_strategy.single_subscription.min_alt.to_w84_m() + max_alt = self.subscription_strategy.single_subscription.max_alt.to_w84_m() + + dss_instance = self.select_dss_instance() + uss_base_url = make_fake_url() + + mutated_sub = dss_instance.upsert_subscription( + area_vertices=area, + start_time=start_time, + end_time=end_time, + base_url=uss_base_url, + sub_id=subscription_id, + notify_for_op_intents=True, + notify_for_constraints=False, + min_alt_m=min_alt, + max_alt_m=max_alt, + ) + success = None + self.user.record_query(mutated_sub, True) + + if mutated_sub.status_code == 409: + # Subscription already exists + success = True + + if success is None and not mutated_sub.success: + success = False + + if success is None: + try: + _ = mutated_sub.subscription + success = True + except ValueError: + success = False + + flight.completed_actions.append( + CompletedFlightAction( + type=FlightActionType.UpsertSCDSubscription, + initiated_at=t0, + causes_flight_failure=not success, + ) + ) + self.subscription_checked = True + if success: + return list(self.get_create_actions(flight)) + else: + return [] + + def get_create_actions(self, flight: Flight) -> Iterable[FlightAction]: + op_intent_id = api.EntityID(uuid.uuid4()) + + if ( + "accept_before_flight_start" in self.op_intent_ref_creation_strategy + and self.op_intent_ref_creation_strategy.accept_before_flight_start + ): + yield FlightAction( + timestamp=flight.start_time + - self.op_intent_ref_creation_strategy.accept_before_flight_start.timedelta, + flight_id=flight.id, + start=partial( + self.upsert_op_intent_ref, + flight, + op_intent_id, + api.OperationalIntentState.Accepted, + ), + run_on_shutdown=False, + ) + + if ( + "activate_before_flight_start" in self.op_intent_ref_creation_strategy + and self.op_intent_ref_creation_strategy.activate_before_flight_start + ): + yield FlightAction( + timestamp=flight.start_time + - self.op_intent_ref_creation_strategy.activate_before_flight_start.timedelta, + flight_id=flight.id, + start=partial( + self.upsert_op_intent_ref, + flight, + op_intent_id, + api.OperationalIntentState.Activated, + ), + run_on_shutdown=False, + ) + + async def upsert_op_intent_ref( + self, + flight: Flight, + op_intent_id: api.EntityID, + state: api.OperationalIntentState, + ) -> list[FlightAction]: + t0 = datetime.now(UTC) + + if state not in ( + api.OperationalIntentState.Accepted, + api.OperationalIntentState.Activated, + ): + raise NotImplementedError(f"Cannot transition op intent to state {state}") + + dss_instance = self.select_dss_instance() + uss_base_url = make_fake_url() + coordination_group = ( + self.op_intent_ref_creation_strategy.ovn_coordination_group + if "ovn_coordination_group" in self.op_intent_ref_creation_strategy + and self.op_intent_ref_creation_strategy.ovn_coordination_group + else None + ) + + kwargs = {} + if self.subscription_strategy.single_subscription: + kwargs["subscription_id"] = ( + self.subscription_strategy.single_subscription.subscription_id + ) + elif self.subscription_strategy.implicit_subscription: + pass + else: + raise ValueError( + f"No valid subscription strategy specified in scd_behavior.subscription_strategy for user `{self.user.user_id}`" + ) + + op_intent_ref = self.op_intent_refs.get(flight.id, None) + old_ovn = op_intent_ref.ovn if op_intent_ref else None + ovn_suffix = None + requested_ovn = None + + attempts = 1 + if ( + "retries" in self.op_intent_ref_creation_strategy + and self.op_intent_ref_creation_strategy.retries + ): + attempts += self.op_intent_ref_creation_strategy.retries + success = False + for attempt in range(attempts): + if ( + "coordinate_requested_ovns" in self.op_intent_ref_creation_strategy + and self.op_intent_ref_creation_strategy.coordinate_requested_ovns + ): + ovn_suffix = str(uuid6.uuid7()) + requested_ovn = api.EntityOVN(f"{op_intent_id}_{ovn_suffix}") + if coordination_group: + self.user.coordinator.publish( + coordination_group, + COORDINATION_SUBJECT_ADD_OVN, + requested_ovn, + ) + + with self.key_lock: + key = list(self.key) + + try: + # TODO: upsert op intent asynchronously + op_intent_ref, _, query = await self.user.run_sync_client_call( + dss_instance.put_op_intent, + extents=flight.volumes.to_f3548v21(), + key=key, + state=state, + base_url=uss_base_url, + oi_id=op_intent_id, + ovn=old_ovn, + requested_ovn_suffix=ovn_suffix, + **kwargs, + ) + self.user.record_query(query) + success = True + break + except QueryError as e: + for q in e.queries: + self.user.record_query(q) + if q.status_code == 409 and q.response.json: + missing_op_intents = q.response.json.get( + "missing_operational_intents" + ) + if missing_op_intents: + logger.warning( + f"{self.user.user_id} missing OVNs for OI {op_intent_id}:\n" + + "\n".join( + f"OI {oi.get('id', None)}: OVN {oi.get('ovn', None)}" + for oi in missing_op_intents + ) + ) + flight.completed_actions.append( + CompletedFlightAction( + type=FlightActionType.UpsertOpIntent, + initiated_at=t0, + causes_flight_failure=attempt >= attempts - 1, + ) + ) + t0 = datetime.now(UTC) + if requested_ovn and coordination_group: + self.user.coordinator.publish( + coordination_group, + COORDINATION_SUBJECT_REMOVE_OVN, + requested_ovn, + ) + + if not success: + return [] + + if op_intent_ref is None: + raise RuntimeError( + "op_intent_ref cannot be None upon successful op intent ref upsertion" + ) + + flight.completed_actions.append( + CompletedFlightAction( + type=FlightActionType.UpsertOpIntent, + initiated_at=t0, + causes_flight_failure=False, + ) + ) + self.op_intent_refs[flight.id] = op_intent_ref + + # Notify other coordinating users + if coordination_group: + if op_intent_ref.ovn != old_ovn and op_intent_ref.ovn != requested_ovn: + self.user.coordinator.publish( + coordination_group, + COORDINATION_SUBJECT_ADD_OVN, + op_intent_ref.ovn, + ) + if old_ovn and op_intent_ref.ovn != old_ovn: + self.user.coordinator.publish( + coordination_group, + COORDINATION_SUBJECT_REMOVE_OVN, + old_ovn, + ) + if requested_ovn and op_intent_ref.ovn != requested_ovn: + logger.warning( + f"Requested OVN {requested_ovn} was not accepted; returned {op_intent_ref.ovn} instead for user {self.user.user_id}" + ) + self.user.coordinator.publish( + coordination_group, + COORDINATION_SUBJECT_REMOVE_OVN, + requested_ovn, + ) + else: + if old_ovn and op_intent_ref.ovn != old_ovn: + self.receive_coordination_message( + CoordinationMessage( + group_id=None, + subject=COORDINATION_SUBJECT_REMOVE_OVN, + content=op_intent_ref.ovn, + ) + ) + if old_ovn is None or op_intent_ref.ovn != old_ovn: + self.receive_coordination_message( + CoordinationMessage( + group_id=None, + subject=COORDINATION_SUBJECT_ADD_OVN, + content=op_intent_ref.ovn, + ) + ) + + return list(self.get_delete_actions(flight, op_intent_id)) + + def get_delete_actions( + self, flight: Flight, op_intent_id: api.EntityID + ) -> Iterable[FlightAction]: + deletion_time = None + if ( + "after_actual_flight_end" in self.op_intent_ref_cleanup_strategy + and self.op_intent_ref_cleanup_strategy.after_actual_flight_end + ): + deletion_time = ( + flight.actual_end_time + + self.op_intent_ref_cleanup_strategy.after_actual_flight_end.timedelta + ) + if ( + "after_planned_flight_end" in self.op_intent_ref_cleanup_strategy + and self.op_intent_ref_cleanup_strategy.after_planned_flight_end + ): + new_time = ( + flight.volumes.time_end_not_none.datetime + + self.op_intent_ref_cleanup_strategy.after_planned_flight_end.timedelta + ) + deletion_time = ( + new_time if deletion_time is None else max(deletion_time, new_time) + ) + if deletion_time is not None: + yield FlightAction( + timestamp=deletion_time, + flight_id=flight.id, + start=partial( + self.delete_op_intent_ref, + flight, + op_intent_id, + ), + run_on_shutdown=True, + ) + + async def delete_op_intent_ref( + self, flight: Flight, op_intent_id: api.EntityID + ) -> list[FlightAction]: + t0 = datetime.now(UTC) + dss_instance = self.select_dss_instance() + + with self.key_lock: + op_intent_ref = self.op_intent_refs.pop(flight.id, None) + + if op_intent_ref: + # TODO: delete op intent ref asynchronously + _, _, query = await self.user.run_sync_client_call( + dss_instance.delete_op_intent, + id=op_intent_id, + ovn=op_intent_ref.ovn, + ) + + success = query.status_code == 200 + self.user.record_query(query, success) + flight.completed_actions.append( + CompletedFlightAction( + type=FlightActionType.DeleteOpIntent, + initiated_at=t0, + causes_flight_failure=not success, + ) + ) + + if ( + "ovn_coordination_group" in self.op_intent_ref_creation_strategy + and self.op_intent_ref_creation_strategy.ovn_coordination_group + ): + self.user.coordinator.publish( + self.op_intent_ref_creation_strategy.ovn_coordination_group, + COORDINATION_SUBJECT_REMOVE_OVN, + op_intent_ref.ovn, + ) + else: + self.receive_coordination_message( + CoordinationMessage( + group_id=None, + subject=COORDINATION_SUBJECT_REMOVE_OVN, + content=op_intent_ref.ovn, + ) + ) + + return [] + + @staticmethod + def enumerate_coordination_groups( + behavior: BehaviorSpecification, + ) -> Iterable[CoordinationGroupID]: + if ( + "ovn_coordination_group" in behavior.op_intent_ref_creation_strategy + and behavior.op_intent_ref_creation_strategy.ovn_coordination_group + ): + yield behavior.op_intent_ref_creation_strategy.ovn_coordination_group diff --git a/monitoring/benchmarker/engine/users/framework.py b/monitoring/benchmarker/engine/users/framework.py index 227af0ea9e..57d5fab486 100644 --- a/monitoring/benchmarker/engine/users/framework.py +++ b/monitoring/benchmarker/engine/users/framework.py @@ -13,7 +13,9 @@ from monitoring.benchmarker.configurations.users import ( BenchmarkUserName, ) +from monitoring.benchmarker.engine.coordination import Coordinator from monitoring.benchmarker.engine.operations import ExecutedOperation +from monitoring.monitorlib.errors import stacktrace_string from monitoring.monitorlib.fetch import Query @@ -28,6 +30,8 @@ class VirtualUser(ABC): executor: ThreadPoolExecutor """Means by which to perform synchronous work in the async context.""" + coordinator: Coordinator + record_operation: Callable[[ExecutedOperation], None] """Means by which to record an operation completed by the virtual user.""" @@ -36,11 +40,13 @@ def __init__( user_id: str, user_type_name: BenchmarkUserName, executor: ThreadPoolExecutor, + coordinator: Coordinator, record_operation: Callable[[ExecutedOperation], None], ): self.user_id = user_id self.user_type_name = user_type_name self.executor = executor + self.coordinator = coordinator self.record_operation = record_operation async def run_sync_client_call( @@ -80,7 +86,9 @@ async def run_workflow(self, stop_event: asyncio.Event) -> None: try: await self.run_custom_workflow(stop_event) except Exception as e: - logger.error(f"Error during user {self.user_id} workflow: {e}") + logger.error( + f"{type(e)}.__name__ during user {self.user_id} workflow: {str(e)}\n{stacktrace_string(e)}" + ) @abstractmethod async def run_custom_workflow(self, stop_event: asyncio.Event) -> None: diff --git a/monitoring/benchmarker/reports/analysis.py b/monitoring/benchmarker/reports/analysis.py index edf7257aae..dabf1aec15 100644 --- a/monitoring/benchmarker/reports/analysis.py +++ b/monitoring/benchmarker/reports/analysis.py @@ -91,7 +91,8 @@ def throughput_of_operations( 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,) + if "outcomes" not in kwargs: + kwargs["outcomes"] = (True,) kwargs["completed_after"] = start_time kwargs["completed_before"] = end_time return ( diff --git a/monitoring/monitorlib/geotemporal.py b/monitoring/monitorlib/geotemporal.py index 4671884ee5..f9ac60d525 100644 --- a/monitoring/monitorlib/geotemporal.py +++ b/monitoring/monitorlib/geotemporal.py @@ -350,6 +350,13 @@ def time_start(self) -> Time | None: else None ) + @property + def time_start_not_none(self) -> Time: + t = self.time_start + if t is None: + raise ValueError("Start times of all volumes must be defined") + return t + @property def time_end(self) -> Time | None: return ( @@ -358,6 +365,13 @@ def time_end(self) -> Time | None: else None ) + @property + def time_end_not_none(self) -> Time: + t = self.time_end + if t is None: + raise ValueError("End times of all volumes must be defined") + return t + @property def altitude_lower(self) -> Altitude | None: return Altitude( @@ -368,6 +382,13 @@ def altitude_lower(self) -> Altitude | None: units=DistanceUnits.M, ) + @property + def altitude_lower_not_none(self) -> Altitude: + alt = self.altitude_lower + if alt is None: + raise ValueError("Lower altitudes of all volumes must be defined") + return alt + @property def altitude_upper(self) -> Altitude | None: return Altitude( @@ -378,6 +399,13 @@ def altitude_upper(self) -> Altitude | None: units=DistanceUnits.M, ) + @property + def altitude_upper_not_none(self) -> Altitude: + alt = self.altitude_upper + if alt is None: + raise ValueError("Upper altitudes of all volumes must be defined") + return alt + def offset_times(self, dt: timedelta) -> Volume4DCollection: return Volume4DCollection([v.offset_time(dt) for v in self]) diff --git a/schemas/monitoring/benchmarker/configurations/loads/LoadCompletionCriteria.json b/schemas/monitoring/benchmarker/configurations/loads/LoadCompletionCriteria.json index 5556ef8559..0d7815a4b7 100644 --- a/schemas/monitoring/benchmarker/configurations/loads/LoadCompletionCriteria.json +++ b/schemas/monitoring/benchmarker/configurations/loads/LoadCompletionCriteria.json @@ -16,6 +16,13 @@ "null" ] }, + "completed_steps": { + "description": "Evaluates true when this many steps have been completed.", + "type": [ + "integer", + "null" + ] + }, "failures_more_than": { "description": "Evaluates true when the number of failures for the specified operations exceeds the specified number.", "oneOf": [ diff --git a/schemas/monitoring/benchmarker/configurations/loads/StepCompletionCriteria.json b/schemas/monitoring/benchmarker/configurations/loads/StepCompletionCriteria.json index 2d520cf403..9355d51fc7 100644 --- a/schemas/monitoring/benchmarker/configurations/loads/StepCompletionCriteria.json +++ b/schemas/monitoring/benchmarker/configurations/loads/StepCompletionCriteria.json @@ -38,6 +38,17 @@ } ] }, + "failures_more_than": { + "description": "Evaluates true when the number of failures for the specified operations exceeds the specified number during this step.", + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "OperationCount.json" + } + ] + }, "sampling_duration_at_least": { "description": "Evaluates true when the step has been collecting valid throughput data for at least this long.", "format": "duration", diff --git a/schemas/monitoring/benchmarker/configurations/loads/UserRampLoad.json b/schemas/monitoring/benchmarker/configurations/loads/UserRampLoad.json index df5535d90c..d51c44bfc2 100644 --- a/schemas/monitoring/benchmarker/configurations/loads/UserRampLoad.json +++ b/schemas/monitoring/benchmarker/configurations/loads/UserRampLoad.json @@ -19,6 +19,13 @@ "$ref": "LoadCompletionCriteria.json", "description": "The load is considered complete if these criteria are met." }, + "random_seed": { + "description": "Seed to use to randomly generate seeds for each user.", + "type": [ + "integer", + "null" + ] + }, "step_completion_criteria": { "$ref": "StepCompletionCriteria.json", "description": "The current step is considered complete once these criteria are met." diff --git a/schemas/monitoring/benchmarker/configurations/scenarios/BenchmarkScenarioSpecification.json b/schemas/monitoring/benchmarker/configurations/scenarios/BenchmarkScenarioSpecification.json index df8118cbd0..2c1c3cca0e 100644 --- a/schemas/monitoring/benchmarker/configurations/scenarios/BenchmarkScenarioSpecification.json +++ b/schemas/monitoring/benchmarker/configurations/scenarios/BenchmarkScenarioSpecification.json @@ -13,7 +13,10 @@ }, "metadata": { "description": "Arbitrary data that may be relevant to the scenario.", - "type": "object" + "type": [ + "object", + "null" + ] }, "name": { "type": "string" @@ -45,7 +48,6 @@ }, "required": [ "load", - "metadata", "name" ], "type": "object" diff --git a/schemas/monitoring/benchmarker/configurations/user/astm/scd/BehaviorSpecification.json b/schemas/monitoring/benchmarker/configurations/user/astm/scd/BehaviorSpecification.json new file mode 100644 index 0000000000..6e1bd2def9 --- /dev/null +++ b/schemas/monitoring/benchmarker/configurations/user/astm/scd/BehaviorSpecification.json @@ -0,0 +1,44 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/user/astm/scd/BehaviorSpecification.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.configurations.user.astm.scd.BehaviorSpecification, as defined in monitoring/benchmarker/configurations/user/astm/scd.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "dss_pool": { + "description": "Means to interact with the ASTM DSS.\n\nBenchmark configuration must contain a `resources.astm.f3548.v21.DSSInstanceResource` resource with each of these names.", + "items": { + "type": "string" + }, + "type": "array" + }, + "dss_selection_strategy": { + "enum": [ + "First", + "Random" + ], + "type": [ + "string", + "null" + ] + }, + "op_intent_ref_cleanup_strategy": { + "$ref": "OpIntentRefCleanupStrategy.json" + }, + "op_intent_ref_creation_strategy": { + "$ref": "OpIntentRefCreationStrategy.json" + }, + "subscription_strategy": { + "$ref": "SubscriptionStrategy.json" + } + }, + "required": [ + "dss_pool", + "op_intent_ref_cleanup_strategy", + "op_intent_ref_creation_strategy", + "subscription_strategy" + ], + "type": "object" +} \ No newline at end of file diff --git a/schemas/monitoring/benchmarker/configurations/user/astm/scd/ImplicitSubscription.json b/schemas/monitoring/benchmarker/configurations/user/astm/scd/ImplicitSubscription.json new file mode 100644 index 0000000000..12baf3b186 --- /dev/null +++ b/schemas/monitoring/benchmarker/configurations/user/astm/scd/ImplicitSubscription.json @@ -0,0 +1,12 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/user/astm/scd/ImplicitSubscription.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.configurations.user.astm.scd.ImplicitSubscription, as defined in monitoring/benchmarker/configurations/user/astm/scd.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + } + }, + "type": "object" +} \ No newline at end of file diff --git a/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCleanupStrategy.json b/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCleanupStrategy.json new file mode 100644 index 0000000000..df3a0e1037 --- /dev/null +++ b/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCleanupStrategy.json @@ -0,0 +1,28 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCleanupStrategy.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.configurations.user.astm.scd.OpIntentRefCleanupStrategy, as defined in monitoring/benchmarker/configurations/user/astm/scd.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "after_actual_flight_end": { + "description": "Delete operational intent reference this amount of time after the actual end of the flight.", + "format": "duration", + "type": [ + "string", + "null" + ] + }, + "after_planned_flight_end": { + "description": "Delete operational intent reference this amount of time after the planned end of the flight (end of last volume).", + "format": "duration", + "type": [ + "string", + "null" + ] + } + }, + "type": "object" +} \ No newline at end of file diff --git a/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCreationStrategy.json b/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCreationStrategy.json new file mode 100644 index 0000000000..faa864fb0c --- /dev/null +++ b/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCreationStrategy.json @@ -0,0 +1,49 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/user/astm/scd/OpIntentRefCreationStrategy.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.configurations.user.astm.scd.OpIntentRefCreationStrategy, as defined in monitoring/benchmarker/configurations/user/astm/scd.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "accept_before_flight_start": { + "description": "Create an operational intent in the Accepted state this long before the flight starts (start of first volume).", + "format": "duration", + "type": [ + "string", + "null" + ] + }, + "activate_before_flight_start": { + "description": "Create or update an operational intent to the Activated state this long before the flight starts (start of first volume).", + "format": "duration", + "type": [ + "string", + "null" + ] + }, + "coordinate_requested_ovns": { + "description": "If specified true, use requested OVNs and share them with the coordination group before requesting.", + "type": [ + "boolean", + "null" + ] + }, + "ovn_coordination_group": { + "description": "If specified, exchange OVNs directly between other virtual users using this coordination group to exchange OVNs.", + "type": [ + "string", + "null" + ] + }, + "retries": { + "description": "If specified, number of times to retry failed operational intent operations before giving up.", + "type": [ + "integer", + "null" + ] + } + }, + "type": "object" +} \ No newline at end of file diff --git a/schemas/monitoring/benchmarker/configurations/user/astm/scd/SingleSubscription.json b/schemas/monitoring/benchmarker/configurations/user/astm/scd/SingleSubscription.json new file mode 100644 index 0000000000..9511f3a8a0 --- /dev/null +++ b/schemas/monitoring/benchmarker/configurations/user/astm/scd/SingleSubscription.json @@ -0,0 +1,40 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/user/astm/scd/SingleSubscription.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.configurations.user.astm.scd.SingleSubscription, as defined in monitoring/benchmarker/configurations/user/astm/scd.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "area": { + "$ref": "../../../../../monitorlib/geo/LatLngBoundingBox.json", + "description": "Horizontal area this subscription should cover." + }, + "duration": { + "description": "Duration of the subscription, from the time it is created.", + "format": "duration", + "type": "string" + }, + "max_alt": { + "$ref": "../../../../../monitorlib/geo/Altitude.json", + "description": "Altitude above which this subscription should not apply." + }, + "min_alt": { + "$ref": "../../../../../monitorlib/geo/Altitude.json", + "description": "Altitude below which this subscription should not apply." + }, + "subscription_id": { + "description": "ID of the single subscription to create, or ensure exists.", + "type": "string" + } + }, + "required": [ + "area", + "duration", + "max_alt", + "min_alt", + "subscription_id" + ], + "type": "object" +} \ No newline at end of file diff --git a/schemas/monitoring/benchmarker/configurations/user/astm/scd/SubscriptionStrategy.json b/schemas/monitoring/benchmarker/configurations/user/astm/scd/SubscriptionStrategy.json new file mode 100644 index 0000000000..6526205ebe --- /dev/null +++ b/schemas/monitoring/benchmarker/configurations/user/astm/scd/SubscriptionStrategy.json @@ -0,0 +1,34 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/user/astm/scd/SubscriptionStrategy.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.configurations.user.astm.scd.SubscriptionStrategy, as defined in monitoring/benchmarker/configurations/user/astm/scd.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "implicit_subscription": { + "description": "Planner has the DSS establish an implicit subscription for each individual flight.", + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "ImplicitSubscription.json" + } + ] + }, + "single_subscription": { + "description": "Planner ensures there is a single subscription, established at the start of operations, covering all their flights.", + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "SingleSubscription.json" + } + ] + } + }, + "type": "object" +} \ No newline at end of file diff --git a/schemas/monitoring/benchmarker/configurations/users/FlightExecutionSpecification.json b/schemas/monitoring/benchmarker/configurations/users/FlightExecutionSpecification.json new file mode 100644 index 0000000000..4ff236fe4d --- /dev/null +++ b/schemas/monitoring/benchmarker/configurations/users/FlightExecutionSpecification.json @@ -0,0 +1,28 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/users/FlightExecutionSpecification.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.configurations.users.FlightExecutionSpecification, as defined in monitoring/benchmarker/configurations/users.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "end_flight_after_start": { + "description": "If specified, the operator ends their flight early this amount of time after the flight began.", + "format": "duration", + "type": [ + "string", + "null" + ] + }, + "end_flight_before_end": { + "description": "If specified, the operator ends their flight early this amount of time before the flight is scheduled to end.", + "format": "duration", + "type": [ + "string", + "null" + ] + } + }, + "type": "object" +} \ No newline at end of file diff --git a/schemas/monitoring/benchmarker/configurations/users/FlightGenerationSpecification.json b/schemas/monitoring/benchmarker/configurations/users/FlightGenerationSpecification.json index b7796cb323..24ae4893e7 100644 --- a/schemas/monitoring/benchmarker/configurations/users/FlightGenerationSpecification.json +++ b/schemas/monitoring/benchmarker/configurations/users/FlightGenerationSpecification.json @@ -7,6 +7,17 @@ "description": "Path to content that replaces the $ref", "type": "string" }, + "execution": { + "description": "How operator actually executes their planned flights.", + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "FlightExecutionSpecification.json" + } + ] + }, "independent_time_location_shape": { "oneOf": [ { diff --git a/schemas/monitoring/benchmarker/configurations/users/FlightPlannerSpecification.json b/schemas/monitoring/benchmarker/configurations/users/FlightPlannerSpecification.json index 1c8bb7c525..4a2b0da966 100644 --- a/schemas/monitoring/benchmarker/configurations/users/FlightPlannerSpecification.json +++ b/schemas/monitoring/benchmarker/configurations/users/FlightPlannerSpecification.json @@ -21,6 +21,17 @@ "flight_generation": { "$ref": "FlightGenerationSpecification.json", "description": "How flight planner generates their flights." + }, + "scd_behavior": { + "description": "How flight planner provides strategic conflict detection service for their flights.", + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "../user/astm/scd/BehaviorSpecification.json" + } + ] } }, "required": [ diff --git a/schemas/monitoring/benchmarker/configurations/users/FlightTimeGenerationSpecification.json b/schemas/monitoring/benchmarker/configurations/users/FlightTimeGenerationSpecification.json index 66367ebae4..9cb6e4d3c3 100644 --- a/schemas/monitoring/benchmarker/configurations/users/FlightTimeGenerationSpecification.json +++ b/schemas/monitoring/benchmarker/configurations/users/FlightTimeGenerationSpecification.json @@ -8,7 +8,15 @@ "type": "string" }, "fixed_spacing": { - "description": "Set delay between the end of the previous flight and the start of the next flight to this value.", + "description": "Add this fixed delay between the end of the previous flight and the start of the next flight.", + "format": "duration", + "type": [ + "string", + "null" + ] + }, + "uniform_random_spacing": { + "description": "Add a random delay uniformly distributed between 0 and this duration between the end of the previous flight and the start of the next flight.", "format": "duration", "type": [ "string", diff --git a/schemas/monitoring/benchmarker/configurations/users/IndependentComponentsFlightGenerationSpecification.json b/schemas/monitoring/benchmarker/configurations/users/IndependentComponentsFlightGenerationSpecification.json index 86ce3a7271..96e26d25e1 100644 --- a/schemas/monitoring/benchmarker/configurations/users/IndependentComponentsFlightGenerationSpecification.json +++ b/schemas/monitoring/benchmarker/configurations/users/IndependentComponentsFlightGenerationSpecification.json @@ -13,7 +13,7 @@ }, "shape": { "$ref": "FlightShapeGenerationSpecification.json", - "description": "Means to generate flight shapes." + "description": "Means to generate flight shapes.\n\nNote that these volumes represent planned volumes and the flight may be ended before the end of the volumes." }, "time": { "$ref": "FlightTimeGenerationSpecification.json", diff --git a/schemas/monitoring/monitorlib/geo/LatLngBoundingBox.json b/schemas/monitoring/monitorlib/geo/LatLngBoundingBox.json new file mode 100644 index 0000000000..da84055d29 --- /dev/null +++ b/schemas/monitoring/monitorlib/geo/LatLngBoundingBox.json @@ -0,0 +1,34 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/monitorlib/geo/LatLngBoundingBox.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "Bounding box in latitude and longitude\n\nmonitoring.monitorlib.geo.LatLngBoundingBox, as defined in monitoring/monitorlib/geo.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "lat_max": { + "description": "Upper latitude bound (degrees)", + "type": "number" + }, + "lat_min": { + "description": "Lower latitude bound (degrees)", + "type": "number" + }, + "lng_max": { + "description": "Upper longitude bound (degrees)", + "type": "number" + }, + "lng_min": { + "description": "Lower longitude bound (degrees)", + "type": "number" + } + }, + "required": [ + "lat_max", + "lat_min", + "lng_max", + "lng_min" + ], + "type": "object" +} \ No newline at end of file From b811298603b173c68e867b1eb3263b1e406a64ec Mon Sep 17 00:00:00 2001 From: Benjamin Pelletier Date: Thu, 23 Jul 2026 11:32:59 -0700 Subject: [PATCH 2/2] Apply suggestions from code review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Mickaël Misbach --- monitoring/benchmarker/engine/loads/criteria.py | 1 - monitoring/benchmarker/engine/users/framework.py | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/monitoring/benchmarker/engine/loads/criteria.py b/monitoring/benchmarker/engine/loads/criteria.py index cb3f25bb55..c9e856ecd3 100644 --- a/monitoring/benchmarker/engine/loads/criteria.py +++ b/monitoring/benchmarker/engine/loads/criteria.py @@ -136,7 +136,6 @@ def check_step_completion_criteria( if not op.successful and op.type in relevant_ops and op.completed_at.datetime >= step_start_time - and op.type in relevant_ops and op.completed_at.datetime <= now ) if fails <= req_count: diff --git a/monitoring/benchmarker/engine/users/framework.py b/monitoring/benchmarker/engine/users/framework.py index 57d5fab486..323c36c95b 100644 --- a/monitoring/benchmarker/engine/users/framework.py +++ b/monitoring/benchmarker/engine/users/framework.py @@ -87,7 +87,7 @@ async def run_workflow(self, stop_event: asyncio.Event) -> None: await self.run_custom_workflow(stop_event) except Exception as e: logger.error( - f"{type(e)}.__name__ during user {self.user_id} workflow: {str(e)}\n{stacktrace_string(e)}" + f"{type(e).__name__} during user {self.user_id} workflow: {str(e)}\n{stacktrace_string(e)}" ) @abstractmethod