[benchmarker] Add virtual user cleanup

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 = (

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(

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

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,

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,

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:

monitoring/benchmarker/reports/report.py

@@ -79,13 +79,24 @@ 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."""
 
     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."""
 

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": [

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"
+}