[orchagent]: Publish identified events via structured-events channel

orchagent/crmorch.cpp

@@ -241,8 +241,9 @@ const map<string, CrmResourceType> crmUsedCntsTableMap =
     { "crm_stats_nexthop_group_map_used", CrmResourceType::CRM_NEXTHOP_GROUP_MAP },
 };
 
-CrmOrch::CrmOrch(DBConnector *db, string tableName):
+CrmOrch::CrmOrch(DBConnector *db, string tableName, event_handle_t handle):
     Orch(db, tableName),
+    m_events_handle(handle),
     m_countersDb(new DBConnector("COUNTERS_DB", 0)),
     m_countersCrmTable(new Table(m_countersDb.get(), COUNTERS_CRM_TABLE)),
     m_timer(new SelectableTimer(timespec { .tv_sec = CRM_POLLING_INTERVAL_DEFAULT, .tv_nsec = 0 }))
@@ -763,9 +764,18 @@ void CrmOrch::checkCrmThresholds()
 
             if ((utilization >= res.highThreshold) && (res.exceededLogCounter < CRM_EXCEEDED_MSG_MAX))
             {
+                event_params_t params = {
+                    { "percent", to_string(percentageUtil) },
+                    { "used_cnt", to_string(cnt.usedCounter) },
+                    { "free_cnt", to_string(cnt.availableCounter) }};
+
                 SWSS_LOG_WARN("%s THRESHOLD_EXCEEDED for %s %u%% Used count %u free count %u",
                               res.name.c_str(), threshType.c_str(), percentageUtil, cnt.usedCounter, cnt.availableCounter);
 
+                if (0 != event_publish(m_events_handle, "chk_crm_threshold", &params)) {
+                    SWSS_LOG_WARN("Failed to publish event for if-state");
+                }
+
                 res.exceededLogCounter++;
             }
             else if ((utilization <= res.lowThreshold) && (res.exceededLogCounter > 0) && (res.highThreshold != res.lowThreshold))

orchagent/crmorch.h

@@ -5,6 +5,7 @@
 #include <map>
 #include "orch.h"
 #include "port.h"
+#include "events.h"
 
 extern "C" {
 #include "sai.h"
@@ -51,7 +52,7 @@ enum class CrmResourceStatus
 class CrmOrch : public Orch
 {
 public:
-    CrmOrch(swss::DBConnector *db, std::string tableName);
+    CrmOrch(swss::DBConnector *db, std::string tableName, event_handle_t handle = NULL);
     void incCrmResUsedCounter(CrmResourceType resource);
     void decCrmResUsedCounter(CrmResourceType resource);
     // Increment "used" counter for the ACL table/group CRM resources
@@ -64,6 +65,7 @@ class CrmOrch : public Orch
     void decCrmAclTableUsedCounter(CrmResourceType resource, sai_object_id_t tableId);
 
 private:
+    event_handle_t m_events_handle;
     std::shared_ptr<swss::DBConnector> m_countersDb = nullptr;
     std::shared_ptr<swss::Table> m_countersCrmTable = nullptr;
     swss::SelectableTimer *m_timer = nullptr;

orchagent/orchdaemon.cpp

@@ -62,6 +62,7 @@ bool gIsNatSupported = false;
 size_t gMaxBulkSize = DEFAULT_MAX_BULK_SIZE;
 
 OrchDaemon::OrchDaemon(DBConnector *applDb, DBConnector *configDb, DBConnector *stateDb, DBConnector *chassisAppDb) :
+        m_events_handle(NULL),
         m_applDb(applDb),
         m_configDb(configDb),
         m_stateDb(stateDb),
@@ -89,6 +90,8 @@ OrchDaemon::~OrchDaemon()
         delete(*it);
     }
     delete m_select;
+
+    events_deinit_publisher(m_events_handle);
 }
 
 bool OrchDaemon::init()
@@ -97,7 +100,9 @@ bool OrchDaemon::init()
 
     string platform = getenv("platform") ? getenv("platform") : "";
 
-    gCrmOrch = new CrmOrch(m_configDb, CFG_CRM_TABLE_NAME);
+    m_events_handle = events_init_publisher("sonic-events-swss");
+
+    gCrmOrch = new CrmOrch(m_configDb, CFG_CRM_TABLE_NAME, m_events_handle);
 
     TableConnector stateDbSwitchTable(m_stateDb, "SWITCH_CAPABILITY");
     TableConnector app_switch_table(m_applDb, APP_SWITCH_TABLE_NAME);
@@ -126,7 +131,7 @@ bool OrchDaemon::init()
         { APP_MCLAG_FDB_TABLE_NAME,  FdbOrch::fdborch_pri}
     };
 
-    gPortsOrch = new PortsOrch(m_applDb, m_stateDb, ports_tables, m_chassisAppDb);
+    gPortsOrch = new PortsOrch(m_applDb, m_stateDb, ports_tables, m_chassisAppDb, m_events_handle);
     TableConnector stateDbFdb(m_stateDb, STATE_FDB_TABLE_NAME);
     TableConnector stateMclagDbFdb(m_stateDb, STATE_MCLAG_REMOTE_FDB_TABLE_NAME);
     gFdbOrch = new FdbOrch(m_applDb, app_fdb_tables, stateDbFdb, stateMclagDbFdb, gPortsOrch);
@@ -508,7 +513,8 @@ bool OrchDaemon::init()
                     portStatIds,
                     queueStatIds,
                     queueAttrIds,
-                    PFC_WD_POLL_MSECS));
+                    PFC_WD_POLL_MSECS,
+                    m_events_handle));
     }
     else if ((platform == INVM_PLATFORM_SUBSTRING)
              || (platform == BFN_PLATFORM_SUBSTRING)
@@ -551,7 +557,8 @@ bool OrchDaemon::init()
                         portStatIds,
                         queueStatIds,
                         queueAttrIds,
-                        PFC_WD_POLL_MSECS));
+                        PFC_WD_POLL_MSECS,
+                        m_events_handle));
         }
         else if (platform == BFN_PLATFORM_SUBSTRING)
         {
@@ -561,7 +568,8 @@ bool OrchDaemon::init()
                         portStatIds,
                         queueStatIds,
                         queueAttrIds,
-                        PFC_WD_POLL_MSECS));
+                        PFC_WD_POLL_MSECS,
+                        m_events_handle));
         }
     }
     else if (platform == BRCM_PLATFORM_SUBSTRING)
@@ -605,7 +613,8 @@ bool OrchDaemon::init()
                         portStatIds,
                         queueStatIds,
                         queueAttrIds,
-                        PFC_WD_POLL_MSECS));
+                        PFC_WD_POLL_MSECS,
+                        m_events_handle));
         }
         else
         {
@@ -615,7 +624,8 @@ bool OrchDaemon::init()
                         portStatIds,
                         queueStatIds,
                         queueAttrIds,
-                        PFC_WD_POLL_MSECS));
+                        PFC_WD_POLL_MSECS,
+                        m_events_handle));
         }
     } else if (platform == CISCO_8000_PLATFORM_SUBSTRING)
     {
@@ -637,7 +647,8 @@ bool OrchDaemon::init()
                     portStatIds,
                     queueStatIds,
                     queueAttrIds,
-                    PFC_WD_POLL_MSECS));
+                    PFC_WD_POLL_MSECS,
+                    m_events_handle));
     }
 
     m_orchList.push_back(&CounterCheckOrch::getInstance(m_configDb));

orchagent/orchdaemon.h

@@ -68,6 +68,7 @@ class OrchDaemon
         m_fabricEnabled = enabled;
     }
 private:
+    event_handle_t m_events_handle;
     DBConnector *m_applDb;
     DBConnector *m_configDb;
     DBConnector *m_stateDb;

orchagent/p4orch/tests/fake_crmorch.cpp

@@ -3,7 +3,7 @@
 
 #include "crmorch.h"
 
-CrmOrch::CrmOrch(swss::DBConnector *db, std::string tableName) : Orch(db, std::vector<std::string>{})
+CrmOrch::CrmOrch(swss::DBConnector *db, std::string tableName, event_handle_t) : Orch(db, std::vector<std::string>{})
 {
 }
 

orchagent/p4orch/tests/fake_portorch.cpp

@@ -13,7 +13,7 @@ extern "C"
 #define QUEUE_STAT_FLEX_COUNTER_POLLING_INTERVAL_MS 10000
 
 PortsOrch::PortsOrch(DBConnector *db, DBConnector *stateDb, vector<table_name_with_pri_t> &tableNames,
-                     DBConnector *chassisAppDb)
+                     DBConnector *chassisAppDb, event_handle_t)
     : Orch(db, tableNames), m_portStateTable(stateDb, STATE_PORT_TABLE_NAME),
       port_stat_manager(PORT_STAT_COUNTER_FLEX_COUNTER_GROUP, StatsMode::READ,
                         PORT_STAT_FLEX_COUNTER_POLLING_INTERVAL_MS, true),

orchagent/pfcwdorch.cpp

@@ -686,8 +686,10 @@ PfcWdSwOrch<DropHandler, ForwardHandler>::PfcWdSwOrch(
         const vector<sai_port_stat_t> &portStatIds,
         const vector<sai_queue_stat_t> &queueStatIds,
         const vector<sai_queue_attr_t> &queueAttrIds,
-        int pollInterval):
+        int pollInterval,
+        event_handle_t handle):
     PfcWdOrch<DropHandler, ForwardHandler>(db, tableNames),
+    m_events_handle(handle),
     m_flexCounterDb(new DBConnector("FLEX_COUNTER_DB", 0)),
     m_flexCounterTable(new ProducerTable(m_flexCounterDb.get(), FLEX_COUNTER_TABLE)),
     m_flexCounterGroupTable(new ProducerTable(m_flexCounterDb.get(), FLEX_COUNTER_GROUP_TABLE)),
@@ -933,6 +935,28 @@ void PfcWdSwOrch<DropHandler, ForwardHandler>::doTask(SelectableTimer &timer)
 
 }
 
+template <typename DropHandler, typename ForwardHandler>
+void PfcWdSwOrch<DropHandler, ForwardHandler>::report_pfc_storm(
+        sai_object_id_t id, const PfcWdQueueEntry *entry)
+{
+    event_params_t params = {
+        { "ifname", entry->portAlias },
+        { "queue_index", to_string(entry->index) },
+        { "queue_id", to_string(id) },
+        { "port_id", to_string(entry->portId) }};
+
+    SWSS_LOG_NOTICE(
+            "PFC Watchdog detected PFC storm on port %s, queue index %d, queue id 0x%" PRIx64 " and port id 0x%" PRIx64 ".",
+            entry->portAlias.c_str(),
+            entry->index,
+            id,
+            entry->portId);
+
+    if (0 != event_publish(m_events_handle, "pfc-storm", &params)) {
+        SWSS_LOG_WARN("Failed to publish event for pfc-storm");
+    }
+}
+
 template <typename DropHandler, typename ForwardHandler>
 bool PfcWdSwOrch<DropHandler, ForwardHandler>::startWdActionOnQueue(const string &event, sai_object_id_t queueId)
 {
@@ -955,12 +979,7 @@ bool PfcWdSwOrch<DropHandler, ForwardHandler>::startWdActionOnQueue(const string
         {
             if (entry->second.handler == nullptr)
             {
-                SWSS_LOG_NOTICE(
-                        "PFC Watchdog detected PFC storm on port %s, queue index %d, queue id 0x%" PRIx64 " and port id 0x%" PRIx64 ".",
-                        entry->second.portAlias.c_str(),
-                        entry->second.index,
-                        entry->first,
-                        entry->second.portId);
+                report_pfc_storm(entry->first, &entry->second);
 
                 entry->second.handler = make_shared<PfcWdActionHandler>(
                         entry->second.portId,
@@ -977,12 +996,7 @@ bool PfcWdSwOrch<DropHandler, ForwardHandler>::startWdActionOnQueue(const string
         {
             if (entry->second.handler == nullptr)
             {
-                SWSS_LOG_NOTICE(
-                        "PFC Watchdog detected PFC storm on port %s, queue index %d, queue id 0x%" PRIx64 " and port id 0x%" PRIx64 ".",
-                        entry->second.portAlias.c_str(),
-                        entry->second.index,
-                        entry->first,
-                        entry->second.portId);
+                report_pfc_storm(entry->first, &entry->second);
 
                 entry->second.handler = make_shared<DropHandler>(
                         entry->second.portId,
@@ -999,12 +1013,7 @@ bool PfcWdSwOrch<DropHandler, ForwardHandler>::startWdActionOnQueue(const string
         {
             if (entry->second.handler == nullptr)
             {
-                SWSS_LOG_NOTICE(
-                        "PFC Watchdog detected PFC storm on port %s, queue index %d, queue id 0x%" PRIx64 " and port id 0x%" PRIx64 ".",
-                        entry->second.portAlias.c_str(),
-                        entry->second.index,
-                        entry->first,
-                        entry->second.portId);
+                report_pfc_storm(entry->first, &entry->second);
 
                 entry->second.handler = make_shared<ForwardHandler>(
                         entry->second.portId,

orchagent/pfcwdorch.h

@@ -7,6 +7,7 @@
 #include "producertable.h"
 #include "notificationconsumer.h"
 #include "timer.h"
+#include "events.h"
 
 extern "C" {
 #include "sai.h"
@@ -55,7 +56,6 @@ class PfcWdOrch: public Orch
 protected:
     virtual bool startWdActionOnQueue(const string &event, sai_object_id_t queueId) = 0;
     string m_platform = "";
-
 private:
 
     shared_ptr<DBConnector> m_countersDb = nullptr;
@@ -73,7 +73,8 @@ class PfcWdSwOrch: public PfcWdOrch<DropHandler, ForwardHandler>
             const vector<sai_port_stat_t> &portStatIds,
             const vector<sai_queue_stat_t> &queueStatIds,
             const vector<sai_queue_attr_t> &queueAttrIds,
-            int pollInterval);
+            int pollInterval,
+            event_handle_t handle = NULL);
     virtual ~PfcWdSwOrch(void);
 
     void doTask(Consumer& consumer) override;
@@ -92,6 +93,8 @@ class PfcWdSwOrch: public PfcWdOrch<DropHandler, ForwardHandler>
     bool startWdActionOnQueue(const string &event, sai_object_id_t queueId) override;
 
 private:
+    event_handle_t m_events_handle;
+
     struct PfcWdQueueEntry
     {
         PfcWdQueueEntry(
@@ -121,6 +124,8 @@ class PfcWdSwOrch: public PfcWdOrch<DropHandler, ForwardHandler>
     void enableBigRedSwitchMode();
     void setBigRedSwitchMode(string value);
 
+    void report_pfc_storm(sai_object_id_t id, const PfcWdQueueEntry *);
+
     map<sai_object_id_t, PfcWdQueueEntry> m_entryMap;
     map<sai_object_id_t, PfcWdQueueEntry> m_brsEntryMap;
 

orchagent/portsorch.cpp

@@ -352,8 +352,10 @@ static bool isValidPortTypeForLagMember(const Port& port)
  *    bridge. By design, SONiC switch starts with all bridge ports removed from
  *    default VLAN and all ports removed from .1Q bridge.
  */
-PortsOrch::PortsOrch(DBConnector *db, DBConnector *stateDb, vector<table_name_with_pri_t> &tableNames, DBConnector *chassisAppDb) :
+PortsOrch::PortsOrch(DBConnector *db, DBConnector *stateDb, vector<table_name_with_pri_t> &tableNames,
+        DBConnector *chassisAppDb, event_handle_t events_handle) :
         Orch(db, tableNames),
+        m_events_handle(events_handle),
         m_portStateTable(stateDb, STATE_PORT_TABLE_NAME),
         port_stat_manager(PORT_STAT_COUNTER_FLEX_COUNTER_GROUP, StatsMode::READ, PORT_STAT_FLEX_COUNTER_POLLING_INTERVAL_MS, false),
         gb_port_stat_manager("GB_FLEX_COUNTER_DB",
@@ -2537,6 +2539,10 @@ bool PortsOrch::setHostIntfsOperStatus(const Port& port, bool isUp) const
     SWSS_LOG_NOTICE("Set operation status %s to host interface %s",
             isUp ? "UP" : "DOWN", port.m_alias.c_str());
 
+    event_params_t params = {{"ifname",port.m_alias},{"status",isUp ? "up" : "down"}};
+    if (0 != event_publish(m_events_handle, "if-state", &params)) {
+        SWSS_LOG_WARN("Failed to publish event for if-state");
+    }
     return true;
 }
 

orchagent/portsorch.h

@@ -14,6 +14,7 @@
 #include "saihelper.h"
 #include "lagid.h"
 #include "flexcounterorch.h"
+#include "events.h"
 
 
 #define FCS_LEN 4
@@ -77,7 +78,7 @@ struct VlanMemberUpdate
 class PortsOrch : public Orch, public Subject
 {
 public:
-    PortsOrch(DBConnector *db, DBConnector *stateDb, vector<table_name_with_pri_t> &tableNames, DBConnector *chassisAppDb);
+    PortsOrch(DBConnector *db, DBConnector *stateDb, vector<table_name_with_pri_t> &tableNames, DBConnector *chassisAppDb, event_handle_t handle = NULL);
 
     bool allPortsReady();
     bool isInitDone();
@@ -188,6 +189,7 @@ class PortsOrch : public Orch, public Subject
     bool isMACsecPort(sai_object_id_t port_id) const;
 
 private:
+    event_handle_t m_events_handle;
     unique_ptr<Table> m_counterTable;
     unique_ptr<Table> m_counterLagTable;
     unique_ptr<Table> m_portTable;