diff --git a/monitoring/benchmarker/engine/engine.py b/monitoring/benchmarker/engine/engine.py index 24e5ae9d0f..58222f2ac1 100644 --- a/monitoring/benchmarker/engine/engine.py +++ b/monitoring/benchmarker/engine/engine.py @@ -89,7 +89,7 @@ async def _run_benchmark_async( f"Scenario load '{scenario_spec.load}' not defined in configuration.loads" ) load_spec = loads_map[scenario_spec.load] - scenario_ops, scenario_steps = await run_scenario_load( + scenario_ops, scenario_steps, cleanup = await run_scenario_load( load_spec, user_specs_map, resource_pool, @@ -102,6 +102,7 @@ async def _run_benchmark_async( scenario_report = BenchmarkScenarioReport( operations=group_operations(scenario_ops), steps=scenario_steps, + cleanup=cleanup, ) if "metadata" in scenario_spec: scenario_report.metadata = ( diff --git a/monitoring/benchmarker/engine/loads/loads.py b/monitoring/benchmarker/engine/loads/loads.py index ec6f35f7ee..71f77904df 100644 --- a/monitoring/benchmarker/engine/loads/loads.py +++ b/monitoring/benchmarker/engine/loads/loads.py @@ -12,7 +12,10 @@ 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 +from monitoring.benchmarker.reports.report import ( + BenchmarkScenarioStepReport, + CleanupReport, +) from monitoring.uss_qualifier.resources.definitions import ResourceID @@ -23,7 +26,9 @@ async def run_scenario_load( executor: ThreadPoolExecutor, coordinator: Coordinator, scenario_name: BenchmarkScenarioName, -) -> tuple[list[ExecutedOperation], list[BenchmarkScenarioStepReport]]: +) -> tuple[ + list[ExecutedOperation], list[BenchmarkScenarioStepReport], CleanupReport | None +]: """Execute a scenario load.""" if "user_ramp" in load_spec and load_spec.user_ramp: return await run_user_ramp_load( diff --git a/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py b/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py index f30f7290ae..4864388c4b 100644 --- a/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py +++ b/monitoring/benchmarker/engine/loads/user_ramp/user_ramp.py @@ -29,6 +29,7 @@ from monitoring.benchmarker.engine.users.framework import VirtualUser from monitoring.benchmarker.reports.report import ( BenchmarkScenarioStepReport, + CleanupReport, StepTerminationReason, ) from monitoring.uss_qualifier.resources.definitions import ResourceID @@ -43,7 +44,7 @@ async def run_user_ramp_load( executor: ThreadPoolExecutor, coordinator: Coordinator, scenario_name: BenchmarkScenarioName, -) -> tuple[list[ExecutedOperation], list[BenchmarkScenarioStepReport]]: +) -> tuple[list[ExecutedOperation], list[BenchmarkScenarioStepReport], CleanupReport]: """Apply a load by driving virtual user workflows and monitoring step criteria.""" if "user_types" in ramp and ramp.user_types: user_types_list = ramp.user_types @@ -356,6 +357,17 @@ async def _periodic_summary_logger() -> None: f"Waiting for {len(active_tasks)} active virtual users to wind down gracefully..." ) await asyncio.gather(*active_tasks, return_exceptions=True) - logger.info("All virtual users have finished.") + logger.info("All virtual users have finished their workflows.") + + logger.info(f"Cleaning up {len(virtual_users)} virtual users...") + cleanup_start = datetime.now(UTC) + for virtual_user in virtual_users: + await virtual_user.cleanup() + cleanup_end = datetime.now(UTC) + logger.info("All virtual users have been cleaned up.") + cleanup_report = CleanupReport( + start_time=StringBasedDateTime(cleanup_start), + end_time=StringBasedDateTime(cleanup_end), + ) - return operations, steps + return operations, steps, cleanup_report diff --git a/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py b/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py index d8244ee2d4..ecbb91c461 100644 --- a/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py +++ b/monitoring/benchmarker/engine/users/flight_planner/flight_planner.py @@ -207,6 +207,10 @@ async def run_custom_workflow(self, stop_event: asyncio.Event) -> None: if next_action.run_on_shutdown: await next_action.start() + async def cleanup(self) -> None: + if self.scd: + await self.scd.cleanup() + @staticmethod def enumerate_coordination_groups( flight_planner: FlightPlannerSpecification, diff --git a/monitoring/benchmarker/engine/users/flight_planner/scd.py b/monitoring/benchmarker/engine/users/flight_planner/scd.py index 5ad38dd17d..d8622ace74 100644 --- a/monitoring/benchmarker/engine/users/flight_planner/scd.py +++ b/monitoring/benchmarker/engine/users/flight_planner/scd.py @@ -547,26 +547,72 @@ async def delete_op_intent_ref( ) ) - 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, + if 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, + ) + ) + else: + with self.key_lock: + self.op_intent_refs[flight.id] = op_intent_ref return [] + async def cleanup(self) -> None: + with self.key_lock: + op_intent_refs = {k: v for k, v in self.op_intent_refs.items()} + self.op_intent_refs.clear() + + undeleted_ids = [] + n_deleted = 0 + n_already_gone = 0 + for flight_id, op_intent_ref in op_intent_refs.items(): + dss_instance = self.select_dss_instance() + try: + _, _, query = await self.user.run_sync_client_call( + dss_instance.delete_op_intent, + id=op_intent_ref.id, + ovn=op_intent_ref.ovn, + ) + self.user.record_query(query, True) + n_deleted += 1 + except QueryError as e: + success = e.queries[0].status_code == 404 + for query in e.queries: + self.user.record_query(query, success) + if success: + n_already_gone += 1 + else: + logger.warning( + f"{self.user.user_id}'s SCDHandler was unable to clean up op intent {op_intent_ref.id} for flight {flight_id} from {dss_instance.participant_id}'s DSS; HTTP code {e.queries[0].status_code}" + ) + undeleted_ids.append(flight_id) + + with self.key_lock: + for flight_id in undeleted_ids: + self.op_intent_refs[flight_id] = op_intent_refs[flight_id] + + if n_deleted + n_already_gone + len(undeleted_ids) > 0: + msg = f"{self.user.user_id}'s SCDHandler deleted {n_deleted} op intents" + if n_already_gone: + msg += f"; {n_already_gone} op intents already gone" + if len(undeleted_ids) > 0: + msg += f"; {len(undeleted_ids)} could not be deleted" + logger.debug(msg) + @staticmethod def enumerate_coordination_groups( behavior: BehaviorSpecification, diff --git a/monitoring/benchmarker/engine/users/framework.py b/monitoring/benchmarker/engine/users/framework.py index 323c36c95b..18120c87c0 100644 --- a/monitoring/benchmarker/engine/users/framework.py +++ b/monitoring/benchmarker/engine/users/framework.py @@ -94,6 +94,10 @@ async def run_workflow(self, stop_event: asyncio.Event) -> None: async def run_custom_workflow(self, stop_event: asyncio.Event) -> None: raise NotImplementedError() + @abstractmethod + async def cleanup(self) -> None: + raise NotImplementedError() + @dataclass(order=True, kw_only=True) class Action: diff --git a/monitoring/benchmarker/reports/report.py b/monitoring/benchmarker/reports/report.py index 409beb81e0..fc97569948 100644 --- a/monitoring/benchmarker/reports/report.py +++ b/monitoring/benchmarker/reports/report.py @@ -79,6 +79,14 @@ class BenchmarkScenarioStepReport(ImplicitDict): """The reason this step terminated.""" +class CleanupReport(ImplicitDict): + start_time: StringBasedDateTime + """Time cleanup started.""" + + end_time: StringBasedDateTime + """Time cleanup ended.""" + + class BenchmarkScenarioReport(ImplicitDict): operations: list[OperationsByType] """All operations that occurred during the benchmark run.""" @@ -86,6 +94,9 @@ class BenchmarkScenarioReport(ImplicitDict): steps: list[BenchmarkScenarioStepReport] """Boundaries of steps within this scenario.""" + cleanup: CleanupReport | None + """Information about cleanup activities for this scenario.""" + metadata: Optional[dict] """Arbitrary metadata copied from the scenario specification.""" diff --git a/schemas/monitoring/benchmarker/reports/report/BenchmarkScenarioReport.json b/schemas/monitoring/benchmarker/reports/report/BenchmarkScenarioReport.json index 55c7d1a68d..1b4aaae6d3 100644 --- a/schemas/monitoring/benchmarker/reports/report/BenchmarkScenarioReport.json +++ b/schemas/monitoring/benchmarker/reports/report/BenchmarkScenarioReport.json @@ -7,6 +7,17 @@ "description": "Path to content that replaces the $ref", "type": "string" }, + "cleanup": { + "description": "Information about cleanup activities for this scenario.", + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "CleanupReport.json" + } + ] + }, "metadata": { "description": "Arbitrary metadata copied from the scenario specification.", "type": [ diff --git a/schemas/monitoring/benchmarker/reports/report/CleanupReport.json b/schemas/monitoring/benchmarker/reports/report/CleanupReport.json new file mode 100644 index 0000000000..2c15078aea --- /dev/null +++ b/schemas/monitoring/benchmarker/reports/report/CleanupReport.json @@ -0,0 +1,26 @@ +{ + "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/reports/report/CleanupReport.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "monitoring.benchmarker.reports.report.CleanupReport, as defined in monitoring/benchmarker/reports/report.py", + "properties": { + "$ref": { + "description": "Path to content that replaces the $ref", + "type": "string" + }, + "end_time": { + "description": "Time cleanup ended.", + "format": "date-time", + "type": "string" + }, + "start_time": { + "description": "Time cleanup started.", + "format": "date-time", + "type": "string" + } + }, + "required": [ + "end_time", + "start_time" + ], + "type": "object" +} \ No newline at end of file