xcvrd: Add new thread to poll temperatures aggressively

For high power modules it is important to query the temperature often as they can
heat up quickly and the cooling algorithm needs to react quickly to avoid damaging
the transceiver modules.

Introducing a new DomThermalUpdateInfoTask thread which is solely polling the
temperature information out of the xcvrs and publishing these in a new
TRANSCEIVER_DOM_TEMPERATURE table in STATE_DB.

This is mostly additional code being added.
To maximize code reuse, the existing DomUpdateInfoTask has been refactored so that
it now share a base with the newly introduced DomThermalInfoUpdateTask.

Benchmarking shows that polling the temperature out of a xcvrand publishing it inr
the databasse takes around 0.0016 for 800G optics.
Considering a worst case of 0.002 and considering a device with 64 ports, this
amounts to around 0.128s for every loop.
With this loop running every 5s this means this represent a ~2.5% of busyness time.

Author: Staphylo

sonic-xcvrd/tests/test_xcvrd.py

@@ -394,33 +394,38 @@ def test_SfpStateUpdateTask_task_run_with_exception(self):
 
     @patch('xcvrd.xcvrd.SfpStateUpdateTask.is_alive', MagicMock(return_value = False))
     @patch('xcvrd.xcvrd.DomInfoUpdateTask.is_alive', MagicMock(return_value = False))
+    @patch('xcvrd.xcvrd.DomThermalInfoUpdateTask.is_alive', MagicMock(return_value = False))
     @patch('xcvrd.xcvrd.CmisManagerTask.is_alive', MagicMock(return_value = False))
     @patch('xcvrd.xcvrd.SffManagerTask.is_alive', MagicMock(return_value=False))
     @patch('xcvrd.xcvrd.CmisManagerTask.join', MagicMock(side_effect=NotImplementedError))
     @patch('xcvrd.xcvrd.CmisManagerTask.start', MagicMock())
     @patch('xcvrd.xcvrd.SffManagerTask.start', MagicMock())
     @patch('xcvrd.xcvrd.DomInfoUpdateTask.start', MagicMock())
+    @patch('xcvrd.xcvrd.DomThermalInfoUpdateTask.start', MagicMock())
     @patch('xcvrd.xcvrd.SfpStateUpdateTask.start', MagicMock())
     @patch('xcvrd.xcvrd.DaemonXcvrd.deinit', MagicMock())
     @patch('os.kill')
     @patch('xcvrd.xcvrd.DaemonXcvrd.init')
     @patch('xcvrd.xcvrd.DomInfoUpdateTask.join')
+    @patch('xcvrd.xcvrd.DomThermalInfoUpdateTask.join')
     @patch('xcvrd.xcvrd.SfpStateUpdateTask.join')
     @patch('xcvrd.xcvrd.SffManagerTask.join')
     def test_DaemonXcvrd_run_with_exception(self, mock_task_join_sff, mock_task_join_sfp,
-                                            mock_task_join_dom, mock_init, mock_os_kill):
+                                            mock_task_join_dom, mock_task_join_dom_thermal,
+                                            mock_init, mock_os_kill):
         mock_init.return_value = PortMapping()
         xcvrd = DaemonXcvrd(SYSLOG_IDENTIFIER)
         xcvrd.enable_sff_mgr = True
         xcvrd.load_feature_flags = MagicMock()
         xcvrd.stop_event.wait = MagicMock()
         xcvrd.run()
 
-        assert len(xcvrd.threads) == 4
+        assert len(xcvrd.threads) == 5
         assert mock_init.call_count == 1
         assert mock_task_join_sff.call_count == 1
         assert mock_task_join_sfp.call_count == 1
         assert mock_task_join_dom.call_count == 1
+        assert mock_task_join_dom_thermal.call_count == 1
         assert mock_os_kill.call_count == 1
 
 class TestXcvrdScript(object):
@@ -644,6 +649,64 @@ def mock_get_transceiver_dom_sensor_real_value(physical_port):
         dom_db_utils.post_port_dom_sensor_info_to_db(logical_port_name, db_cache=db_cache)
         assert dom_tbl.get_size_for_key(logical_port_name) == 27
 
+    def test_post_port_dom_temperature_info_to_db(self):
+        def mock_get_transceiver_dom_temperature(physical_port):
+            return {
+                'temperature': '68.75',
+            }
+
+        logical_port_name = "Ethernet0"
+        port_mapping = PortMapping()
+        port_mapping.get_logical_to_physical = MagicMock(return_value=[0])
+        xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE)
+        stop_event = threading.Event()
+        mock_sfp_obj_dict = {0 : MagicMock()}
+
+        dom_db_utils = DOMDBUtils(mock_sfp_obj_dict, port_mapping, xcvr_table_helper, stop_event, helper_logger)
+        dom_db_utils.dom_utils = MagicMock()
+        dom_db_utils.xcvrd_utils.get_transceiver_presence = MagicMock(return_value=False)
+        dom_db_utils.xcvrd_utils.is_transceiver_flat_memory = MagicMock(return_value=False)
+        dom_tbl = Table("STATE_DB", TRANSCEIVER_DOM_TEMPERATURE_TABLE)
+        dom_db_utils.xcvr_table_helper.get_dom_temperature_tbl = MagicMock(return_value=dom_tbl)
+        dom_db_utils.dom_utils.get_transceiver_dom_temperature = MagicMock(return_value=None)
+        assert dom_tbl.get_size() == 0
+
+        # Ensure table is empty asic_index is None
+        port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=None)
+        dom_db_utils.post_port_dom_temperature_info_to_db(logical_port_name)
+        assert dom_tbl.get_size() == 0
+
+        # Set asic_index to 0
+        port_mapping.get_asic_id_for_logical_port = MagicMock(return_value=0)
+
+        # Ensure table is empty if stop_event is set
+        stop_event.set()
+        dom_db_utils.post_port_dom_temperature_info_to_db(logical_port_name)
+        assert dom_tbl.get_size() == 0
+        stop_event.clear()
+
+        # Ensure table is empty if transceiver is not present
+        dom_db_utils.post_port_dom_temperature_info_to_db(logical_port_name)
+        assert dom_tbl.get_size() == 0
+        dom_db_utils.return_value = True
+
+        # Ensure table is empty if get_values_func returns None
+        dom_db_utils.xcvrd_utils.get_transceiver_presence = MagicMock(return_value=True)
+        dom_db_utils.post_port_dom_temperature_info_to_db(logical_port_name)
+        assert dom_tbl.get_size() == 0
+
+        # Ensure table is populated if get_values_func returns valid values
+        db_cache = {}
+        dom_db_utils.dom_utils.get_transceiver_dom_temperature = MagicMock(side_effect=mock_get_transceiver_dom_temperature)
+        dom_db_utils.post_port_dom_temperature_info_to_db(logical_port_name, db_cache=db_cache)
+        assert dom_tbl.get_size_for_key(logical_port_name) == 2
+
+        # Ensure db_cache is populated correctly
+        assert db_cache.get(0) is not None
+        dom_db_utils.dom_utils.get_transceiver_dom_temperature = MagicMock(return_value=None)
+        dom_db_utils.post_port_dom_temperature_info_to_db(logical_port_name, db_cache=db_cache)
+        assert dom_tbl.get_size_for_key(logical_port_name) == 2
+
     def test_post_port_dom_flags_to_db(self):
         def mock_get_transceiver_dom_flags(physical_port):
             return {
@@ -4582,6 +4645,7 @@ def test_DaemonXcvrd_init_deinit_fastboot_enabled(self, mock_del_port_sfp_dom_in
             status_sw_tbl = MagicMock()
             xcvrd.xcvr_table_helper.get_status_sw_tbl = MagicMock(return_value=status_sw_tbl)
             xcvrd.xcvr_table_helper.get_dom_tbl = MagicMock(return_value=MagicMock)
+            xcvrd.xcvr_table_helper.get_dom_temperature_tbl = MagicMock(return_value=MagicMock)
             xcvrd.xcvr_table_helper.get_dom_flag_tbl = MagicMock()
             xcvrd.xcvr_table_helper.get_dom_flag_change_count_tbl = MagicMock()
             xcvrd.xcvr_table_helper.get_dom_flag_set_time_tbl = MagicMock()
@@ -4629,6 +4693,7 @@ def test_DaemonXcvrd_init_deinit_cold(self, mock_del_port_sfp_dom_info_from_db):
             status_sw_tbl = MagicMock()
             xcvrdaemon.xcvr_table_helper.get_status_sw_tbl = MagicMock(return_value=status_sw_tbl)
             xcvrdaemon.xcvr_table_helper.get_dom_tbl = MagicMock(return_value=MagicMock)
+            xcvrdaemon.xcvr_table_helper.get_dom_temperature_tbl = MagicMock(return_value=MagicMock)
             xcvrdaemon.xcvr_table_helper.get_dom_threshold_tbl = MagicMock(return_value=MagicMock)
             xcvrdaemon.xcvr_table_helper.get_dom_flag_tbl = MagicMock()
             xcvrdaemon.xcvr_table_helper.get_dom_flag_change_count_tbl = MagicMock()

sonic-xcvrd/xcvrd/dom/dom_mgr.py

@@ -13,6 +13,7 @@
     import copy
     import sys
     import re
+    import time
 
     from natsort import natsorted
     from sonic_py_common import syslogger
@@ -33,32 +34,19 @@
 
 SYSLOG_IDENTIFIER_DOMINFOUPDATETASK = "DomInfoUpdateTask"
 
-class DomInfoUpdateTask(threading.Thread):
-    DOM_INFO_UPDATE_PERIOD_SECS = 60
-    DIAG_DB_UPDATE_TIME_AFTER_LINK_CHANGE = 1
-    DOM_PORT_CHG_OBSERVER_TBL_MAP = [
-        {'APPL_DB': 'PORT_TABLE', 'FILTER': ['flap_count']},
-    ]
+class DomInfoUpdateBase(threading.Thread):
 
-    def __init__(self, namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event, skip_cmis_mgr):
+    name = ''
+
+    def __init__(self, namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event):
         threading.Thread.__init__(self)
-        self.name = "DomInfoUpdateTask"
         self.exc = None
         self.task_stopping_event = threading.Event()
         self.main_thread_stop_event = main_thread_stop_event
         self.helper_logger = syslogger.SysLogger(SYSLOG_IDENTIFIER_DOMINFOUPDATETASK, enable_runtime_config=True)
         self.port_mapping = copy.deepcopy(port_mapping)
         self.namespaces = namespaces
-        self.skip_cmis_mgr = skip_cmis_mgr
         self.sfp_obj_dict = sfp_obj_dict
-        self.link_change_affected_ports = {}
-        self.xcvr_table_helper = XcvrTableHelper(self.namespaces)
-        self.xcvrd_utils = XCVRDUtils(self.sfp_obj_dict, helper_logger)
-        self.dom_db_utils = DOMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
-        self.db_utils = self.dom_db_utils
-        self.vdm_utils = VDMUtils(self.sfp_obj_dict, self.helper_logger)
-        self.vdm_db_utils = VDMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
-        self.status_db_utils = StatusDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
 
     def log_debug(self, message):
         self.helper_logger.log_debug("{}".format(message))
@@ -75,6 +63,14 @@ def log_warning(self, message):
     def log_error(self, message):
         self.helper_logger.log_error("{}".format(message))
 
+    def on_port_config_change(self, port_change_event):
+        if port_change_event.event_type == port_event_helper.PortChangeEvent.PORT_REMOVE:
+            self.on_remove_logical_port(port_change_event)
+        self.port_mapping.handle_port_change_event(port_change_event)
+
+    def on_remove_logical_port(self, port_change_event):
+        pass
+
     def get_dom_polling_from_config_db(self, lport):
         """
             Returns the value of dom_polling field from PORT table in CONFIG_DB
@@ -110,6 +106,57 @@ def get_dom_polling_from_config_db(self, lport):
 
         return dom_polling
 
+
+    def is_port_dom_monitoring_disabled(self, logical_port_name):
+        return self.get_dom_polling_from_config_db(logical_port_name) == 'disabled'
+
+    def task_worker(self):
+        pass
+
+    def run(self):
+        if self.task_stopping_event.is_set():
+            return
+        try:
+            self.task_worker()
+        except Exception as e:
+            self.log_error("Exception occured at {} thread due to {}".format(threading.current_thread().getName(), repr(e)))
+            xcvrd.log_exception_traceback()
+            self.exc = e
+            self.main_thread_stop_event.set()
+
+    def join(self):
+        self.task_stopping_event.set()
+        threading.Thread.join(self)
+        if self.exc:
+            raise self.exc
+
+    def update_log_level(self):
+        """Call the logger's update log level method.
+        """
+        return self.helper_logger.update_log_level()
+
+
+class DomInfoUpdateTask(DomInfoUpdateBase):
+    name = "DomInfoUpdateTask"
+
+    DOM_INFO_UPDATE_PERIOD_SECS = 60
+    DIAG_DB_UPDATE_TIME_AFTER_LINK_CHANGE = 1
+    DOM_PORT_CHG_OBSERVER_TBL_MAP = [
+        {'APPL_DB': 'PORT_TABLE', 'FILTER': ['flap_count']},
+    ]
+
+    def __init__(self, namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event, skip_cmis_mgr):
+        super().__init__(namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event)
+        self.skip_cmis_mgr = skip_cmis_mgr
+        self.link_change_affected_ports = {}
+        self.xcvr_table_helper = XcvrTableHelper(self.namespaces)
+        self.xcvrd_utils = XCVRDUtils(self.sfp_obj_dict, helper_logger)
+        self.dom_db_utils = DOMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+        self.db_utils = self.dom_db_utils
+        self.vdm_utils = VDMUtils(self.sfp_obj_dict, self.helper_logger)
+        self.vdm_db_utils = VDMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+        self.status_db_utils = StatusDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+
     """
     Checks if the port is going through CMIS initialization process
     This API assumes CMIS_STATE_UNKNOWN as a transitional state since it is the
@@ -339,23 +386,6 @@ def task_worker(self):
 
         self.log_notice("Stop DOM monitoring loop")
 
-    def run(self):
-        if self.task_stopping_event.is_set():
-            return
-        try:
-            self.task_worker()
-        except Exception as e:
-            self.log_error("Exception occured at {} thread due to {}".format(threading.current_thread().getName(), repr(e)))
-            common.log_exception_traceback()
-            self.exc = e
-            self.main_thread_stop_event.set()
-
-    def join(self):
-        self.task_stopping_event.set()
-        threading.Thread.join(self)
-        if self.exc:
-            raise self.exc
-
     def on_port_update_event(self, port_change_event):
         """Called when a port change event is received
 
@@ -427,11 +457,6 @@ def update_port_db_diagnostics_on_link_change(self, physical_port):
                 self.log_warning(f"Update DB diagnostics during link change: Got exception {repr(e)} while processing vdm flags for port {first_logical_port}, ignored")
                 return
 
-    def on_port_config_change(self, port_change_event):
-        if port_change_event.event_type == port_event_helper.PortChangeEvent.PORT_REMOVE:
-            self.on_remove_logical_port(port_change_event)
-        self.port_mapping.handle_port_change_event(port_change_event)
-
     def on_remove_logical_port(self, port_change_event):
         """Called when a logical port is removed from CONFIG_DB
 
@@ -444,6 +469,7 @@ def on_remove_logical_port(self, port_change_event):
         common.del_port_sfp_dom_info_from_db(port_change_event.port_name,
                                       self.port_mapping,
                                       [self.xcvr_table_helper.get_dom_tbl(port_change_event.asic_id),
+                                      self.xcvr_table_helper.get_dom_temperature_tbl(port_change_event.asic_id),
                                       self.xcvr_table_helper.get_dom_flag_tbl(port_change_event.asic_id),
                                       self.xcvr_table_helper.get_dom_flag_change_count_tbl(port_change_event.asic_id),
                                       self.xcvr_table_helper.get_dom_flag_set_time_tbl(port_change_event.asic_id),
@@ -461,7 +487,61 @@ def on_remove_logical_port(self, port_change_event):
                                       self.xcvr_table_helper.get_pm_tbl(port_change_event.asic_id),
                                       self.xcvr_table_helper.get_firmware_info_tbl(port_change_event.asic_id)
                                       ])
-    def update_log_level(self):
-        """Call the logger's update log level method.
-        """
-        return self.helper_logger.update_log_level()
+
+
+class DomThermalInfoUpdateTask(DomInfoUpdateBase):
+    name = 'DomThermalInfoUpdateTask'
+
+    DOM_INFO_UPDATE_PERIOD_SECS = 5
+
+    def __init__(self, namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event):
+        super().__init__(namespaces, port_mapping, sfp_obj_dict, main_thread_stop_event)
+        self.xcvr_table_helper = XcvrTableHelper(self.namespaces)
+        self.dom_db_utils = DOMDBUtils(self.sfp_obj_dict, self.port_mapping, self.xcvr_table_helper, self.task_stopping_event, self.helper_logger)
+
+    def task_worker(self):
+        self.log_notice("Start DOM thermal monitoring loop")
+
+        # Set the periodic db update time
+        dom_info_update_periodic_secs = self.DOM_INFO_UPDATE_PERIOD_SECS
+
+        # Poll transceiver temperature as soon as possible
+        next_periodic_db_update_time = datetime.datetime.now()
+
+        # Start loop to update dom info in DB periodically and handle port change events
+        while not self.task_stopping_event.is_set():
+            # Check if periodic db update is needed
+            now = datetime.datetime.now()
+            if next_periodic_db_update_time > now:
+               # Sleep for 1 second or less depending on the remaining time
+               time.sleep(min(1, (next_periodic_db_update_time - now).total_seconds()))
+               continue
+
+            for physical_port, logical_ports in self.port_mapping.physical_to_logical.items():
+                # Get the first logical port name since it corresponds to the first subport
+                # of the breakout group
+                logical_port_name = logical_ports[0]
+
+                if self.is_port_dom_monitoring_disabled(logical_port_name):
+                    continue
+
+                # Get the asic to which this port belongs
+                asic_index = self.port_mapping.get_asic_id_for_logical_port(logical_port_name)
+                if asic_index is None:
+                    self.log_warning("Got invalid asic index for {}, ignored".format(logical_port_name))
+                    continue
+
+                if not sfp_status_helper.detect_port_in_error_status(logical_port_name, self.xcvr_table_helper.get_status_sw_tbl(asic_index)):
+                    if not xcvrd._wrapper_get_presence(physical_port):
+                        continue
+
+                try:
+                    self.dom_db_utils.post_port_dom_temperature_info_to_db(logical_port_name)
+                except (KeyError, TypeError) as e:
+                    #continue to process next port since exception could be raised due to port reset, transceiver removal
+                    self.log_warning("Got exception {} while processing dom info for port {}, ignored".format(repr(e), logical_port_name))
+
+            # Set the periodic db update time after all the ports are processed
+            next_periodic_db_update_time = now + datetime.timedelta(seconds=dom_info_update_periodic_secs)
+
+        self.log_notice("Stop DOM thermal monitoring loop")

sonic-xcvrd/xcvrd/dom/utilities/dom_sensor/db_utils.py

@@ -24,6 +24,19 @@ def __init__(self, sfp_obj_dict, port_mapping, xcvr_table_helper, task_stopping_
         self.dom_utils = DOMUtils(self.sfp_obj_dict, logger)
         self.logger = logger
 
+    def post_port_dom_temperature_info_to_db(self, logical_port_name, db_cache=None):
+        asic_index = self.port_mapping.get_asic_id_for_logical_port(logical_port_name)
+        if asic_index is None:
+            self.logger.log_error(f"Post port dom sensor info to db failed for {logical_port_name} "
+                                  "as no asic index found")
+            return
+
+        return self.post_diagnostic_values_to_db(logical_port_name,
+                                                 self.xcvr_table_helper.get_dom_temperature_tbl(asic_index),
+                                                 self.dom_utils.get_transceiver_dom_temperature,
+                                                 db_cache=db_cache,
+                                                 beautify_func=self._beautify_dom_info_dict)
+
     def post_port_dom_sensor_info_to_db(self, logical_port_name, db_cache=None):
         asic_index = self.port_mapping.get_asic_id_for_logical_port(logical_port_name)
         if asic_index is None:

sonic-xcvrd/xcvrd/dom/utilities/dom_sensor/utils.py

@@ -7,6 +7,14 @@ def __init__(self, sfp_obj_dict, logger):
         self.sfp_obj_dict = sfp_obj_dict
         self.logger = logger
 
+    def get_transceiver_dom_temperature(self, physical_port):
+        try:
+            return {
+                'temperature': self.sfp_obj_dict[physical_port].get_temperature(),
+            }
+        except (NotImplementedError):
+            return {}
+
     def get_transceiver_dom_sensor_real_value(self, physical_port):
         try:
             return self.sfp_obj_dict[physical_port].get_transceiver_dom_real_value()