[watermarkorch] add watermarkorch, extend queue and pg counters with …

…wat… (#629)

* [watermarks] add watermarkorch, extend queue and pg counters with watermark stats

Signed-off-by: Mykola Faryma <mykolaf@mellanox.com>

* [watermark] add VS tests

Signed-off-by: Mykola Faryma <mykolaf@mellanox.com>

* [watermark] resolve conflict

Signed-off-by: Mykola Faryma <mykolaf@mellanox.com>

* address review comments

Signed-off-by: Mykola Faryma <mykolaf@mellanox.com>

* update VS test to use set counters and cover the flex counter flow

Signed-off-by: Mykola Faryma <mykolaf@mellanox.com>

orchagent/Makefile.am

@@ -8,7 +8,9 @@ dist_swss_DATA = \
 		 pfc_detect_mellanox.lua  \
 		 pfc_detect_broadcom.lua \
 		 pfc_detect_barefoot.lua \
-		 pfc_restore.lua
+		 pfc_restore.lua \
+		 watermark_queue.lua \
+		 watermark_pg.lua
 
 bin_PROGRAMS = orchagent routeresync orchagent_restart_check
 
@@ -46,6 +48,7 @@ orchagent_SOURCES = \
             vnetorch.cpp \
             dtelorch.cpp \
             flexcounterorch.cpp \
+            watermarkorch.cpp  \
             acltable.h \
             aclorch.h \
             bufferorch.h \
@@ -76,7 +79,8 @@ orchagent_SOURCES = \
             countercheckorch.h \
             vxlanorch.h \
             vnetorch.h \
-            flexcounterorch.h
+            flexcounterorch.h   \
+            watermarkorch.h
 
 orchagent_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
 orchagent_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)

orchagent/flexcounterorch.cpp

@@ -16,6 +16,8 @@ unordered_map<string, string> flexCounterGroupMap =
     {"PORT", PORT_STAT_COUNTER_FLEX_COUNTER_GROUP},
     {"QUEUE", QUEUE_STAT_COUNTER_FLEX_COUNTER_GROUP},
     {"PFCWD", PFC_WD_FLEX_COUNTER_GROUP},
+    {"QUEUE_WATERMARK", QUEUE_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP},
+    {"PG_WATERMARK", PG_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP},
 };
 
 
@@ -75,6 +77,7 @@ void FlexCounterOrch::doTask(Consumer &consumer)
                     // Currently the counters are disabled by default
                     // The queue maps will be generated as soon as counters are enabled
                     gPortsOrch->generateQueueMap();
+                    gPortsOrch->generatePriorityGroupMap();
 
                     vector<FieldValueTuple> fieldValues;
                     fieldValues.emplace_back(FLEX_COUNTER_STATUS_FIELD, value);

orchagent/orchdaemon.cpp

@@ -138,6 +138,8 @@ bool OrchDaemon::init()
         CFG_DTEL_EVENT_TABLE_NAME
     };
 
+    WatermarkOrch *wm_orch = new WatermarkOrch(m_configDb, CFG_WATERMARK_TABLE_NAME);
+
     /*
      * The order of the orch list is important for state restore of warm start and
      * the queued processing in m_toSync map after gPortsOrch->isInitDone() is set.
@@ -146,7 +148,8 @@ bool OrchDaemon::init()
      * when iterating ConsumerMap.
      * That is ensured implicitly by the order of map key, "LAG_TABLE" is smaller than "VLAN_TABLE" in lexicographic order.
      */
-    m_orchList = { gSwitchOrch, gCrmOrch, gBufferOrch, gPortsOrch, gIntfsOrch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch};
+    m_orchList = { gSwitchOrch, gCrmOrch, gBufferOrch, gPortsOrch, gIntfsOrch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch, wm_orch };
+
 
     bool initialize_dtel = false;
     if (platform == BFN_PLATFORM_SUBSTRING || platform == VS_PLATFORM_SUBSTRING)

orchagent/orchdaemon.h

@@ -25,6 +25,7 @@
 #include "vnetorch.h"
 #include "countercheckorch.h"
 #include "flexcounterorch.h"
+#include "watermarkorch.h"
 #include "directory.h"
 
 using namespace swss;

orchagent/pfcwdorch.cpp

@@ -650,6 +650,7 @@ PfcWdSwOrch<DropHandler, ForwardHandler>::PfcWdSwOrch(
         vector<FieldValueTuple> fieldValues;
         fieldValues.emplace_back(QUEUE_PLUGIN_FIELD, detectSha + "," + restoreSha);
         fieldValues.emplace_back(POLL_INTERVAL_FIELD, to_string(m_pollInterval));
+        fieldValues.emplace_back(STATS_MODE_FIELD, STATS_MODE_READ);
         m_flexCounterGroupTable->set(PFC_WD_FLEX_COUNTER_GROUP, fieldValues);
     }
     catch (...)

orchagent/portsorch.cpp

@@ -16,6 +16,7 @@
 
 #include "logger.h"
 #include "schema.h"
+#include "redisapi.h"
 #include "converter.h"
 #include "sai_serialize.h"
 #include "crmorch.h"
@@ -40,6 +41,9 @@ extern BufferOrch *gBufferOrch;
 #define DEFAULT_VLAN_ID     1
 #define PORT_FLEX_STAT_COUNTER_POLL_MSECS "1000"
 #define QUEUE_FLEX_STAT_COUNTER_POLL_MSECS "10000"
+#define QUEUE_WATERMARK_FLEX_STAT_COUNTER_POLL_MSECS "1000"
+#define PG_WATERMARK_FLEX_STAT_COUNTER_POLL_MSECS "1000"
+
 
 static map<string, sai_port_fec_mode_t> fec_mode_map =
 {
@@ -105,6 +109,17 @@ static const vector<sai_queue_stat_t> queueStatIds =
     SAI_QUEUE_STAT_DROPPED_BYTES,
 };
 
+static const vector<sai_queue_stat_t> queueWatermarkStatIds =
+{
+    SAI_QUEUE_STAT_SHARED_WATERMARK_BYTES,
+};
+
+static const vector<sai_ingress_priority_group_stat_t> ingressPriorityGroupWatermarkStatIds =
+{
+    SAI_INGRESS_PRIORITY_GROUP_STAT_XOFF_ROOM_WATERMARK_BYTES,
+    SAI_INGRESS_PRIORITY_GROUP_STAT_SHARED_WATERMARK_BYTES,
+};
+
 static char* hostif_vlan_tag[] = {
     [SAI_HOSTIF_VLAN_TAG_STRIP]     = "SAI_HOSTIF_VLAN_TAG_STRIP",
     [SAI_HOSTIF_VLAN_TAG_KEEP]      = "SAI_HOSTIF_VLAN_TAG_KEEP",
@@ -141,17 +156,53 @@ PortsOrch::PortsOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames)
     m_queueIndexTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_QUEUE_INDEX_MAP));
     m_queueTypeTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_QUEUE_TYPE_MAP));
 
+    /* Initialize ingress priority group tables */
+    m_pgTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_PG_NAME_MAP));
+    m_pgPortTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_PG_PORT_MAP));
+    m_pgIndexTable = unique_ptr<Table>(new Table(m_counter_db.get(), COUNTERS_PG_INDEX_MAP));
+
     m_flex_db = shared_ptr<DBConnector>(new DBConnector(FLEX_COUNTER_DB, DBConnector::DEFAULT_UNIXSOCKET, 0));
     m_flexCounterTable = unique_ptr<ProducerTable>(new ProducerTable(m_flex_db.get(), FLEX_COUNTER_TABLE));
     m_flexCounterGroupTable = unique_ptr<ProducerTable>(new ProducerTable(m_flex_db.get(), FLEX_COUNTER_GROUP_TABLE));
 
     vector<FieldValueTuple> fields;
     fields.emplace_back(POLL_INTERVAL_FIELD, PORT_FLEX_STAT_COUNTER_POLL_MSECS);
+    fields.emplace_back(STATS_MODE_FIELD, STATS_MODE_READ);
     m_flexCounterGroupTable->set(PORT_STAT_COUNTER_FLEX_COUNTER_GROUP, fields);
 
     fields.emplace_back(POLL_INTERVAL_FIELD, QUEUE_FLEX_STAT_COUNTER_POLL_MSECS);
+    fields.emplace_back(STATS_MODE_FIELD, STATS_MODE_READ);
     m_flexCounterGroupTable->set(QUEUE_STAT_COUNTER_FLEX_COUNTER_GROUP, fields);
 
+    string queueWmSha, pgWmSha;
+    string queueWmPluginName = "watermark_queue.lua";
+    string pgWmPluginName = "watermark_pg.lua";
+
+    try
+    {
+        string queueLuaScript = swss::loadLuaScript(queueWmPluginName);
+        queueWmSha = swss::loadRedisScript(m_counter_db.get(), queueLuaScript);
+
+        string pgLuaScript = swss::loadLuaScript(pgWmPluginName);
+        pgWmSha = swss::loadRedisScript(m_counter_db.get(), pgLuaScript);
+
+        vector<FieldValueTuple> fieldValues;
+        fieldValues.emplace_back(QUEUE_PLUGIN_FIELD, queueWmSha);
+        fieldValues.emplace_back(POLL_INTERVAL_FIELD, QUEUE_WATERMARK_FLEX_STAT_COUNTER_POLL_MSECS);
+        fieldValues.emplace_back(STATS_MODE_FIELD, STATS_MODE_READ_AND_CLEAR);
+        m_flexCounterGroupTable->set(QUEUE_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP, fieldValues);
+
+        fieldValues.clear();
+        fieldValues.emplace_back(PG_PLUGIN_FIELD, pgWmSha);
+        fieldValues.emplace_back(POLL_INTERVAL_FIELD, PG_WATERMARK_FLEX_STAT_COUNTER_POLL_MSECS);
+        fieldValues.emplace_back(STATS_MODE_FIELD, STATS_MODE_READ_AND_CLEAR);
+        m_flexCounterGroupTable->set(PG_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP, fieldValues);
+    }
+    catch (...)
+    {
+        SWSS_LOG_WARN("Watermark flex counter groups were not set successfully");
+    }
+
     uint32_t i, j;
     sai_status_t status;
     sai_attribute_t attr;
@@ -1248,6 +1299,16 @@ string PortsOrch::getQueueFlexCounterTableKey(string key)
     return string(QUEUE_STAT_COUNTER_FLEX_COUNTER_GROUP) + ":" + key;
 }
 
+string PortsOrch::getQueueWatermarkFlexCounterTableKey(string key)
+{
+    return string(QUEUE_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP) + ":" + key;
+}
+
+string PortsOrch::getPriorityGroupWatermarkFlexCounterTableKey(string key)
+{
+    return string(PG_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP) + ":" + key;
+}
+
 bool PortsOrch::initPort(const string &alias, const set<int> &lane_set)
 {
     SWSS_LOG_ENTER();
@@ -1287,7 +1348,7 @@ bool PortsOrch::initPort(const string &alias, const set<int> &lane_set)
                 for (const auto &id: portStatIds)
                 {
                     counters_stream << delimiter << sai_serialize_port_stat(id);
-                    delimiter = ",";
+                    delimiter = comma;
                 }
 
                 fields.clear();
@@ -2873,20 +2934,37 @@ void PortsOrch::generateQueueMapPerPort(const Port& port)
             queueIndexVector.emplace_back(id, to_string(queueRealIndex));
         }
 
+        /* add ordinary Queue stat counters */
         string key = getQueueFlexCounterTableKey(id);
 
         std::string delimiter = "";
         std::ostringstream counters_stream;
         for (const auto& it: queueStatIds)
         {
             counters_stream << delimiter << sai_serialize_queue_stat(it);
-            delimiter = ",";
+            delimiter = comma;
         }
 
         vector<FieldValueTuple> fieldValues;
         fieldValues.emplace_back(QUEUE_COUNTER_ID_LIST, counters_stream.str());
 
         m_flexCounterTable->set(key, fieldValues);
+
+        /* add watermark queue counters */
+        key = getQueueWatermarkFlexCounterTableKey(id);
+
+        delimiter = "";
+        counters_stream.str("");
+        for (const auto& it: queueWatermarkStatIds)
+        {
+            counters_stream << delimiter << sai_serialize_queue_stat(it);
+            delimiter = comma;
+        }
+
+        fieldValues.clear();
+        fieldValues.emplace_back(QUEUE_COUNTER_ID_LIST, counters_stream.str());
+
+        m_flexCounterTable->set(key, fieldValues);
     }
 
     m_queueTable->set("", queueVector);
@@ -2897,6 +2975,67 @@ void PortsOrch::generateQueueMapPerPort(const Port& port)
     CounterCheckOrch::getInstance().addPort(port);
 }
 
+void PortsOrch::generatePriorityGroupMap()
+{
+    if (m_isPriorityGroupMapGenerated)
+    {
+        return;
+    }
+
+    for (const auto& it: m_portList)
+    {
+        if (it.second.m_type == Port::PHY)
+        {
+            generatePriorityGroupMapPerPort(it.second);
+        }
+    }
+
+    m_isPriorityGroupMapGenerated = true;
+}
+
+void PortsOrch::generatePriorityGroupMapPerPort(const Port& port)
+{
+    /* Create the PG map in the Counter DB */
+    /* Add stat counters to flex_counter */
+    vector<FieldValueTuple> pgVector;
+    vector<FieldValueTuple> pgPortVector;
+    vector<FieldValueTuple> pgIndexVector;
+
+    for (size_t pgIndex = 0; pgIndex < port.m_priority_group_ids.size(); ++pgIndex)
+    {
+        std::ostringstream name;
+        name << port.m_alias << ":" << pgIndex;
+
+        const auto id = sai_serialize_object_id(port.m_priority_group_ids[pgIndex]);
+
+        pgVector.emplace_back(name.str(), id);
+        pgPortVector.emplace_back(id, sai_serialize_object_id(port.m_port_id));
+        pgIndexVector.emplace_back(id, to_string(pgIndex));
+
+        string key = getPriorityGroupWatermarkFlexCounterTableKey(id);
+
+        std::string delimiter = "";
+        std::ostringstream counters_stream;
+        /* Add watermark counters to flex_counter */
+        for (const auto& it: ingressPriorityGroupWatermarkStatIds)
+        {
+            counters_stream << delimiter << sai_serialize_ingress_priority_group_stat(it);
+            delimiter = comma;
+        }
+
+        vector<FieldValueTuple> fieldValues;
+        fieldValues.emplace_back(PG_COUNTER_ID_LIST, counters_stream.str());
+
+        m_flexCounterTable->set(key, fieldValues);
+    }
+
+    m_pgTable->set("", pgVector);
+    m_pgPortTable->set("", pgPortVector);
+    m_pgIndexTable->set("", pgIndexVector);
+
+    CounterCheckOrch::getInstance().addPort(port);
+}
+
 void PortsOrch::doTask(NotificationConsumer &consumer)
 {
     SWSS_LOG_ENTER();

orchagent/portsorch.h

@@ -14,6 +14,8 @@
 #define VLAN_TAG_LEN 4
 #define PORT_STAT_COUNTER_FLEX_COUNTER_GROUP "PORT_STAT_COUNTER"
 #define QUEUE_STAT_COUNTER_FLEX_COUNTER_GROUP "QUEUE_STAT_COUNTER"
+#define QUEUE_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP "QUEUE_WATERMARK_STAT_COUNTER"
+#define PG_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP "PG_WATERMARK_STAT_COUNTER"
 
 
 typedef std::vector<sai_uint32_t> PortSupportedSpeeds;
@@ -76,19 +78,27 @@ class PortsOrch : public Orch, public Subject
     bool setPortPfc(sai_object_id_t portId, uint8_t pfc_bitmask);
 
     void generateQueueMap();
+    void generatePriorityGroupMap();
+
     void refreshPortStatus();
+
 private:
     unique_ptr<Table> m_counterTable;
     unique_ptr<Table> m_portTable;
     unique_ptr<Table> m_queueTable;
     unique_ptr<Table> m_queuePortTable;
     unique_ptr<Table> m_queueIndexTable;
     unique_ptr<Table> m_queueTypeTable;
+    unique_ptr<Table> m_pgTable;
+    unique_ptr<Table> m_pgPortTable;
+    unique_ptr<Table> m_pgIndexTable;
     unique_ptr<ProducerTable> m_flexCounterTable;
     unique_ptr<ProducerTable> m_flexCounterGroupTable;
 
     std::string getQueueFlexCounterTableKey(std::string s);
+    std::string getQueueWatermarkFlexCounterTableKey(std::string s);
     std::string getPortFlexCounterTableKey(std::string s);
+    std::string getPriorityGroupWatermarkFlexCounterTableKey(std::string s);
 
     shared_ptr<DBConnector> m_counter_db;
     shared_ptr<DBConnector> m_flex_db;
@@ -166,6 +176,9 @@ class PortsOrch : public Orch, public Subject
     bool m_isQueueMapGenerated = false;
     void generateQueueMapPerPort(const Port& port);
 
+    bool m_isPriorityGroupMapGenerated = false;
+    void generatePriorityGroupMapPerPort(const Port& port);
+
     bool setPortAutoNeg(sai_object_id_t id, int an);
     bool setPortFecMode(sai_object_id_t id, int fec);