Experiment server

ert_gui/simulation/run_dialog.py

@@ -17,6 +17,8 @@
 )
 from ert_shared.status.tracker.factory import create_tracker
 from ert_shared.status.utils import format_running_time
+from ert_shared.ensemble_evaluator.evaluator import EnsembleEvaluatorService
+from ert_shared.feature_toggling import FeatureToggling
 from qtpy.QtCore import QModelIndex, QSize, Qt, QThread, QTimer, Signal, Slot
 from qtpy.QtWidgets import (
     QDialog,
@@ -270,10 +272,16 @@ def run():
 
         self._ticker.start(1000)
 
+        ee_config = (
+            EnsembleEvaluatorService.get_config()
+            if FeatureToggling.is_enabled("ensemble-evaluator")
+            else None
+        )
+
         tracker = create_tracker(
             self._run_model,
             num_realizations=self._simulations_argments["active_realizations"].count(),
-            ee_config=self._simulations_argments.get("ee_config", None),
+            ee_config=ee_config,
         )
 
         worker = TrackerWorker(tracker)

ert_gui/simulation/simulation_panel.py

@@ -23,18 +23,13 @@
     SimulationConfigPanel,
 )
 from ert_gui.simulation import RunDialog
-from ert_shared.feature_toggling import FeatureToggling
 from collections import OrderedDict
-from ert_shared.ensemble_evaluator.config import EvaluatorServerConfig
 
 
 class SimulationPanel(QWidget):
     def __init__(self, config_file):
         QWidget.__init__(self)
         self._config_file = config_file
-        self._ee_config = None
-        if FeatureToggling.is_enabled("ensemble-evaluator"):
-            self._ee_config = EvaluatorServerConfig()
 
         self.setObjectName("Simulation_panel")
         layout = QVBoxLayout()
@@ -110,8 +105,6 @@ def getSimulationArguments(self):
         """@rtype: dict[str,object]"""
         simulation_widget = self._simulation_widgets[self.getCurrentSimulationModel()]
         args = simulation_widget.getSimulationArguments()
-        if self._ee_config is not None:
-            args.update({"ee_config": self._ee_config})
         return args
 
     def runSimulation(self):

ert_shared/cli/main.py

@@ -15,7 +15,7 @@
 from ert_shared.cli.monitor import Monitor
 from ert_shared.cli.notifier import ErtCliNotifier
 from ert_shared.cli.workflow import execute_workflow
-from ert_shared.ensemble_evaluator.config import EvaluatorServerConfig
+from ert_shared.ensemble_evaluator.evaluator import EnsembleEvaluatorService
 from ert_shared.feature_toggling import FeatureToggling
 from ert_shared.status.tracker.factory import create_tracker
 from res.enkf import EnKFMain, ResConfig
@@ -63,10 +63,11 @@ def run_cli(args):
         clear_global_state()
         raise ErtCliError(msg)
 
-    ee_config = None
-    if FeatureToggling.is_enabled("ensemble-evaluator"):
-        ee_config = EvaluatorServerConfig(custom_port_range=args.port_range)
-        argument.update({"ee_config": ee_config})
+    ee_config = (
+        EnsembleEvaluatorService.get_config()
+        if FeatureToggling.is_enabled("ensemble-evaluator")
+        else None
+    )
 
     thread = threading.Thread(
         name="ert_cli_simulation_thread",

ert_shared/ensemble_evaluator/ensemble/legacy.py

@@ -51,7 +51,7 @@ def on_timeout(callback_args):
             )
             timeout_queue.put_nowait(timeout_cloudevent)
 
-        dispatch_url = self._config.dispatch_uri
+        dispatch_url = f"{self._config.dispatch_uri}/{self._ee_id}"
         cert = self._config.cert
         token = self._config.token
 
@@ -86,7 +86,7 @@ def evaluate(self, config, ee_id):
     def _evaluate(self):
         asyncio.set_event_loop(asyncio.new_event_loop())
 
-        dispatch_url = self._config.dispatch_uri
+        dispatch_url = f"{self._config.dispatch_uri}/{self._ee_id}"
         cert = self._config.cert
         token = self._config.token
         try:
@@ -216,7 +216,7 @@ def _cancel(self):
         )
         asyncio.get_event_loop().run_until_complete(
             self.send_cloudevent(
-                self._config.dispatch_uri,
+                f"{self._config.dispatch_uri}/{self._ee_id}",
                 out_cloudevent,
                 token=self._config.token,
                 cert=self._config.cert,

ert_shared/ensemble_evaluator/ensemble/prefect.py

@@ -210,7 +210,11 @@ def evaluate(self, config: EvaluatorServerConfig, ee_id: str):
     def _evaluate(self, ee_config: EvaluatorServerConfig, ee_id):
         asyncio.set_event_loop(asyncio.get_event_loop())
         try:
-            with Client(ee_config.dispatch_uri, ee_config.token, ee_config.cert) as c:
+            with Client(
+                f"{ee_config.dispatch_uri}/{self._ee_id}",
+                ee_config.token,
+                ee_config.cert,
+            ) as c:
                 event = CloudEvent(
                     {
                         "type": ids.EVTYPE_ENSEMBLE_STARTED,
@@ -219,11 +223,17 @@ def _evaluate(self, ee_config: EvaluatorServerConfig, ee_id):
                 )
                 c.send(to_json(event).decode())
             with prefect.context(
-                url=ee_config.dispatch_uri, token=ee_config.token, cert=ee_config.cert
+                url=f"{ee_config.dispatch_uri}/{self._ee_id}",
+                token=ee_config.token,
+                cert=ee_config.cert,
             ):
                 self.run_flow(ee_id)
 
-            with Client(ee_config.dispatch_uri, ee_config.token, ee_config.cert) as c:
+            with Client(
+                f"{ee_config.dispatch_uri}/{self._ee_id}",
+                ee_config.token,
+                ee_config.cert,
+            ) as c:
                 event = CloudEvent(
                     {
                         "type": ids.EVTYPE_ENSEMBLE_STOPPED,
@@ -238,7 +248,11 @@ def _evaluate(self, ee_config: EvaluatorServerConfig, ee_id):
                 "An exception occurred while starting the ensemble evaluation",
                 exc_info=True,
             )
-            with Client(ee_config.dispatch_uri, ee_config.token, ee_config.cert) as c:
+            with Client(
+                f"{ee_config.dispatch_uri}/{self._ee_id}",
+                ee_config.token,
+                ee_config.cert,
+            ) as c:
                 event = CloudEvent(
                     {
                         "type": ids.EVTYPE_ENSEMBLE_FAILED,
@@ -303,7 +317,7 @@ def _cancel(self):
         loop = asyncio.new_event_loop()
         loop.run_until_complete(
             self.send_cloudevent(
-                self._ee_config.dispatch_uri,
+                f"{ee_config.dispatch_uri}/{self._ee_id}",
                 event,
                 token=self._ee_config.token,
                 cert=self._ee_config.cert,

ert_shared/ensemble_evaluator/entity/identifiers.py

@@ -101,6 +101,15 @@
 EVTYPE_EE_USER_CANCEL = "com.equinor.ert.ee.user_cancel"
 EVTYPE_EE_USER_DONE = "com.equinor.ert.ee.user_done"
 
+EVTYPE_EE_EXPERIMENT_UPDATE = "com.equinor.ert.ee.experiment.update"
+EVTYPE_EE_EXPERIMENT_TERMINATED = "com.equinor.ert.ee.experiment.terminated"
+
+EVTYPE_EXPERIMENT_SUBMITTED = "com.equinor.ert.experiment.submitted"
+EVTYPE_EXPERIMENT_RUNNING = "com.equinor.ert.experiment.running"
+EVTYPE_EXPERIMENT_SUCCESS = "com.equinor.ert.experiment.success"
+EVTYPE_EXPERIMENT_FAILURE = "com.equinor.ert.experiment.failure"
+
+
 EVTYPE_ENSEMBLE_STARTED = "com.equinor.ert.ensemble.started"
 EVTYPE_ENSEMBLE_STOPPED = "com.equinor.ert.ensemble.stopped"
 EVTYPE_ENSEMBLE_CANCELLED = "com.equinor.ert.ensemble.cancelled"

ert_shared/ensemble_evaluator/entity/tool.py

@@ -37,3 +37,7 @@ def get_step_id(source):
 
 def get_job_id(source):
     return _match_token("job", source)
+
+
+def get_evaluation_id(source):
+    return _match_token("ee", source)