Create end point for events #9691
Conversation
CodSpeed Performance ReportMerging #9691 will not alter performanceComparing Summary
|
oyvindeide
force-pushed
the
add_websockets_endpoint
branch
4 times, most recently
from
06dfbd1 to
79c008e
Compare
oyvindeide
force-pushed
the
add_websockets_endpoint
branch
2 times, most recently
from
c396805 to
e734a28
Compare
oyvindeide
force-pushed
the
add_websockets_endpoint
branch
from
e734a28 to
c373dd0
Compare
| if message: | ||
| event_dict = json.loads(message) | ||
| if "snapshot" in event_dict: | ||
| event_dict["snapshot"] = EnsembleSnapshot.from_nested_dict( | ||
| event_dict["snapshot"] | ||
| ) | ||
| try: | ||
| event = EventWrapper(event=event_dict).event |
There was a problem hiding this comment.
If you rewrite this:
if message:
event_dict = json.loads(message)
if "snapshot" in event_dict:
event_dict["snapshot"] = EnsembleSnapshot.from_nested_dict(
event_dict["snapshot"]
)
try:
event = EventWrapper(event=event_dict).eventas
if message:
try:
event = status_event_from_json(message)will make it a bit more clear and you no longer need EventWrapper
There was a problem hiding this comment.
I get this then 🤔
pydantic_core._pydantic_core.ValidationError: 1 validation error for tagged-union[ForwardModelStepStart,ForwardModelStepRunning,ForwardModelStepSuccess,ForwardModelStepFailure,ForwardModelStepChecksum,RealizationPending,RealizationRunning,RealizationSuccess,RealizationFailed,RealizationTimeout,RealizationUnknown,RealizationWaiting,RealizationResubmit,EESnapshot,EESnapshotUpdate,EETerminated,EEUserCancel,EEUserDone,EnsembleStarted,EnsembleSucceeded,EnsembleFailed,EnsembleCancelled]
Input tag 'FullSnapshotEvent' found using 'event_type' does not match any of the expected tags: 'forward_model_step.start', 'forward_model_step.running', 'forward_model_step.success', 'forward_model_step.failure', 'forward_model_step.checksum', 'realization.pending', 'realization.running', 'realization.success', 'realization.failure', 'realization.timeout', 'realization.unknown', 'realization.waiting', 'realization.resubmit', 'ee.snapshot', 'ee.snapshot_update', 'ee.terminated', 'ee.user_cancel', 'ee.user_done', 'ensemble.started', 'ensemble.succeeded', 'ensemble.failed', 'ensemble.cancelled' [type=union_tag_invalid, input_value={'iteration_label': 'Runn...e': 'FullSnapshotEvent'}, input_type=dict]
| new_opt_status = _query_server(cert, auth, opt_endpoint) | ||
| if new_opt_status != opt_status: | ||
| opt_status = new_opt_status | ||
| ret = bool(callback({OPT_PROGRESS_ID: opt_status})) | ||
| stop |= ret |
There was a problem hiding this comment.
Do you think we should also add an issue to have a websocket endpoint for the optimization progress updates?
There was a problem hiding this comment.
I think we should just add them to the same queue, and they will automatically be sent through the same end-point. Can create that issue.
| continue | ||
|
|
||
| event = jsonable_encoder(item) | ||
| self.shared_data["events"].append(event) |
There was a problem hiding this comment.
I am a bit confused by the subscriber pattern you are using to notify the websocket clients about the new events is it not possible to just sent the events to the same websocket endpoint once you get them from the status_queue ?
There was a problem hiding this comment.
That would be possible, but the subscribers are in case you connect a second monitor, or in cases where you dont connect a monitor before you start getting events. I have solved that here by subscribers where we know which events the have received, and if you connect at a later time, all the previous events are played back to you.
We could just cache the latest full snapshot and all the changes to it, but there are not that many events, so seemed more future proof to just save all of them and replay. Could start merging partial snapshots in the list, but then keeping track of where the subscribers are is more difficult.
oyvindeide
force-pushed
the
add_websockets_endpoint
branch
3 times, most recently
from
b9d4f73 to
66155f7
Compare
oyvindeide
force-pushed
the
add_websockets_endpoint
branch
from
66155f7 to
6221630
Compare
oyvindeide
force-pushed
the
add_websockets_endpoint
branch
from
6221630 to
c92e178
Compare
| @@ -345,7 +273,4 @@ async def test_status_contains_max_runtime_failure( | |||
|
|
|||
| assert status["status"] == ServerStatus.failed | |||
| print(status["message"]) | |||
| @@ -244,42 +192,22 @@ def mocked_server(url, verify, auth, timeout, proxies): | |||
| # The server should fail and store a user-friendly message. | |||
| assert status["status"] == ServerStatus.failed | |||
| assert OPT_FAILURE_REALIZATIONS in status["message"] | |||
| assert "job1 Failed with: job 1 error 1" in status["message"] | |||
| assert "job1 Failed with: job 1 error 2" in status["message"] | |||
| assert "job2 Failed with: job 2 error 1" in status["message"] | |||
|
|
|||
There was a problem hiding this comment.
Do we not have this information about the failure anymore?
|
Seems like quite a few of the tests are testing less (asserting less) are we sure the changed tests tests everything we want ? |
|
|
||
| if isinstance(item, EndEvent): | ||
| break | ||
| await asyncio.sleep(0.1) |
There was a problem hiding this comment.
what is this sleep needed for?
Issue
Resolves #9001
Approach
Short description of the approach
(Screenshot of new behavior in GUI if applicable)
git rebase -i main --exec 'pytest tests/ert/unit_tests -n logical -m "not integration_test"')When applicable