CPU Spike because of redundant and flooded keyspace notifis handled

src/sonic_ax_impl/mibs/ieee802_1ab.py

@@ -96,6 +96,18 @@ def poll_lldp_entry_updates(pubsub):
         return ret
     return data, interface, if_index
 
+def get_latest_notification(pubsub):
+    """
+    Fetches the latest notification recorded on a lldp entry.
+    """
+    latest_update_map = {}
+    while True:
+        data, interface, if_index = poll_lldp_entry_updates(pubsub)
+        if not data:
+            break
+        latest_update_map[interface] = (data, if_index)
+    return latest_update_map
+
 def parse_sys_capability(sys_cap):
     return bytearray([int (x, 16) for x in sys_cap.split()])
 
@@ -542,19 +554,17 @@ def _update_per_namespace_data(self, pubsub):
         """
         Listen to updates in APP DB, update local cache
         """
-        while True:
-            data, interface, if_index = poll_lldp_entry_updates(pubsub)
-
-            if not data:
-                break
-
+        event_cache = get_latest_notification(pubsub)
+        for interface in event_cache:
+            data = event_cache[interface][0]
+            if_index = event_cache[interface][1]
+
             if "set" in data:
                 self.update_rem_if_mgmt(if_index, interface)
             elif "del" in data:
-                # some remote data about that neighbor is gone, del it and try to query again
+                # if del is the latest notification, then delete it from the local cache
                 self.if_range = [sub_oid for sub_oid in self.if_range if sub_oid[0] != if_index]
-                self.update_rem_if_mgmt(if_index, interface)
-
+
     def update_data(self):
         for i in range(len(self.db_conn)):
             if not self.pubsub[i]:

tests/test_lldp.py

@@ -17,7 +17,12 @@
 from ax_interface.pdu import PDU, PDUHeader
 from ax_interface.mib import MIBTable
 from sonic_ax_impl.mibs import ieee802_1ab
+from mock import patch
 
+def mock_poll_lldp_notif(mock_lldp_polled_entries):
+    if not mock_lldp_polled_entries:
+        return None, None, None
+    return mock_lldp_polled_entries.pop(0)
 
 class TestLLDPMIB(TestCase):
     @classmethod
@@ -314,3 +319,27 @@ def test_getnextpdu_lldpRemSysCapEnabled(self):
         self.assertEqual(value0.type_, ValueType.OCTET_STRING)
         self.assertEqual(str(value0.name), str(ObjectIdentifier(12, 0, 1, 0, (1, 0, 8802, 1, 1, 2, 1, 4, 1, 1, 12, 1, 1))))
         self.assertEqual(str(value0.data), "\x28\x00")
+
+    @patch("sonic_ax_impl.mibs.ieee802_1ab.poll_lldp_entry_updates", mock_poll_lldp_notif)
+    def test_get_latest_notification(self):
+        mock_lldp_polled_entries = []
+        mock_lldp_polled_entries.extend([("hset", "Ethernet0", "123"),
+                                        ("hset", "Ethernet4", "124"),
+                                        ("del", "Ethernet4", "124"),
+                                        ("del", "Ethernet8", "125"),
+                                        ("hset", "Ethernet8", "125"),
+                                        ("hset", "Ethernet4", "124"),
+                                        ("del", "Ethernet0", "123"),
+                                        ("hset", "Ethernet12", "126"),
+                                        ("del", "Ethernet12", "126"),
+                                        ("hset", "Ethernet0", "123"),
+                                        ("del", "Ethernet16", "127")])
+        event_cache = ieee802_1ab.get_latest_notification(mock_lldp_polled_entries)
+        expect = {"Ethernet0" : ("hset", "123"),
+                  "Ethernet4" : ("hset", "124"),
+                  "Ethernet8" : ("hset", "125"),
+                  "Ethernet12" : ("del", "126"),
+                  "Ethernet16" : ("del", "127")}
+        for key in expect.keys():
+            assert key in event_cache
+            self.assertEqual(expect[key], event_cache[key])