Avoid notifying counter polling for duplicated priority groups or queues (#3885)

* Avoid notifying counter polling for duplicated priority groups or queues
What I did

Avoid notifying counter polling for duplicated PG or queue when create_only_config_db_buffers is False in the flow:

Flex counter orchagent enables counter polling for PG or queue. It notifies counter polling for all PGs or queues since create_only_config_db_buffers is False
Particular items in BUFFER_PG or BUFFER_QUEUE is configured. It notifies counter polling for the items again.
In most cases, BUFFER_PG and BUFFER_QUEUE are handled ahead of counter polling (i.e., the order is 2, 1), and the issue won't occur. But in a rare scenario, it can be 1 and then 2.

Author: stephenxs

orchagent/bufferorch.cpp

@@ -1103,18 +1103,21 @@ task_process_status BufferOrch::processQueuePost(const QueueTask& task)
                 else if (gMySwitchType != "voq")
                 {
                     auto flexCounterOrch = gDirectory.get<FlexCounterOrch*>();
-                    auto queues = tokens[1];
-                    if (!queueContext.counter_was_added && queueContext.counter_needs_to_add &&
-                        (flexCounterOrch->getQueueCountersState() || flexCounterOrch->getQueueWatermarkCountersState()))
+                    if (flexCounterOrch->isCreateOnlyConfigDbBuffers())
                     {
-                        SWSS_LOG_INFO("Creating counters for %s %zd", port_name.c_str(), ind);
-                        gPortsOrch->createPortBufferQueueCounters(port, queues);
-                    }
-                    else if (queueContext.counter_was_added && !queueContext.counter_needs_to_add &&
-                                (flexCounterOrch->getQueueCountersState() || flexCounterOrch->getQueueWatermarkCountersState()))
-                    {
-                        SWSS_LOG_INFO("Removing counters for %s %zd", port_name.c_str(), ind);
-                        gPortsOrch->removePortBufferQueueCounters(port, queues);
+                        auto queues = tokens[1];
+                        if (!queueContext.counter_was_added && queueContext.counter_needs_to_add &&
+                            (flexCounterOrch->getQueueCountersState() || flexCounterOrch->getQueueWatermarkCountersState()))
+                        {
+                            SWSS_LOG_INFO("Creating counters for %s %zd", port_name.c_str(), ind);
+                            gPortsOrch->createPortBufferQueueCounters(port, queues);
+                        }
+                        else if (queueContext.counter_was_added && !queueContext.counter_needs_to_add &&
+                                    (flexCounterOrch->getQueueCountersState() || flexCounterOrch->getQueueWatermarkCountersState()))
+                        {
+                            SWSS_LOG_INFO("Removing counters for %s %zd", port_name.c_str(), ind);
+                            gPortsOrch->removePortBufferQueueCounters(port, queues);
+                        }
                     }
                 }
             }
@@ -1471,18 +1474,21 @@ task_process_status BufferOrch::processPriorityGroupPost(const PriorityGroupTask
                 else
                 {
                     auto flexCounterOrch = gDirectory.get<FlexCounterOrch*>();
-                    auto pgs = tokens[1];
-                    if (!pg.counter_was_added && pg.counter_needs_to_add &&
-                        (flexCounterOrch->getPgCountersState() || flexCounterOrch->getPgWatermarkCountersState()))
-                    {
-                        SWSS_LOG_INFO("Creating counters for priority group %s %zd", port_name.c_str(), ind);
-                        gPortsOrch->createPortBufferPgCounters(port, pgs);
-                    }
-                    else if (pg.counter_was_added && !pg.counter_needs_to_add &&
-                                (flexCounterOrch->getPgCountersState() || flexCounterOrch->getPgWatermarkCountersState()))
+                    if (flexCounterOrch->isCreateOnlyConfigDbBuffers())
                     {
-                        SWSS_LOG_INFO("Removing counters for priority group %s %zd", port_name.c_str(), ind);
-                        gPortsOrch->removePortBufferPgCounters(port, pgs);
+                        auto pgs = tokens[1];
+                        if (!pg.counter_was_added && pg.counter_needs_to_add &&
+                            (flexCounterOrch->getPgCountersState() || flexCounterOrch->getPgWatermarkCountersState()))
+                        {
+                            SWSS_LOG_INFO("Creating counters for priority group %s %zd", port_name.c_str(), ind);
+                            gPortsOrch->createPortBufferPgCounters(port, pgs);
+                        }
+                        else if (pg.counter_was_added && !pg.counter_needs_to_add &&
+                                    (flexCounterOrch->getPgCountersState() || flexCounterOrch->getPgWatermarkCountersState()))
+                        {
+                            SWSS_LOG_INFO("Removing counters for priority group %s %zd", port_name.c_str(), ind);
+                            gPortsOrch->removePortBufferPgCounters(port, pgs);
+                        }
                     }
                 }
             }

orchagent/flexcounterorch.cpp

@@ -98,6 +98,24 @@ FlexCounterOrch::FlexCounterOrch(DBConnector *db, vector<string> &tableNames):
     m_deviceMetadataConfigTable(db, CFG_DEVICE_METADATA_TABLE_NAME)
 {
     SWSS_LOG_ENTER();
+
+    // Read create_only_config_db_buffers configuration once during initialization
+    std::string createOnlyConfigDbBuffersValue;
+    try
+    {
+        if (m_deviceMetadataConfigTable.hget("localhost", "create_only_config_db_buffers", createOnlyConfigDbBuffersValue))
+        {
+            if (createOnlyConfigDbBuffersValue == "true")
+            {
+                m_createOnlyConfigDbBuffers = true;
+            }
+        }
+    }
+    catch(const std::system_error& e)
+    {
+        SWSS_LOG_ERROR("System error reading create_only_config_db_buffers: %s", e.what());
+    }
+
     m_delayTimer = std::make_unique<SelectableTimer>(timespec{.tv_sec = FLEX_COUNTER_DELAY_SEC, .tv_nsec = 0});
     if (WarmStart::isWarmStart())
     {
@@ -120,6 +138,13 @@ void FlexCounterOrch::doTask(Consumer &consumer)
 {
     SWSS_LOG_ENTER();
 
+    // Handle DEVICE_METADATA table changes for create_only_config_db_buffers
+    if (consumer.getTableName() == CFG_DEVICE_METADATA_TABLE_NAME)
+    {
+        handleDeviceMetadataTable(consumer);
+        return;
+    }
+
     if (!m_delayTimerExpired)
     {
         return;
@@ -394,39 +419,59 @@ bool FlexCounterOrch::getWredPortCountersState() const
     return m_wred_port_counter_enabled;
 }
 
-bool FlexCounterOrch::bake()
+bool FlexCounterOrch::isCreateOnlyConfigDbBuffers() const
 {
-    /*
-     * bake is called during warmreboot reconciling procedure.
-     * By default, it should fetch items from the tables the sub agents listen to,
-     * and then push them into m_toSync of each sub agent.
-     * The motivation is to make sub agents handle the saved entries first and then handle the upcoming entries.
-     * The FCs are not data plane configuration required during reconciling process, hence don't do anything in bake.
-     */
-
-    return true;
+    return m_createOnlyConfigDbBuffers;
 }
 
-static bool isCreateOnlyConfigDbBuffers(Table& deviceMetadataConfigTable)
+void FlexCounterOrch::handleDeviceMetadataTable(Consumer &consumer)
 {
-    std::string createOnlyConfigDbBuffersValue;
+    SWSS_LOG_ENTER();
 
-    try
+    auto it = consumer.m_toSync.begin();
+    while (it != consumer.m_toSync.end())
     {
-        if (deviceMetadataConfigTable.hget("localhost", "create_only_config_db_buffers", createOnlyConfigDbBuffersValue))
+        KeyOpFieldsValuesTuple t = it->second;
+        string key = kfvKey(t);
+        string op = kfvOp(t);
+        auto data = kfvFieldsValues(t);
+
+        // Only process localhost entries
+        if (key == "localhost" && op == SET_COMMAND)
         {
-            if (createOnlyConfigDbBuffersValue == "true")
+            for (auto valuePair : data)
             {
-                return true;
+                const auto &field = fvField(valuePair);
+                const auto &value = fvValue(valuePair);
+
+                if (field == "create_only_config_db_buffers")
+                {
+                    bool newValue = (value == "true");
+                    if (m_createOnlyConfigDbBuffers != newValue)
+                    {
+                        SWSS_LOG_NOTICE("Updating create_only_config_db_buffers from %s to %s",
+                                       m_createOnlyConfigDbBuffers ? "true" : "false",
+                                       value.c_str());
+                        m_createOnlyConfigDbBuffers = newValue;
+                    }
+                }
             }
         }
+        consumer.m_toSync.erase(it++);
     }
-    catch(const std::system_error& e)
-    {
-        SWSS_LOG_ERROR("System error: %s", e.what());
-    }
+}
 
-    return false;
+bool FlexCounterOrch::bake()
+{
+    /*
+     * bake is called during warmreboot reconciling procedure.
+     * By default, it should fetch items from the tables the sub agents listen to,
+     * and then push them into m_toSync of each sub agent.
+     * The motivation is to make sub agents handle the saved entries first and then handle the upcoming entries.
+     * The FCs are not data plane configuration required during reconciling process, hence don't do anything in bake.
+     */
+
+    return true;
 }
 
 map<string, FlexCounterQueueStates> FlexCounterOrch::getQueueConfigurations()
@@ -435,7 +480,7 @@ map<string, FlexCounterQueueStates> FlexCounterOrch::getQueueConfigurations()
 
     map<string, FlexCounterQueueStates> queuesStateVector;
 
-    if (!isCreateOnlyConfigDbBuffers(m_deviceMetadataConfigTable))
+    if (!isCreateOnlyConfigDbBuffers())
     {
         FlexCounterQueueStates flexCounterQueueState(0);
         queuesStateVector.insert(make_pair(createAllAvailableBuffersStr, flexCounterQueueState));
@@ -504,7 +549,7 @@ map<string, FlexCounterPgStates> FlexCounterOrch::getPgConfigurations()
 
     map<string, FlexCounterPgStates> pgsStateVector;
 
-    if (!isCreateOnlyConfigDbBuffers(m_deviceMetadataConfigTable))
+    if (!isCreateOnlyConfigDbBuffers())
     {
         FlexCounterPgStates flexCounterPgState(0);
         pgsStateVector.insert(make_pair(createAllAvailableBuffersStr, flexCounterPgState));

orchagent/flexcounterorch.h

@@ -56,9 +56,11 @@ class FlexCounterOrch: public Orch
     bool getRouteFlowCountersState() const {return m_route_flow_counter_enabled;}
     bool getWredQueueCountersState() const;
     bool getWredPortCountersState() const;
+    bool isCreateOnlyConfigDbBuffers() const;
     bool bake() override;
 
 private:
+    void handleDeviceMetadataTable(Consumer &consumer);
     bool m_port_counter_enabled = false;
     bool m_port_buffer_drop_counter_enabled = false;
     bool m_queue_enabled = false;
@@ -76,6 +78,8 @@ class FlexCounterOrch: public Orch
     std::unique_ptr<SelectableTimer> m_delayTimer;
     std::unique_ptr<Executor> m_delayExecutor;
     std::unordered_set<std::string> m_groupsWithBulkChunkSize;
+
+    bool m_createOnlyConfigDbBuffers = false;
 };
 
 #endif

orchagent/orchdaemon.cpp

@@ -592,7 +592,8 @@ bool OrchDaemon::init()
     }
 
     vector<string> flex_counter_tables = {
-        CFG_FLEX_COUNTER_TABLE_NAME
+        CFG_FLEX_COUNTER_TABLE_NAME,
+        CFG_DEVICE_METADATA_TABLE_NAME
     };
 
     auto* flexCounterOrch = new FlexCounterOrch(m_configDb, flex_counter_tables);