From 6bb3674406f3be817026949271acabbd7575dadf Mon Sep 17 00:00:00 2001
From: vzhestkov <vzhestkov@suse.com>
Date: Fri, 9 Feb 2024 09:44:46 +0100
Subject: [PATCH 1/3] Improve broken events catching and reporting

---
 salt/utils/event.py | 23 +++++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/salt/utils/event.py b/salt/utils/event.py
index e6d7b00520..ef048335ae 100644
--- a/salt/utils/event.py
+++ b/salt/utils/event.py
@@ -75,6 +75,7 @@
 import salt.utils.process
 import salt.utils.stringutils
 import salt.utils.zeromq
+from salt.exceptions import SaltDeserializationError
 
 log = logging.getLogger(__name__)
 
@@ -461,7 +462,13 @@ def unpack(cls, raw):
             salt.utils.stringutils.to_bytes(TAGEND)
         )  # split tag from data
         mtag = salt.utils.stringutils.to_str(mtag)
-        data = salt.payload.loads(mdata, encoding="utf-8")
+        try:
+            data = salt.payload.loads(mdata, encoding="utf-8")
+        except SaltDeserializationError:
+            log.warning(
+                "SaltDeserializationError on unpacking data, the payload could be incomplete"
+            )
+            raise
         return mtag, data
 
     def _get_match_func(self, match_type=None):
@@ -583,6 +590,9 @@ def _get_event(self, wait, tag, match_func=None, no_block=False):
                     raise
                 else:
                     return None
+            except SaltDeserializationError:
+                log.error("Unable to deserialize received event")
+                return None
             except RuntimeError:
                 return None
 
@@ -889,6 +899,14 @@ def _fire_ret_load_specific_fun(self, load, fun_index=0):
             ret = load.get("return", {})
             retcode = load["retcode"]
 
+        if not isinstance(ret, dict):
+            log.error(
+                "Event with bad payload received from '%s': %s",
+                load.get("id", "UNKNOWN"),
+                "".join(ret) if isinstance(ret, list) else ret,
+            )
+            return
+
         try:
             for tag, data in ret.items():
                 data["retcode"] = retcode
@@ -910,7 +928,8 @@ def _fire_ret_load_specific_fun(self, load, fun_index=0):
                     )
         except Exception as exc:  # pylint: disable=broad-except
             log.error(
-                "Event iteration failed with exception: %s",
+                "Event from '%s' iteration failed with exception: %s",
+                load.get("id", "UNKNOWN"),
                 exc,
                 exc_info_on_loglevel=logging.DEBUG,
             )

From 0ec60c8fd8333e60cb2c4688221ba98d39813bb6 Mon Sep 17 00:00:00 2001
From: vzhestkov <vzhestkov@suse.com>
Date: Fri, 9 Feb 2024 14:55:00 +0100
Subject: [PATCH 2/3] Add test of catching SaltDeserializationError on reading
 event

---
 tests/pytests/unit/utils/event/test_event.py | 26 ++++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/tests/pytests/unit/utils/event/test_event.py b/tests/pytests/unit/utils/event/test_event.py
index e289e72dad..d4d70c9d5e 100644
--- a/tests/pytests/unit/utils/event/test_event.py
+++ b/tests/pytests/unit/utils/event/test_event.py
@@ -12,6 +12,7 @@
 import salt.ext.tornado.iostream
 import salt.utils.event
 import salt.utils.stringutils
+from salt.exceptions import SaltDeserializationError
 from salt.utils.event import SaltEvent
 from tests.support.events import eventpublisher_process, eventsender_process
 from tests.support.mock import patch
@@ -340,3 +341,28 @@ def test_master_pub_permissions(sock_dir):
         assert bool(os.lstat(p).st_mode & stat.S_IRUSR)
         assert not bool(os.lstat(p).st_mode & stat.S_IRGRP)
         assert not bool(os.lstat(p).st_mode & stat.S_IROTH)
+
+
+def test_event_unpack_with_SaltDeserializationError(sock_dir):
+    with eventpublisher_process(str(sock_dir)), salt.utils.event.MasterEvent(
+        str(sock_dir), listen=True
+    ) as me, patch.object(
+        salt.utils.event.log, "warning", autospec=True
+    ) as mock_log_warning, patch.object(
+        salt.utils.event.log, "error", autospec=True
+    ) as mock_log_error:
+        me.fire_event({"data": "foo1"}, "evt1")
+        me.fire_event({"data": "foo2"}, "evt2")
+        evt2 = me.get_event(tag="")
+        with patch("salt.payload.loads", side_effect=SaltDeserializationError):
+            evt1 = me.get_event(tag="")
+        _assert_got_event(evt2, {"data": "foo2"}, expected_failure=True)
+        assert evt1 is None
+        assert (
+            mock_log_warning.mock_calls[0].args[0]
+            == "SaltDeserializationError on unpacking data, the payload could be incomplete"
+        )
+        assert (
+            mock_log_error.mock_calls[0].args[0]
+            == "Unable to deserialize received event"
+        )

From e7a8a76e98b02029d40c52babfe2c678b4aeb633 Mon Sep 17 00:00:00 2001
From: vzhestkov <vzhestkov@suse.com>
Date: Tue, 13 Feb 2024 14:56:17 +0100
Subject: [PATCH 3/3] Add test for fire_ret_load

---
 tests/pytests/unit/utils/event/test_event.py | 81 ++++++++++++++++++++
 1 file changed, 81 insertions(+)

diff --git a/tests/pytests/unit/utils/event/test_event.py b/tests/pytests/unit/utils/event/test_event.py
index d4d70c9d5e..793157c3e8 100644
--- a/tests/pytests/unit/utils/event/test_event.py
+++ b/tests/pytests/unit/utils/event/test_event.py
@@ -366,3 +366,84 @@ def test_event_unpack_with_SaltDeserializationError(sock_dir):
             mock_log_error.mock_calls[0].args[0]
             == "Unable to deserialize received event"
         )
+
+
+def test_event_fire_ret_load():
+    event = SaltEvent(node=None)
+    test_load = {
+        "id": "minion_id.example.org",
+        "jid": "20240212095247760376",
+        "fun": "state.highstate",
+        "retcode": 254,
+        "return": {
+            "saltutil_|-sync_states_|-sync_states_|-sync_states": {
+                "result": True,
+            },
+            "saltutil_|-sync_modules_|-sync_modules_|-sync_modules": {
+                "result": False,
+            },
+        },
+    }
+    test_fire_event_data = {
+        "result": False,
+        "retcode": 254,
+        "jid": "20240212095247760376",
+        "id": "minion_id.example.org",
+        "success": False,
+        "return": "Error: saltutil.sync_modules",
+        "fun": "state.highstate",
+    }
+    test_unhandled_exc = "Unhandled exception running state.highstate"
+    test_traceback = [
+        "Traceback (most recent call last):\n",
+        "    Just an example of possible return as a list\n",
+    ]
+    with patch.object(
+        event, "fire_event", side_effect=[None, None, Exception]
+    ) as mock_fire_event, patch.object(
+        salt.utils.event.log, "error", autospec=True
+    ) as mock_log_error:
+        event.fire_ret_load(test_load)
+        assert len(mock_fire_event.mock_calls) == 2
+        assert mock_fire_event.mock_calls[0].args[0] == test_fire_event_data
+        assert mock_fire_event.mock_calls[0].args[1] == "saltutil.sync_modules"
+        assert mock_fire_event.mock_calls[1].args[0] == test_fire_event_data
+        assert (
+            mock_fire_event.mock_calls[1].args[1]
+            == "salt/job/20240212095247760376/sub/minion_id.example.org/error/state.highstate"
+        )
+        assert not mock_log_error.mock_calls
+
+        mock_log_error.reset_mock()
+
+        event.fire_ret_load(test_load)
+        assert (
+            mock_log_error.mock_calls[0].args[0]
+            == "Event from '%s' iteration failed with exception: %s"
+        )
+        assert mock_log_error.mock_calls[0].args[1] == "minion_id.example.org"
+
+        mock_log_error.reset_mock()
+        test_load["return"] = test_unhandled_exc
+
+        event.fire_ret_load(test_load)
+        assert (
+            mock_log_error.mock_calls[0].args[0]
+            == "Event with bad payload received from '%s': %s"
+        )
+        assert mock_log_error.mock_calls[0].args[1] == "minion_id.example.org"
+        assert (
+            mock_log_error.mock_calls[0].args[2]
+            == "Unhandled exception running state.highstate"
+        )
+
+        mock_log_error.reset_mock()
+        test_load["return"] = test_traceback
+
+        event.fire_ret_load(test_load)
+        assert (
+            mock_log_error.mock_calls[0].args[0]
+            == "Event with bad payload received from '%s': %s"
+        )
+        assert mock_log_error.mock_calls[0].args[1] == "minion_id.example.org"
+        assert mock_log_error.mock_calls[0].args[2] == "".join(test_traceback)
