Optimize counter initialization by reducing the number of bulk counter poll calls and communication between swss/sairedis (#49)

<!--
Please make sure you have read and understood the contribution guildlines:
https://github.com/Azure/SONiC/blob/gh-pages/CONTRIBUTING.md

1. Make sure your commit includes a signature generted with `git commit -s`
2. Make sure your commit title follows the correct format: [component]: description
3. Make sure your commit message contains enough details about the change and related tests
4. Make sure your pull request adds related reviewers, asignees, labels

Please also provide the following information in this pull request:
-->

**What I did**

Optimize counter initialization by reducing the number of bulk counter poll calls and communication between swss(orchagent)/sairedis(syncd) during initialization.

Originally, `orchagent` notifies `syncd` to initialize the counter using an extended sairedis call `SAI_REDIS_SWITCH_ATTR_FLEX_COUNTER` for each SAI object with the object ID as the key, which means the number of the extended sairedis calls is identical as the number of objects. It takes time to finish all the extended sairedis calls.

Now, for counter groups that have many objects (e.g., port, PG, queues, etc), `orchagent` notifies `syncd` to initialize the counter using a single extend sairedis call with many objects' ID as the key (format: `<key1>,<key2>,...<keyn>`). So, it takes much less time to initialize the counters because fewer extend sairedis calls are required.

HLD sonic-net/SONiC#1862
Depends on sonic-net/sonic-sairedis#1527

**Why I did it**

**How I verified it**

Unit test (mock test) and regression.

**Details if related**
- In sairedis, the bulk counter is supported for all counter groups except `Buffer Pool Counter` and `DASH ENI counter`.
- In swss, bulk counter for the following counter groups

 - priority group watermark
 - priority group drop
 - queue watermark
 - queue stat
 - PFC watchdog
 - WRED/ECN counter

Author: mssonicbld

orchagent/bufferorch.cpp

@@ -1858,6 +1858,7 @@ void BufferOrch::doTask()
             continue;
         consumer->drain();
     }
+    gPortsOrch->flushCounters();
 }
 
 void BufferOrch::doTask(Consumer &consumer)
@@ -1921,4 +1922,6 @@ void BufferOrch::doTask(Consumer &consumer)
     {
         (this->*(m_bufferFlushHandlerMap[map_type_name]))(consumer);
     }
+
+    gPortsOrch->flushCounters();
 }

orchagent/flex_counter/flex_counter_manager.cpp

@@ -5,7 +5,6 @@
 #include "schema.h"
 #include "rediscommand.h"
 #include "logger.h"
-#include "sai_serialize.h"
 
 #include <macsecorch.h>
 
@@ -20,14 +19,13 @@ using swss::ProducerTable;
 
 extern sai_switch_api_t *sai_switch_api;
 
-extern sai_object_id_t gSwitchId;
-
 const string FLEX_COUNTER_ENABLE("enable");
 const string FLEX_COUNTER_DISABLE("disable");
 
 const unordered_map<StatsMode, string> FlexCounterManager::stats_mode_lookup =
 {
     { StatsMode::READ, STATS_MODE_READ },
+    { StatsMode::READ_AND_CLEAR, STATS_MODE_READ_AND_CLEAR },
 };
 
 const unordered_map<bool, string> FlexCounterManager::status_lookup =
@@ -42,6 +40,8 @@ const unordered_map<CounterType, string> FlexCounterManager::counter_id_field_lo
     { CounterType::SWITCH_DEBUG,    SWITCH_DEBUG_COUNTER_ID_LIST },
     { CounterType::PORT,            PORT_COUNTER_ID_LIST },
     { CounterType::QUEUE,           QUEUE_COUNTER_ID_LIST },
+    { CounterType::QUEUE_ATTR,      QUEUE_ATTR_ID_LIST },
+    { CounterType::PRIORITY_GROUP,  PG_COUNTER_ID_LIST },
     { CounterType::MACSEC_SA_ATTR,  MACSEC_SA_ATTR_ID_LIST },
     { CounterType::MACSEC_SA,       MACSEC_SA_COUNTER_ID_LIST },
     { CounterType::MACSEC_FLOW,     MACSEC_FLOW_COUNTER_ID_LIST },
@@ -259,7 +259,7 @@ string FlexCounterManager::getFlexCounterTableKey(
 
 // serializeCounterStats turns a set of stats into a format suitable for FLEX_COUNTER_DB.
 string FlexCounterManager::serializeCounterStats(
-        const unordered_set<string>& counter_stats) const
+        const unordered_set<string>& counter_stats)
 {
     SWSS_LOG_ENTER();
 

orchagent/flex_counter/flex_counter_manager.h

@@ -9,20 +9,26 @@
 #include "producertable.h"
 #include "table.h"
 #include <inttypes.h>
+#include <type_traits>
+#include "sai_serialize.h"
+#include "saihelper.h"
 
 extern "C" {
 #include "sai.h"
 }
 
 enum class StatsMode
 {
-    READ
+    READ,
+    READ_AND_CLEAR
 };
 
 enum class CounterType
 {
     PORT,
     QUEUE,
+    QUEUE_ATTR,
+    PRIORITY_GROUP,
     PORT_DEBUG,
     SWITCH_DEBUG,
     MACSEC_SA_ATTR,
@@ -35,13 +41,18 @@ enum class CounterType
     ENI
 };
 
+extern bool gTraditionalFlexCounter;
+extern sai_object_id_t gSwitchId;
+
+struct CachedObjects;
 // FlexCounterManager allows users to manage a group of flex counters.
 //
 // TODO: FlexCounterManager doesn't currently support the full range of
 // flex counter features. In particular, support for standard (i.e. non-debug)
 // counters and support for plugins needs to be added.
 class FlexCounterManager
 {
+    friend struct CachedObjects;
     public:
         FlexCounterManager(
                 const std::string& group_name,
@@ -69,12 +80,12 @@ class FlexCounterManager
         void enableFlexCounterGroup();
         void disableFlexCounterGroup();
 
-        void setCounterIdList(
+        virtual void setCounterIdList(
                 const sai_object_id_t object_id,
                 const CounterType counter_type,
                 const std::unordered_set<std::string>& counter_stats,
                 const sai_object_id_t switch_id=SAI_NULL_OBJECT_ID);
-        void clearCounterIdList(const sai_object_id_t object_id);
+        virtual void clearCounterIdList(const sai_object_id_t object_id);
 
         const std::string& getGroupName() const
         {
@@ -99,12 +110,9 @@ class FlexCounterManager
     protected:
         void applyGroupConfiguration();
 
-    private:
         std::string getFlexCounterTableKey(
                 const std::string& group_name,
                 const sai_object_id_t object_id) const;
-        std::string serializeCounterStats(
-                const std::unordered_set<std::string>& counter_stats) const;
 
         std::string group_name;
         StatsMode stats_mode;
@@ -114,11 +122,235 @@ class FlexCounterManager
         std::unordered_map<sai_object_id_t, sai_object_id_t> installed_counters;
         bool is_gearbox;
 
+        static std::string serializeCounterStats(
+                const std::unordered_set<std::string>& counter_stats);
+
         static const std::unordered_map<StatsMode, std::string> stats_mode_lookup;
         static const std::unordered_map<bool, std::string> status_lookup;
         static const std::unordered_map<CounterType, std::string> counter_id_field_lookup;
 };
 
+struct CachedObjects
+{
+    CounterType pending_counter_type;
+    sai_object_id_t pending_switch_id;
+    std::unordered_set<std::string> pending_counter_stats;
+    std::unordered_set<sai_object_id_t> pending_sai_objects;
+
+    bool try_cache(const sai_object_id_t object_id,
+                   const CounterType counter_type,
+                   const std::unordered_set<std::string>& counter_stats,
+                   sai_object_id_t switch_id)
+    {
+        if (pending_sai_objects.empty())
+        {
+            pending_counter_type = counter_type;
+            pending_switch_id = switch_id;
+            // Just to avoid recreating counter IDs
+            if (pending_counter_stats != counter_stats)
+            {
+                pending_counter_stats = counter_stats;
+            }
+        }
+        else if (counter_type != pending_counter_type ||
+                 switch_id != pending_switch_id ||
+                 counter_stats != pending_counter_stats)
+        {
+            return false;
+        }
+
+        cache(object_id);
+
+        return true;
+    }
+
+    bool is_cached(const sai_object_id_t object_id)
+    {
+        return pending_sai_objects.find(object_id) != pending_sai_objects.end();
+    }
+
+    void flush(const std::string &group_name)
+    {
+        if (pending_sai_objects.empty())
+        {
+            return;
+        }
+
+        auto counter_ids = FlexCounterManager::serializeCounterStats(pending_counter_stats);
+        auto counter_type_it = FlexCounterManager::counter_id_field_lookup.find(pending_counter_type);
+
+        auto counter_keys = group_name + ":";
+        for (const auto& oid: pending_sai_objects)
+        {
+            counter_keys += sai_serialize_object_id(oid) + ",";
+        }
+        counter_keys.pop_back();
+
+        startFlexCounterPolling(pending_switch_id, counter_keys, counter_ids, counter_type_it->second);
+
+        pending_sai_objects.clear();
+    }
+
+    void cache(sai_object_id_t object_id)
+    {
+        pending_sai_objects.emplace(object_id);
+    }
+};
+
+class FlexCounterCachedManager : public FlexCounterManager
+{
+    public:
+        FlexCounterCachedManager(
+                const std::string& group_name,
+                const StatsMode stats_mode,
+                const uint polling_interval,
+                const bool enabled,
+                swss::FieldValueTuple fv_plugin = std::make_pair("","")) :
+            FlexCounterManager(group_name, stats_mode, polling_interval, enabled, fv_plugin)
+        {
+        }
+
+        virtual void flush()
+        {
+        }
+
+    protected:
+        void flush(const std::string &group_name, struct CachedObjects &cached_objects)
+        {
+            cached_objects.flush(group_name);
+        }
+
+        void setCounterIdList(
+            struct CachedObjects &cached_objects,
+            const sai_object_id_t object_id,
+            const CounterType counter_type,
+            const std::unordered_set<std::string>& counter_stats,
+            const sai_object_id_t switch_id=SAI_NULL_OBJECT_ID)
+        {
+            if (gTraditionalFlexCounter)
+            {
+                // Unable to cache an object and initialize in bulk in traditional flex counter mode
+                FlexCounterManager::setCounterIdList(object_id, counter_type, counter_stats, switch_id);
+                return;
+            }
+
+            auto effective_switch_id = switch_id == SAI_NULL_OBJECT_ID ? gSwitchId : switch_id;
+            installed_counters[object_id] = effective_switch_id;
+            if (cached_objects.try_cache(object_id, counter_type, counter_stats, effective_switch_id))
+            {
+                return;
+            }
+            else
+            {
+                flush(group_name, cached_objects);
+                cached_objects.cache(object_id);
+            }
+        }
+
+        void clearCounterIdList(
+            struct CachedObjects &cached_objects,
+            const sai_object_id_t object_id)
+        {
+            auto search = cached_objects.pending_sai_objects.find(object_id);
+            if (search == cached_objects.pending_sai_objects.end())
+            {
+                FlexCounterManager::clearCounterIdList(object_id);
+            }
+            else
+            {
+                installed_counters.erase(object_id);
+                cached_objects.pending_sai_objects.erase(search);
+            }
+        }
+};
+
+template <typename TagType, typename Enable=void>
+class FlexCounterTaggedCachedManager : public FlexCounterCachedManager
+{
+    public:
+        FlexCounterTaggedCachedManager(
+                const std::string& group_name,
+                const StatsMode stats_mode,
+                const uint polling_interval,
+                const bool enabled,
+                swss::FieldValueTuple fv_plugin = std::make_pair("","")) :
+            FlexCounterCachedManager(group_name, stats_mode, polling_interval, enabled, fv_plugin)
+        {
+        }
+
+        void flush()
+        {
+            FlexCounterCachedManager::flush(group_name, cached_objects);
+        }
+
+        virtual void setCounterIdList(
+            const sai_object_id_t object_id,
+            const CounterType counter_type,
+            const std::unordered_set<std::string>& counter_stats,
+            const sai_object_id_t switch_id=SAI_NULL_OBJECT_ID)
+        {
+            FlexCounterCachedManager::setCounterIdList(cached_objects,
+                                                       object_id,
+                                                       counter_type,
+                                                       counter_stats);
+        }
+
+        virtual void clearCounterIdList(
+            const sai_object_id_t object_id)
+        {
+            FlexCounterCachedManager::clearCounterIdList(cached_objects, object_id);
+        }
+
+    private:
+        struct CachedObjects cached_objects;
+};
+
+template <typename TagType>
+class FlexCounterTaggedCachedManager<TagType, typename std::enable_if_t<std::is_enum<TagType>::value>> : public FlexCounterCachedManager
+{
+    public:
+        FlexCounterTaggedCachedManager(
+                const std::string& group_name,
+                const StatsMode stats_mode,
+                const uint polling_interval,
+                const bool enabled,
+                swss::FieldValueTuple fv_plugin = std::make_pair("","")) :
+            FlexCounterCachedManager(group_name, stats_mode, polling_interval, enabled, fv_plugin)
+        {
+        }
+
+        void flush()
+        {
+            for(auto &it : cached_objects)
+            {
+                FlexCounterCachedManager::flush(group_name, it.second);
+            }
+        }
+
+        void setCounterIdList(
+            const sai_object_id_t object_id,
+            const CounterType counter_type,
+            const std::unordered_set<std::string>& counter_stats,
+            const TagType tag,
+            const sai_object_id_t switch_id=SAI_NULL_OBJECT_ID)
+        {
+            FlexCounterCachedManager::setCounterIdList(cached_objects[tag],
+                                                       object_id,
+                                                       counter_type,
+                                                       counter_stats);
+        }
+
+        void clearCounterIdList(
+            const sai_object_id_t object_id,
+            const TagType tag)
+        {
+            FlexCounterCachedManager::clearCounterIdList(cached_objects[tag], object_id);
+        }
+
+    private:
+        std::map<TagType, struct CachedObjects> cached_objects;
+};
+
 class FlexManagerDirectory
 {
     public:

orchagent/flexcounterorch.cpp

@@ -271,6 +271,7 @@ void FlexCounterOrch::doTask(Consumer &consumer)
                         }
                     }
 
+                    gPortsOrch->flushCounters();
                     setFlexCounterGroupOperation(flexCounterGroupMap[key], value);
 
                     if (gPortsOrch && gPortsOrch->isGearboxEnabled())