[hostcfgd] Move hostcfgd back to ConfigDBConnector for subscribing to updates

src/sonic-host-services/scripts/hostcfgd

@@ -11,7 +11,6 @@ import signal
 
 import jinja2
 from sonic_py_common import device_info
-from swsscommon.swsscommon import SubscriberStateTable, DBConnector, Select
 from swsscommon.swsscommon import ConfigDBConnector, TableConsumable
 
 # FILE
@@ -195,6 +194,20 @@ class FeatureHandler(object):
             else:
                 self.resync_feature_state(self._cached_config[feature_name])
 
+    def update_all_features_config(self, feature_table):
+        for feature_name in feature_table.keys():
+            if not feature_name:
+                syslog.syslog(syslog.LOG_WARNING, "Feature is None")
+                continue
+
+            feature = Feature(feature_name, feature_table[feature_name], self._device_config)
+            self._cached_config.setdefault(feature_name, feature)
+
+            self.update_feature_auto_restart(feature)
+
+            self.update_feature_state(feature)
+            self.resync_feature_state(feature)
+
     def sync_state_field(self):
         """
         Summary:
@@ -956,14 +969,8 @@ class HostConfigDaemon:
         # before moving forward
         self.config_db = ConfigDBConnector()
         self.config_db.connect(wait_for_init=True, retry_on=True)
-        self.dbconn = DBConnector(CFG_DB, 0)
-        self.selector = Select()
         syslog.syslog(syslog.LOG_INFO, 'ConfigDB connect success')
 
-        self.select = Select()
-        self.callbacks = dict()
-        self.subscriber_map = dict()
-
         # Load DEVICE metadata configurations
         self.device_config = {}
         self.device_config['DEVICE_METADATA'] = self.config_db.get_table('DEVICE_METADATA')
@@ -987,16 +994,32 @@ class HostConfigDaemon:
         # Initialize AAACfg
         self.hostname_cache=""
         self.aaacfg = AaaCfg()
+
+        # Initialize cache
+        self.cache = {}
 
 
     def load(self):
+        features = self.config_db.get_table('FEATURE')
+        self.feature_handler.update_all_features_config(features)
+
         aaa = self.config_db.get_table('AAA')
         tacacs_global = self.config_db.get_table('TACPLUS')
         tacacs_server = self.config_db.get_table('TACPLUS_SERVER')
         radius_global = self.config_db.get_table('RADIUS')
         radius_server = self.config_db.get_table('RADIUS_SERVER')
+        self.cache = {'AAA': aaa, 'TACPLUS': tacacs_global, 'TACPLUS_SERVER': tacacs_server,
+                        'RADIUS': radius_global, 'RADIUS_SERVER': radius_server}
         self.aaacfg.load(aaa, tacacs_global, tacacs_server, radius_global, radius_server)
 
+        lpbk_table = self.config_db.get_table('LOOPBACK_INTERFACE')
+        self.iptables.load(lpbk_table)
+
+        # Load NTP configurations
+        ntp_server = self.config_db.get_table('NTP_SERVER')
+        ntp_global = self.config_db.get_table('NTP')
+        self.ntpcfg.load(ntp_global, ntp_server)
+
         try:
             dev_meta = self.config_db.get_table('DEVICE_METADATA')
             if 'localhost' in dev_meta:
@@ -1097,68 +1120,48 @@ class HostConfigDaemon:
         systemctl_cmd = "sudo systemctl is-system-running --wait --quiet"
         subprocess.call(systemctl_cmd, shell=True)
 
-    def subscribe(self, table, callback, pri):
-        try:
-            if table not in self.callbacks:
-                self.callbacks[table] = []
-                subscriber = SubscriberStateTable(self.dbconn, table, TableConsumable.DEFAULT_POP_BATCH_SIZE, pri)
-                self.selector.addSelectable(subscriber) # Add to the Selector
-                self.subscriber_map[subscriber.getFd()] = (subscriber, table) # Maintain a mapping b/w subscriber & fd
+    def register_callbacks(self):
 
-            self.callbacks[table].append(callback)
-        except Exception as err:
-            syslog.syslog(syslog.LOG_ERR, "Subscribe to table {} failed with error {}".format(table, err))
+        def make_callback(func):
+            def callback(table, key, data):
+                if data is None:
+                    op = "DEL"
+                else:
+                    op = "SET"
+                return func(key, op, data)
+            return callback
 
-    def register_callbacks(self):
-        self.subscribe('KDUMP', lambda table, key, op, data: self.kdump_handler(key, op, data), HOSTCFGD_MAX_PRI)
+        self.config_db.subscribe('KDUMP', make_callback(self.kdump_handler))
         # Handle FEATURE updates before other tables
-        self.subscribe('FEATURE', lambda table, key, op, data: self.feature_handler.handle(key, op, data), HOSTCFGD_MAX_PRI-1)
+        self.config_db.subscribe('FEATURE', make_callback(self.feature_handler.handle))
         # Handle AAA, TACACS and RADIUS related tables
-        self.subscribe('AAA', lambda table, key, op, data: self.aaa_handler(key, op, data), HOSTCFGD_MAX_PRI-2)
-        self.subscribe('TACPLUS', lambda table, key, op, data: self.tacacs_global_handler(key, op, data), HOSTCFGD_MAX_PRI-2)
-        self.subscribe('TACPLUS_SERVER', lambda table, key, op, data: self.tacacs_server_handler(key, op, data), HOSTCFGD_MAX_PRI-2)
-        self.subscribe('RADIUS', lambda table, key, op, data: self.radius_global_handler(key, op, data), HOSTCFGD_MAX_PRI-2)
-        self.subscribe('RADIUS_SERVER', lambda table, key, op, data: self.radius_server_handler(key, op, data), HOSTCFGD_MAX_PRI-2)
+        self.config_db.subscribe('AAA', make_callback(self.aaa_handler))
+        self.config_db.subscribe('TACPLUS', make_callback(self.tacacs_global_handler))
+        self.config_db.subscribe('TACPLUS_SERVER', make_callback(self.tacacs_server_handler))
+        self.config_db.subscribe('RADIUS', make_callback(self.radius_global_handler))
+        self.config_db.subscribe('RADIUS_SERVER', make_callback(self.radius_server_handler))
         # Handle IPTables configuration
-        self.subscribe('LOOPBACK_INTERFACE', lambda table, key, op, data: self.lpbk_handler(key, op, data), HOSTCFGD_MAX_PRI-3)
+        self.config_db.subscribe('LOOPBACK_INTERFACE', make_callback(self.lpbk_handler))
         # Handle NTP & NTP_SERVER updates
-        self.subscribe('NTP', lambda table, key, op, data: self.ntp_global_handler(key, op, data), HOSTCFGD_MAX_PRI-4)
-        self.subscribe('NTP_SERVER', lambda table, key, op, data: self.ntp_server_handler(key, op, data), HOSTCFGD_MAX_PRI-4)
+        self.config_db.subscribe('NTP', make_callback(self.ntp_global_handler))
+        self.config_db.subscribe('NTP_SERVER', make_callback(self.ntp_server_handler))
         # Handle updates to src intf changes in radius
-        self.subscribe('MGMT_INTERFACE', lambda table, key, op, data: self.mgmt_intf_handler(key, op, data), HOSTCFGD_MAX_PRI-5)
-        self.subscribe('VLAN_INTERFACE', lambda table, key, op, data: self.vlan_intf_handler(key, op, data), HOSTCFGD_MAX_PRI-5)
-        self.subscribe('VLAN_SUB_INTERFACE', lambda table, key, op, data: self.vlan_sub_intf_handler(key, op, data), HOSTCFGD_MAX_PRI-5)
-        self.subscribe('PORTCHANNEL_INTERFACE', lambda table, key, op, data: self.portchannel_intf_handler(key, op, data), HOSTCFGD_MAX_PRI-5)
-        self.subscribe('INTERFACE', lambda table, key, op, data: self.phy_intf_handler(key, op, data), HOSTCFGD_MAX_PRI-5)
+        self.config_db.subscribe('MGMT_INTERFACE', make_callback(self.mgmt_intf_handler))
+        self.config_db.subscribe('VLAN_INTERFACE', make_callback(self.vlan_intf_handler))
+        self.config_db.subscribe('VLAN_SUB_INTERFACE', make_callback(self.vlan_sub_intf_handler))
+        self.config_db.subscribe('PORTCHANNEL_INTERFACE', make_callback(self.portchannel_intf_handler))
+        self.config_db.subscribe('INTERFACE', make_callback(self.phy_intf_handler))
 
         syslog.syslog(syslog.LOG_INFO,
                       "Waiting for systemctl to finish initialization")
         self.wait_till_system_init_done()
         syslog.syslog(syslog.LOG_INFO,
                       "systemctl has finished initialization -- proceeding ...")
 
-    def start(self):
-        while True:
-            state, selectable_ = self.selector.select(DEFAULT_SELECT_TIMEOUT)
-            if state == self.selector.TIMEOUT:
-                continue
-            elif state == self.selector.ERROR:
-                syslog.syslog(syslog.LOG_ERR,
-                        "error returned by select")
-                continue
+        self.config_db.listen(start=False)
 
-            fd = selectable_.getFd()
-            # Get the Corresponding subscriber & table
-            subscriber, table = self.subscriber_map.get(fd, (None, ""))
-            if not subscriber:
-                syslog.syslog(syslog.LOG_ERR,
-                        "No Subscriber object found for fd: {}, subscriber map: {}".format(fd, subscriber_map))
-                continue
-            key, op, fvs = subscriber.pop()
-            # Get the registered callback
-            cbs = self.callbacks.get(table, None)
-            for callback in cbs:
-                callback(table, key, op, dict(fvs))
+    def start(self):
+        self.config_db.process(cache=self.cache)
 
 
 def main():

src/sonic-host-services/tests/common/mock_configdb.py

@@ -4,9 +4,10 @@ class MockConfigDb(object):
     """
     STATE_DB = None
     CONFIG_DB = None
+    event_queue = []
 
     def __init__(self, **kwargs):
-        pass
+        self.handlers = {}
 
     @staticmethod
     def set_config_db(test_config_db):
@@ -44,75 +45,13 @@ def set_entry(self, key, field, data):
     def get_table(self, table_name):
         return MockConfigDb.CONFIG_DB[table_name]
 
+    def subscribe(self, table_name, callback):
+        self.handlers[table_name] = callback
 
-class MockSelect():
-
-    event_queue = []
-
-    @staticmethod
-    def set_event_queue(Q):
-        MockSelect.event_queue = Q
-
-    @staticmethod
-    def get_event_queue():
-        return MockSelect.event_queue
-
-    @staticmethod
-    def reset_event_queue():
-        MockSelect.event_queue = []
-
-    def __init__(self):
-        self.sub_map = {}
-        self.TIMEOUT = "TIMEOUT"
-        self.ERROR = "ERROR"
-
-    def addSelectable(self, subscriber):
-        self.sub_map[subscriber.table] = subscriber
-
-    def select(self, TIMEOUT):
-        if not MockSelect.get_event_queue():
-            raise TimeoutError
-        table, key = MockSelect.get_event_queue().pop(0)
-        self.sub_map[table].nextKey(key)
-        return "OBJECT", self.sub_map[table]
-
-
-class MockSubscriberStateTable():
-
-    FD_INIT = 0
-
-    @staticmethod
-    def generate_fd():
-        curr = MockSubscriberStateTable.FD_INIT
-        MockSubscriberStateTable.FD_INIT = curr + 1
-        return curr
-
-    @staticmethod
-    def reset_fd():
-        MockSubscriberStateTable.FD_INIT = 0
-
-    def __init__(self, conn, table, pop, pri):
-        self.fd = MockSubscriberStateTable.generate_fd()
-        self.next_key = ''
-        self.table = table
-
-    def getFd(self):
-        return self.fd
-
-    def nextKey(self, key):
-        self.next_key = key
-
-    def pop(self):
-        table = MockConfigDb.CONFIG_DB.get(self.table, {})
-        if self.next_key not in table:
-            op = "DEL"
-            fvs = {}
-        else:
-            op = "SET"
-            fvs = table.get(self.next_key, {})
-        return self.next_key, op, fvs
+    def listen(self, start):
+        pass
 
+    def process(self, cache):
+        for e in MockConfigDb.event_queue:
+            self.handlers[e[0]](e[0], e[1], self.get_entry(e[0], e[1]))
 
-class MockDBConnector():
-    def __init__(self, db, val):
-        pass

src/sonic-host-services/tests/hostcfgd/hostcfgd_radius_test.py

@@ -10,8 +10,7 @@
 from parameterized import parameterized
 from unittest import TestCase, mock
 from tests.hostcfgd.test_radius_vectors import HOSTCFGD_TEST_RADIUS_VECTOR
-from tests.common.mock_configdb import MockConfigDb, MockSubscriberStateTable
-from tests.common.mock_configdb import MockSelect, MockDBConnector
+from tests.common.mock_configdb import MockConfigDb
 
 
 test_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@@ -33,9 +32,6 @@
 
 # Mock swsscommon classes
 hostcfgd.ConfigDBConnector = MockConfigDb
-hostcfgd.SubscriberStateTable = MockSubscriberStateTable
-hostcfgd.Select = MockSelect
-hostcfgd.DBConnector = MockDBConnector
 
 
 class TestHostcfgdRADIUS(TestCase):

src/sonic-host-services/tests/hostcfgd/hostcfgd_tacacs_test.py

@@ -10,8 +10,7 @@
 from parameterized import parameterized
 from unittest import TestCase, mock
 from tests.hostcfgd.test_tacacs_vectors import HOSTCFGD_TEST_TACACS_VECTOR
-from tests.common.mock_configdb import MockConfigDb, MockSubscriberStateTable
-from tests.common.mock_configdb import MockSelect, MockDBConnector
+from tests.common.mock_configdb import MockConfigDb
 
 test_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 modules_path = os.path.dirname(test_path)
@@ -32,9 +31,6 @@
 
 # Mock swsscommon classes
 hostcfgd.ConfigDBConnector = MockConfigDb
-hostcfgd.SubscriberStateTable = MockSubscriberStateTable
-hostcfgd.Select = MockSelect
-hostcfgd.DBConnector = MockDBConnector
 
 class TestHostcfgdTACACS(TestCase):
     """

src/sonic-host-services/tests/hostcfgd/hostcfgd_test.py

@@ -8,8 +8,7 @@
 from unittest import TestCase, mock
 
 from .test_vectors import HOSTCFGD_TEST_VECTOR, HOSTCFG_DAEMON_CFG_DB
-from tests.common.mock_configdb import MockConfigDb, MockSubscriberStateTable
-from tests.common.mock_configdb import MockSelect, MockDBConnector
+from tests.common.mock_configdb import MockConfigDb
 
 from pyfakefs.fake_filesystem_unittest import patchfs
 from deepdiff import DeepDiff
@@ -24,9 +23,6 @@
 hostcfgd_path = os.path.join(scripts_path, 'hostcfgd')
 hostcfgd = load_module_from_source('hostcfgd', hostcfgd_path)
 hostcfgd.ConfigDBConnector = MockConfigDb
-hostcfgd.SubscriberStateTable = MockSubscriberStateTable
-hostcfgd.Select = MockSelect
-hostcfgd.DBConnector = MockDBConnector
 
 
 class TestHostcfgd(TestCase):
@@ -209,7 +205,7 @@ def tearDown(self):
     @patchfs
     def test_feature_events(self, fs):
         fs.create_dir(hostcfgd.FeatureHandler.SYSTEMD_SYSTEM_DIR)
-        MockSelect.event_queue = [('FEATURE', 'dhcp_relay'),
+        MockConfigDb.event_queue = [('FEATURE', 'dhcp_relay'),
                                 ('FEATURE', 'mux'),
                                 ('FEATURE', 'telemetry')]
         daemon = hostcfgd.HostConfigDaemon()
@@ -240,7 +236,7 @@ def test_feature_events(self, fs):
 
             # Change the state to disabled
             MockConfigDb.CONFIG_DB['FEATURE']['telemetry']['state'] = 'disabled'
-            MockSelect.event_queue = [('FEATURE', 'telemetry')]
+            MockConfigDb.event_queue = [('FEATURE', 'telemetry')]
             try:
                 daemon.start()
             except TimeoutError:
@@ -255,7 +251,7 @@ def test_feature_events(self, fs):
 
     def test_loopback_events(self):
         MockConfigDb.set_config_db(HOSTCFG_DAEMON_CFG_DB)
-        MockSelect.event_queue = [('NTP', 'global'),
+        MockConfigDb.event_queue = [('NTP', 'global'),
                                   ('NTP_SERVER', '0.debian.pool.ntp.org'),
                                   ('LOOPBACK_INTERFACE', 'Loopback0|10.184.8.233/32')]
         daemon = hostcfgd.HostConfigDaemon()
@@ -279,7 +275,7 @@ def test_kdump_event(self):
         daemon = hostcfgd.HostConfigDaemon()
         daemon.register_callbacks()
         assert MockConfigDb.CONFIG_DB['KDUMP']['config']
-        MockSelect.event_queue = [('KDUMP', 'config')]
+        MockConfigDb.event_queue = [('KDUMP', 'config')]
         with mock.patch('hostcfgd.subprocess') as mocked_subprocess:
             popen_mock = mock.Mock()
             attrs = {'communicate.return_value': ('output', 'error')}