[orchagent]: Publish identified events via structured-events channel (sonic-net#2446)

* Add events publish

* Added header file

* signature fix

* syntax

* syntax

* syntax

* syntax

* syntax

* syntax

* Updated fake code

* Remove if and log messages for event_publish

* Remove if and log messages for event_publish (#1)

* Remove event_handle_t from signature and add globally

* Remove extern orchdaemon.cpp

* Revert unneeded changes

Co-authored-by: zbud-msft <zainbudhwani@microsoft.com>
Co-authored-by: Zain Budhwani <99770260+zbud-msft@users.noreply.github.com>

Author: renukamanavalan

orchagent/crmorch.cpp

@@ -18,6 +18,7 @@
 extern sai_object_id_t gSwitchId;
 extern sai_switch_api_t *sai_switch_api;
 extern sai_acl_api_t *sai_acl_api;
+extern event_handle_t g_events_handle;
 
 using namespace std;
 using namespace swss;
@@ -763,9 +764,15 @@ 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);
 
+                event_publish(g_events_handle, "chk_crm_threshold", &params);
                 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"

orchagent/orchdaemon.cpp

@@ -57,6 +57,7 @@ FlowCounterRouteOrch *gFlowCounterRouteOrch;
 DebugCounterOrch *gDebugCounterOrch;
 
 bool gIsNatSupported = false;
+event_handle_t g_events_handle;
 
 #define DEFAULT_MAX_BULK_SIZE 1000
 size_t gMaxBulkSize = DEFAULT_MAX_BULK_SIZE;
@@ -89,6 +90,8 @@ OrchDaemon::~OrchDaemon()
         delete(*it);
     }
     delete m_select;
+
+    events_deinit_publisher(g_events_handle);
 }
 
 bool OrchDaemon::init()
@@ -97,6 +100,8 @@ bool OrchDaemon::init()
 
     string platform = getenv("platform") ? getenv("platform") : "";
 
+    g_events_handle = events_init_publisher("sonic-events-swss");
+
     gCrmOrch = new CrmOrch(m_configDb, CFG_CRM_TABLE_NAME);
 
     TableConnector stateDbSwitchTable(m_stateDb, "SWITCH_CAPABILITY");

orchagent/pfcwdorch.cpp

@@ -32,6 +32,8 @@ extern sai_switch_api_t* sai_switch_api;
 extern sai_port_api_t *sai_port_api;
 extern sai_queue_api_t *sai_queue_api;
 
+extern event_handle_t g_events_handle;
+
 extern SwitchOrch *gSwitchOrch;
 extern PortsOrch *gPortsOrch;
 
@@ -933,6 +935,26 @@ 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);
+
+    event_publish(g_events_handle, "pfc-storm", &params);
+}
+
 template <typename DropHandler, typename ForwardHandler>
 bool PfcWdSwOrch<DropHandler, ForwardHandler>::startWdActionOnQueue(const string &event, sai_object_id_t queueId)
 {
@@ -955,12 +977,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 +994,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 +1011,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;
@@ -121,6 +121,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

@@ -55,6 +55,7 @@ extern string gMySwitchType;
 extern int32_t gVoqMySwitchId;
 extern string gMyHostName;
 extern string gMyAsicName;
+extern event_handle_t g_events_handle;
 
 #define DEFAULT_SYSTEM_PORT_MTU 9100
 #define VLAN_PREFIX         "Vlan"
@@ -2537,6 +2538,8 @@ 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"}};
+    event_publish(g_events_handle, "if-state", &params);
     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