Update port and queue stat counters by Flex counter (sonic-net#358)

* Update port and queue stat counters by Flex counter

* Fixed review remarks and rebase issue

orchagent/orchdaemon.cpp

@@ -11,6 +11,7 @@ using namespace swss;
 
 /* select() function timeout retry time */
 #define SELECT_TIMEOUT 1000
+#define FLEX_COUNTER_POLL_MSECS 100
 
 extern sai_switch_api_t*           sai_switch_api;
 extern sai_object_id_t             gSwitchId;
@@ -140,7 +141,8 @@ bool OrchDaemon::init()
                     pfc_wd_tables,
                     portStatIds,
                     queueStatIds,
-                    queueAttrIds));
+                    queueAttrIds,
+                    FLEX_COUNTER_POLL_MSECS));
     }
     else if (platform == BRCM_PLATFORM_SUBSTRING)
     {
@@ -180,7 +182,8 @@ bool OrchDaemon::init()
                     pfc_wd_tables,
                     portStatIds,
                     queueStatIds,
-                    queueAttrIds));
+                    queueAttrIds,
+                    FLEX_COUNTER_POLL_MSECS));
     }
 
     return true;

orchagent/pfcwdorch.cpp

@@ -273,13 +273,12 @@ void PfcWdSwOrch<DropHandler, ForwardHandler>::registerInWdDb(const Port& port,
 
     if (!c_portStatIds.empty())
     {
+        string key = sai_serialize_object_id(port.m_port_id) + ":" + std::to_string(m_pollInterval);
         vector<FieldValueTuple> fieldValues;
         string str = counterIdsToStr(c_portStatIds, &sai_serialize_port_stat);
         fieldValues.emplace_back(PFC_WD_PORT_COUNTER_ID_LIST, str);
 
-        m_pfcWdTable->set(
-                sai_serialize_object_id(port.m_port_id),
-                fieldValues);
+        m_pfcWdTable->set(key, fieldValues);
     }
 
     uint8_t pfcMask = attr.value.u8;
@@ -319,7 +318,9 @@ void PfcWdSwOrch<DropHandler, ForwardHandler>::registerInWdDb(const Port& port,
         // Create internal entry
         m_entryMap.emplace(queueId, PfcWdQueueEntry(action, port.m_port_id, i));
 
-        m_pfcWdTable->set(queueIdStr, queueFieldValues);
+        string key = queueIdStr + ":" + std::to_string(m_pollInterval);
+
+        m_pfcWdTable->set(key, queueFieldValues);
 
         // Initialize PFC WD related counters
         PfcWdActionHandler::initWdCounters(
@@ -336,9 +337,10 @@ void PfcWdSwOrch<DropHandler, ForwardHandler>::unregisterFromWdDb(const Port& po
     for (uint8_t i = 0; i < PFC_WD_TC_MAX; i++)
     {
         sai_object_id_t queueId = port.m_queue_ids[i];
+        string key = sai_serialize_object_id(queueId) + ":" + std::to_string(m_pollInterval);
 
         // Unregister in syncd
-        m_pfcWdTable->del(sai_serialize_object_id(queueId));
+        m_pfcWdTable->del(key);
         m_entryMap.erase(queueId);
     }
 }
@@ -349,13 +351,15 @@ PfcWdSwOrch<DropHandler, ForwardHandler>::PfcWdSwOrch(
         vector<string> &tableNames,
         const vector<sai_port_stat_t> &portStatIds,
         const vector<sai_queue_stat_t> &queueStatIds,
-        const vector<sai_queue_attr_t> &queueAttrIds):
+        const vector<sai_queue_attr_t> &queueAttrIds, 
+        int pollInterval):
     PfcWdOrch<DropHandler, ForwardHandler>(db, tableNames),
     m_pfcWdDb(new DBConnector(PFC_WD_DB, DBConnector::DEFAULT_UNIXSOCKET, 0)),
     m_pfcWdTable(new ProducerStateTable(m_pfcWdDb.get(), PFC_WD_STATE_TABLE)),
     c_portStatIds(portStatIds),
     c_queueStatIds(queueStatIds),
-    c_queueAttrIds(queueAttrIds)
+    c_queueAttrIds(queueAttrIds),
+    m_pollInterval(pollInterval)
 {
     SWSS_LOG_ENTER();
 

orchagent/pfcwdorch.h

@@ -62,7 +62,8 @@ class PfcWdSwOrch: public PfcWdOrch<DropHandler, ForwardHandler>
             vector<string> &tableNames,
             const vector<sai_port_stat_t> &portStatIds,
             const vector<sai_queue_stat_t> &queueStatIds,
-            const vector<sai_queue_attr_t> &queueAttrIds);
+            const vector<sai_queue_attr_t> &queueAttrIds,
+            int pollInterval);
     virtual ~PfcWdSwOrch(void);
 
     virtual bool startWdOnPort(const Port& port,
@@ -106,6 +107,8 @@ class PfcWdSwOrch: public PfcWdOrch<DropHandler, ForwardHandler>
 
     atomic_bool m_runPfcWdSwOrchThread = { false };
     shared_ptr<thread> m_pfcWatchdogThread = nullptr;
+
+    int m_pollInterval;
 };
 
 #endif

orchagent/portsorch.cpp

@@ -6,6 +6,7 @@
 #include <set>
 #include <algorithm>
 #include <tuple>
+#include <sstream>
 
 #include <netinet/if_ether.h>
 #include "net/if.h"
@@ -29,6 +30,7 @@ extern sai_object_id_t gSwitchId;
 
 #define VLAN_PREFIX         "Vlan"
 #define DEFAULT_VLAN_ID     1
+#define FLEX_STAT_COUNTER_POLL_MSECS "1000"
 
 static map<string, sai_port_fec_mode_t> fec_mode_map =
 {
@@ -56,16 +58,19 @@ PortsOrch::PortsOrch(DBConnector *db, vector<string> tableNames) :
     SWSS_LOG_ENTER();
 
     /* Initialize counter table */
-    DBConnector *counter_db(new DBConnector(COUNTERS_DB, DBConnector::DEFAULT_UNIXSOCKET, 0));
-    m_counterTable = unique_ptr<Table>(new Table(counter_db, COUNTERS_PORT_NAME_MAP));
+    m_counter_db = shared_ptr<DBConnector>(new DBConnector(COUNTERS_DB, DBConnector::DEFAULT_UNIXSOCKET, 0));
+    m_counterTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_PORT_NAME_MAP));
 
     /* Initialize port table */
     m_portTable = unique_ptr<Table>(new Table(db, APP_PORT_TABLE_NAME));
 
     /* Initialize queue tables */
-    m_queueTable = unique_ptr<Table>(new Table(counter_db, COUNTERS_QUEUE_NAME_MAP));
-    m_queuePortTable = unique_ptr<Table>(new Table(counter_db, COUNTERS_QUEUE_PORT_MAP));
-    m_queueIndexTable = unique_ptr<Table>(new Table(counter_db, COUNTERS_QUEUE_INDEX_MAP));
+    m_queueTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_QUEUE_NAME_MAP));
+    m_queuePortTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_QUEUE_PORT_MAP));
+    m_queueIndexTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_QUEUE_INDEX_MAP));
+
+    m_flex_db = shared_ptr<DBConnector>(new DBConnector(PFC_WD_DB, DBConnector::DEFAULT_UNIXSOCKET, 0));
+    m_flexCounterTable = unique_ptr<ProducerStateTable>(new ProducerStateTable(m_flex_db.get(), PFC_WD_STATE_TABLE));
 
     uint32_t i, j;
     sai_status_t status;
@@ -643,9 +648,25 @@ bool PortsOrch::initPort(const string &alias, const set<int> &lane_set)
                 m_portList[alias] = p;
                 /* Add port name map to counter table */
                 FieldValueTuple tuple(p.m_alias, sai_serialize_object_id(p.m_port_id));
-                vector<FieldValueTuple> vector;
-                vector.push_back(tuple);
-                m_counterTable->set("", vector);
+                vector<FieldValueTuple> fields;
+                fields.push_back(tuple);
+                m_counterTable->set("", fields);
+
+                /* Add port to flex_counter for updating stat counters  */
+                string key = sai_serialize_object_id(p.m_port_id) + ":" + FLEX_STAT_COUNTER_POLL_MSECS;
+
+                std::string delimiter = "";
+                std::ostringstream counters_stream;
+                for (int cntr = SAI_PORT_STAT_IF_IN_OCTETS; cntr <= SAI_PORT_STAT_PFC_7_ON2OFF_RX_PKTS; ++cntr)
+                {
+                    counters_stream << delimiter << sai_serialize_port_stat(static_cast<sai_port_stat_t>(cntr));
+                    delimiter = ",";
+                }
+
+                fields.clear();
+                fields.emplace_back(PFC_WD_PORT_COUNTER_ID_LIST, counters_stream.str());
+
+                m_flexCounterTable->set(key, fields);
 
                 SWSS_LOG_NOTICE("Initialized port %s", alias.c_str());
             }
@@ -1362,6 +1383,7 @@ void PortsOrch::initializeQueues(Port &port)
     SWSS_LOG_INFO("Get queues for port %s", port.m_alias.c_str());
 
     /* Create the Queue map in the Counter DB */
+    /* Add stat counters to flex_counter */
     vector<FieldValueTuple> queueVector;
     vector<FieldValueTuple> queuePortVector;
     vector<FieldValueTuple> queueIndexVector;
@@ -1382,6 +1404,21 @@ void PortsOrch::initializeQueues(Port &port)
                 sai_serialize_object_id(port.m_queue_ids[queueIndex]),
                 to_string(queueIndex));
         queueIndexVector.push_back(queueIndexTuple);
+
+        string key = sai_serialize_object_id(port.m_queue_ids[queueIndex]) + ":" + FLEX_STAT_COUNTER_POLL_MSECS;
+
+        std::string delimiter = "";
+        std::ostringstream counters_stream;
+        for (int cntr = SAI_QUEUE_STAT_PACKETS; cntr <= SAI_QUEUE_STAT_SHARED_WATERMARK_BYTES ; ++cntr)
+        {
+            counters_stream << delimiter << sai_serialize_queue_stat(static_cast<sai_queue_stat_t>(cntr));
+            delimiter = ",";
+        }
+
+        vector<FieldValueTuple> fieldValues;
+        fieldValues.emplace_back(PFC_WD_QUEUE_COUNTER_ID_LIST, counters_stream.str());
+
+        m_flexCounterTable->set(key, fieldValues);
     }
 
     m_queueTable->set("", queueVector);

orchagent/portsorch.h

@@ -7,6 +7,7 @@
 #include "port.h"
 #include "observer.h"
 #include "macaddress.h"
+#include "producerstatetable.h"
 
 #include <map>
 
@@ -61,6 +62,10 @@ class PortsOrch : public Orch, public Subject
     unique_ptr<Table> m_queueTable;
     unique_ptr<Table> m_queuePortTable;
     unique_ptr<Table> m_queueIndexTable;
+    unique_ptr<ProducerStateTable> m_flexCounterTable;
+
+    shared_ptr<DBConnector> m_counter_db;
+    shared_ptr<DBConnector> m_flex_db;
 
     std::map<sai_object_id_t, PortSupportedSpeeds> m_portSupportedSpeeds;