Enable runtime config log level

src/sonic-py-common/sonic_py_common/daemon_base.py

@@ -35,10 +35,10 @@ def db_connect(db_name, namespace=EMPTY_NAMESPACE):
 
 
 class DaemonBase(Logger):
-    def __init__(self, log_identifier, use_syslogger=True):
+    def __init__(self, log_identifier, use_syslogger=True, enable_runtime_log_config=False):
         super().__init__()
         if use_syslogger:
-            self.logger_instance = SysLogger(log_identifier)
+            self.logger_instance = SysLogger(log_identifier, enable_runtime_config=enable_runtime_log_config)
         else:
             self.logger_instance = Logger(
                 log_identifier=log_identifier,

src/sonic-py-common/sonic_py_common/syslogger.py

@@ -4,6 +4,11 @@
 import socket
 import sys
 
+CONFIG_DB = 'CONFIG_DB'
+FIELD_LOG_LEVEL = 'LOGLEVEL'
+FIELD_REQUIRE_REFRESH = 'require_manual_refresh'
+
+
 # customize python logging to support notice logger
 logging.NOTICE = logging.INFO + 1
 logging.addLevelName(logging.NOTICE, "NOTICE")
@@ -18,12 +23,11 @@ class SysLogger:
     DEFAULT_LOG_FACILITY = SysLogHandler.LOG_USER
     DEFAULT_LOG_LEVEL = logging.NOTICE
 
-    def __init__(self, log_identifier=None, log_facility=DEFAULT_LOG_FACILITY, log_level=DEFAULT_LOG_LEVEL):
-        if log_identifier is None:
-            log_identifier = os.path.basename(sys.argv[0])
+    def __init__(self, log_identifier=None, log_facility=DEFAULT_LOG_FACILITY, log_level=DEFAULT_LOG_LEVEL, enable_runtime_config=False):
+        self.log_identifier = log_identifier if log_identifier else os.path.basename(sys.argv[0]) 
 
         # Initialize SysLogger
-        self.logger = logging.getLogger(log_identifier)
+        self.logger = logging.getLogger(self.log_identifier)
 
         # Reset all existing handlers
         for handler in self.logger.handlers[:]:
@@ -35,6 +39,74 @@ def __init__(self, log_identifier=None, log_facility=DEFAULT_LOG_FACILITY, log_l
         self.logger.addHandler(handler)
 
         self.set_min_log_priority(log_level)
+
+        if enable_runtime_config:
+            self.update_log_level()
+
+    def update_log_level(self):
+        """Refresh log level.
+
+        Returns:
+            tuple: (refresh result, fail reason)
+        """        
+        from swsscommon import swsscommon
+        try:
+            config_db = swsscommon.SonicV2Connector(use_unix_socket_path=True)
+            config_db.connect(CONFIG_DB)
+            log_level_in_db = config_db.get(CONFIG_DB, f'{swsscommon.CFG_LOGGER_TABLE_NAME}|{self.log_identifier}', FIELD_LOG_LEVEL)
+            if log_level_in_db:
+                self.set_min_log_priority(self.log_priority_from_str(log_level_in_db))
+            else:
+                data = {
+                    FIELD_LOG_LEVEL: self.log_priority_to_str(self._min_log_level),
+                    FIELD_REQUIRE_REFRESH: 'true'
+                }
+                config_db.hmset(CONFIG_DB, f'{swsscommon.CFG_LOGGER_TABLE_NAME}|{self.log_identifier}', data)
+            return True, ''
+        except Exception as e:
+            return False, f'Failed to refresh log configuration - {e}'
+
+    def log_priority_to_str(self, priority):
+        """Convert log priority to string.
+        Args:
+            priority (int): log priority.
+        Returns:
+            str: log priority in string.
+        """
+        if priority == logging.INFO:
+            return 'INFO'
+        elif priority == logging.NOTICE:
+            return 'NOTICE'
+        elif priority == logging.DEBUG:
+            return 'DEBUG'
+        elif priority == logging.WARNING:
+            return 'WARN'
+        elif priority == logging.ERROR:
+            return 'ERROR'
+        else:
+            self.log_error(f'Invalid log priority: {priority}')
+            return 'WARN'
+
+    def log_priority_from_str(self, priority_in_str):
+        """Convert log priority from string.
+        Args:
+            priority_in_str (str): log priority in string.
+        Returns:
+            _type_: log priority.
+        """
+        if priority_in_str == 'DEBUG':
+            return logging.DEBUG
+        elif priority_in_str == 'INFO':
+            return logging.INFO
+        elif priority_in_str == 'NOTICE':
+            return logging.NOTICE
+        elif priority_in_str == 'WARN':
+            return logging.WARNING
+        elif priority_in_str == 'ERROR':
+            return logging.ERROR
+        else:
+            self.log_error(f'Invalid log priority string: {priority_in_str}')
+            return logging.WARNING
 
     def set_min_log_priority(self, priority):
         """

src/sonic-py-common/tests/test_syslogger.py

@@ -32,3 +32,60 @@ def test_notice_log(self, capsys):
         log.log_notice('this is a message')
         captured = capsys.readouterr() 
         assert 'NOTICE' in captured.out
+
+    def test_basic(self):
+        log = syslogger.SysLogger()
+        log.logger.log = mock.MagicMock()
+        log.log_error('error message')
+        log.log_warning('warning message')
+        log.log_notice('notice message')
+        log.log_info('info message')
+        log.log_debug('debug message')
+        log.log(logging.ERROR, 'error msg', also_print_to_console=True)
+
+    def test_log_priority(self):
+        log = syslogger.SysLogger()
+        log.set_min_log_priority(logging.ERROR)
+        assert log.logger.level == logging.ERROR
+
+    def test_log_priority_from_str(self):
+        log = syslogger.SysLogger()
+        assert log.log_priority_from_str('ERROR') == logging.ERROR
+        assert log.log_priority_from_str('INFO') == logging.INFO
+        assert log.log_priority_from_str('NOTICE') == logging.NOTICE
+        assert log.log_priority_from_str('WARN') == logging.WARN
+        assert log.log_priority_from_str('DEBUG') == logging.DEBUG
+        assert log.log_priority_from_str('invalid') == logging.WARN
+
+    def test_log_priority_to_str(self):
+        log = syslogger.SysLogger()
+        assert log.log_priority_to_str(logging.NOTICE) == 'NOTICE'
+        assert log.log_priority_to_str(logging.INFO) == 'INFO'
+        assert log.log_priority_to_str(logging.DEBUG) == 'DEBUG'
+        assert log.log_priority_to_str(logging.WARN) == 'WARN'
+        assert log.log_priority_to_str(logging.ERROR) == 'ERROR'
+        assert log.log_priority_to_str(-1) == 'WARN'
+
+    @mock.patch('swsscommon.swsscommon.SonicV2Connector')
+    def test_runtime_config(self, mock_connector):
+        mock_db = mock.MagicMock()
+        mock_db.get = mock.MagicMock(return_value='DEBUG')
+        mock_connector.return_value = mock_db
+        log = syslogger.SysLogger(log_identifier='log1', enable_runtime_config=True, log_level=logging.INFO)
+        assert log.logger.level == logging.DEBUG
+
+        mock_db.get.return_value = 'ERROR'
+        ret, msg = log.update_log_level()
+        assert ret
+        assert not msg
+
+    @mock.patch('swsscommon.swsscommon.SonicV2Connector')
+    def test_runtime_config_negative(self, mock_connector):
+        mock_db = mock.MagicMock()
+        mock_db.get = mock.MagicMock(side_effect=Exception(''))
+        mock_connector.return_value = mock_db
+        log = syslogger.SysLogger(log_identifier='log', enable_runtime_config=True)
+
+        ret, msg = log.update_log_level()
+        assert not ret
+        assert msg