Remove ZMQ client from LegacyEnsemble

This commit changes LegacyEnsemble to put events directly in scheduler queue rather than sending it over ZMQ.

Author: jonathan-eq

src/ert/ensemble_evaluator/_ensemble.py

@@ -16,9 +16,7 @@
     FMEvent,
     ForwardModelStepFailure,
     ForwardModelStepSuccess,
-    event_to_json,
 )
-from _ert.forward_model_runner.client import Client
 from ert.config import ForwardModelStep, QueueConfig
 from ert.run_arg import RunArg
 from ert.scheduler import Scheduler, create_driver
@@ -185,15 +183,6 @@ def update_snapshot(self, events: Sequence[Event]) -> EnsembleSnapshot:
 
         return snapshot_mutate_event
 
-    async def send_event(
-        self,
-        event: Event,
-        retries: int = 10,
-    ) -> None:
-        assert self._config is not None
-        async with Client(self._config.get_uri(), token=self._config.token) as client:
-            await client.send(event_to_json(event), retries)
-
     async def evaluate(
         self,
         config: EvaluatorServerConfig,
@@ -209,6 +198,7 @@ async def evaluate(
         the final result of executing all its jobs through a scheduler and driver.
         """
         self._config = config
+        outbound_event_queue = scheduler_queue or asyncio.Queue()
 
         if not self.id_:
             raise ValueError("Ensemble id not set")
@@ -229,7 +219,7 @@ async def evaluate(
                 ee_token=self._config.token,
             )
 
-            await self.send_event(EnsembleStarted(ensemble=self.id_))
+            await outbound_event_queue.put(EnsembleStarted(ensemble=self.id_))
 
             min_required_realizations = (
                 self.min_required_realizations
@@ -243,7 +233,7 @@ async def evaluate(
             )
         except PermissionError as error:
             logger.exception(f"Unexpected exception in ensemble: \n {error!s}")
-            await self.send_event(EnsembleFailed(ensemble=self.id_))
+            await outbound_event_queue.put(EnsembleFailed(ensemble=self.id_))
             return
         except Exception as exc:
             logger.exception(
@@ -253,16 +243,16 @@ async def evaluate(
                     )
                 ),
             )
-            await self.send_event(EnsembleFailed(ensemble=self.id_))
+            await outbound_event_queue.put(EnsembleFailed(ensemble=self.id_))
             return
         except asyncio.CancelledError:
             print("Cancelling evaluator task!")
             return
         # Dispatch final result from evaluator - SUCCEEDED or CANCELLED
         if scheduler_finished_successfully:
-            await self.send_event(EnsembleSucceeded(ensemble=self.id_))
+            await outbound_event_queue.put(EnsembleSucceeded(ensemble=self.id_))
         else:
-            await self.send_event(EnsembleCancelled(ensemble=self.id_))
+            await outbound_event_queue.put(EnsembleCancelled(ensemble=self.id_))
 
     @property
     def cancellable(self) -> bool:

tests/ert/unit_tests/ensemble_evaluator/test_ensemble_legacy.py

@@ -114,14 +114,6 @@ async def test_queue_config_properties_propagated_to_scheduler(
     monkeypatch.setattr(QueueConfig, "max_running", 44)
     ensemble._queue_config.max_submit = 55
 
-    async def mock_send_event_method(*args, **kwargs):
-        return
-
-    monkeypatch.setattr(
-        "ert.ensemble_evaluator._ensemble.LegacyEnsemble.send_event",
-        mock_send_event_method,
-    )
-
     # The function under test:
     await ensemble.evaluate(config=MagicMock())