[syncd][PFC WD]: Add support for plugin subscription (sonic-net#229)

* [syncd][PFC WD]: Add support for SAI PFC storm notification

This commit implements native support for
SAI_SWITCH_ATTRIBUTE_QUEUE_PFC_DEADLOCK_NOTIFY.
Calling lua scripts is moved from orchagent to syncd, which uses native
notification by a SW watchdog as well.

Signed-off-by: marian-pritsak <marianp@mellanox.com>

* [syncd_pfcwatchdog]: Add plugin subscription

Signed-off-by: marian-pritsak <marianp@mellanox.com>

* [syncd]: Remove native notidications

Signed-off-by: marian-pritsak <marianp@mellanox.com>

* [syncd.cpp]: Fix typo

Signed-off-by: marian-pritsak <marianp@mellanox.com>

* Address comments

Signed-off-by: marian-pritsak <marianp@mellanox.com>

syncd/syncd.cpp

@@ -2358,9 +2358,10 @@ void processPfcWdEvent(
     const auto &key = kfvKey(kco);
     const auto &op = kfvOp(kco);
 
-    sai_object_id_t queueVid = SAI_NULL_OBJECT_ID;
-    sai_deserialize_object_id(key, queueVid);
-    sai_object_id_t queueId = translate_vid_to_rid(queueVid);
+    sai_object_id_t vid = SAI_NULL_OBJECT_ID;
+    sai_deserialize_object_id(key, vid);
+    sai_object_id_t rid = translate_vid_to_rid(vid);
+    sai_object_type_t objectType = sai_object_type_query(rid);
 
     const auto values = kfvFieldsValues(kco);
     for (const auto& valuePair : values)
@@ -2370,33 +2371,94 @@ void processPfcWdEvent(
 
         if (op == DEL_COMMAND)
         {
-            PfcWatchdog::removeQueue(queueVid);
-            continue;
-        }
-
-        auto idStrings  = swss::tokenize(value, ',');
-
-        if (field == PFC_WD_PORT_COUNTER_ID_LIST)
-        {
-            std::vector<sai_port_stat_t> portCounterIds;
-            for (const auto &str : idStrings)
+            if (objectType == SAI_OBJECT_TYPE_PORT)
+            {
+                PfcWatchdog::removePort(vid);
+            }
+            else if (objectType == SAI_OBJECT_TYPE_QUEUE)
             {
-                sai_port_stat_t stat;
-                sai_deserialize_port_stat(str, stat);
-                portCounterIds.push_back(stat);
+                PfcWatchdog::removeQueue(vid);
+            }
+            else
+            {
+                SWSS_LOG_ERROR("Object type for removal not supported");
             }
-            PfcWatchdog::setPortCounterList(queueVid, queueId, portCounterIds);
         }
-        else if (field == PFC_WD_QUEUE_COUNTER_ID_LIST)
+        else if (op == SET_COMMAND)
         {
-            std::vector<sai_queue_stat_t> queueCounterIds;
-            for (const auto &str : idStrings)
+            auto idStrings  = swss::tokenize(value, ',');
+
+            if (objectType == SAI_OBJECT_TYPE_PORT && field == PFC_WD_PORT_COUNTER_ID_LIST)
             {
-                sai_queue_stat_t stat;
-                sai_deserialize_queue_stat(str, stat);
-                queueCounterIds.push_back(stat);
+                std::vector<sai_port_stat_t> portCounterIds;
+                for (const auto &str : idStrings)
+                {
+                    sai_port_stat_t stat;
+                    sai_deserialize_port_stat(str, stat);
+                    portCounterIds.push_back(stat);
+                }
+                PfcWatchdog::setPortCounterList(vid, rid, portCounterIds);
             }
-            PfcWatchdog::setQueueCounterList(queueVid, queueId, queueCounterIds);
+            else if (objectType == SAI_OBJECT_TYPE_QUEUE && field == PFC_WD_QUEUE_COUNTER_ID_LIST)
+            {
+                std::vector<sai_queue_stat_t> queueCounterIds;
+                for (const auto &str : idStrings)
+                {
+                    sai_queue_stat_t stat;
+                    sai_deserialize_queue_stat(str, stat);
+                    queueCounterIds.push_back(stat);
+                }
+                PfcWatchdog::setQueueCounterList(vid, rid, queueCounterIds);
+            }
+            else
+            {
+                SWSS_LOG_ERROR("Object type not supported");
+            }
+        }
+    }
+}
+
+void processPfcWdPluginEvent(
+        _In_ swss::ConsumerStateTable &consumer)
+{
+    std::lock_guard<std::mutex> lock(g_mutex);
+
+    SWSS_LOG_ENTER();
+
+    swss::KeyOpFieldsValuesTuple kco;
+    consumer.pop(kco);
+
+    const auto &key = kfvKey(kco);
+    const auto &op = kfvOp(kco);
+
+    if (op == DEL_COMMAND)
+    {
+        PfcWatchdog::removeCounterPlugin(key);
+        return;
+    }
+
+    const auto values = kfvFieldsValues(kco);
+    for (const auto& valuePair : values)
+    {
+        const auto field = fvField(valuePair);
+        const auto value = fvValue(valuePair);
+
+        if (field != SAI_OBJECT_TYPE)
+        {
+            continue;
+        }
+
+        if (value == sai_serialize_object_type(SAI_OBJECT_TYPE_PORT))
+        {
+            PfcWatchdog::addPortCounterPlugin(key);
+        }
+        else if (value == sai_serialize_object_type(SAI_OBJECT_TYPE_QUEUE))
+        {
+            PfcWatchdog::addQueueCounterPlugin(key);
+        }
+        else
+        {
+            SWSS_LOG_ERROR("Plugin for %s is not supported", value.c_str());
         }
     }
 }
@@ -2981,6 +3043,7 @@ int main(int argc, char **argv)
     std::shared_ptr<swss::ConsumerTable> asicState = std::make_shared<swss::ConsumerTable>(dbAsic.get(), ASIC_STATE_TABLE);
     std::shared_ptr<swss::NotificationConsumer> restartQuery = std::make_shared<swss::NotificationConsumer>(dbAsic.get(), "RESTARTQUERY");
     std::shared_ptr<swss::ConsumerStateTable> pfcWdState = std::make_shared<swss::ConsumerStateTable>(dbPfcWatchdog.get(), PFC_WD_STATE_TABLE);
+    std::shared_ptr<swss::ConsumerStateTable> pfcWdPlugin = std::make_shared<swss::ConsumerStateTable>(dbPfcWatchdog.get(), PLUGIN_TABLE);
 
     /*
      * At the end we cant use producer consumer concept since if one proces
@@ -3077,6 +3140,7 @@ int main(int argc, char **argv)
         s.addSelectable(asicState.get());
         s.addSelectable(restartQuery.get());
         s.addSelectable(pfcWdState.get());
+        s.addSelectable(pfcWdPlugin.get());
 
         SWSS_LOG_NOTICE("starting main loop");
 
@@ -3105,6 +3169,10 @@ int main(int argc, char **argv)
             {
                 processPfcWdEvent(*(swss::ConsumerStateTable*)sel);
             }
+            else if (sel == pfcWdPlugin.get())
+            {
+                processPfcWdPluginEvent(*(swss::ConsumerStateTable*)sel);
+            }
             else if (result == swss::Select::OBJECT)
             {
                 processEvent(*(swss::ConsumerTable*)sel);