[counters] Improve performance by polling only configured ports buffer queue/pg counters (#2360)

Propagating #2143 with resolved merge conflicts
Depends on: sonic-net/sonic-utilities#2239

- What I did
Currently in SONiC all ports queue and pg counters are created by default with the max possible amount of counters.
This feature change this behavior to poll only configured counters provided by the config DB BUFFER_PG and BUFFER_QUEUE tables.
If no tables are present in the DB, no counters will be created for ports.
Filter the unwanted queues/pgs returned by SAI API calls and skip the creation of these queue/pg counters.
Also allow creating/removing counters on runtime if buffer PG/Queue is configured or removed.

- Why I did it
Improve performance by filtering unconfigured queue/pg counters on init.

- How I verified it
Check after enabling the counters, if configured counters created in Counters DB according to the configurations.
Add/Remove buffer PG/Queue configurations and observe the corresponding counters created/removed accordingly.
New UT added to verify this flow.

Signed-off-by: Nazarii Hnydyn <[email protected]>

Author: nazariig

orchagent/bufferorch.cpp

@@ -1,5 +1,6 @@
 #include "tokenize.h"
 #include "bufferorch.h"
+#include "directory.h"
 #include "logger.h"
 #include "sai_serialize.h"
 #include "warm_restart.h"
@@ -16,6 +17,7 @@ extern sai_switch_api_t *sai_switch_api;
 extern sai_buffer_api_t *sai_buffer_api;
 
 extern PortsOrch *gPortsOrch;
+extern Directory<Orch*> gDirectory;
 extern sai_object_id_t gSwitchId;
 
 #define BUFFER_POOL_WATERMARK_FLEX_STAT_COUNTER_POLL_MSECS  "60000"
@@ -812,6 +814,20 @@ task_process_status BufferOrch::processQueue(KeyOpFieldsValuesTuple &tuple)
                         return handle_status;
                     }
                 }
+                // create/remove a port queue counter for the queue buffer
+                else
+                {
+                    auto flexCounterOrch = gDirectory.get<FlexCounterOrch*>();
+                    auto queues = tokens[1];
+                    if (op == SET_COMMAND && flexCounterOrch->getQueueCountersState())
+                    {
+                        gPortsOrch->createPortBufferQueueCounters(port, queues);
+                    }
+                    else if (op == DEL_COMMAND && flexCounterOrch->getQueueCountersState())
+                    {
+                        gPortsOrch->removePortBufferQueueCounters(port, queues);
+                    }
+                }
             }
         }
     }
@@ -871,7 +887,7 @@ task_process_status BufferOrch::processPriorityGroup(KeyOpFieldsValuesTuple &tup
     if (op == SET_COMMAND)
     {
         ref_resolve_status  resolve_result = resolveFieldRefValue(m_buffer_type_maps, buffer_profile_field_name,
-                                             buffer_to_ref_table_map.at(buffer_profile_field_name), tuple, 
+                                             buffer_to_ref_table_map.at(buffer_profile_field_name), tuple,
                                              sai_buffer_profile, buffer_profile_name);
         if (ref_resolve_status::success != resolve_result)
         {
@@ -944,6 +960,20 @@ task_process_status BufferOrch::processPriorityGroup(KeyOpFieldsValuesTuple &tup
                             return handle_status;
                         }
                     }
+                    // create or remove a port PG counter for the PG buffer
+                    else
+                    {
+                        auto flexCounterOrch = gDirectory.get<FlexCounterOrch*>();
+                        auto pgs = tokens[1];
+                        if (op == SET_COMMAND && flexCounterOrch->getPgWatermarkCountersState())
+                        {
+                            gPortsOrch->createPortBufferPgCounters(port, pgs);
+                        }
+                        else if (op == DEL_COMMAND && flexCounterOrch->getPgWatermarkCountersState())
+                        {
+                            gPortsOrch->removePortBufferPgCounters(port, pgs);
+                        }
+                    }
                 }
             }
         }

orchagent/flexcounterorch.cpp

@@ -10,6 +10,7 @@
 #include "debugcounterorch.h"
 #include "directory.h"
 #include "copporch.h"
+#include <swss/tokenize.h>
 #include "routeorch.h"
 #include "macsecorch.h"
 #include "flowcounterrouteorch.h"
@@ -62,6 +63,8 @@ unordered_map<string, string> flexCounterGroupMap =
 FlexCounterOrch::FlexCounterOrch(DBConnector *db, vector<string> &tableNames):
     Orch(db, tableNames),
     m_flexCounterConfigTable(db, CFG_FLEX_COUNTER_TABLE_NAME),
+    m_bufferQueueConfigTable(db, CFG_BUFFER_QUEUE_TABLE_NAME),
+    m_bufferPgConfigTable(db, CFG_BUFFER_PG_TABLE_NAME),
     m_flexCounterDb(new DBConnector("FLEX_COUNTER_DB", 0)),
     m_flexCounterGroupTable(new ProducerTable(m_flexCounterDb.get(), FLEX_COUNTER_GROUP_TABLE)),
     m_gbflexCounterDb(new DBConnector("GB_FLEX_COUNTER_DB", 0)),
@@ -157,11 +160,13 @@ void FlexCounterOrch::doTask(Consumer &consumer)
                         }
                         else if(key == QUEUE_KEY)
                         {
-                            gPortsOrch->generateQueueMap();
+                            gPortsOrch->generateQueueMap(getQueueConfigurations());
+                            m_queue_enabled = true;
                         }
                         else if(key == PG_WATERMARK_KEY)
                         {
-                            gPortsOrch->generatePriorityGroupMap();
+                            gPortsOrch->generatePriorityGroupMap(getPgConfigurations());
+                            m_pg_watermark_enabled = true;
                         }
                     }
                     if(gIntfsOrch && (key == RIF_KEY) && (value == "enable"))
@@ -245,6 +250,16 @@ bool FlexCounterOrch::getPortBufferDropCountersState() const
     return m_port_buffer_drop_counter_enabled;
 }
 
+bool FlexCounterOrch::getPgWatermarkCountersState() const
+{
+    return m_pg_watermark_enabled;
+}
+
+bool FlexCounterOrch::getQueueCountersState() const
+{
+    return m_queue_enabled;
+}
+
 bool FlexCounterOrch::bake()
 {
     /*
@@ -286,3 +301,165 @@ bool FlexCounterOrch::bake()
     Consumer* consumer = dynamic_cast<Consumer *>(getExecutor(CFG_FLEX_COUNTER_TABLE_NAME));
     return consumer->addToSync(entries);
 }
+
+map<string, FlexCounterQueueStates> FlexCounterOrch::getQueueConfigurations()
+{
+    SWSS_LOG_ENTER();
+
+    map<string, FlexCounterQueueStates> queuesStateVector;
+    std::vector<std::string> portQueueKeys;
+    m_bufferQueueConfigTable.getKeys(portQueueKeys);
+
+    for (const auto& portQueueKey : portQueueKeys)
+    {
+        auto toks = tokenize(portQueueKey, '|');
+        if (toks.size() != 2)
+        {
+            SWSS_LOG_ERROR("Invalid BUFFER_QUEUE key: [%s]", portQueueKey.c_str());
+            continue;
+        }
+
+        auto configPortNames = tokenize(toks[0], ',');
+        auto configPortQueues = toks[1];
+        toks = tokenize(configPortQueues, '-');
+
+        for (const auto& configPortName : configPortNames)
+        {
+            uint32_t maxQueueNumber = gPortsOrch->getNumberOfPortSupportedQueueCounters(configPortName);
+            uint32_t maxQueueIndex = maxQueueNumber - 1;
+            uint32_t minQueueIndex = 0;
+
+            if (!queuesStateVector.count(configPortName))
+            {
+                FlexCounterQueueStates flexCounterQueueState(maxQueueNumber);
+                queuesStateVector.insert(make_pair(configPortName, flexCounterQueueState));
+            }
+
+            try {
+                auto startIndex = to_uint<uint32_t>(toks[0], minQueueIndex, maxQueueIndex);
+                if (toks.size() > 1)
+                {
+                    auto endIndex = to_uint<uint32_t>(toks[1], minQueueIndex, maxQueueIndex);
+                    queuesStateVector.at(configPortName).enableQueueCounters(startIndex, endIndex);
+                }
+                else
+                {
+                    queuesStateVector.at(configPortName).enableQueueCounter(startIndex);
+                }
+            } catch (std::invalid_argument const& e) {
+                    SWSS_LOG_ERROR("Invalid queue index [%s] for port [%s]", configPortQueues.c_str(), configPortName.c_str());
+                    continue;
+            }
+        }
+    }
+
+    return queuesStateVector;
+}
+
+map<string, FlexCounterPgStates> FlexCounterOrch::getPgConfigurations()
+{
+    SWSS_LOG_ENTER();
+
+    map<string, FlexCounterPgStates> pgsStateVector;
+    std::vector<std::string> portPgKeys;
+    m_bufferPgConfigTable.getKeys(portPgKeys);
+
+    for (const auto& portPgKey : portPgKeys)
+    {
+        auto toks = tokenize(portPgKey, '|');
+        if (toks.size() != 2)
+        {
+            SWSS_LOG_ERROR("Invalid BUFFER_PG key: [%s]", portPgKey.c_str());
+            continue;
+        }
+
+        auto configPortNames = tokenize(toks[0], ',');
+        auto configPortPgs = toks[1];
+        toks = tokenize(configPortPgs, '-');
+
+        for (const auto& configPortName : configPortNames)
+        {
+            uint32_t maxPgNumber = gPortsOrch->getNumberOfPortSupportedPgCounters(configPortName);
+            uint32_t maxPgIndex = maxPgNumber - 1;
+            uint32_t minPgIndex = 0;
+
+            if (!pgsStateVector.count(configPortName))
+            {
+                FlexCounterPgStates flexCounterPgState(maxPgNumber);
+                pgsStateVector.insert(make_pair(configPortName, flexCounterPgState));
+            }
+
+            try {
+                auto startIndex = to_uint<uint32_t>(toks[0], minPgIndex, maxPgIndex);
+                if (toks.size() > 1)
+                {
+                    auto endIndex = to_uint<uint32_t>(toks[1], minPgIndex, maxPgIndex);
+                    pgsStateVector.at(configPortName).enablePgCounters(startIndex, endIndex);
+                }
+                else
+                {
+                    pgsStateVector.at(configPortName).enablePgCounter(startIndex);
+                }
+            } catch (std::invalid_argument const& e) {
+                    SWSS_LOG_ERROR("Invalid pg index [%s] for port [%s]", configPortPgs.c_str(), configPortName.c_str());
+                    continue;
+            }
+        }
+    }
+
+    return pgsStateVector;
+}
+
+FlexCounterQueueStates::FlexCounterQueueStates(uint32_t maxQueueNumber)
+{
+    SWSS_LOG_ENTER();
+    m_queueStates.resize(maxQueueNumber, false);
+}
+
+bool FlexCounterQueueStates::isQueueCounterEnabled(uint32_t index) const
+{
+    SWSS_LOG_ENTER();
+    return m_queueStates[index];
+}
+
+void FlexCounterQueueStates::enableQueueCounters(uint32_t startIndex, uint32_t endIndex)
+{
+    SWSS_LOG_ENTER();
+    for (uint32_t queueIndex = startIndex; queueIndex <= endIndex; queueIndex++)
+    {
+        enableQueueCounter(queueIndex);
+    }
+}
+
+void FlexCounterQueueStates::enableQueueCounter(uint32_t queueIndex)
+{
+    SWSS_LOG_ENTER();
+    m_queueStates[queueIndex] = true;
+}
+
+FlexCounterPgStates::FlexCounterPgStates(uint32_t maxPgNumber)
+{
+    SWSS_LOG_ENTER();
+    m_pgStates.resize(maxPgNumber, false);
+}
+
+bool FlexCounterPgStates::isPgCounterEnabled(uint32_t index) const
+{
+    SWSS_LOG_ENTER();
+    return m_pgStates[index];
+}
+
+void FlexCounterPgStates::enablePgCounters(uint32_t startIndex, uint32_t endIndex)
+{
+    SWSS_LOG_ENTER();
+    for (uint32_t pgIndex = startIndex; pgIndex <= endIndex; pgIndex++)
+    {
+        enablePgCounter(pgIndex);
+    }
+}
+
+void FlexCounterPgStates::enablePgCounter(uint32_t pgIndex)
+{
+    SWSS_LOG_ENTER();
+    m_pgStates[pgIndex] = true;
+}

orchagent/flexcounterorch.h

@@ -10,6 +10,30 @@ extern "C" {
 #include "sai.h"
 }
 
+class FlexCounterQueueStates
+{
+public:
+    FlexCounterQueueStates(uint32_t maxQueueNumber);
+    bool isQueueCounterEnabled(uint32_t index) const;
+    void enableQueueCounters(uint32_t startIndex, uint32_t endIndex);
+    void enableQueueCounter(uint32_t queueIndex);
+
+private:
+    std::vector<bool> m_queueStates{};
+};
+
+class FlexCounterPgStates
+{
+public:
+    FlexCounterPgStates(uint32_t maxPgNumber);
+    bool isPgCounterEnabled(uint32_t index) const;
+    void enablePgCounters(uint32_t startIndex, uint32_t endIndex);
+    void enablePgCounter(uint32_t pgIndex);
+
+private:
+    std::vector<bool> m_pgStates{};
+};
+
 class FlexCounterOrch: public Orch
 {
 public:
@@ -18,6 +42,10 @@ class FlexCounterOrch: public Orch
     virtual ~FlexCounterOrch(void);
     bool getPortCountersState() const;
     bool getPortBufferDropCountersState() const;
+    bool getPgWatermarkCountersState() const;
+    bool getQueueCountersState() const;
+    std::map<std::string, FlexCounterQueueStates> getQueueConfigurations();
+    std::map<std::string, FlexCounterPgStates> getPgConfigurations();
     bool getHostIfTrapCounterState() const {return m_hostif_trap_counter_enabled;}
     bool getRouteFlowCountersState() const {return m_route_flow_counter_enabled;}
     bool bake() override;
@@ -29,9 +57,13 @@ class FlexCounterOrch: public Orch
     std::shared_ptr<ProducerTable> m_gbflexCounterGroupTable = nullptr;
     bool m_port_counter_enabled = false;
     bool m_port_buffer_drop_counter_enabled = false;
+    bool m_pg_watermark_enabled = false;
+    bool m_queue_enabled = false;
     bool m_hostif_trap_counter_enabled = false;
     bool m_route_flow_counter_enabled = false;
     Table m_flexCounterConfigTable;
+    Table m_bufferQueueConfigTable;
+    Table m_bufferPgConfigTable;
 };
 
 #endif

orchagent/p4orch/tests/fake_flexcounterorch.cpp

@@ -1,8 +1,11 @@
 #include "copporch.h"
 #include "flexcounterorch.h"
 
-FlexCounterOrch::FlexCounterOrch(swss::DBConnector *db, std::vector<std::string> &tableNames)
-    : Orch(db, tableNames), m_flexCounterConfigTable(db, CFG_FLEX_COUNTER_TABLE_NAME)
+FlexCounterOrch::FlexCounterOrch(swss::DBConnector *db, std::vector<std::string> &tableNames) :
+    Orch(db, tableNames),
+    m_flexCounterConfigTable(db, CFG_FLEX_COUNTER_TABLE_NAME),
+    m_bufferQueueConfigTable(db, CFG_BUFFER_QUEUE_TABLE_NAME),
+    m_bufferPgConfigTable(db, CFG_BUFFER_PG_TABLE_NAME)
 {
 }