fix(backtrace_decoding): make sure event are published

Events from the thread processing the queue weren't published
and backtraces weren't reported as events.

* `task_done()` isn't on multiprocessing.Queue, but on `queues.Quere`
  we move to use multiprocessing since the thread reading the log
  moved into it's own process
* events clone via multiprocessing lost thier `_ready_to_publish`
  flag set by `add_info` on a differnt process, so we need to
  out it back, so publish can work.
* unittest wasn't changed and was still using `queues.Quere`,
  hence not catching this.

sdcm/cluster.py

@@ -1476,20 +1476,19 @@ def decode_backtrace(self):
             try:
                 obj = self.test_config.DECODING_QUEUE.get(timeout=5)
                 if obj is None:
-                    self.test_config.DECODING_QUEUE.task_done()
                     break
                 event = obj["event"]
                 if not scylla_debug_file:
                     scylla_debug_file = self.copy_scylla_debug_info(obj["node"], obj["debug_file"])
                 output = self.decode_raw_backtrace(scylla_debug_file, " ".join(event.raw_backtrace.split('\n')))
                 event.backtrace = output.stdout
-                self.test_config.DECODING_QUEUE.task_done()
             except queue.Empty:
                 pass
             except Exception as details:  # pylint: disable=broad-except
                 self.log.error("failed to decode backtrace %s", details)
             finally:
                 if event:
+                    event.ready_to_publish()
                     event.publish()
 
             if self.termination_event.is_set() and self.test_config.DECODING_QUEUE.empty():

sdcm/sct_events/base.py

@@ -228,6 +228,9 @@ def dont_publish(self):
         self._ready_to_publish = False
         LOGGER.debug("%s marked to not publish", self)
 
+    def ready_to_publish(self):
+        self._ready_to_publish = True
+
     def to_json(self, encoder: Type[JSONEncoder] = JSONEncoder) -> str:
         return json.dumps({
             "base": self.base,

unit_tests/test_decode_backtrace.py

@@ -13,7 +13,7 @@
 
 import os
 import json
-import queue
+from multiprocessing import Queue
 import unittest
 from functools import cached_property
 
@@ -125,7 +125,7 @@ def test_01_reactor_stall_is_not_decoded_if_disabled(self):
     def test_02_reactor_stalls_is_decoded_if_enabled(self):
         self.test_config.BACKTRACE_DECODING = True
 
-        self.test_config.DECODING_QUEUE = queue.Queue()
+        self.test_config.DECODING_QUEUE = Queue()
 
         self.monitor_node.start_decode_on_monitor_node_thread()
         self._read_and_publish_events()
@@ -146,7 +146,7 @@ def test_02_reactor_stalls_is_decoded_if_enabled(self):
 
     def test_03_decode_interlace_reactor_stall(self):  # pylint: disable=invalid-name
 
-        self.test_config.DECODING_QUEUE = queue.Queue()
+        self.test_config.DECODING_QUEUE = Queue()
         self.test_config.BACKTRACE_DECODING = True
 
         self.monitor_node.start_decode_on_monitor_node_thread()
@@ -171,7 +171,7 @@ def test_03_decode_interlace_reactor_stall(self):  # pylint: disable=invalid-nam
 
     def test_04_decode_backtraces_core(self):
 
-        self.test_config.DECODING_QUEUE = queue.Queue()
+        self.test_config.DECODING_QUEUE = Queue()
         self.test_config.BACKTRACE_DECODING = True
 
         self.monitor_node.start_decode_on_monitor_node_thread()