Sflow orchagent changes

cfgmgr/Makefile.am

@@ -3,7 +3,7 @@ CFLAGS_SAI = -I /usr/include/sai
 LIBNL_CFLAGS = -I/usr/include/libnl3
 LIBNL_LIBS = -lnl-genl-3 -lnl-route-3 -lnl-3
 
-bin_PROGRAMS = vlanmgrd teammgrd portmgrd intfmgrd buffermgrd vrfmgrd nbrmgrd vxlanmgrd
+bin_PROGRAMS = vlanmgrd teammgrd portmgrd intfmgrd buffermgrd vrfmgrd nbrmgrd vxlanmgrd sflowmgrd
 
 if DEBUG
 DBGFLAGS = -ggdb -DDEBUG
@@ -49,4 +49,9 @@ nbrmgrd_LDADD = -lswsscommon $(LIBNL_LIBS)
 vxlanmgrd_SOURCES = vxlanmgrd.cpp vxlanmgr.cpp $(top_srcdir)/orchagent/orch.cpp $(top_srcdir)/orchagent/request_parser.cpp shellcmd.h
 vxlanmgrd_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
 vxlanmgrd_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
-vxlanmgrd_LDADD = -lswsscommon
+vxlanmgrd_LDADD = -lswsscommon
+
+sflowmgrd_SOURCES = sflowmgrd.cpp sflowmgr.cpp $(top_srcdir)/orchagent/orch.cpp $(top_srcdir)/orchagent/request_parser.cpp shellcmd.h
+sflowmgrd_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
+sflowmgrd_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
+sflowmgrd_LDADD = -lswsscommon

cfgmgr/sflowmgr.cpp

@@ -0,0 +1,117 @@
+#include "logger.h"
+#include "dbconnector.h"
+#include "producerstatetable.h"
+#include "tokenize.h"
+#include "ipprefix.h"
+#include "sflowmgr.h"
+#include "exec.h"
+#include "shellcmd.h"
+
+using namespace std;
+using namespace swss;
+
+SflowMgr::SflowMgr(DBConnector *cfgDb, DBConnector *appDb, const vector<string> &tableNames) :
+        Orch(cfgDb, tableNames),
+        m_cfgSflowTable(cfgDb, CFG_SFLOW_TABLE_NAME),
+        m_cfgSflowSessionTable(cfgDb, CFG_SFLOW_SESSION_TABLE_NAME),
+        m_appSflowTable(appDb, APP_SFLOW_TABLE_NAME),
+        m_appSflowSessionTable(appDb, APP_SFLOW_SESSION_TABLE_NAME),
+        m_appSflowSpeedRateTable(appDb, APP_SFLOW_SAMPLE_RATE_TABLE_NAME)
+{
+    vector<FieldValueTuple> fieldValues;
+
+    fieldValues.emplace_back(SFLOW_SAMPLE_RATE_KEY_400G, SFLOW_SAMPLE_RATE_VALUE_400G);
+    fieldValues.emplace_back(SFLOW_SAMPLE_RATE_KEY_100G, SFLOW_SAMPLE_RATE_VALUE_100G);
+    fieldValues.emplace_back(SFLOW_SAMPLE_RATE_KEY_50G, SFLOW_SAMPLE_RATE_VALUE_50G);
+    fieldValues.emplace_back(SFLOW_SAMPLE_RATE_KEY_40G, SFLOW_SAMPLE_RATE_VALUE_40G);
+    fieldValues.emplace_back(SFLOW_SAMPLE_RATE_KEY_25G, SFLOW_SAMPLE_RATE_VALUE_25G);
+    fieldValues.emplace_back(SFLOW_SAMPLE_RATE_KEY_10G, SFLOW_SAMPLE_RATE_VALUE_10G);
+    fieldValues.emplace_back(SFLOW_SAMPLE_RATE_KEY_1G, SFLOW_SAMPLE_RATE_VALUE_1G);
+
+    m_appSflowSpeedRateTable.set("global", fieldValues);
+}
+
+void SflowMgr::handleSflowTableConfig(Consumer &consumer)
+{
+    stringstream cmd;
+    string res;
+
+    auto it = consumer.m_toSync.begin();
+
+    while (it != consumer.m_toSync.end())
+    {
+        auto t = it->second;
+
+        string key = kfvKey(t);
+        string op = kfvOp(t);
+        auto values = kfvFieldsValues(t);
+
+        if (op == SET_COMMAND)
+        {
+            for (auto i : kfvFieldsValues(t))
+            {
+                if (fvField(i) == "admin_state")
+                {
+                    if (fvValue(i) == "enable")
+                    {
+                        cmd << "service hsflowd restart";
+                    }
+                    else
+                    {
+                        cmd << "service hsflowd stop";
+                    }
+
+                    int ret = swss::exec(cmd.str(), res);
+                    if (ret)
+                    {
+                        SWSS_LOG_ERROR("Command '%s' failed with rc %d", cmd.str().c_str(), ret);
+                    }
+                    else
+                    {
+                        SWSS_LOG_INFO("Command '%s' succeeded", cmd.str().c_str());
+                    }
+                }
+            }
+            m_appSflowTable.set(key, values);
+        }
+        else if(op == DEL_COMMAND)
+        {
+            m_appSflowTable.del(key);
+        }
+        it = consumer.m_toSync.erase(it);
+    }
+}
+
+void SflowMgr::doTask(Consumer &consumer)
+{
+    SWSS_LOG_ENTER();
+
+    auto table = consumer.getTableName();
+
+    if(table == CFG_SFLOW_TABLE_NAME)
+    {
+        handleSflowTableConfig(consumer);
+        return;
+    }
+
+    auto it = consumer.m_toSync.begin();
+    while (it != consumer.m_toSync.end())
+    {
+        KeyOpFieldsValuesTuple t = it->second;
+
+        string key = kfvKey(t);
+        string op = kfvOp(t);
+        auto values = kfvFieldsValues(t);
+
+        if (op == SET_COMMAND)
+        {
+            m_appSflowSessionTable.set(key, values);
+        }
+        else if (op == DEL_COMMAND)
+        {
+            m_appSflowSessionTable.del(key);
+        }
+
+        it = consumer.m_toSync.erase(it);
+    }
+}

cfgmgr/sflowmgr.h

@@ -0,0 +1,50 @@
+#pragma once
+
+#include "dbconnector.h"
+#include "orch.h"
+#include "producerstatetable.h"
+
+#include <map>
+#include <set>
+#include <string>
+
+namespace swss {
+
+/* Port default admin status is down */
+#define DEFAULT_ADMIN_STATUS_STR    "down"
+#define DEFAULT_MTU_STR             "9100"
+
+#define SFLOW_SAMPLE_RATE_KEY_400G "400000"
+#define SFLOW_SAMPLE_RATE_KEY_100G "100000"
+#define SFLOW_SAMPLE_RATE_KEY_50G  "50000"
+#define SFLOW_SAMPLE_RATE_KEY_40G  "40000"
+#define SFLOW_SAMPLE_RATE_KEY_25G  "25000"
+#define SFLOW_SAMPLE_RATE_KEY_10G  "10000"
+#define SFLOW_SAMPLE_RATE_KEY_1G   "1000"
+
+#define SFLOW_SAMPLE_RATE_VALUE_400G "40000"
+#define SFLOW_SAMPLE_RATE_VALUE_100G "10000"
+#define SFLOW_SAMPLE_RATE_VALUE_50G  "5000"
+#define SFLOW_SAMPLE_RATE_VALUE_40G  "4000"
+#define SFLOW_SAMPLE_RATE_VALUE_25G  "2500"
+#define SFLOW_SAMPLE_RATE_VALUE_10G  "1000"
+#define SFLOW_SAMPLE_RATE_VALUE_1G   "100"
+
+class SflowMgr : public Orch
+{
+public:
+    SflowMgr(DBConnector *cfgDb, DBConnector *appDb, const vector<string> &tableNames);
+
+    using Orch::doTask;
+private:
+    Table m_cfgSflowTable;
+    Table m_cfgSflowSessionTable;
+    Table m_appSflowSpeedRateTable;
+    ProducerStateTable m_appSflowTable;
+    ProducerStateTable m_appSflowSessionTable;
+
+    void doTask(Consumer &consumer);
+    void handleSflowTableConfig(Consumer &consumer);
+};
+
+}

cfgmgr/sflowmgrd.cpp

@@ -0,0 +1,87 @@
+#include <fstream>
+#include <iostream>
+#include <mutex>
+#include <unistd.h>
+#include <vector>
+
+#include "exec.h"
+#include "sflowmgr.h"
+#include "schema.h"
+#include "select.h"
+
+using namespace std;
+using namespace swss;
+
+/* select() function timeout retry time, in millisecond */
+#define SELECT_TIMEOUT 1000
+
+/*
+ * Following global variables are defined here for the purpose of
+ * using existing Orch class which is to be refactored soon to
+ * eliminate the direct exposure of the global variables.
+ *
+ * Once Orch class refactoring is done, these global variables
+ * should be removed from here.
+ */
+int gBatchSize = 0;
+bool gSwssRecord = false;
+bool gLogRotate = false;
+ofstream gRecordOfs;
+string gRecordFile;
+/* Global database mutex */
+mutex gDbMutex;
+
+int main(int argc, char **argv)
+{
+    Logger::linkToDbNative("sflowmgrd");
+    SWSS_LOG_ENTER();
+
+    SWSS_LOG_NOTICE("--- Starting sflowmgrd ---");
+
+    try
+    {
+        vector<string> cfg_sflow_tables = {
+            CFG_SFLOW_TABLE_NAME,
+            CFG_SFLOW_SESSION_TABLE_NAME
+        };
+
+        DBConnector cfgDb(CONFIG_DB, DBConnector::DEFAULT_UNIXSOCKET, 0);
+        DBConnector appDb(APPL_DB, DBConnector::DEFAULT_UNIXSOCKET, 0);
+
+        SflowMgr sflowmgr(&cfgDb, &appDb, cfg_sflow_tables);
+
+        vector<Orch *> cfgOrchList = {&sflowmgr};
+
+        swss::Select s;
+        for (Orch *o : cfgOrchList)
+        {
+            s.addSelectables(o->getSelectables());
+        }
+
+        while (true)
+        {
+            Selectable *sel;
+            int ret;
+
+            ret = s.select(&sel, SELECT_TIMEOUT);
+            if (ret == Select::ERROR)
+            {
+                SWSS_LOG_NOTICE("Error: %s!", strerror(errno));
+                continue;
+            }
+            if (ret == Select::TIMEOUT)
+            {
+                sflowmgr.doTask();
+                continue;
+            }
+
+            auto *c = (Executor *)sel;
+            c->execute();
+        }
+    }
+    catch (const exception &e)
+    {
+        SWSS_LOG_ERROR("Runtime error: %s", e.what());
+    }
+    return -1;
+}

orchagent/Makefile.am

@@ -52,7 +52,8 @@ orchagent_SOURCES = \
             dtelorch.cpp \
             flexcounterorch.cpp \
             watermarkorch.cpp \
-            policerorch.cpp
+            policerorch.cpp \
+            sfloworch.cpp
 
 orchagent_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
 orchagent_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)

orchagent/orchdaemon.cpp

@@ -177,6 +177,12 @@ bool OrchDaemon::init()
 
     WatermarkOrch *wm_orch = new WatermarkOrch(m_configDb, wm_tables);
 
+    vector<string> sflow_tables = {
+            APP_SFLOW_TABLE_NAME,
+            APP_SFLOW_SESSION_TABLE_NAME
+    };
+    SflowOrch *sflow_orch = new SflowOrch(m_applDb,  sflow_tables);
+
     /*
      * The order of the orch list is important for state restore of warm start and
      * the queued processing in m_toSync map after gPortsOrch->allPortsReady() is set.
@@ -185,7 +191,7 @@ 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, wm_orch, policer_orch };
+    m_orchList = { gSwitchOrch, gCrmOrch, gBufferOrch, gPortsOrch, gIntfsOrch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch, wm_orch, policer_orch, sflow_orch};
 
 
     bool initialize_dtel = false;

orchagent/orchdaemon.h

@@ -27,6 +27,7 @@
 #include "flexcounterorch.h"
 #include "watermarkorch.h"
 #include "policerorch.h"
+#include "sfloworch.h"
 #include "directory.h"
 
 using namespace swss;

orchagent/saihelper.cpp

@@ -41,6 +41,7 @@ sai_mirror_api_t*           sai_mirror_api;
 sai_fdb_api_t*              sai_fdb_api;
 sai_dtel_api_t*             sai_dtel_api;
 sai_bmtor_api_t*            sai_bmtor_api;
+sai_samplepacket_api_t*     sai_samplepacket_api;
 
 extern sai_object_id_t gSwitchId;
 extern bool gSairedisRecord;
@@ -130,6 +131,7 @@ void initSaiApi()
     sai_api_query(SAI_API_ACL,                  (void **)&sai_acl_api);
     sai_api_query(SAI_API_DTEL,                 (void **)&sai_dtel_api);
     sai_api_query((sai_api_t)SAI_API_BMTOR,     (void **)&sai_bmtor_api);
+    sai_api_query(SAI_API_SAMPLEPACKET,         (void **)&sai_samplepacket_api);
 
     sai_log_set(SAI_API_SWITCH,                 SAI_LOG_LEVEL_NOTICE);
     sai_log_set(SAI_API_BRIDGE,                 SAI_LOG_LEVEL_NOTICE);
@@ -156,6 +158,7 @@ void initSaiApi()
     sai_log_set(SAI_API_ACL,                    SAI_LOG_LEVEL_NOTICE);
     sai_log_set(SAI_API_DTEL,                   SAI_LOG_LEVEL_NOTICE);
     sai_log_set((sai_api_t)SAI_API_BMTOR,       SAI_LOG_LEVEL_NOTICE);
+    sai_log_set(SAI_API_SAMPLEPACKET,           SAI_LOG_LEVEL_NOTICE);
 }
 
 void initSaiRedis(const string &record_location)